|
#!/usr/bin/env python3
|
|
"""Claude Code custom statusline — minimal 2-line: context bar + rate limits, git + session.
|
|
Works on macOS and Linux. Drop into ~/.claude/statusline.py
|
|
"""
|
|
|
|
import sys
|
|
import io
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import urllib.request
|
|
|
|
# Force UTF-8 output (safe on all platforms)
|
|
if hasattr(sys.stdout, "buffer"):
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
|
sys.stdin = io.TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace")
|
|
|
|
# ANSI colors
|
|
E = "\033"
|
|
GREEN = f"{E}[38;2;0;200;0m"
|
|
CYAN = f"{E}[38;2;100;200;200m"
|
|
RED = f"{E}[38;2;255;85;85m"
|
|
YELLOW = f"{E}[38;2;230;200;0m"
|
|
WHITE = f"{E}[38;2;220;220;220m"
|
|
GRAY = f"{E}[38;2;140;140;140m"
|
|
DIM = f"{E}[2m"
|
|
RST = f"{E}[0m"
|
|
FILLED = chr(0x25B0) # ▰
|
|
EMPTY = chr(0x25B1) # ▱
|
|
|
|
|
|
def fmt_tokens(num):
|
|
if num >= 1_000_000:
|
|
return f"{num / 1_000_000:.1f}m"
|
|
if num >= 1_000:
|
|
return f"{round(num / 1000)}k"
|
|
return str(num)
|
|
|
|
|
|
def bar(pct, width=20):
|
|
pct = max(0, min(100, pct))
|
|
filled = round(pct * width / 100)
|
|
empty = width - filled
|
|
c = RED if pct >= 75 else YELLOW if pct >= 50 else GREEN
|
|
return f"{c}{FILLED * filled}{DIM}{EMPTY * empty}{RST}"
|
|
|
|
|
|
def fmt_reset(iso, style="time"):
|
|
if not iso:
|
|
return ""
|
|
try:
|
|
utc = datetime.fromisoformat(iso.replace("Z", "+00:00"))
|
|
local = utc.astimezone()
|
|
if style == "time":
|
|
return local.strftime("%I:%M%p").lstrip("0").lower()
|
|
if style == "datetime":
|
|
m = local.strftime("%b").lower()
|
|
return f"{m} {local.day}, {local.strftime('%I:%M%p').lstrip('0').lower()}"
|
|
m = local.strftime("%b").lower()
|
|
return f"{m} {local.day}"
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def fmt_reset_compact(iso, style="time"):
|
|
"""Compact reset: '9pm', 'feb17', 'mar1'."""
|
|
raw = fmt_reset(iso, style)
|
|
if not raw:
|
|
return ""
|
|
if style == "time":
|
|
return re.sub(r':00(am|pm)', r'\1', raw)
|
|
return raw.replace(" ", "")
|
|
|
|
|
|
def get_git_info(cwd):
|
|
"""Get git branch and dirty file count."""
|
|
if not cwd:
|
|
return None, 0
|
|
try:
|
|
branch = subprocess.check_output(
|
|
["git", "rev-parse", "--abbrev-ref", "HEAD"],
|
|
cwd=cwd, stderr=subprocess.DEVNULL, timeout=3
|
|
).decode().strip()
|
|
dirty_out = subprocess.check_output(
|
|
["git", "status", "--porcelain"],
|
|
cwd=cwd, stderr=subprocess.DEVNULL, timeout=3
|
|
).decode().strip()
|
|
dirty_count = len(dirty_out.splitlines()) if dirty_out else 0
|
|
return branch, dirty_count
|
|
except Exception:
|
|
return None, 0
|
|
|
|
|
|
def short_path(cwd):
|
|
"""Abbreviate path: replace home dir with ~."""
|
|
if not cwd:
|
|
return ""
|
|
cwd = cwd.replace("\\", "/")
|
|
home = str(Path.home()).replace("\\", "/")
|
|
if cwd.lower().startswith(home.lower()):
|
|
return "~" + cwd[len(home):]
|
|
return cwd
|
|
|
|
|
|
def fmt_duration(secs):
|
|
"""Format seconds into human-readable duration."""
|
|
secs = max(0, int(secs))
|
|
if secs < 60:
|
|
return f"{secs}s"
|
|
if secs < 3600:
|
|
return f"{secs // 60}m"
|
|
h = secs // 3600
|
|
m = (secs % 3600) // 60
|
|
return f"{h}h{m}m"
|
|
|
|
|
|
def get_session_tracking(session_id, used_tokens):
|
|
"""Track session start time and approximate turn count."""
|
|
if not session_id:
|
|
return 0, 0
|
|
cache_dir = Path.home() / ".claude" / "cache"
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
sfile = cache_dir / f"sl-session-{session_id[:16]}.json"
|
|
now = time.time()
|
|
sd = {"start": now, "turns": 0, "last_tokens": 0}
|
|
if sfile.exists():
|
|
try:
|
|
sd = json.loads(sfile.read_text())
|
|
except Exception:
|
|
sd = {"start": now, "turns": 0, "last_tokens": 0}
|
|
prev = sd.get("last_tokens", 0)
|
|
if used_tokens - prev > 1000:
|
|
sd["turns"] = sd.get("turns", 0) + 1
|
|
sd["last_tokens"] = used_tokens
|
|
if "start" not in sd:
|
|
sd["start"] = now
|
|
try:
|
|
sfile.write_text(json.dumps(sd))
|
|
except Exception:
|
|
pass
|
|
return now - sd.get("start", now), sd.get("turns", 0)
|
|
|
|
|
|
def estimate_cost(usg):
|
|
"""Estimate session cost from token usage (Opus 4.6 pricing)."""
|
|
inp = int(usg.get("input_tokens") or 0)
|
|
cc = int(usg.get("cache_creation_input_tokens") or 0)
|
|
cr = int(usg.get("cache_read_input_tokens") or 0)
|
|
out = int(usg.get("output_tokens") or 0)
|
|
return (
|
|
inp * 15 / 1_000_000
|
|
+ cc * 18.75 / 1_000_000
|
|
+ cr * 1.50 / 1_000_000
|
|
+ out * 75 / 1_000_000
|
|
)
|
|
|
|
|
|
def fetch_usage():
|
|
cache_dir = Path.home() / ".claude" / "cache"
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
cache_file = cache_dir / "statusline-usage.json"
|
|
|
|
usage_data = None
|
|
needs_refresh = True
|
|
|
|
if cache_file.exists():
|
|
if time.time() - cache_file.stat().st_mtime < 60:
|
|
needs_refresh = False
|
|
try:
|
|
usage_data = json.loads(cache_file.read_text())
|
|
except Exception:
|
|
needs_refresh = True
|
|
|
|
if needs_refresh:
|
|
try:
|
|
creds_path = Path.home() / ".claude" / ".credentials.json"
|
|
if creds_path.exists():
|
|
creds = json.loads(creds_path.read_text())
|
|
token = creds.get("claudeAiOauth", {}).get("accessToken", "")
|
|
if token:
|
|
req = urllib.request.Request(
|
|
"https://api.anthropic.com/api/oauth/usage",
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {token}",
|
|
"anthropic-beta": "oauth-2025-04-20",
|
|
"User-Agent": "claude-code/2.1.34",
|
|
},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
usage_data = json.loads(resp.read())
|
|
cache_file.write_text(json.dumps(usage_data))
|
|
except Exception:
|
|
if cache_file.exists():
|
|
try:
|
|
usage_data = json.loads(cache_file.read_text())
|
|
except Exception:
|
|
pass
|
|
|
|
return usage_data
|
|
|
|
|
|
def main():
|
|
try:
|
|
raw = sys.stdin.read().strip()
|
|
if not raw:
|
|
sys.stdout.write("Claude")
|
|
return
|
|
|
|
data = json.loads(raw)
|
|
|
|
# ── Parse input ──
|
|
model = (data.get("model") or {}).get("display_name") or "Claude"
|
|
|
|
cw = data.get("context_window") or {}
|
|
size = int(cw.get("context_window_size") or 0) or 200_000
|
|
usg = cw.get("current_usage") or {}
|
|
used = (
|
|
int(usg.get("input_tokens") or 0)
|
|
+ int(usg.get("cache_creation_input_tokens") or 0)
|
|
+ int(usg.get("cache_read_input_tokens") or 0)
|
|
)
|
|
remain = size - used
|
|
pct_used = round((used / size) * 100) if size else 0
|
|
|
|
cwd = ((data.get("workspace") or {}).get("current_dir") or "").replace("\\", "/")
|
|
session_id = data.get("session_id") or ""
|
|
|
|
# Effort level → model name color
|
|
effort = ""
|
|
settings_path = Path.home() / ".claude" / "settings.json"
|
|
if settings_path.exists():
|
|
try:
|
|
effort = json.loads(settings_path.read_text()).get("effortLevel", "")
|
|
except Exception:
|
|
pass
|
|
model_clr = {"low": RED, "medium": YELLOW}.get(effort, GREEN)
|
|
|
|
# ── LINE 1: Context Window + Rate Limits ──
|
|
hero = bar(pct_used)
|
|
left = (
|
|
f"{model_clr}{model}{RST} {hero} "
|
|
f"{WHITE}{pct_used}%{RST} {GRAY}{fmt_tokens(remain)} left{RST}"
|
|
)
|
|
|
|
# Right zone: compact rate limits
|
|
ud = fetch_usage()
|
|
rl_parts = []
|
|
|
|
if ud:
|
|
fh = ud.get("five_hour") or {}
|
|
fh_pct = round(float(fh.get("utilization") or 0))
|
|
fh_rst = fmt_reset_compact(fh.get("resets_at"), "time")
|
|
fh_clr = RED if fh_pct >= 75 else YELLOW if fh_pct >= 50 else GREEN
|
|
rl_parts.append(
|
|
f"{GRAY}5hr{RST} {fh_clr}{fh_pct}%{RST} {DIM}\u21BB {fh_rst}{RST}"
|
|
)
|
|
|
|
sd = ud.get("seven_day") or {}
|
|
sd_pct = round(float(sd.get("utilization") or 0))
|
|
sd_rst = fmt_reset_compact(sd.get("resets_at"), "date")
|
|
sd_clr = RED if sd_pct >= 75 else YELLOW if sd_pct >= 50 else GREEN
|
|
rl_parts.append(
|
|
f"{GRAY}7d{RST} {sd_clr}{sd_pct}%{RST} {DIM}\u21BB {sd_rst}{RST}"
|
|
)
|
|
|
|
extra = ud.get("extra_usage") or {}
|
|
if extra.get("is_enabled"):
|
|
ex_used = round(float(extra.get("used_credits") or 0) / 100, 2)
|
|
ex_rst = fmt_reset_compact(extra.get("resets_at"), "date")
|
|
if not ex_rst:
|
|
now = datetime.now()
|
|
nxt = now.replace(
|
|
year=now.year + (1 if now.month == 12 else 0),
|
|
month=(now.month % 12) + 1, day=1,
|
|
)
|
|
ex_rst = f"{nxt.strftime('%b').lower()}{nxt.day}"
|
|
rl_parts.append(
|
|
f"{GRAY}extra{RST} {YELLOW}${ex_used:.2f}{RST} {DIM}\u21BB {ex_rst}{RST}"
|
|
)
|
|
|
|
rate_limits = f" {DIM}\u00B7{RST} ".join(rl_parts)
|
|
|
|
# ── LINE 2: Path + Git + Session ──
|
|
l2_parts = []
|
|
|
|
sp = short_path(cwd)
|
|
if sp:
|
|
l2_parts.append(f"{GRAY}{sp}{RST}")
|
|
|
|
branch, dirty = get_git_info(cwd)
|
|
if branch:
|
|
gs = f"{CYAN}{branch}{RST}"
|
|
if dirty > 0:
|
|
gs += f" {RED}+{dirty}{RST}"
|
|
else:
|
|
gs += f" {GREEN}\u2713{RST}"
|
|
l2_parts.append(gs)
|
|
|
|
sess = []
|
|
cost = estimate_cost(usg)
|
|
if cost >= 0.01:
|
|
sess.append(f"{YELLOW}~${cost:.2f}{RST}")
|
|
elapsed, turns = get_session_tracking(session_id, used)
|
|
if turns > 0:
|
|
sess.append(f"{WHITE}{turns}t{RST}")
|
|
if elapsed > 10:
|
|
sess.append(f"{GRAY}{fmt_duration(elapsed)}{RST}")
|
|
if sess:
|
|
l2_parts.append(f" {DIM}\u00B7{RST} ".join(sess))
|
|
|
|
line2 = " ".join(l2_parts)
|
|
|
|
# ── Output ──
|
|
sys.stdout.write(left)
|
|
if rate_limits:
|
|
sys.stdout.write(f"\n{rate_limits}")
|
|
if line2:
|
|
sys.stdout.write(f"\n{line2}")
|
|
|
|
except Exception as e:
|
|
sys.stdout.write(f"Claude | err: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|