#!/usr/bin/env python3 """plan-usage.py — snapshot each plan's REAL usage, straight from the vendor, into files meter.py reads. Reference copy from jacobkunselman.com/tips/real-usage-and-a-spare-capacity-queue (2026-10-13). Codex `codex app-server`, JSON-RPC `account/rateLimits/read`: the 5-hour and weekly used percent and resets, the same numbers the ChatGPT usage page shows (read 56 % / 11 % against the page's 44 % and 89 % left). Before this the meter saw Codex only through the rollouts Codex writes on THIS Pi, so use on the TUF, the MSI or ChatGPT's own apps was invisible until the next run here. Grok `grok agent stdio`, ACP extension `_x.ai/billing`: creditUsagePercent and the current weekly period, the account page's "Weekly Super Limit". Before this Grok was an estimate from consult dollars and the owner's readings, and its reset time had to be read by hand (it moved on 2026-10-13 without anyone noticing). Gemini `agy -p /usage`, through gemini-usage.py, as before (already exact). Claude GET api.anthropic.com/api/oauth/usage with Claude Code's own login (~/.claude/.credentials.json): the endpoint `/usage` itself calls (found in the CLI, 2026-10-13). The 5-hour session, the week and Fable's week, each with its reset. ⚠ It only READS the login: it never refreshes it, since a refresh rotates the token under your running sessions. An expired token is skipped; the next Claude session refreshes it. ⭐ No model call and no tokens: each is a read of the account's own counters (~2-5 s). ⛔ Never call Codex's `account/rateLimitResetCredit/consume`, which sits beside the read in the same protocol: it spends a reset credit. Writes one JSON line per snapshot to ~/.local/state/ai-meter/{codex,grok}-usage.jsonl and keeps 30 days. Silent no-op for a CLI that is missing or not signed in. Never raises. ./plan-usage.py snapshot all three and append ./plan-usage.py --print snapshot and print, append nothing """ import datetime, json, os, pathlib, select, subprocess, sys, time HOME = pathlib.Path.home() OUT = HOME / ".local/state/ai-meter" CODEX = HOME / ".local/bin/codex" CLAUDE_LOGIN = HOME / ".claude/.credentials.json" GROK = HOME / ".grok/bin/grok" KEEP_DAYS = 30 HERE = pathlib.Path(__file__).resolve().parent def rpc(argv, calls, timeout=30): """start a JSON-RPC-over-stdio server, send `calls` [(method, params)] in order, return {method: result}. A notification is a (method, None) pair.""" env = {k: v for k, v in os.environ.items() if k not in ("DISPLAY", "WAYLAND_DISPLAY", "OPENAI_API_KEY", "XAI_API_KEY", "GROK_API_KEY")} env["BROWSER"] = "/bin/false" # never a browser from a timer try: p = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1, env=env, cwd=str(HOME)) except OSError: return {} out, end = {}, time.time() + timeout try: for i, (method, params) in enumerate(calls, 1): msg = {"jsonrpc": "2.0", "method": method} if params is not None: msg.update(id=i, params=params) p.stdin.write(json.dumps(msg) + "\n"); p.stdin.flush() if params is None: continue while time.time() < end: r, _, _ = select.select([p.stdout], [], [], 1) if not r: continue line = p.stdout.readline() if not line: return out try: m = json.loads(line) except ValueError: continue if m.get("id") == i: if "result" in m: out[method] = m["result"] break except (OSError, ValueError): pass finally: p.kill() p.wait() return out def codex(): """{"at", "windows": [{"window_min", "used", "resets"}], "plan"} or None""" if not CODEX.exists(): return None r = rpc([str(CODEX), "app-server"], [ ("initialize", {"clientInfo": {"name": "shop-meter", "version": "1"}}), ("initialized", None), ("account/rateLimits/read", {"excludeResetCreditDetails": True}), # a READ. Never .../consume. ]).get("account/rateLimits/read") rl = (r or {}).get("rateLimits") or {} wins = [] for k in ("primary", "secondary"): w = rl.get(k) if isinstance(w, dict) and w.get("windowDurationMins"): wins.append({"window_min": int(w["windowDurationMins"]), "used": float(w.get("usedPercent", 0)), "resets": int(w.get("resetsAt") or 0)}) if not wins: return None return {"at": datetime.datetime.now().astimezone().isoformat(timespec="seconds"), "windows": wins, "plan": rl.get("planType"), "allowed": r.get("ordinaryUsageAllowed")} def grok(): """{"at", "used", "start", "end", "tier"} or None""" if not GROK.exists(): return None r = rpc([str(GROK), "agent", "--no-leader", "stdio"], [ ("initialize", {"protocolVersion": 1, "clientCapabilities": {}, "clientInfo": {"name": "shop-meter", "version": "1"}}), ("_x.ai/billing", {}), ], timeout=40).get("_x.ai/billing") c = (r or {}).get("config") or {} per = c.get("currentPeriod") or {} if c.get("creditUsagePercent") is None or not per.get("end"): return None return {"at": datetime.datetime.now().astimezone().isoformat(timespec="seconds"), "used": float(c["creditUsagePercent"]), "start": per.get("start"), "end": per["end"], "period": per.get("type"), "tier": r.get("subscription_tier")} def claude(): """{"at", "session": {used, resets}, "week": {...}, "fable": {...}} or None""" import urllib.request try: o = json.loads(CLAUDE_LOGIN.read_text()).get("claudeAiOauth") or {} except (OSError, ValueError): return None if not o.get("accessToken") or o.get("expiresAt", 0) / 1000 < time.time() + 60: return None # expired: leave the refresh to Claude Code req = urllib.request.Request("https://api.anthropic.com/api/oauth/usage", headers={ "Authorization": "Bearer " + o["accessToken"], "anthropic-beta": "oauth-2025-04-20", "User-Agent": "shop-meter/1"}) try: with urllib.request.urlopen(req, timeout=15) as r: d = json.loads(r.read()) except (OSError, ValueError): return None def win(w): return {"used": float(w["utilization"]), "resets": w["resets_at"]} if w and w.get("utilization") is not None \ and w.get("resets_at") else None fable = next((l for l in d.get("limits") or [] if l.get("kind") == "weekly_scoped" and "fable" in str(((l.get("scope") or {}).get("model") or {}).get("display_name", "")).lower()), None) out = {"at": datetime.datetime.now().astimezone().isoformat(timespec="seconds"), "session": win(d.get("five_hour")), "week": win(d.get("seven_day")), "fable": {"used": float(fable["percent"]), "resets": fable["resets_at"]} if fable else None, "plan": {"type": o.get("subscriptionType"), "tier": o.get("rateLimitTier")}} # the plan as the account sees it (/api/oauth/profile): only these fields, never the name, email or ids req = urllib.request.Request("https://api.anthropic.com/api/oauth/profile", headers=dict(req.headers)) try: with urllib.request.urlopen(req, timeout=15) as r: org = (json.loads(r.read()).get("organization") or {}) out["plan"].update({k: org.get(k) for k in ("organization_type", "rate_limit_tier", "billing_type", "subscription_status", "has_extra_usage_enabled")}) except (OSError, ValueError): pass return out if out["week"] else None def append(name, snap): OUT.mkdir(parents=True, exist_ok=True) f = OUT / f"{name}-usage.jsonl" cut = (datetime.datetime.now().astimezone() - datetime.timedelta(days=KEEP_DAYS)).isoformat() keep = [] for line in f.read_text().splitlines() if f.exists() else []: try: if json.loads(line)["at"] >= cut: keep.append(line) except (ValueError, KeyError, TypeError): pass keep.append(json.dumps(snap)) tmp = f.with_suffix(f".tmp{os.getpid()}") tmp.write_text("\n".join(keep) + "\n") os.replace(tmp, f) def main(): show = "--print" in sys.argv got = {} for name, fn in (("claude", claude), ("codex", codex), ("grok", grok)): try: s = fn() except Exception: # noqa: BLE001 — a meter must never take its caller down s = None got[name] = s if s and not show: append(name, s) try: subprocess.run([sys.executable, str(HERE / "gemini-usage.py"), *(["--print"] if show else [])], timeout=45, stdin=subprocess.DEVNULL, capture_output=not show) except Exception: # noqa: BLE001 pass if show: print(json.dumps(got, indent=1)) if __name__ == "__main__": main()