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>
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""Wiring: the daily recalibration loop and the shift notification."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
|
|
from . import calibrate
|
|
|
|
CHECK_S = 3600
|
|
EVERY_S = 86400
|
|
|
|
|
|
def _notify(deps, account: str, text: str) -> None:
|
|
from notifications.service import _create_notify
|
|
try:
|
|
_create_notify(deps, {"title": "📏 Claude plan rate shifted", "message": text,
|
|
"type": "ai", "url": "/dashboards/plan"})
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def calibrate_all(deps, notify: bool = True) -> tuple[dict, list[str]]:
|
|
cals, shifted = {}, []
|
|
for acct in deps.usage_poller.dirs:
|
|
cal, moved = calibrate.run(deps, acct, (lambda a, t: _notify(deps, a, t))
|
|
if notify else None)
|
|
if cal:
|
|
cals[acct] = cal
|
|
if moved:
|
|
shifted.append(acct)
|
|
return cals, shifted
|
|
|
|
|
|
def _loop(deps) -> None:
|
|
time.sleep(120) # let the first index pass land
|
|
while True:
|
|
try:
|
|
for acct in deps.usage_poller.dirs:
|
|
fits = deps.usage_store.calibrations(acct)
|
|
if not fits:
|
|
calibrate.backfill(deps, acct)
|
|
last = max((c["fittedAt"] for c in deps.usage_store.calibrations(acct)),
|
|
default=0)
|
|
if time.time() - last >= EVERY_S:
|
|
calibrate.run(deps, acct, lambda a, t: _notify(deps, a, t))
|
|
except Exception:
|
|
pass
|
|
time.sleep(CHECK_S)
|
|
|
|
|
|
def start(deps) -> None:
|
|
deps.usage_poller.start()
|
|
threading.Thread(target=_loop, args=(deps,), daemon=True,
|
|
name="usage-calibration").start()
|