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>
70 lines
2.9 KiB
Python
Executable File
70 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Backfill the ai-agent's plan-usage history from Home Assistant.
|
|
|
|
Before the backend recorded the /usage % itself (usage/poller.py), the only
|
|
history was in HA: the sensors `claude-usage-notifier` pushes, kept by HA's
|
|
recorder as hourly long-term statistics (min / max per hour) since 2026-07-17.
|
|
This reads them straight from the recorder SQLite db (read-only) and POSTs one
|
|
point per hour — the hour's max, stamped at its end — to /api/usage/import.
|
|
|
|
A window resets inside an hour when the hour's min falls below the previous
|
|
hour's max; those points get a new synthetic ``resetsAt`` (the epoch of the
|
|
reset hour) so the calibration skips the hour like a live reset. Idempotent:
|
|
re-running inserts nothing new.
|
|
|
|
scripts/usage-backfill-ha.py # both accounts → :8096
|
|
scripts/usage-backfill-ha.py --api http://127.0.0.1:8197 --account personal
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import pathlib
|
|
import sqlite3
|
|
import urllib.request
|
|
|
|
ACCOUNTS = {"personal": "sensor.claude_usage", "work": "sensor.claude_work_usage"}
|
|
WINDOWS = {"5h": "five_hour", "week": "seven_day"}
|
|
|
|
|
|
def points(db: pathlib.Path, prefix: str) -> list[dict]:
|
|
con = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
|
|
out = []
|
|
for suffix, window in WINDOWS.items():
|
|
rows = con.execute(
|
|
"SELECT s.start_ts, s.min, s.max FROM statistics s JOIN statistics_meta m"
|
|
" ON s.metadata_id = m.id WHERE m.statistic_id = ? ORDER BY s.start_ts",
|
|
(f"{prefix}_{suffix}",)).fetchall()
|
|
prev, epoch = None, None
|
|
for start, lo, hi in rows:
|
|
if hi is None:
|
|
continue
|
|
if epoch is None or (prev is not None and lo is not None and lo < prev - 0.5):
|
|
epoch = start
|
|
out.append({"window": window, "t": start + 3599, "pct": hi, "resetsAt": epoch})
|
|
prev = hi
|
|
return out
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
|
|
ap.add_argument("--ha-db", type=pathlib.Path, default=pathlib.Path.home()
|
|
/ "homelab/services/home-assistant/config/home-assistant_v2.db")
|
|
ap.add_argument("--api", default="http://127.0.0.1:8096")
|
|
ap.add_argument("--account", choices=list(ACCOUNTS), action="append")
|
|
args = ap.parse_args()
|
|
for acct in args.account or list(ACCOUNTS):
|
|
pts = points(args.ha_db, ACCOUNTS[acct])
|
|
total = 0
|
|
for i in range(0, len(pts), 2000):
|
|
body = json.dumps({"account": acct, "source": "ha",
|
|
"points": pts[i:i + 2000]}).encode()
|
|
req = urllib.request.Request(f"{args.api}/api/usage/import", data=body,
|
|
headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=120) as r:
|
|
total += json.load(r)["imported"]
|
|
print(f"{acct}: {len(pts)} hourly points read, {total} new")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|