v2 Phase 4: Session screen — today's tokens and equivalent cost

Daemon: compute_today_usage() sums today's Claude Code token usage across the
local project transcripts (~/.claude/projects/**/*.jsonl), reading only files
modified today so the scan stays cheap, and prices it with Anthropic list rates
to show the equivalent API cost (the ccusage-style flex for subscription users).
The compact fields tk (total tokens), tc (cost in cents) and tn (message count)
piggyback onto the existing 60s BLE payload.

Firmware: UsageData gains tokens_today (64-bit) / cost_cents_today /
messages_today; parse_json reads tk/tc/tn; a new Session screen shows the cost
as the hero with total tokens and message count below, refreshed from ui_update
so it's current whenever opened. Wire SCREEN_SESSION to the real screen (was a
stub). Verified end to end: daemon sends e.g. tk=105374109 tc=26838 tn=663.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wenil
2026-06-20 15:03:23 +03:00
co-authored by Claude Opus 4.8
parent abe72eca0a
commit 56b49898a3
4 changed files with 159 additions and 1 deletions
+87
View File
@@ -57,6 +57,90 @@ API_BODY = {
"messages": [{"role": "user", "content": "hi"}],
}
# Anthropic list prices, USD per million tokens, keyed by a substring of the
# model id. Used to show the "equivalent API cost" of today's Claude Code usage
# — subscription users don't actually pay this; it's the ccusage-style flex.
PRICING = {
"opus": {"in": 15.0, "out": 75.0, "cache_w": 18.75, "cache_r": 1.50},
"sonnet": {"in": 3.0, "out": 15.0, "cache_w": 3.75, "cache_r": 0.30},
"haiku": {"in": 1.0, "out": 5.0, "cache_w": 1.25, "cache_r": 0.10},
}
_DEFAULT_PRICE = PRICING["opus"]
def _price_for(model: str) -> dict:
m = (model or "").lower()
for key, price in PRICING.items():
if key in m:
return price
return _DEFAULT_PRICE
def compute_today_usage() -> dict:
"""Sum today's Claude Code token usage + equivalent API cost across all local
project transcripts (~/.claude/projects/**/*.jsonl).
Reads only files modified today (a session that touched today has mtime
today), so the scan stays cheap even with a large transcript history. Each
assistant line carries message.usage (input/output/cache token counts) and a
UTC timestamp; per-line timestamps gate to today so a session spanning
midnight is split correctly. Returns the compact BLE fields tk/tc/tn.
"""
base = Path.home() / ".claude" / "projects"
today = datetime.date.today()
midnight = datetime.datetime.combine(today, datetime.time.min)
total_tokens = 0
cost = 0.0
messages = 0
try:
files = list(base.glob("**/*.jsonl"))
except OSError:
return {"tk": 0, "tc": 0, "tn": 0}
for f in files:
try:
if datetime.datetime.fromtimestamp(f.stat().st_mtime) < midnight:
continue
except OSError:
continue
try:
with open(f, encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except (json.JSONDecodeError, ValueError):
continue
msg = obj.get("message") if isinstance(obj, dict) else None
if not isinstance(msg, dict):
continue
usage = msg.get("usage")
if not isinstance(usage, dict):
continue
ts = obj.get("timestamp")
if not ts:
continue
try:
d = datetime.datetime.fromisoformat(
ts.replace("Z", "+00:00")).astimezone().date()
except ValueError:
continue
if d != today:
continue
inp = usage.get("input_tokens", 0) or 0
out = usage.get("output_tokens", 0) or 0
cw = usage.get("cache_creation_input_tokens", 0) or 0
cr = usage.get("cache_read_input_tokens", 0) or 0
total_tokens += inp + out + cw + cr
messages += 1
p = _price_for(msg.get("model"))
cost += (inp * p["in"] + out * p["out"]
+ cw * p["cache_w"] + cr * p["cache_r"]) / 1_000_000
except OSError:
continue
return {"tk": int(total_tokens), "tc": int(round(cost * 100)), "tn": messages}
def _build_file_logger() -> logging.Logger | None:
"""Create a rotating file logger for field diagnostics, or None.
@@ -158,6 +242,9 @@ async def poll_api(token: str) -> dict | None:
"st": hdr("anthropic-ratelimit-unified-5h-status", "unknown"),
"ok": True,
}
# Piggyback today's local token usage/cost onto the same 60s payload. Local
# file read (~tens of ms once per minute) — fine to do inline in the poller.
payload.update(compute_today_usage())
return payload