#!/usr/bin/env python3 """ mimo-usage — local token usage tracker for cc-mimo Scans ~/.claude-envs/mimo/.claude/projects/**/*.jsonl session files, sums input/output/cache tokens per time window (today/week/all), writes to ~/.codexbar/mimo-local-usage.json, and prints a human-readable summary by default. Usage: mimo-usage # show summary (also refreshes cache) mimo-usage --update # refresh cache only, no output (for LaunchAgent/wrapper) mimo-usage --json # JSON output mimo-usage --short # 1-line status (for status line / widget) """ import json import os import sys from pathlib import Path from datetime import datetime, timedelta, timezone MIMO_HOME = Path(os.environ.get("MIMO_CLAUDE_HOME", Path.home() / ".claude-envs" / "mimo")).expanduser() PROJECTS_DIR = MIMO_HOME / ".claude" / "projects" CACHE_PATH = Path( os.environ.get("MIMO_LOCAL_USAGE_PATH", Path.home() / ".codexbar" / "mimo-local-usage.json") ).expanduser() def parse_session_usage(jsonl_path: Path): """Yield (identity, timestamp_iso, usage_dict) for each assistant message with usage.""" try: with jsonl_path.open() as f: for line in f: try: d = json.loads(line) ts = d.get("timestamp") msg = d.get("message") if not isinstance(msg, dict): continue usage = msg.get("usage") if not isinstance(usage, dict): continue if not ts: continue metadata = d.get("metadata") message_metadata = msg.get("metadata") session_id = d.get("sessionId") or d.get("session_id") if not session_id and isinstance(metadata, dict): session_id = metadata.get("sessionId") if not session_id and isinstance(message_metadata, dict): session_id = message_metadata.get("sessionId") message_id = msg.get("id") request_id = d.get("requestId") or d.get("request_id") identity = None if all(isinstance(value, str) and value for value in (message_id, request_id)): identity = ("request", message_id, request_id) elif ( request_id is None and isinstance(session_id, str) and session_id and isinstance(message_id, str) and message_id ): identity = ("legacy", session_id, message_id) yield identity, ts, usage except (json.JSONDecodeError, ValueError): continue except (OSError, IOError): return def aggregate_usage(): """Scan all mimo session jsonls and return windowed token sums.""" now = datetime.now(timezone.utc) today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) # Week starts on Monday 00:00 UTC week_start = today_start - timedelta(days=today_start.weekday()) windows = { "today": {"input": 0, "output": 0, "cache_read": 0, "cache_create": 0, "messages": 0}, "week": {"input": 0, "output": 0, "cache_read": 0, "cache_create": 0, "messages": 0}, "all_time": {"input": 0, "output": 0, "cache_read": 0, "cache_create": 0, "messages": 0}, } sessions_scanned = 0 last_activity = None keyed_rows = {} unkeyed_rows = [] if not PROJECTS_DIR.exists(): return windows, sessions_scanned, last_activity for jsonl in PROJECTS_DIR.rglob("*.jsonl"): sessions_scanned += 1 for identity, ts_str, usage in parse_session_usage(jsonl): try: # Parse ISO timestamp (may end with Z) ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) except (ValueError, TypeError): continue row = (ts, usage) if identity is None: unkeyed_rows.append(row) else: previous = keyed_rows.get(identity) if previous is None or ts >= previous[0]: keyed_rows[identity] = row for ts, usage in [*keyed_rows.values(), *unkeyed_rows]: input_t = int(usage.get("input_tokens", 0) or 0) output_t = int(usage.get("output_tokens", 0) or 0) cache_read_t = int(usage.get("cache_read_input_tokens", 0) or 0) cache_create_t = int(usage.get("cache_creation_input_tokens", 0) or 0) if last_activity is None or ts > last_activity: last_activity = ts # all_time w = windows["all_time"] w["input"] += input_t w["output"] += output_t w["cache_read"] += cache_read_t w["cache_create"] += cache_create_t w["messages"] += 1 if ts >= week_start: w = windows["week"] w["input"] += input_t w["output"] += output_t w["cache_read"] += cache_read_t w["cache_create"] += cache_create_t w["messages"] += 1 if ts >= today_start: w = windows["today"] w["input"] += input_t w["output"] += output_t w["cache_read"] += cache_read_t w["cache_create"] += cache_create_t w["messages"] += 1 return windows, sessions_scanned, last_activity def write_cache(windows, sessions_scanned, last_activity): CACHE_PATH.parent.mkdir(parents=True, exist_ok=True) payload = { "updated_at": datetime.now(timezone.utc).isoformat(), "last_activity": last_activity.isoformat() if last_activity else None, "sessions_scanned": sessions_scanned, "windows": windows, "source": "local-jsonl-scan", "note": "Local token accounting from cc-mimo session jsonl. Not a quota; mimo platform.xiaomimimo.com SSO cookie required for real quota.", } tmp = CACHE_PATH.with_suffix(".json.tmp") tmp.write_text(json.dumps(payload, indent=2)) tmp.replace(CACHE_PATH) return payload def fmt_tokens(n: int) -> str: if n >= 1_000_000: return f"{n / 1_000_000:.1f}M" if n >= 1_000: return f"{n / 1_000:.1f}k" return str(n) def short_status(payload): """1-line status line.""" w = payload["windows"]["week"] total = w["input"] + w["output"] + w["cache_read"] + w["cache_create"] return f"mimo: {fmt_tokens(total)} tok this week ({w['messages']} msg)" def human_summary(payload): """Multi-line human-readable summary.""" last = payload.get("last_activity") if last: try: last_dt = datetime.fromisoformat(last) ago = datetime.now(timezone.utc) - last_dt if ago.total_seconds() < 60: ago_str = "just now" elif ago.total_seconds() < 3600: ago_str = f"{int(ago.total_seconds() / 60)}m ago" elif ago.total_seconds() < 86400: ago_str = f"{int(ago.total_seconds() / 3600)}h ago" else: ago_str = f"{ago.days}d ago" except (ValueError, TypeError): ago_str = last else: ago_str = "never" lines = [ "== MiMo (local tracker) ==", f"Sessions scanned: {payload['sessions_scanned']}", f"Last activity: {ago_str}", "", ] for window_name, label in [("today", "Today"), ("week", "This week"), ("all_time", "All time")]: w = payload["windows"][window_name] in_t = fmt_tokens(w["input"]) out_t = fmt_tokens(w["output"]) cr_t = fmt_tokens(w["cache_read"]) cc_t = fmt_tokens(w["cache_create"]) total = w["input"] + w["output"] + w["cache_read"] + w["cache_create"] lines.append(f"{label:>10}: {fmt_tokens(total):>8} total | in={in_t} out={out_t} cache_r={cr_t} cache_c={cc_t} | msg={w['messages']}") lines.append("") lines.append("Note: this is local accounting from cc-mimo session jsonl.") lines.append("Real platform quota requires Chrome cookie (cookieSource=manual).") return "\n".join(lines) def main(): args = sys.argv[1:] quiet = "--update" in args json_out = "--json" in args short = "--short" in args windows, sessions_scanned, last_activity = aggregate_usage() payload = write_cache(windows, sessions_scanned, last_activity) if quiet: return 0 if json_out: print(json.dumps(payload, indent=2)) return 0 if short: print(short_status(payload)) return 0 print(human_summary(payload)) return 0 if __name__ == "__main__": sys.exit(main())