fuzzyworld-tools / files / lib / usage.py
usage.py
| 1 | #!/usr/bin/env python3 |
| 2 | """fw usage: token use per Claude Code session, from ~/.claude/projects/*/*.jsonl. |
| 3 | |
| 4 | fw usage sessions of the current directory's project, newest first |
| 5 | fw usage --all every project |
| 6 | fw usage --now one line about the newest session here: messages, current |
| 7 | context size, model share (use it to decide on "put down") |
| 8 | fw usage --trailers [--session ID] |
| 9 | the commit-message trailers for the newest session here (or the |
| 10 | one named): exact integers, plus one FW-Model-Tokens line per model |
| 11 | (in, out, cache read, 5-minute and 1-hour cache writes) so the |
| 12 | site can price a commit. Subagents are included, each message is |
| 13 | counted once. Warns on stderr when the transcript is over 10 |
| 14 | minutes old (it may be the wrong session). FW-Tokens-Method: 2 |
| 15 | marks this counting; older trailers double counted transcript lines. |
| 16 | """ |
| 17 | import glob, json, os, re, subprocess, sys, time |
| 18 | |
| 19 | root = os.path.expanduser('~/.claude/projects') |
| 20 | |
| 21 | def _load_over100k(): |
| 22 | """{model: True} for models whose price table has an above_100k block.""" |
| 23 | try: |
| 24 | with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'server', 'prices.json')) as f: |
| 25 | return {k: True for k, v in json.load(f).get('models', {}).items() if 'above_100k' in v} |
| 26 | except (OSError, ValueError): |
| 27 | print('warning: server/prices.json is missing or unreadable; long-prompt (over 100K) requests ' |
| 28 | 'will not be filed separately and may be underpriced later', file=sys.stderr) |
| 29 | return {} |
| 30 | |
| 31 | OVER_100K = _load_over100k() |
| 32 | |
| 33 | def scan(path): |
| 34 | """Totals include the session's subagent transcripts (<session>/subagents/*.jsonl); |
| 35 | the context size and message count are the main conversation's. per_model is |
| 36 | {model: [in, out, read, write5m, write1h]} over main and subagents.""" |
| 37 | msgs, tot, ctx, models, per = scan_one(path) |
| 38 | for sub in glob.glob(path[:-6] + '/subagents/*.jsonl'): |
| 39 | _, t, _, _, p = scan_one(sub) |
| 40 | tot = [a + b for a, b in zip(tot, t)] |
| 41 | for k, v in p.items(): |
| 42 | per[k] = [a + b for a, b in zip(per.get(k, [0] * 5), v)] |
| 43 | return msgs, tot, ctx, models, per |
| 44 | |
| 45 | def scan_one(path): |
| 46 | """Each message id counts once: Claude Code writes one line per content block, each repeating the usage. |
| 47 | per_model keys are the model id, or '<model>@over100k' for a request whose prompt (input + cache read + |
| 48 | cache write) is over 100,000 tokens when the model's price table has a higher tier.""" |
| 49 | seen, msgs, tot, ctx, models, per = set(), 0, [0, 0, 0, 0], 0, {}, {} |
| 50 | with open(path, errors='ignore') as fh: |
| 51 | for line in fh: |
| 52 | try: d = json.loads(line) |
| 53 | except ValueError: continue |
| 54 | m = d.get('message') |
| 55 | u = m.get('usage') if isinstance(m, dict) else None |
| 56 | if not u or d.get('type') != 'assistant': continue |
| 57 | k = m.get('id') or d.get('uuid') |
| 58 | if k in seen: continue |
| 59 | seen.add(k) |
| 60 | v = [u.get('input_tokens', 0), u.get('output_tokens', 0), |
| 61 | u.get('cache_read_input_tokens', 0), u.get('cache_creation_input_tokens', 0)] |
| 62 | tot = [a + b for a, b in zip(tot, v)] |
| 63 | ctx = v[0] + v[2] + v[3] |
| 64 | msgs += 1 |
| 65 | model = m.get('model', '?') |
| 66 | models[model] = models.get(model, 0) + 1 |
| 67 | cc = u.get('cache_creation') |
| 68 | if isinstance(cc, dict) and ('ephemeral_5m_input_tokens' in cc or 'ephemeral_1h_input_tokens' in cc): |
| 69 | w5, w1 = cc.get('ephemeral_5m_input_tokens', 0), cc.get('ephemeral_1h_input_tokens', 0) |
| 70 | else: |
| 71 | w5, w1 = v[3], 0 |
| 72 | key = model + '@over100k' if ctx > 100000 and model in OVER_100K else model |
| 73 | per[key] = [a + b for a, b in zip(per.get(key, [0] * 5), [v[0], v[1], v[2], w5, w1])] |
| 74 | return msgs, tot, ctx, models, per |
| 75 | |
| 76 | def trailers(path): |
| 77 | """The commit-message trailer lines for one session transcript.""" |
| 78 | _, _, _, _, per = scan(path) |
| 79 | per = {k: v for k, v in per.items() if k != '<synthetic>' and any(v)} |
| 80 | lines = [f'FW-Tokens-In: {sum(v[0] for v in per.values())}', |
| 81 | f'FW-Tokens-Out: {sum(v[1] for v in per.values())}', |
| 82 | f'FW-Tokens-Cached: {sum(v[2] for v in per.values())}', |
| 83 | 'FW-Tokens-Method: 2'] |
| 84 | for k in sorted(per): |
| 85 | v = per[k] |
| 86 | lines.append(f'FW-Model-Tokens: {k} in={v[0]} out={v[1]} read={v[2]} write5m={v[3]} write1h={v[4]}') |
| 87 | lines.append(f'FW-Session: {os.path.basename(path)[:-6]}') |
| 88 | try: |
| 89 | r = subprocess.run(['rtk', 'gain', '--format', 'json'], capture_output=True, text=True, timeout=30, check=True) |
| 90 | s = json.loads(r.stdout)['summary'] |
| 91 | lines += [f'FW-Rtk-Saved: {s["total_saved"]}', f'FW-Rtk-Cmds: {s["total_commands"]}'] |
| 92 | except (OSError, subprocess.SubprocessError, ValueError, KeyError): |
| 93 | pass |
| 94 | return lines |
| 95 | |
| 96 | def fmt(n): return f'{n:,}' |
| 97 | |
| 98 | def main(argv): |
| 99 | args = argv |
| 100 | proj = os.path.join(root, re.sub(r'[^a-zA-Z0-9]', '-', os.environ.get('INIT_CWD') or os.getcwd())) |
| 101 | dirs = sorted(glob.glob(root + '/*')) if '--all' in args else [proj] |
| 102 | files = [f for d in dirs for f in glob.glob(d + '/*.jsonl')] |
| 103 | files.sort(key=os.path.getmtime, reverse=True) |
| 104 | if not files: |
| 105 | sys.exit('no sessions found for ' + proj) |
| 106 | |
| 107 | if '--trailers' in args: |
| 108 | f = files[0] |
| 109 | if '--session' in args: |
| 110 | i = args.index('--session') + 1 |
| 111 | if i >= len(args): |
| 112 | sys.exit('usage: usage.py --trailers [--session ID]') |
| 113 | sid = args[i] |
| 114 | f = os.path.join(proj, sid + '.jsonl') |
| 115 | if not os.path.exists(f): |
| 116 | sys.exit('no such session: ' + f) |
| 117 | if time.time() - os.path.getmtime(f) > 600: |
| 118 | print(f'warning: {f} was last written over 10 minutes ago; is it the current session?', file=sys.stderr) |
| 119 | print('\n'.join(trailers(f))) |
| 120 | return |
| 121 | |
| 122 | if '--now' in args: |
| 123 | f = files[0] |
| 124 | msgs, tot, ctx, models, _ = scan(f) |
| 125 | share = ', '.join(f'{k.replace("claude-", "")} {v}' for k, v in sorted(models.items()) if k != '<synthetic>') |
| 126 | print(f'{os.path.basename(f)[:8]}: {msgs} msgs, context now {fmt(ctx)} tokens, ' |
| 127 | f'cache-read so far {fmt(tot[2])}, output {fmt(tot[1])}; models: {share}') |
| 128 | sys.exit() |
| 129 | |
| 130 | print(f'{"session":9}{"project":28}{"msgs":>6}{"context":>12}{"cache-read":>16}{"output":>12} last used') |
| 131 | for f in files: |
| 132 | msgs, tot, ctx, _, _ = scan(f) |
| 133 | p = os.path.basename(os.path.dirname(f)).replace('-Users-pavel-', '')[:27] |
| 134 | print(f'{os.path.basename(f)[:8]:9}{p:28}{msgs:>6}{fmt(ctx):>12}{fmt(tot[2]):>16}{fmt(tot[1]):>12} ' |
| 135 | + time.strftime('%Y-%m-%d %H:%M', time.localtime(os.path.getmtime(f)))) |
| 136 | |
| 137 | if __name__ == "__main__": |
| 138 | main(sys.argv[1:]) |