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>
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
"""Plan usage: live /usage %, its history and the calibration (usage/)."""
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
import accounts as accounts_mod
|
|
import schemas
|
|
from core.http import _json_in_worker, _r
|
|
from core.state import State
|
|
from users.caller import CallerDep
|
|
|
|
from . import calibrate, service
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _owner(caller) -> None:
|
|
if not caller.is_owner:
|
|
raise HTTPException(403, "owner only")
|
|
|
|
|
|
def _account(account: str) -> str:
|
|
if account not in accounts_mod.ids():
|
|
raise HTTPException(400, f"unknown account {account!r}")
|
|
return account
|
|
|
|
|
|
@router.get("/api/usage/limits", responses=_r(schemas.UsageLimitsResponse))
|
|
def usage_limits(deps: State, caller: CallerDep):
|
|
"""The latest /usage % per account, with the rate in force."""
|
|
_owner(caller)
|
|
out = {}
|
|
for acct, lim in deps.usage_poller.limits().items():
|
|
a = accounts_mod.get(acct) or {}
|
|
out[acct] = {**lim, "label": a.get("label"),
|
|
"calibration": calibrate.current(deps, acct)}
|
|
return {"accounts": out}
|
|
|
|
|
|
@router.get("/api/usage/history", responses=_r(schemas.UsageHistory))
|
|
@_json_in_worker
|
|
def usage_history(deps: State, caller: CallerDep, account: str = "personal",
|
|
days: int = 14):
|
|
"""Raw % series, hourly meters and per-day observed / predicted /
|
|
untracked % for one account over the last ``days`` (≤ 120)."""
|
|
_owner(caller)
|
|
return calibrate.history(deps, _account(account), max(1, min(days, 120)))
|
|
|
|
|
|
@router.get("/api/usage/calibration", responses=_r(schemas.CalibrationHistory))
|
|
def usage_calibration(deps: State, caller: CallerDep, account: str = "personal"):
|
|
"""Every stored fit of an account (oldest first) and the one in force."""
|
|
_owner(caller)
|
|
acct = _account(account)
|
|
return {"account": acct, "current": calibrate.current(deps, acct),
|
|
"history": deps.usage_store.calibrations(acct)}
|
|
|
|
|
|
class CalibrateBody(BaseModel):
|
|
backfill: bool = False
|
|
|
|
|
|
@router.post("/api/usage/calibrate", responses=_r(schemas.CalibrateResult))
|
|
def usage_calibrate(deps: State, caller: CallerDep, body: CalibrateBody | None = None):
|
|
"""Fit every account now (``backfill``: also the weekly fits over the whole
|
|
history, for the $/% over time chart)."""
|
|
_owner(caller)
|
|
if body and body.backfill:
|
|
for acct in deps.usage_poller.dirs:
|
|
calibrate.backfill(deps, acct)
|
|
cals, shifted = service.calibrate_all(deps)
|
|
return {"calibrations": cals, "shifted": shifted}
|
|
|
|
|
|
class ImportBody(BaseModel):
|
|
account: str
|
|
source: str = "ha"
|
|
# [{t, window, pct, resetsAt?}]
|
|
points: list[dict]
|
|
|
|
|
|
@router.post("/api/usage/import", responses=_r(schemas.UsageImportResult))
|
|
def usage_import(deps: State, caller: CallerDep, body: ImportBody):
|
|
"""Backfill historical % points (scripts/usage-backfill-ha.py)."""
|
|
_owner(caller)
|
|
acct = _account(body.account)
|
|
rows = [(acct, str(p["window"]), float(p["t"]), float(p["pct"]),
|
|
p.get("resetsAt"), body.source) for p in body.points
|
|
if p.get("window") and p.get("t") is not None and p.get("pct") is not None]
|
|
n = deps.usage_store.add_points(rows)
|
|
if n:
|
|
deps.hub.publish({"type": "usage"})
|
|
return {"imported": n}
|