Gabriel Vidal
4bedccc005
backend/usage/ polls each account's rate-limits.json (read-only mount,
USAGE_STATE_DIRS) and keeps every change (+ a 30 min heartbeat) in
usage_limits. A daily fit over 14 days of hourly rises vs the runs'
meterBins gives plan units per 1 % of the 5h / week window; only trusted
fits (n>=24, R2>=0.5, within x3 of the seed) replace the rate, and a >20 %
shift pushes a notification. Cards carry planPct, --final notifications
get ' · $X · ≈Y% wk', and /api/usage/{limits,history,calibration,
calibrate,import} serve the Plan usage page (owner only).
scripts/usage-backfill-ha.py imports the HA hourly statistics since July.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
87 lines
3.4 KiB
Python
87 lines
3.4 KiB
Python
"""SQLite tables for plan usage: every observed /usage % point and every
|
|
calibration fit. Shares ai-agent.db with the main Store (own connection; WAL
|
|
lets both write)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import threading
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS usage_limits (
|
|
account TEXT NOT NULL,
|
|
window TEXT NOT NULL, -- five_hour | seven_day | seven_day_<model>
|
|
t REAL NOT NULL, -- epoch secs the % was observed
|
|
pct REAL NOT NULL,
|
|
resets_at REAL, -- when that window frees up (a change = a reset)
|
|
source TEXT NOT NULL DEFAULT 'live', -- live | ha (backfill)
|
|
PRIMARY KEY (account, window, t)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS usage_calibration (
|
|
account TEXT NOT NULL,
|
|
fitted_at REAL NOT NULL, -- end of the fitted window
|
|
data_json TEXT NOT NULL, -- the Calibration dict
|
|
PRIMARY KEY (account, fitted_at)
|
|
);
|
|
"""
|
|
|
|
|
|
class UsageStore:
|
|
def __init__(self, path: str):
|
|
self._lock = threading.Lock()
|
|
self._db = sqlite3.connect(path, check_same_thread=False)
|
|
self._db.execute("PRAGMA journal_mode=WAL")
|
|
self._db.executescript(_SCHEMA)
|
|
self.version = 0 # bumped on every write; memo keys read it
|
|
|
|
def add_points(self, rows: list[tuple]) -> int:
|
|
"""rows = (account, window, t, pct, resets_at, source); duplicates ignored."""
|
|
if not rows:
|
|
return 0
|
|
with self._lock:
|
|
cur = self._db.executemany(
|
|
"INSERT OR IGNORE INTO usage_limits VALUES (?,?,?,?,?,?)", rows)
|
|
self._db.commit()
|
|
n = cur.rowcount
|
|
if n:
|
|
self.version += 1
|
|
return n
|
|
|
|
def last_point(self, account: str, window: str) -> tuple | None:
|
|
with self._lock:
|
|
return self._db.execute(
|
|
"SELECT t, pct, resets_at FROM usage_limits WHERE account=? AND window=?"
|
|
" ORDER BY t DESC LIMIT 1", (account, window)).fetchone()
|
|
|
|
def series(self, account: str, since: float, until: float | None = None
|
|
) -> dict[str, list[tuple[float, float, float | None]]]:
|
|
q = ("SELECT window, t, pct, resets_at FROM usage_limits WHERE account=?"
|
|
" AND t>=?" + (" AND t<=?" if until else "") + " ORDER BY t")
|
|
args = (account, since, until) if until else (account, since)
|
|
out: dict[str, list] = {}
|
|
with self._lock:
|
|
for w, t, p, r in self._db.execute(q, args):
|
|
out.setdefault(w, []).append((t, p, r))
|
|
return out
|
|
|
|
def first_t(self, account: str) -> float | None:
|
|
with self._lock:
|
|
row = self._db.execute("SELECT MIN(t) FROM usage_limits WHERE account=?",
|
|
(account,)).fetchone()
|
|
return row[0] if row else None
|
|
|
|
def add_calibration(self, cal: dict) -> None:
|
|
with self._lock:
|
|
self._db.execute("INSERT OR REPLACE INTO usage_calibration VALUES (?,?,?)",
|
|
(cal["account"], cal["fittedAt"], json.dumps(cal)))
|
|
self._db.commit()
|
|
self.version += 1
|
|
|
|
def calibrations(self, account: str) -> list[dict]:
|
|
with self._lock:
|
|
rows = self._db.execute(
|
|
"SELECT data_json FROM usage_calibration WHERE account=? ORDER BY fitted_at",
|
|
(account,)).fetchall()
|
|
return [json.loads(r[0]) for r in rows]
|