Claude Code Usage

| #!/usr/bin/env -S uv run | | | | | | | """ | | claude-usage.py - Captures the output of the /usage command from the Claude CLI. | | | | Drives the interactive claude TUI through a pseudo-terminal (PTY): launches it, | | sends /usage, snapshots the rendered screen, then shuts the process down cleanly. | | | | Usage: | | uv run claude-usage.py | | | | Steps performed: | |

1. Launch claude attached to a PTY so it behaves as an interactive session. | |
2. Wait for it to reach the input prompt. | |
3. Type /usage and press ENTER. | |
4. Wait, then capture the rendered screen. | |
5. Shut down: ESC -> wait -> /exit ENTER -> wait -> kill if still alive. | |
6. Print the captured screen from step 4. | | """ | | | | from future import annotations | | | | import argparse | | import fcntl | | import json | | import os | | import re | | import select | | import shutil | | import signal | | import struct | | import sys | | import termios | | import threading | | import time | | from datetime import datetime, timedelta | | from zoneinfo import ZoneInfo | | | | import pyte | | | | CLAUDE_COMMAND = "claude" | | | | TERM_ROWS = 150 | | TERM_COLS = 160 | | | | WAIT_AFTER_START = 3.0 | | WAIT_AFTER_USAGE = 3.0 | | WAIT_AFTER_ESC = 2.0 | | WAIT_AFTER_EXIT = 3.0 | | TYPE_DELAY = 0.4 | | | | ENTER = b"\r" | | ESC = b"\x1b" | | | | | | VERBOSE = True | | | | | | def log(message: str) -> None: | | if VERBOSE: | | print(f"[claude-usage] {message}", file=sys.stderr, flush=True) | | | | | | def require_command(name: str) -> None: | | if shutil.which(name) is None: | | print(f"{name} is not installed or not on PATH", file=sys.stderr) | | sys.exit(1) | | | | | | def set_window_size(fd: int, rows: int, cols: int) -> None: | | winsize = struct.pack("HHHH", rows, cols, 0, 0) | | fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize) | | | | | | def render_screen(screen: pyte.Screen) -> str: | | lines = [line.rstrip() for line in screen.display] | | while lines and not lines[-1]: | | lines.pop() | | return "\n".join(lines) | | | | | | MONTHS = { | | "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6, | | "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12, | | } | | | | | | def parse_clock(text: str) -> tuple[int, int]: | | """Parses a clock string like '7:30pm', '4am' or '12pm' into (hour, minute).""" | | match = re.match(r"^(\d{1,2})(?::(\d{2}))?\s*([ap]m)$", text.strip().lower()) | | if not match: | | raise ValueError(f"Unrecognized time format: {text!r}") | | hour = int(match.group(1)) | | minute = int(match.group(2)) if match.group(2) else 0 | | meridiem = match.group(3) | | if meridiem == "pm" and hour != 12: | | hour += 12 | | elif meridiem == "am" and hour == 12: | | hour = 0 | | return hour, minute | | | | | | def parse_reset(text: str) -> str | None: | | """Resolves a reset string into an ISO 8601 timestamp. | | | | Accepts 'time (Zone)' (next occurrence of that time) or | | 'Month Day at time (Zone)' (next occurrence of that date). | | """ | | if not text: | | return None | | | | tz_match = re.search(r"(([^)]+))\s*$", text) | | try: | | tz = ZoneInfo(tz_match.group(1)) if tz_match else ZoneInfo("UTC") | | except Exception: | | tz = ZoneInfo("UTC") | | | | body = re.sub(r"\s*([^)]+)\s*$", "", text).strip() | | now = datetime.now(tz) | | | | dated = re.match(r"^([A-Za-z]+)\s+(\d+)\s+at\s+(.+)$", body) | | if dated: | | month = MONTHS.get(dated.group(1)[:3].lower()) | | day = int(dated.group(2)) | | hour, minute = parse_clock(dated.group(3)) | | if month is None: | | return None | | dt = datetime(now.year, month, day, hour, minute, tzinfo=tz) | | if dt < now: | | dt = dt.replace(year=now.year + 1) | | return dt.isoformat() | | | | try: | | hour, minute = parse_clock(body) | | except ValueError: | | return None | | dt = now.replace(hour=hour, minute=minute, second=0, microsecond=0) | | if dt <= now: | | dt += timedelta(days=1) | | return dt.isoformat() | | | | | | def parse_usage_block(lines: list[str], label: str) -> dict | None: | | """Finds a usage label and extracts its '<n>% used' value and reset time.""" | | for i, line in enumerate(lines): | | if line.strip() == label: | | usage = None | | resets = None | | for row in lines[i + 1:i + 5]: | | if usage is None: | | used = re.search(r"(\d+)%\s+used", row) | | if used: | | usage = int(used.group(1)) | | if resets is None: | | reset_line = re.search(r"Resets\s+(.+)", row) | | if reset_line: | | resets = reset_line.group(1).strip() | | return {"usage": usage, "resets": parse_reset(resets)} | | return None | | | | | | def parse_percentage_table(lines: list[str], header: str) -> dict: | | """Parses a '<Header> % of usage' table into {name: percentage}.""" | | result: dict[str, int] = {} | | for i, line in enumerate(lines): | | if re.match(rf"\s*{re.escape(header)}\s+% of usage", line): | | for row in lines[i + 1:]: | | if not row.strip(): | | break | | cell = re.match(r"^(.?)\s+(\d+)%\s$", row.strip()) | | if not cell: | | break | | result[cell.group(1).strip()] = int(cell.group(2)) | | break | | return result | | | | | | def build_usage_json(captured: str) -> dict: | | lines = captured.splitlines() | | return { | | "current_usage": { | | "current_session": parse_usage_block(lines, "Current session"), | | "current_week": parse_usage_block(lines, "Current week (all models)"), | | "current_week_sonnet": parse_usage_block(lines, "Current week (Sonnet only)"), | | }, | | "skills": parse_percentage_table(lines, "Skills"), | | "subagents": parse_percentage_table(lines, "Subagents"), | | "plugins": parse_percentage_table(lines, "Plugins"), | | } | | | | | | HELP_DESCRIPTION = """\ | | Captures the /usage output of the Claude CLI and emits it as JSON. | | | | Drives the interactive claude TUI through a pseudo-terminal (PTY): launches it, | | types /usage, snapshots the rendered screen, shuts the process down cleanly, | | parses the screen and prints structured JSON to stdout. | | | | Output streams: | | stdout Structured JSON result (single compact line by default). | | stderr Progress logs and the raw captured screen (suppressed with --quiet). | | | | JSON shape: | | { | | "current_usage": { | | "current_session": { "usage": <int %>, "resets": <ISO 8601> }, | | "current_week": { "usage": <int %>, "resets": <ISO 8601> }, | | "current_week_sonnet": { "usage": <int %>, "resets": <ISO 8601> } | | }, | | "skills": { "<name>": <int %>, ... }, | | "subagents": { "<name>": <int %>, ... }, | | "plugins": { "<name>": <int %>, ... } | | } | | """ | | | | HELP_EPILOG = """\ | | Examples: | | uv run claude-usage.py Compact JSON on stdout, logs on stderr. | | uv run claude-usage.py -f Pretty-printed JSON, logs on stderr. | | uv run claude-usage.py -V Compact JSON only; no stderr output. | | uv run claude-usage.py -V -f Pretty-printed JSON only; no stderr output. | | """ | | | | | | def parse_args() -> argparse.Namespace: | | parser = argparse.ArgumentParser( | | prog="claude-usage.py", | | description=HELP_DESCRIPTION, | | epilog=HELP_EPILOG, | | formatter_class=argparse.RawDescriptionHelpFormatter, | | ) | | verbosity = parser.add_mutually_exclusive_group() | | verbosity.add_argument( | | "-v", "--verbose", action="store_true", | | help="Print progress logs and the captured screen to stderr (default).", | | ) | | verbosity.add_argument( | | "-V", "--quiet", action="store_true", | | help="Suppress all stderr log output.", | | ) | | parser.add_argument( | | "-f", "--format", action="store_true", | | help="Pretty-print the JSON (indented). Default is a single compact line.", | | ) | | return parser.parse_args() | | | | | | def main() -> None: | | args = parse_args() | | global VERBOSE | | VERBOSE = not args.quiet | | | | require_command(CLAUDE_COMMAND) | | | | master_fd, slave_fd = os.openpty() | | set_window_size(slave_fd, TERM_ROWS, TERM_COLS) | | | | screen = pyte.Screen(TERM_COLS, TERM_ROWS) | | stream = pyte.ByteStream(screen) | | lock = threading.Lock() | | stop = threading.Event() | | | | def reader() -> None: | | while not stop.is_set(): | | try: | | ready, _, _ = select.select([master_fd], [], [], 0.2) | | except (OSError, ValueError): | | break | | if not ready: | | continue | | try: | | data = os.read(master_fd, 65536) | | except OSError: | | break | | if not data: | | break | | with lock: | | stream.feed(data) | | | | def preexec() -> None: | | os.setsid() | | fcntl.ioctl(slave_fd, termios.TIOCSCTTY, 0) | | | | reader_thread = threading.Thread(target=reader, daemon=True) | | reader_thread.start() | | | | env = dict(os.environ) | | env["TERM"] = env.get("TERM", "xterm-256color") | | | | import subprocess | | | | log(f"Launching {CLAUDE_COMMAND} on a {TERM_COLS}x{TERM_ROWS} PTY...") | | proc = subprocess.Popen( | | [CLAUDE_COMMAND], | | stdin=slave_fd, | | stdout=slave_fd, | | stderr=slave_fd, | | preexec_fn=preexec, | | env=env, | | close_fds=True, | | ) | | os.close(slave_fd) | | log(f"Process started (pid {proc.pid}).") | | | | def send(data: bytes) -> None: | | try: | | os.write(master_fd, data) | | except OSError: | | pass | | | | captured = "" | | try: | | | log(f"Waiting {WAIT_AFTER_START:g}s for the input prompt...") | | time.sleep(WAIT_AFTER_START) | | | | | log("Sending /usage + ENTER.") | | send(b"/usage") | | time.sleep(TYPE_DELAY) | | send(ENTER) | | | | | log(f"Waiting {WAIT_AFTER_USAGE:g}s for /usage output...") | | time.sleep(WAIT_AFTER_USAGE) | | log("Capturing rendered screen.") | | with lock: | | captured = render_screen(screen) | | | | | log("Shutting down: ESC.") | | send(ESC) | | time.sleep(WAIT_AFTER_ESC) | | log("Sending /exit + ENTER.") | | send(b"/exit") | | time.sleep(TYPE_DELAY) | | send(ENTER) | | | | log(f"Waiting up to {WAIT_AFTER_EXIT:g}s for clean exit...") | | try: | | proc.wait(timeout=WAIT_AFTER_EXIT) | | log("Process exited on its own.") | | except subprocess.TimeoutExpired: | | log("Still alive; terminating.") | | proc.terminate() | | try: | | proc.wait(timeout=2.0) | | except subprocess.TimeoutExpired: | | log("Did not terminate; killing.") | | proc.kill() | | finally: | | stop.set() | | if proc.poll() is None: | | proc.send_signal(signal.SIGKILL) | | try: | | os.close(master_fd) | | except OSError: | | pass | | reader_thread.join(timeout=2.0) | | | | | if VERBOSE: | | log("Captured screen:") | | print(captured, file=sys.stderr) | | log("Done. Emitting JSON to stdout.") | | | | data = build_usage_json(captured) | | indent = 2 if args.format else None | | print(json.dumps(data, indent=indent, ensure_ascii=False)) | | | | | | if name == "main": | | try: | | main() | | except KeyboardInterrupt: | | sys.exit(130) |