live_dashboard.py

#!/usr/bin/env python3
"""Live terminal dashboard widget for Claude Code View — rotates through metrics screens."""

import json
import os
import re
import shutil
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path

# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------

STATS_FILE = Path.home() / ".claude" / "stats-cache.json"
CONFIG_FILE = Path(__file__).resolve().parent / "config.json"
ROTATE_SECONDS = 10
DATA_REFRESH_SECONDS = 300  # re-read data every 5 min

# ANSI helpers
BOLD = "\033[1m"
DIM = "\033[2m"
RESET = "\033[0m"
INVERT = "\033[7m"
CLEAR = "\033[2J\033[H"

# Box drawing
TL, TR, BL, BR = "╔", "╗", "╚", "╝"
H, V = "═", "║"
ML, MR = "╠", "╣"

# Block characters for charts
BLOCKS = " ▏▎▍▌▋▊█"
SPARK = "▁▂▃▄▅▆▇█"


# ---------------------------------------------------------------------------
# Data loading
# ---------------------------------------------------------------------------

def load_stats():
    """Read Claude Code stats-cache.json."""
    if not STATS_FILE.is_file():
        return None
    try:
        return json.loads(STATS_FILE.read_text(errors="replace"))
    except (json.JSONDecodeError, OSError):
        return None


AI_MARKERS = re.compile(
    r"co-authored-by.*(?:claude|copilot|anthropic|openai|cursor|aider)"
    r"|generated.*(?:by|with|using).*(?:claude|copilot|gpt|ai)"
    r"|cursor.*composer",
    re.IGNORECASE,
)


def scan_project_git(folder):
    """Get AI code stats for a project."""
    if not (folder / ".git").is_dir():
        return None
    try:
        result = subprocess.run(
            ["git", "-C", str(folder), "log", "--all",
             "--format=COMMIT:%H%n%b%nENDBODY", "--shortstat"],
            capture_output=True, text=True, timeout=10,
        )
        if result.returncode != 0:
            return None
    except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
        return None

    total_commits = 0
    ai_commits = 0
    total_ins = 0
    ai_ins = 0
    current_is_ai = False

    for line in result.stdout.split("\n"):
        line = line.strip()
        if line.startswith("COMMIT:"):
            total_commits += 1
            current_is_ai = False
        elif AI_MARKERS.search(line):
            if not current_is_ai:
                current_is_ai = True
                ai_commits += 1
        elif "insertion" in line or "deletion" in line:
            m = re.search(r"(\d+) insertion", line)
            ins = int(m.group(1)) if m else 0
            total_ins += ins
            if current_is_ai:
                ai_ins += ins

    if not total_commits:
        return None

    # Get last commit date
    try:
        r2 = subprocess.run(
            ["git", "-C", str(folder), "log", "-1", "--format=%cr"],
            capture_output=True, text=True, timeout=5,
        )
        last_commit = r2.stdout.strip() if r2.returncode == 0 else ""
    except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
        last_commit = ""

    return {
        "name": folder.name,
        "total_commits": total_commits,
        "ai_commits": ai_commits,
        "total_ins": total_ins,
        "ai_ins": ai_ins,
        "ai_pct": round(ai_ins / total_ins * 100) if total_ins else 0,
        "last_commit": last_commit,
    }


def load_config():
    """Load folder paths from config.json."""
    if not CONFIG_FILE.is_file():
        return []
    try:
        data = json.loads(CONFIG_FILE.read_text(errors="replace"))
        return [Path(f["path"]).expanduser() for f in data.get("folders", [])]
    except (json.JSONDecodeError, OSError, KeyError):
        return []


def scan_all_projects():
    """Scan git stats for all configured project folders."""
    projects = []
    dirs = load_config()
    if not dirs:
        # Fallback: scan current directory parent
        dirs = [Path.cwd()]
    for d in dirs:
        if not d.is_dir():
            continue
        try:
            for child in sorted(d.iterdir()):
                if child.is_dir() and not child.name.startswith("."):
                    info = scan_project_git(child)
                    if info:
                        projects.append(info)
        except PermissionError:
            pass
    return projects


def count_dirs(d):
    """Count immediate subdirectories."""
    if not d.is_dir():
        return 0
    try:
        return sum(1 for c in d.iterdir() if c.is_dir() and not c.name.startswith("."))
    except PermissionError:
        return 0


def count_ollama():
    """Count ollama models."""
    try:
        r = subprocess.run(["ollama", "list"], capture_output=True, text=True, timeout=10)
        if r.returncode != 0:
            return 0
        lines = r.stdout.strip().split("\n")
        return max(0, len(lines) - 1)
    except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
        return 0


def load_all_data():
    """Load all data sources."""
    stats = load_stats()
    projects = scan_all_projects()
    dirs = load_config()
    folder_count = sum(count_dirs(d) for d in dirs if d.is_dir())
    model_count = count_ollama()
    return {
        "stats": stats,
        "projects": projects,
        "folder_count": folder_count,
        "model_count": model_count,
    }


# ---------------------------------------------------------------------------
# Rendering helpers
# ---------------------------------------------------------------------------

def get_terminal_width():
    """Get full terminal width."""
    return shutil.get_terminal_size((40, 10)).columns


def get_width():
    """Get box width, capped for readability."""
    w = get_terminal_width()
    return max(28, min(w, 50))


def box_top(w, title=""):
    """Draw box top with optional title."""
    if title:
        title = f" {title} "
        fill = w - 4 - len(title)
        return f"{TL}{H}{H}{BOLD}{title}{RESET}{H * max(0, fill)}{TR}"
    return f"{TL}{H * (w - 2)}{TR}"


def box_mid(w, title=""):
    """Draw box middle separator."""
    if title:
        title = f" {title} "
        fill = w - 4 - len(title)
        return f"{ML}{H}{H}{title}{H * max(0, fill)}{MR}"
    return f"{ML}{H * (w - 2)}{MR}"


def box_bot(w):
    """Draw box bottom."""
    return f"{BL}{H * (w - 2)}{BR}"


def box_row(w, text):
    """Draw a box row with text, padded to width."""
    inner = w - 4  # 2 for borders + 2 for padding
    # Strip ANSI for length calculation
    visible = re.sub(r"\033\[[0-9;]*m", "", text)
    pad = max(0, inner - len(visible))
    return f"{V} {text}{' ' * pad} {V}"


def bar_chart(value, max_val, width):
    """Render a horizontal bar using block characters."""
    if max_val <= 0:
        return " " * width
    ratio = min(value / max_val, 1.0)
    full_blocks = int(ratio * width)
    remainder = (ratio * width) - full_blocks
    partial = BLOCKS[int(remainder * 7)] if full_blocks < width else ""
    empty = width - full_blocks - (1 if partial.strip() else 0)
    return "█" * full_blocks + partial + " " * empty


def sparkline(values, width):
    """Render a sparkline from a list of values."""
    if not values:
        return " " * width
    # Resample to fit width
    if len(values) > width:
        step = len(values) / width
        sampled = [values[int(i * step)] for i in range(width)]
    else:
        sampled = values + [0] * (width - len(values))
    mx = max(sampled) if sampled else 1
    if mx == 0:
        return SPARK[0] * width
    return "".join(SPARK[min(7, int(v / mx * 7))] for v in sampled[:width])


def fmt_tokens(n):
    """Format token count."""
    if n >= 1e9:
        return f"{n / 1e9:.1f}B"
    if n >= 1e6:
        return f"{n / 1e6:.1f}M"
    if n >= 1e3:
        return f"{n / 1e3:.1f}K"
    return str(n)


def fmt_cost(n):
    """Format USD cost."""
    if n >= 1000:
        return f"${n:,.0f}"
    return f"${n:.2f}"


# ---------------------------------------------------------------------------
# Screens
# ---------------------------------------------------------------------------

def screen_overview(data, w):
    """Screen 1: Overview pulse."""
    stats = data["stats"] or {}
    total_tokens = sum(
        u.get("inputTokens", 0) + u.get("outputTokens", 0)
        + u.get("cacheReadInputTokens", 0) + u.get("cacheCreationInputTokens", 0)
        for u in (stats.get("modelUsage", {}) or {}).values()
    )
    sessions = stats.get("totalSessions", 0)
    messages = stats.get("totalMessages", 0)
    now = datetime.now().strftime("%H:%M:%S")

    lines = [
        box_top(w, "CC VIEW"),
        box_row(w, ""),
        box_row(w, f"{BOLD}{data['folder_count']}{RESET} projects  {BOLD}{data['model_count']}{RESET} models"),
        box_row(w, f"{BOLD}{fmt_tokens(total_tokens)}{RESET} tokens   {BOLD}{sessions}{RESET} sessions"),
        box_row(w, f"{BOLD}{messages:,}{RESET} messages"),
        box_row(w, ""),
        box_row(w, f"{DIM}● LIVE    {now}{RESET}"),
        box_bot(w),
    ]
    return "\n".join(lines)


def screen_token_burn(data, w):
    """Screen 2: Token burn rate sparkline."""
    stats = data["stats"] or {}
    daily = stats.get("dailyActivity", [])
    # Get daily message counts for sparkline
    msg_counts = [d.get("messageCount", 0) for d in daily]
    total_tokens = sum(
        u.get("inputTokens", 0) + u.get("outputTokens", 0)
        + u.get("cacheReadInputTokens", 0) + u.get("cacheCreationInputTokens", 0)
        for u in (stats.get("modelUsage", {}) or {}).values()
    )
    days = len(daily) or 1
    avg_per_day = total_tokens / days

    # Estimate cost
    cost_per_token = 0.00000376  # rough weighted average
    daily_cost = avg_per_day * cost_per_token

    spark_w = w - 6
    spark = sparkline(msg_counts[-spark_w:], spark_w)

    lines = [
        box_top(w, "TOKEN BURN"),
        box_row(w, f"{DIM}Daily messages (last {len(msg_counts)}d){RESET}"),
        box_row(w, ""),
        box_row(w, f"{BOLD}{spark}{RESET}"),
        box_row(w, ""),
        box_row(w, f"Total: {BOLD}{fmt_tokens(total_tokens)}{RESET} tokens"),
        box_row(w, f"Avg:   {BOLD}{fmt_tokens(int(avg_per_day))}{RESET}/day"),
        box_row(w, f"Cost:  {BOLD}~{fmt_cost(daily_cost)}{RESET}/day"),
        box_bot(w),
    ]
    return "\n".join(lines)


def screen_ai_leaderboard(data, w):
    """Screen 3: AI code percentage leaderboard."""
    projects = sorted(data["projects"], key=lambda p: p["ai_pct"], reverse=True)
    top = projects[:6]
    bar_w = w - 22  # space for name + percentage

    lines = [box_top(w, "AI CODE %")]
    lines.append(box_row(w, ""))
    for p in top:
        name = p["name"][:12].ljust(12)
        bar = bar_chart(p["ai_pct"], 100, max(4, bar_w))
        lines.append(box_row(w, f"{name} {BOLD}{bar}{RESET} {p['ai_pct']:3d}"))
    # Pad if fewer than 6
    for _ in range(6 - len(top)):
        lines.append(box_row(w, ""))
    lines.append(box_row(w, ""))
    lines.append(box_bot(w))
    return "\n".join(lines)


def screen_model_mix(data, w):
    """Screen 4: Model usage breakdown."""
    stats = data["stats"] or {}
    model_usage = stats.get("modelUsage", {}) or {}
    models = []
    for name, u in model_usage.items():
        total = (u.get("inputTokens", 0) + u.get("outputTokens", 0)
                 + u.get("cacheReadInputTokens", 0) + u.get("cacheCreationInputTokens", 0))
        models.append({"name": name, "tokens": total})
    models.sort(key=lambda m: m["tokens"], reverse=True)
    grand_total = sum(m["tokens"] for m in models) or 1

    # Cache hit rate
    total_cache_read = sum(u.get("cacheReadInputTokens", 0) for u in model_usage.values())
    total_cache_create = sum(u.get("cacheCreationInputTokens", 0) for u in model_usage.values())
    cache_total = total_cache_read + total_cache_create
    cache_hit = (total_cache_read / cache_total * 100) if cache_total else 0

    bar_w = w - 22
    lines = [box_top(w, "MODEL USAGE")]
    lines.append(box_row(w, ""))
    for m in models[:4]:
        short = m["name"].split("/")[-1][:12].ljust(12)
        pct = round(m["tokens"] / grand_total * 100)
        bar = bar_chart(pct, 100, max(4, bar_w))
        lines.append(box_row(w, f"{short} {BOLD}{bar}{RESET}{pct:3d}%"))
    for _ in range(4 - len(models[:4])):
        lines.append(box_row(w, ""))
    lines.append(box_row(w, ""))
    lines.append(box_row(w, f"Cache hit: {BOLD}{cache_hit:.1f}%{RESET}"))
    lines.append(box_bot(w))
    return "\n".join(lines)


def screen_activity(data, w):
    """Screen 5: Activity heatmap by hour."""
    stats = data["stats"] or {}
    hour_counts = stats.get("hourCounts", {}) or {}
    # Convert to int keys
    hours = {int(k): v for k, v in hour_counts.items()}
    mx = max(hours.values()) if hours else 1
    shades = " ░▒▓█"

    # Show hours 7-23
    hour_range = range(7, 24, 2)
    header = "  " + "".join(f"{h:>3}" for h in hour_range)

    lines = [box_top(w, "CODING HOURS")]
    lines.append(box_row(w, header[:w - 4]))
    lines.append(box_row(w, ""))
    # Single row since we only have aggregate hours
    row = "  "
    for h in hour_range:
        val = hours.get(h, 0) + hours.get(h + 1, 0)
        idx = min(4, int(val / (mx * 2 + 1) * 4)) if mx else 0
        row += f" {shades[idx]} "
    lines.append(box_row(w, f"{BOLD}{row}{RESET}"[:w - 4]))
    lines.append(box_row(w, ""))

    # Find peak
    peak_h = max(hours, key=hours.get) if hours else 0
    total_sessions = stats.get("totalSessions", 0)
    lines.append(box_row(w, f"Peak hour: {BOLD}{peak_h}:00{RESET}"))
    lines.append(box_row(w, f"Sessions:  {BOLD}{total_sessions}{RESET}"))
    lines.append(box_row(w, ""))
    lines.append(box_bot(w))
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------------

def main():
    screens = [
        screen_overview,
        screen_token_burn,
        screen_ai_leaderboard,
        screen_model_mix,
        screen_activity,
    ]

    print(f"{DIM}Loading data...{RESET}", flush=True)
    data = load_all_data()
    last_refresh = time.time()
    screen_idx = 0

    try:
        while True:
            if time.time() - last_refresh > DATA_REFRESH_SECONDS:
                data = load_all_data()
                last_refresh = time.time()

            w = get_width()
            term_w = get_terminal_width()
            output = screens[screen_idx](data, w)

            # Center the box horizontally in the terminal
            pad = max(0, (term_w - w) // 2)
            if pad > 0:
                centered = "\n".join(" " * pad + line for line in output.split("\n"))
            else:
                centered = output

            print(CLEAR + centered, flush=True)

            screen_idx = (screen_idx + 1) % len(screens)
            time.sleep(ROTATE_SECONDS)

    except KeyboardInterrupt:
        print(CLEAR + f"{DIM}Dashboard stopped.{RESET}")
        sys.exit(0)


if __name__ == "__main__":
    main()