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>
110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
"""Record the official /usage percentages as they change.
|
|
|
|
Claude Code hands its status line a ``rate_limits`` block (used % + reset time
|
|
per window); the homelab status line persists it to
|
|
``<config dir>/usage-state/rate-limits.json`` on every render, and a cron keeps
|
|
it fresh while no terminal is open (services/claude-usage-notifier). Those dirs
|
|
are mounted read-only per account (``USAGE_STATE_DIRS``); this thread reads them
|
|
every ``POLL_S`` and stores a point whenever a window's % or reset time moves —
|
|
plus a heartbeat every ``HEARTBEAT_S`` so an idle hour still has a level.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import threading
|
|
import time
|
|
|
|
POLL_S = 20
|
|
HEARTBEAT_S = 1800
|
|
STALE_S = 1800
|
|
FILE = "rate-limits.json"
|
|
|
|
|
|
def state_dirs() -> dict[str, pathlib.Path]:
|
|
"""``{account: dir}`` from ``USAGE_STATE_DIRS`` (``personal=/usage-state,…``)."""
|
|
raw = os.environ.get("USAGE_STATE_DIRS", "personal=/usage-state,work=/usage-state-work")
|
|
out = {}
|
|
for part in raw.split(","):
|
|
acct, _, path = part.partition("=")
|
|
if acct.strip() and path.strip():
|
|
out[acct.strip()] = pathlib.Path(path.strip())
|
|
return out
|
|
|
|
|
|
def read_snapshot(d: pathlib.Path) -> dict | None:
|
|
"""``{updatedAt, windows: {name: {pct, resetsAt}}}`` or None."""
|
|
try:
|
|
raw = json.loads((d / FILE).read_text())
|
|
except (OSError, ValueError):
|
|
return None
|
|
rl = raw.get("rate_limits") or {}
|
|
windows = {}
|
|
for name, w in rl.items():
|
|
if isinstance(w, dict) and isinstance(w.get("used_percentage"), (int, float)):
|
|
windows[name] = {"pct": float(w["used_percentage"]),
|
|
"resetsAt": w.get("resets_at")}
|
|
return {"updatedAt": raw.get("updated_at"), "windows": windows}
|
|
|
|
|
|
class UsagePoller:
|
|
def __init__(self, store, publish=None, dirs: dict | None = None):
|
|
self.store = store
|
|
self.publish = publish
|
|
self.dirs = dirs if dirs is not None else state_dirs()
|
|
self.latest: dict[str, dict] = {}
|
|
self._last: dict[tuple, tuple] = {}
|
|
|
|
def start(self) -> None:
|
|
if self.dirs:
|
|
threading.Thread(target=self._loop, daemon=True, name="usage-poller").start()
|
|
|
|
def _loop(self) -> None:
|
|
while True:
|
|
try:
|
|
self.poll_once()
|
|
except Exception:
|
|
pass
|
|
time.sleep(POLL_S)
|
|
|
|
def poll_once(self) -> int:
|
|
rows = []
|
|
for acct, d in self.dirs.items():
|
|
snap = read_snapshot(d)
|
|
if not snap or not snap["updatedAt"]:
|
|
continue
|
|
self.latest[acct] = snap
|
|
t = float(snap["updatedAt"])
|
|
for win, w in snap["windows"].items():
|
|
key = (acct, win)
|
|
last = self._last.get(key)
|
|
if last is None:
|
|
p = self.store.last_point(acct, win)
|
|
last = (p[0], p[1], p[2]) if p else None
|
|
if last:
|
|
self._last[key] = last
|
|
if last and t <= last[0]:
|
|
continue
|
|
moved = (last is None or w["pct"] != last[1]
|
|
or (w["resetsAt"] or 0) != (last[2] or 0))
|
|
if moved or t - last[0] >= HEARTBEAT_S:
|
|
rows.append((acct, win, t, w["pct"], w["resetsAt"], "live"))
|
|
self._last[key] = (t, w["pct"], w["resetsAt"])
|
|
n = self.store.add_points(rows)
|
|
if n and self.publish:
|
|
self.publish({"type": "usage"})
|
|
return n
|
|
|
|
def limits(self) -> dict[str, dict]:
|
|
"""Latest snapshot per account (``stale`` past ``STALE_S``)."""
|
|
now = time.time()
|
|
out = {}
|
|
for acct in self.dirs:
|
|
snap = self.latest.get(acct) or read_snapshot(self.dirs[acct]) or {}
|
|
up = snap.get("updatedAt")
|
|
out[acct] = {"account": acct, "windows": snap.get("windows") or {},
|
|
"updatedAt": up, "stale": not up or now - up > STALE_S}
|
|
return out
|