coja
a700e23e0a
Mirrors the private dots tree at 900bdda: one shared base plus per-host overlays, replacing the old flat .config/ layout (last synced 2026-06-28). - packages: common/ gui/ wm/ lw/ fl/ + install.sh and bin/ tooling (dotsync, reconcile-hyde.sh) - new README (layout, deploy order, HyDE dependency), plus ToDo.md and HYDE-UPDATE.md - current HyDE waybar rig (layouts/, cava), pi agent extensions, claude/ config, tmux, presenterm, aichat roles - drops stale duplicates and generated cruft that should never have been tracked: the second top-level .pi/ copy, btop.log, zellij config.kdl.bak, fish_variables, nvim codecompanion.lua - .pi/agent/auth.json is gitignored now; auth.json.example ships instead - fl/ and wm/ hypr themes/ stay untracked (HyDE-generated per machine, per the root .gitignore)
518 lines
16 KiB
Python
Executable File
518 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Claude Code status line.
|
|
|
|
One line: vim mode | dir (repo-relative, hidden at the root) |
|
|
git(repo/branch +dirty +ahead/behind) | model (+effort) |
|
|
context% (+compact warn) | 5h usage (+reset eta) | cost (+burn rate) | disk. If the
|
|
rendered line would overflow the terminal, trailing (lowest-priority) segments are
|
|
dropped until it fits.
|
|
(Dormant, re-addable in main(): ram, cpu%, temp -- retired 2026-07-26 because the
|
|
status line only refreshes on message events, so system metrics sat visibly stale;
|
|
tmux's status bar owns those now -- plus velocity, cache hit %, api ratio, version,
|
|
style.)
|
|
|
|
Reads the status JSON from stdin (Claude Code statusLine command). Every segment
|
|
is wrapped defensively so the status line can never crash the UI -- on any error
|
|
a segment is simply omitted.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import unicodedata
|
|
|
|
# --- ANSI helpers -----------------------------------------------------------
|
|
RESET = "\033[0m"
|
|
DIM = "" # disabled: "\033[2m" blended the gray detail text into dark backgrounds
|
|
BOLD = "\033[1m"
|
|
C = {
|
|
"cyan": "\033[36m",
|
|
"green": "\033[32m",
|
|
"yellow": "\033[33m",
|
|
"red": "\033[31m",
|
|
"magenta": "\033[35m",
|
|
"blue": "\033[34m",
|
|
"gray": "\033[37m", # light gray (was "\033[90m" bright-black, too dark to read)
|
|
}
|
|
|
|
|
|
def color(text, name="", dim=False, bold=False):
|
|
pre = ""
|
|
if dim:
|
|
pre += DIM
|
|
if bold:
|
|
pre += BOLD
|
|
pre += C.get(name, "")
|
|
return f"{pre}{text}{RESET}" if pre else str(text)
|
|
|
|
|
|
def bucket(value, lo, hi, invert=False):
|
|
"""green/yellow/red by threshold. invert=True -> high is good."""
|
|
if invert:
|
|
return "green" if value >= hi else ("yellow" if value >= lo else "red")
|
|
return "green" if value < lo else ("yellow" if value < hi else "red")
|
|
|
|
|
|
DIV = color(" │ ", "gray", dim=True)
|
|
|
|
|
|
def join_line(segs):
|
|
return DIV.join(s for s in segs if s)
|
|
|
|
|
|
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
|
|
|
|
|
|
def vlen(text):
|
|
"""Visible width of a rendered string: ANSI stripped, emoji counted as 2."""
|
|
text = _ANSI_RE.sub("", text)
|
|
w = 0
|
|
for ch in text:
|
|
if ch in ("\u200d", "\ufe0f", "\ufe0e") or unicodedata.combining(ch):
|
|
continue # ZWJ / variation selectors / combining marks: zero width
|
|
o = ord(ch)
|
|
if o >= 0x1F000 or 0x2600 <= o <= 0x27BF or unicodedata.east_asian_width(ch) in ("W", "F"):
|
|
w += 2
|
|
else:
|
|
w += 1
|
|
return w
|
|
|
|
|
|
def term_cols():
|
|
"""Terminal width. Claude Code exports COLUMNS; 0 if unknown -> stay one line."""
|
|
try:
|
|
return shutil.get_terminal_size(fallback=(0, 0)).columns
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
# --- shared transcript usage ------------------------------------------------
|
|
def last_usage(data):
|
|
"""Most recent `message.usage` block from the transcript, or {}."""
|
|
try:
|
|
path = data.get("transcript_path")
|
|
if not path or not os.path.exists(path):
|
|
return {}
|
|
usage = {}
|
|
with open(path, "r", errors="ignore") as fh:
|
|
for line in fh:
|
|
line = line.strip()
|
|
if not line or '"usage"' not in line:
|
|
continue
|
|
try:
|
|
obj = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
msg = obj.get("message")
|
|
if isinstance(msg, dict) and isinstance(msg.get("usage"), dict):
|
|
usage = msg["usage"]
|
|
return usage
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
# --- row 1: work ------------------------------------------------------------
|
|
def vim_segment(data):
|
|
"""Vim editor mode as a compact [I]/[N]/[V] (only present with editorMode: vim).
|
|
|
|
Pair with `"hideVimModeIndicator": true` in the statusLine settings block so
|
|
Claude's own built-in indicator doesn't duplicate this one.
|
|
"""
|
|
mode = (data.get("vim") or {}).get("mode")
|
|
if not mode:
|
|
return ""
|
|
m = mode.upper()
|
|
if m.startswith("INSERT"):
|
|
return color("[I]", "green", bold=True)
|
|
if m.startswith("VISUAL"):
|
|
return color("[V]", "magenta", bold=True)
|
|
if m.startswith("NORMAL"):
|
|
return color("[N]", "blue", bold=True)
|
|
return color(f"[{m[:1]}]", "gray")
|
|
|
|
|
|
def abbrev(path):
|
|
"""fish-style prompt_pwd: shrink every component but the last to one char,
|
|
keeping a leading dot (`.config` -> `.c`)."""
|
|
parts = path.split(os.sep)
|
|
out = []
|
|
for i, p in enumerate(parts):
|
|
if i == len(parts) - 1 or p in ("", "~"):
|
|
out.append(p)
|
|
elif p.startswith("."):
|
|
out.append(p[:2])
|
|
else:
|
|
out.append(p[:1])
|
|
return os.sep.join(out)
|
|
|
|
|
|
def dir_segment(cwd, top=None):
|
|
"""Path *inside* the repo -- git_segment already names the repo itself, so
|
|
repeating the full ~-path here was pure duplication. Empty at the repo root;
|
|
a fish-shortened ~-path when we're not in a repo at all."""
|
|
try:
|
|
rel = os.path.relpath(cwd, top) if top else None
|
|
if rel and not rel.startswith(os.pardir): # `..` -> cwd is outside `top` (symlinked in); use the ~-path
|
|
if rel == os.curdir:
|
|
return ""
|
|
shown = rel if len(rel) <= 28 else abbrev(rel)
|
|
else:
|
|
home = os.path.expanduser("~")
|
|
shown = cwd
|
|
if cwd == home:
|
|
shown = "~"
|
|
elif cwd.startswith(home + os.sep):
|
|
shown = "~" + cwd[len(home):]
|
|
shown = abbrev(shown)
|
|
return "📁 " + color(shown, "cyan")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _git(cwd, *args):
|
|
return subprocess.run(
|
|
["git", "-C", cwd, *args], capture_output=True, text=True, timeout=1
|
|
)
|
|
|
|
|
|
def git_top(cwd):
|
|
"""Absolute path of the enclosing work tree, or None."""
|
|
try:
|
|
r = _git(cwd, "rev-parse", "--show-toplevel")
|
|
return r.stdout.strip() or None if r.returncode == 0 else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def git_segment(cwd, top):
|
|
try:
|
|
if not top:
|
|
return ""
|
|
repo = os.path.basename(top) or "repo"
|
|
branch = _git(cwd, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip() or "?"
|
|
dirty = bool(_git(cwd, "status", "--porcelain").stdout.strip())
|
|
|
|
# ahead/behind vs upstream (omitted when no upstream is configured)
|
|
ab = ""
|
|
rl = _git(cwd, "rev-list", "--left-right", "--count", "@{upstream}...HEAD")
|
|
if rl.returncode == 0 and rl.stdout.strip():
|
|
try:
|
|
behind, ahead = (int(x) for x in rl.stdout.split())
|
|
bits = []
|
|
if ahead:
|
|
bits.append(color(f"↑{ahead}", "cyan"))
|
|
if behind:
|
|
bits.append(color(f"↓{behind}", "yellow"))
|
|
if bits:
|
|
ab = " " + "".join(bits)
|
|
except Exception:
|
|
ab = ""
|
|
|
|
mark = color("*", "red") if dirty else ""
|
|
label = color(repo, "magenta", bold=True)
|
|
br = color(branch, "yellow" if dirty else "green")
|
|
return f"🌿 {label} {br}{mark}{ab}"
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def model_segment(data):
|
|
name = (data.get("model") or {}).get("display_name")
|
|
if not name:
|
|
return ""
|
|
name = name.replace(" context)", ")") # "Opus 4.8 (1M context)" -> "Opus 4.8 (1M)"
|
|
out = "🤖 " + color(name, "blue")
|
|
level = (data.get("effort") or {}).get("level")
|
|
if level:
|
|
out += color(f" · {level}", "gray")
|
|
return out
|
|
|
|
|
|
def context_segment(data, usage):
|
|
try:
|
|
if not usage:
|
|
return ""
|
|
used = (
|
|
usage.get("input_tokens", 0)
|
|
+ usage.get("cache_creation_input_tokens", 0)
|
|
+ usage.get("cache_read_input_tokens", 0)
|
|
)
|
|
model_id = (data.get("model") or {}).get("id", "")
|
|
window = 1_000_000 if "1m" in model_id.lower() else 200_000
|
|
pct = used / window * 100 if window else 0
|
|
col = bucket(pct, 50, 80)
|
|
out = "📝 " + color(f"{pct:.0f}%", col) + color(f" ({used / 1000:.0f}k)", "gray", dim=True)
|
|
if pct >= 80:
|
|
out += color(" ⚠compact", "red", bold=True)
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def usage_segment(data):
|
|
"""5-hour rolling usage window: used % (+ time until it resets)."""
|
|
try:
|
|
five = (data.get("rate_limits") or {}).get("five_hour") or {}
|
|
pct = five.get("used_percentage")
|
|
if pct is None:
|
|
return ""
|
|
out = "📊 " + color(f"{pct:.0f}%", bucket(pct, 50, 80))
|
|
resets = five.get("resets_at")
|
|
if resets:
|
|
secs = int(resets) - int(time.time())
|
|
if secs > 0:
|
|
h, m = divmod(secs // 60, 60)
|
|
out += color(f" {h}h{m:02d}m" if h else f" {m}m", "gray")
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def cost_segment(data):
|
|
try:
|
|
cost = data.get("cost") or {}
|
|
usd = cost.get("total_cost_usd")
|
|
if usd is None:
|
|
return ""
|
|
out = "💰 " + color(f"${usd:.2f}", "yellow")
|
|
dur_ms = cost.get("total_duration_ms") or 0
|
|
if dur_ms > 30_000: # need a meaningful window before extrapolating
|
|
per_hr = usd / (dur_ms / 3_600_000)
|
|
out += color(f" ${per_hr:.2f}/h", "gray", dim=True)
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def velocity_segment(data):
|
|
try:
|
|
cost = data.get("cost") or {}
|
|
added = cost.get("total_lines_added", 0)
|
|
removed = cost.get("total_lines_removed", 0)
|
|
if not (added or removed):
|
|
return ""
|
|
out = "✏️ " + color(f"+{added}", "green") + color("/", "gray", dim=True) + color(f"-{removed}", "red")
|
|
dur_ms = cost.get("total_duration_ms") or 0
|
|
if dur_ms > 30_000:
|
|
per_min = added / (dur_ms / 60_000)
|
|
out += color(f" {per_min:.0f}/m", "gray", dim=True)
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def cache_segment(usage):
|
|
try:
|
|
if not usage:
|
|
return ""
|
|
read = usage.get("cache_read_input_tokens", 0)
|
|
total = (
|
|
usage.get("input_tokens", 0)
|
|
+ usage.get("cache_creation_input_tokens", 0)
|
|
+ read
|
|
)
|
|
if total <= 0:
|
|
return ""
|
|
pct = read / total * 100
|
|
return "♻️ " + color(f"{pct:.0f}%", bucket(pct, 50, 80, invert=True))
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def api_segment(data):
|
|
"""Share of wall-clock time spent in API/inference (rough 'busy' ratio)."""
|
|
try:
|
|
cost = data.get("cost") or {}
|
|
wall = cost.get("total_duration_ms") or 0
|
|
api = cost.get("total_api_duration_ms") or 0
|
|
if wall <= 0 or api <= 0:
|
|
return ""
|
|
pct = min(api / wall * 100, 100)
|
|
return "⚙️ " + color(f"{pct:.0f}%", "blue")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
# --- row 2: system ----------------------------------------------------------
|
|
def ram_segment():
|
|
try:
|
|
info = {}
|
|
with open("/proc/meminfo") as fh:
|
|
for line in fh:
|
|
k, _, v = line.partition(":")
|
|
info[k.strip()] = int(v.split()[0]) # kB
|
|
total = info.get("MemTotal", 0)
|
|
avail = info.get("MemAvailable", info.get("MemFree", 0))
|
|
if total <= 0:
|
|
return ""
|
|
used = total - avail
|
|
pct = used / total * 100
|
|
g = 1024 * 1024
|
|
return (
|
|
"🧠 " + color(f"{pct:.0f}%", bucket(pct, 70, 85))
|
|
+ color(f" {used / g:.1f}/{total / g:.0f}G", "gray", dim=True)
|
|
)
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def disk_segment(cwd):
|
|
try:
|
|
st = os.statvfs(cwd)
|
|
total = st.f_blocks
|
|
if total <= 0:
|
|
return ""
|
|
used = total - st.f_bfree
|
|
pct = used / total * 100
|
|
return "💾 " + color(f"{pct:.0f}%", bucket(pct, 75, 90))
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _cpu_pct():
|
|
"""Live CPU% diffed against a cached /proc/stat snapshot (no sleeping)."""
|
|
try:
|
|
with open("/proc/stat") as fh:
|
|
parts = fh.readline().split()
|
|
if not parts or parts[0] != "cpu":
|
|
return None
|
|
vals = [int(x) for x in parts[1:]]
|
|
idle = vals[3] + (vals[4] if len(vals) > 4 else 0) # idle + iowait
|
|
total = sum(vals)
|
|
|
|
state = os.path.expanduser("~/.claude/cache/statusline.cpu")
|
|
prev = None
|
|
try:
|
|
with open(state) as fh:
|
|
pt, pi = fh.read().split()
|
|
prev = (int(pt), int(pi))
|
|
except Exception:
|
|
prev = None
|
|
try:
|
|
os.makedirs(os.path.dirname(state), exist_ok=True)
|
|
with open(state, "w") as fh:
|
|
fh.write(f"{total} {idle}")
|
|
except Exception:
|
|
pass
|
|
|
|
if not prev:
|
|
return None
|
|
dt = total - prev[0]
|
|
di = idle - prev[1]
|
|
if dt <= 0 or di < 0:
|
|
return None
|
|
return max(0.0, min(100.0, (1 - di / dt) * 100))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def cpu_segment():
|
|
try:
|
|
load1 = os.getloadavg()[0]
|
|
cores = os.cpu_count() or 1
|
|
pct = _cpu_pct()
|
|
if pct is not None:
|
|
head = color(f"{pct:.0f}%", bucket(pct, 60, 85))
|
|
tail = color(f" {load1:.1f} {cores}c", "gray", dim=True)
|
|
else:
|
|
head = color(f"{load1:.2f}", bucket(load1 / cores, 0.7, 1.0))
|
|
tail = color(f" {cores}c", "gray", dim=True)
|
|
return "🖥️ " + head + tail
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def temp_segment():
|
|
"""Hottest CPU-ish thermal zone in °C, falling back to the max zone."""
|
|
try:
|
|
base = "/sys/class/thermal"
|
|
if not os.path.isdir(base):
|
|
return ""
|
|
prefer = ("x86_pkg_temp", "coretemp", "cpu", "k10temp", "tctl")
|
|
chosen = None
|
|
fallback = None
|
|
for name in os.listdir(base):
|
|
if not name.startswith("thermal_zone"):
|
|
continue
|
|
d = os.path.join(base, name)
|
|
try:
|
|
with open(os.path.join(d, "temp")) as fh:
|
|
milli = int(fh.read().strip())
|
|
except Exception:
|
|
continue
|
|
ztype = ""
|
|
try:
|
|
with open(os.path.join(d, "type")) as fh:
|
|
ztype = fh.read().strip().lower()
|
|
except Exception:
|
|
pass
|
|
if fallback is None or milli > fallback:
|
|
fallback = milli
|
|
if any(p in ztype for p in prefer) and (chosen is None or milli > chosen):
|
|
chosen = milli
|
|
milli = chosen if chosen is not None else fallback
|
|
if milli is None:
|
|
return ""
|
|
c = milli / 1000.0
|
|
return "🌡️ " + color(f"{c:.0f}°C", bucket(c, 60, 80))
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def version_segment(data):
|
|
v = data.get("version")
|
|
return color(f"v{v}", "gray", dim=True) if v else ""
|
|
|
|
|
|
def style_segment(data):
|
|
name = (data.get("output_style") or {}).get("name")
|
|
return color(f"🎨{name}", "gray", dim=True) if name else ""
|
|
|
|
|
|
def main():
|
|
try:
|
|
data = json.load(sys.stdin)
|
|
except Exception:
|
|
data = {}
|
|
|
|
cwd = (
|
|
(data.get("workspace") or {}).get("current_dir")
|
|
or data.get("cwd")
|
|
or os.getcwd()
|
|
)
|
|
usage = last_usage(data)
|
|
top = git_top(cwd)
|
|
|
|
# One row: work (left) then system (right), joined. If it would overflow the
|
|
# terminal, trailing (lowest-priority) segments are dropped so it stays one line.
|
|
# Dormant helpers kept above for easy re-add: ram_segment, cpu_segment,
|
|
# temp_segment (retired -- they only refresh on message events, so they sat
|
|
# stale; the tmux status bar owns system metrics now), velocity_segment,
|
|
# cache_segment, api_segment, version_segment, style_segment.
|
|
work = [s for s in (
|
|
vim_segment(data),
|
|
dir_segment(cwd, top),
|
|
git_segment(cwd, top),
|
|
model_segment(data),
|
|
context_segment(data, usage),
|
|
usage_segment(data),
|
|
cost_segment(data),
|
|
) if s]
|
|
system = [s for s in (
|
|
disk_segment(cwd),
|
|
) if s]
|
|
|
|
segs = work + system
|
|
cols = term_cols()
|
|
if cols: # trim from the right until it fits one line (cols=0 -> width unknown, keep all)
|
|
while len(segs) > 1 and vlen(join_line(segs)) > cols:
|
|
segs.pop()
|
|
sys.stdout.write(join_line(segs))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|