|
#!/usr/bin/env python3 |
|
"""Claude Code status line: model, context %, and 5h usage with a projection. |
|
|
|
Reads the status line JSON on stdin and prints one line. The 5h segment is |
|
colored by the usage projected at the end of the window, from the usage rate |
|
over the last --avg minutes. |
|
|
|
rate_limits comes from each session's most recent API response, so an idle |
|
session keeps reporting an old value. Usage only grows within a window, so |
|
the history is shared across sessions and records the time each new highest |
|
percentage was first seen; the highest one is the current value. |
|
|
|
It also shows this month's extra usage spend. Claude Code does not pass that |
|
to status lines, so at most every 10 minutes one session starts |
|
`claude -p /usage` in the background and caches the result; every run shows |
|
the cached value. With --no-extra-usage the script runs nothing else. |
|
|
|
Configure in ~/.claude/settings.json: |
|
"statusLine": {"type": "command", "command": "~/bin/claude-statusline", "refreshInterval": 60} |
|
""" |
|
|
|
import argparse |
|
import json |
|
import os |
|
import shutil |
|
import subprocess |
|
import sys |
|
import tempfile |
|
import time |
|
from pathlib import Path |
|
|
|
CACHE_DIR = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "claude-statusline" |
|
STATE_PATH = CACHE_DIR / "five_hour.json" |
|
EXTRA_PATH = CACHE_DIR / "extra_usage.json" |
|
EXTRA_LOCK = CACHE_DIR / "extra_usage.lock" |
|
EXTRA_INTERVAL_S = 600 |
|
EXTRA_MAX_BACKOFF_S = 3600 |
|
EXTRA_TIMEOUT_S = 60 |
|
# /usage runs locally without a model request. Hooks and MCP servers are |
|
# skipped since they are not needed, and the run is not saved as a session. |
|
USAGE_ARGS = ["-p", "/usage", "--output-format", "stream-json", "--verbose", |
|
"--settings", '{"disableAllHooks": true}', "--strict-mcp-config", |
|
"--no-session-persistence"] |
|
WINDOW_S = 5 * 3600 |
|
# resets_at can shift slightly between responses. |
|
RESET_JITTER_S = 60 |
|
# History is kept for this many averaging periods, beyond what the rate needs. |
|
HISTORY_PERIODS = 10 |
|
|
|
# Projected final usage (%) -> color. |
|
COLOR_STOPS = [ |
|
(50, (0x33, 0xD1, 0x7A)), # green |
|
(65, (0xF6, 0xD3, 0x2D)), # yellow |
|
(100, (0xE0, 0x1B, 0x24)), # red |
|
] |
|
# Extra usage spend (cents) -> color, highest threshold first; no color at $0. |
|
EXTRA_COLORS = [ |
|
(20000, (0xFF, 0x78, 0x00)), # orange, over $200 |
|
(0, (0xF6, 0xD3, 0x2D)), # yellow, over $0 |
|
] |
|
DIM = "\033[2m" |
|
RESET = "\033[0m" |
|
|
|
|
|
def projection_color(projected): |
|
if projected <= COLOR_STOPS[0][0]: |
|
return COLOR_STOPS[0][1] |
|
for (lo, lo_rgb), (hi, hi_rgb) in zip(COLOR_STOPS, COLOR_STOPS[1:]): |
|
if projected <= hi: |
|
f = (projected - lo) / (hi - lo) |
|
return tuple(round(a + f * (b - a)) for a, b in zip(lo_rgb, hi_rgb)) |
|
return COLOR_STOPS[-1][1] |
|
|
|
|
|
def fg(rgb): |
|
return "\033[38;2;{};{};{}m".format(*rgb) |
|
|
|
|
|
def format_remaining(seconds): |
|
total_min = int(seconds // 60) |
|
if total_min < 0: |
|
return "now" |
|
days, rem = divmod(total_min, 1440) |
|
hours, minutes = divmod(rem, 60) |
|
if days: |
|
return f"{days}d{hours}h" |
|
if hours: |
|
return f"{hours}h{minutes}m" |
|
return f"{minutes}m" |
|
|
|
|
|
def load_json(path): |
|
try: |
|
return json.loads(path.read_text()) |
|
except (OSError, ValueError): |
|
return {} |
|
|
|
|
|
def save_json(path, obj): |
|
path.parent.mkdir(parents=True, exist_ok=True) |
|
fd, tmp = tempfile.mkstemp(dir=path.parent, prefix="." + path.stem + ".") |
|
with os.fdopen(fd, "w") as f: |
|
json.dump(obj, f) |
|
os.replace(tmp, path) |
|
|
|
|
|
def live_window(resets_at, now): |
|
"""A window resets in the future, and no more than its length from now.""" |
|
return resets_at is not None and now < resets_at <= now + WINDOW_S + RESET_JITTER_S |
|
|
|
|
|
def record(percent, resets_at, now, keep_s): |
|
"""Adds this reading to the shared history. |
|
|
|
Returns (resets_at, steps) for the window being tracked, or None if there |
|
is none. Each step is [time, percent]: when that percentage was first seen. |
|
The tracked window is replaced only after it has ended, so a reading for |
|
any other window, from a stale session or bad input, cannot clear it. |
|
""" |
|
state = load_json(STATE_PATH) |
|
tracked, steps = state.get("resets_at"), state.get("steps") or [] |
|
if not (steps and live_window(tracked, now)): |
|
if not live_window(resets_at, now): |
|
return None |
|
tracked, steps = resets_at, [] |
|
elif abs(tracked - resets_at) > RESET_JITTER_S: |
|
return tracked, steps |
|
if not steps or percent > steps[-1][1]: |
|
# A new history may start from a stale session's lower value, so its |
|
# first step is replaced until it has stood for a minute. |
|
if len(steps) == 1 and now - steps[0][0] < 60: |
|
steps = [] |
|
steps.append([now, percent]) |
|
# Steps older than keep_s are dropped, except the newest of them, |
|
# which still gives the value at the start of the averaging window. |
|
old = [s for s in steps if now - s[0] > keep_s] |
|
steps = old[-1:] + [s for s in steps if now - s[0] <= keep_s] |
|
save_json(STATE_PATH, {"resets_at": tracked, "steps": steps}) |
|
return tracked, steps |
|
|
|
|
|
def tracked_history(now): |
|
"""Returns (resets_at, steps) for the tracked window if it is still live.""" |
|
state = load_json(STATE_PATH) |
|
tracked, steps = state.get("resets_at"), state.get("steps") or [] |
|
if steps and live_window(tracked, now): |
|
return tracked, steps |
|
return None |
|
|
|
|
|
def project(steps, resets_at, now, window_s): |
|
"""Returns (current, projected, rate per hour), or None without enough history.""" |
|
current = steps[-1][1] |
|
before = [s for s in steps if now - s[0] >= window_s] |
|
if before: |
|
start_time, start_pct = now - window_s, before[-1][1] |
|
else: |
|
start_time, start_pct = steps[0] |
|
span = now - start_time |
|
if span < 60: |
|
return None |
|
rate = (current - start_pct) / span |
|
return current, current + rate * max(0, resets_at - now), rate * 3600 |
|
|
|
|
|
def extra_lock(blocking): |
|
"""Takes the extra usage lock; returns the open lock file, or None if it is |
|
held elsewhere or locking is unavailable (fcntl is POSIX only).""" |
|
try: |
|
import fcntl |
|
except ImportError: |
|
return None |
|
EXTRA_LOCK.parent.mkdir(parents=True, exist_ok=True) |
|
f = open(EXTRA_LOCK, "a") |
|
try: |
|
fcntl.flock(f, fcntl.LOCK_EX | (0 if blocking else fcntl.LOCK_NB)) |
|
except OSError: |
|
f.close() |
|
return None |
|
return f |
|
|
|
|
|
def maybe_start_extra_refresh(now): |
|
"""Starts a background refresh if one is due and no other session has.""" |
|
if now < load_json(EXTRA_PATH).get("next_attempt", 0): |
|
return |
|
lock = extra_lock(blocking=False) |
|
if lock is None: |
|
return |
|
with lock: |
|
cache = load_json(EXTRA_PATH) |
|
if now < cache.get("next_attempt", 0): |
|
return |
|
# Claimed before starting the refresh, so no other session starts one; |
|
# the refresh sets the real next time when it finishes. |
|
cache["next_attempt"] = now + EXTRA_TIMEOUT_S + EXTRA_INTERVAL_S |
|
save_json(EXTRA_PATH, cache) |
|
subprocess.Popen([sys.executable, os.path.abspath(__file__), "--refresh-extra-usage"], |
|
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, |
|
stderr=subprocess.DEVNULL, start_new_session=True) |
|
|
|
|
|
def find_key(obj, key): |
|
if isinstance(obj, dict): |
|
if key in obj: |
|
return obj[key] |
|
obj = list(obj.values()) |
|
if isinstance(obj, list): |
|
for v in obj: |
|
found = find_key(v, key) |
|
if found is not None: |
|
return found |
|
return None |
|
|
|
|
|
def refresh_extra_usage(): |
|
"""Runs `claude -p /usage` and caches its extra usage report.""" |
|
claude = shutil.which("claude") or str(Path.home() / ".local" / "bin" / "claude") |
|
extra, error = None, None |
|
try: |
|
out = subprocess.run([claude] + USAGE_ARGS, stdin=subprocess.DEVNULL, |
|
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, |
|
timeout=EXTRA_TIMEOUT_S, cwd=str(CACHE_DIR)).stdout |
|
for line in out.decode("utf-8", "replace").splitlines(): |
|
try: |
|
msg = json.loads(line) |
|
except ValueError: |
|
continue |
|
report = find_key(msg, "usage_report") |
|
if report: |
|
extra = (report.get("rate_limits") or {}).get("extra_usage") |
|
break |
|
if extra is None: |
|
error = "no extra_usage in /usage output" |
|
except (OSError, subprocess.SubprocessError) as e: |
|
error = str(e)[:200] |
|
|
|
now = time.time() |
|
lock = extra_lock(blocking=True) |
|
try: |
|
cache = load_json(EXTRA_PATH) |
|
if extra is not None: |
|
cache.update(extra_usage=extra, fetched_at=now, failures=0, error=None, |
|
next_attempt=now + EXTRA_INTERVAL_S) |
|
else: |
|
failures = cache.get("failures", 0) + 1 |
|
cache.update(failures=failures, error=error, next_attempt=now + min( |
|
EXTRA_INTERVAL_S * 2 ** failures, EXTRA_MAX_BACKOFF_S)) |
|
save_json(EXTRA_PATH, cache) |
|
finally: |
|
if lock: |
|
lock.close() |
|
|
|
|
|
def extra_usage_segment(now): |
|
cache = load_json(EXTRA_PATH) |
|
extra = cache.get("extra_usage") |
|
if not extra or not extra.get("is_enabled"): |
|
return None |
|
# Amounts are in cents; spend is rounded up to the next dollar. |
|
used_cents = extra.get("used_credits") or 0 |
|
used = -(-used_cents // 100) |
|
limit = (extra.get("monthly_limit") or 0) / 100 |
|
currency = extra.get("currency") or "USD" |
|
sym, suffix = ("$", "") if currency == "USD" else ("", " " + currency) |
|
text = f"extra {sym}{used:,.0f}/{sym}{limit:,.0f}{suffix}" |
|
# Older than a few refreshes: shown dimmed with a marker. |
|
if now - cache.get("fetched_at", 0) > 3 * EXTRA_INTERVAL_S: |
|
return f"{DIM}{text}?{RESET}" |
|
for threshold, rgb in EXTRA_COLORS: |
|
if used_cents > threshold: |
|
return f"{fg(rgb)}{text}{RESET}" |
|
return text |
|
|
|
|
|
def main(): |
|
ap = argparse.ArgumentParser() |
|
ap.add_argument("--avg", type=float, default=10, help="averaging period in minutes (default 10)") |
|
ap.add_argument("--no-extra-usage", dest="extra_usage", action="store_false", |
|
help="don't show this month's extra usage spend, which is otherwise " |
|
"refreshed in the background with `claude -p /usage`") |
|
ap.add_argument("--refresh-extra-usage", action="store_true", help=argparse.SUPPRESS) |
|
args = ap.parse_args() |
|
if args.refresh_extra_usage: |
|
refresh_extra_usage() |
|
return |
|
|
|
try: |
|
data = json.load(sys.stdin) |
|
except ValueError: |
|
data = {} |
|
now = time.time() |
|
parts = [] |
|
|
|
model = (data.get("model") or {}).get("display_name") |
|
if model: |
|
parts.append(model) |
|
|
|
ctx = (data.get("context_window") or {}).get("used_percentage") |
|
if ctx is not None: |
|
parts.append(f"ctx {ctx:.0f}%") |
|
|
|
five = (data.get("rate_limits") or {}).get("five_hour") |
|
window_s = args.avg * 60 |
|
if five and five.get("resets_at"): |
|
rec = record(five.get("used_percentage", 0), five["resets_at"], now, |
|
keep_s=window_s * HISTORY_PERIODS) |
|
if not rec: |
|
rec = (five["resets_at"], [[now, five.get("used_percentage", 0)]]) |
|
else: |
|
# A session that has had no API response in the current window gets no |
|
# five_hour; other sessions may have seen it. |
|
rec = tracked_history(now) |
|
if rec: |
|
resets_at, steps = rec |
|
current = steps[-1][1] |
|
p = project(steps, resets_at, now, window_s) |
|
left = format_remaining(resets_at - now) |
|
if p: |
|
_, projected, rate = p |
|
parts.append(f"{fg(projection_color(projected))}5h {current:.0f}% · {left} " |
|
f"→ {projected:.0f}%{RESET}{DIM} ({rate:.1f}%/h){RESET}") |
|
else: |
|
parts.append(f"5h {current:.0f}% · {left}") |
|
|
|
seven = (data.get("rate_limits") or {}).get("seven_day") |
|
if seven: |
|
parts.append(f"7d {seven.get('used_percentage', 0):.0f}%") |
|
|
|
if args.extra_usage: |
|
maybe_start_extra_refresh(now) |
|
seg = extra_usage_segment(now) |
|
if seg: |
|
parts.append(seg) |
|
|
|
print(" · ".join(parts)) |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |