A single-command dashboard that queries your local OpenCode database and renders a box-drawn executive summary of token burn, cost, model mix, tool usage, and session trends — so you know exactly where your API budget is going without opening a spreadsheet.

opencode-stats.py

#!/usr/bin/env python3
"""OpenCode Dashboard — value-first token & usage intelligence."""

import json
import sqlite3
import sys
import unicodedata
from pathlib import Path
from datetime import datetime, timezone

DB_PATH = Path.home() / ".local" / "share" / "opencode" / "opencode.db"

# ── Formatting ────────────────────────────────────────────────────────

def display_len(s: str) -> int:
    """Return the visual display width of *s*, accounting for wide chars."""
    w = 0
    for ch in s:
        eaw = unicodedata.east_asian_width(ch)
        w += 2 if eaw in ("W", "F") else 1
    return w


def truncate(s: str, max_width: int) -> str:
    """Truncate *s* to *max_width* display columns, appending … if needed."""
    if display_len(s) <= max_width:
        return s
    out = []
    used = 0
    limit = max_width - 1  # reserve 1 for …
    for ch in s:
        cw = 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1
        if used + cw > limit:
            break
        out.append(ch)
        used += cw
    return "".join(out) + "…"


def fmt_tokens(n: int | float) -> str:
    n = n or 0
    if n >= 1_000_000_000:
        return f"{n / 1_000_000_000:.1f}B"
    if n >= 1_000_000:
        return f"{n / 1_000_000:.1f}M"
    if n >= 1_000:
        return f"{n / 1_000:.1f}K"
    return str(int(n))


def fmt_cost(v: float) -> str:
    v = v or 0
    if v == 0:
        return "$0.00"
    if v < 0.01:
        return f"${v:.4f}"
    return f"${v:,.2f}"


def bar(value: float, max_value: float, width: int = 24) -> str:
    filled = int(value / max_value * width) if max_value else 0
    return "█" * filled + "░" * (width - filled)


def spark(values: list[int | float]) -> str:
    blocks = " ░▒▓█"
    if not values:
        return ""
    lo, hi = min(values), max(values)
    rng = hi - lo or 1
    return "".join(blocks[min(int((v - lo) / rng * 4.99), 4)] for v in values)


def print_header(title: str):
    w = 66
    print()
    print(f"  ╔{'═' * (w - 4)}╗")
    print(f"  ║  {title:<{w - 6}}║")
    print(f"  ╠{'═' * (w - 4)}╣")


def print_section(title: str):
    w = 66
    print(f"  ║{'':^{w - 4}}║")
    print(f"  ║  ▸ {title:<{w - 8}}║")
    print(f"  ║{'':^{w - 4}}║")


def print_kv(key: str, val: str):
    w = 66
    inner = w - 8
    content_len = display_len(key) + display_len(val) + 2  # min 2 dots
    dots = inner - content_len
    if dots < 2:
        dots = 2
    line = f"{key}{'·' * dots}{val}"
    padding = inner - display_len(line)
    if padding < 0:
        padding = 0
    print(f"  ║  {line}{' ' * padding}  ║")


def print_bar_row(label: str, value: float, max_value: float, pct: float | None = None):
    w = 66
    b = bar(value, max_value, 16)
    val_str = fmt_tokens(value)
    suffix = f" {pct:5.1f}%" if pct is not None else ""
    inner = w - 8  # 2 spaces + ║ + 2 spaces ... 2 spaces + ║
    # fixed portion: val_str(7) + spaces(3) + bar(16) + suffix(~7) = ~33
    fixed = 7 + 3 + 16 + display_len(suffix)
    label_max = inner - fixed
    if label_max < 10:
        label_max = 10
    if display_len(label) > label_max:
        label = truncate(label, label_max)
    line = f"{label:<{label_max}} {val_str:>7}  {b}{suffix}"
    padding = inner - display_len(line)
    if padding < 0:
        padding = 0
    print(f"  ║  {line}{' ' * padding}  ║")


def print_footer():
    w = 66
    print(f"  ╚{'═' * (w - 4)}╝")


# ── Dashboard ─────────────────────────────────────────────────────────

def main():
    if not DB_PATH.exists():
        print(f"Database not found: {DB_PATH}", file=sys.stderr)
        sys.exit(1)

    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row

    # ── Core aggregations ─────────────────────────────────────────────
    ov = conn.execute("""
        SELECT
            COUNT(*) as sessions,
            SUM(tokens_input) as inp,
            SUM(tokens_output) as out,
            SUM(tokens_reasoning) as reason,
            SUM(tokens_cache_read) as cache_r,
            SUM(tokens_cache_write) as cache_w,
            SUM(cost) as cost,
            MIN(time_created) as first_ts,
            MAX(time_updated) as last_ts
        FROM session
    """).fetchone()

    msg_stats = conn.execute("""
        SELECT
            COUNT(*) as total_msgs,
            SUM(CASE WHEN json_extract(data, '$.role') = 'assistant' THEN 1 ELSE 0 END) as assistant_msgs,
            SUM(CASE WHEN json_extract(data, '$.role') = 'user' THEN 1 ELSE 0 END) as user_msgs
        FROM message
    """).fetchone()

    io_total = (ov["inp"] or 0) + (ov["out"] or 0)
    cache_total = (ov["cache_r"] or 0) + (ov["cache_w"] or 0)
    total_ctx = io_total + (ov["reason"] or 0) + cache_total

    first_dt = datetime.fromtimestamp(ov["first_ts"] / 1000, tz=timezone.utc) if ov["first_ts"] else None
    last_dt = datetime.fromtimestamp(ov["last_ts"] / 1000, tz=timezone.utc) if ov["last_ts"] else None

    days = (last_dt - first_dt).days + 1 if first_dt and last_dt else 1
    sessions_per_day = ov["sessions"] / max(days, 1)
    tokens_per_session = total_ctx / max(ov["sessions"], 1)
    output_ratio = (ov["out"] or 0) / max(ov["inp"] or 1, 1)
    cache_hit = (ov["cache_r"] or 0) / max(io_total, 1)
    reasoning_pct = (ov["reason"] or 0) / max(io_total, 1)

    # Median tokens per session (total context including cache)
    med_row = conn.execute("""
        SELECT tokens_input + tokens_output + tokens_cache_read + tokens_cache_write as total
        FROM session ORDER BY total
    """).fetchall()
    all_totals = [r["total"] for r in med_row if r["total"] is not None]
    n = len(all_totals)
    if n:
        median_tokens = all_totals[n // 2] if n % 2 else (all_totals[n // 2 - 1] + all_totals[n // 2]) / 2
    else:
        median_tokens = 0

    avg_cost_per_day = (ov["cost"] or 0) / max(days, 1)

    # ── Print Dashboard ───────────────────────────────────────────────
    print_header("OPENCODE CEO DASHBOARD")

    print_section("OVERVIEW")
    if first_dt and last_dt:
        print_kv("Period", f"{first_dt:%b %d} – {last_dt:%b %d, %Y}  ({days}d)")
    print_kv("Sessions", f"{ov['sessions']:,}")
    print_kv("Messages", f"{msg_stats['total_msgs']:,}  ({msg_stats['assistant_msgs']:,} assistant)")
    print_kv("Total tokens (I/O)", fmt_tokens(io_total))
    print_kv("Total context processed", fmt_tokens(total_ctx))
    print_kv("Reported cost", fmt_cost(ov["cost"] or 0))
    print_kv("Avg Cost/Day", fmt_cost(avg_cost_per_day))

    # Count sessions with null model
    null_model = conn.execute(
        "SELECT COUNT(*) FROM session WHERE json_extract(model, '$.id') IS NULL"
    ).fetchone()[0]
    if null_model:
        print_kv("Note", f"{null_model} sessions backfilled from messages")

    print_section("EFFICIENCY SCORECARD")
    print_kv("Avg Tokens/Session", f"{fmt_tokens(tokens_per_session)}")
    print_kv("Median Tokens/Session", f"{fmt_tokens(median_tokens)}")
    print_kv("Sessions / day", f"{sessions_per_day:.1f}")
    eff_label = "⚡ efficient" if output_ratio > 0.15 else "📦 context-heavy"
    print_kv("Output / Input ratio", f"{output_ratio:.1%}  {eff_label}")
    cache_label = "✅ excellent" if cache_hit > 10 else "⚠️  low" if cache_hit < 5 else "👍 good"
    print_kv("Cache hit ratio", f"{cache_hit:.0f}x  {cache_label}")
    reason_label = "🧠 heavy" if reasoning_pct > 0.20 else "✅ lean"
    print_kv("Reasoning overhead", f"{reasoning_pct:.1%}  {reason_label}")

    # ── Model Mix ─────────────────────────────────────────────────────
    # Sessions before v1.14.50 have NULL model — backfill from messages
    print_section("MODEL MIX  (tokens I/O, share of total)")
    models = conn.execute("""
        SELECT
            COALESCE(
                json_extract(s.model, '$.id'),
                (SELECT json_extract(m.data, '$.modelID')
                 FROM message m
                 WHERE m.session_id = s.id
                   AND json_extract(m.data, '$.role') = 'assistant'
                   AND json_extract(m.data, '$.modelID') IS NOT NULL
                 LIMIT 1)
            ) as mid,
            COALESCE(
                json_extract(s.model, '$.providerID'),
                (SELECT json_extract(m.data, '$.providerID')
                 FROM message m
                 WHERE m.session_id = s.id
                   AND json_extract(m.data, '$.role') = 'assistant'
                   AND json_extract(m.data, '$.providerID') IS NOT NULL
                 LIMIT 1)
            ) as pid,
            json_extract(s.model, '$.variant') as var,
            COUNT(*) as sess,
            SUM(s.tokens_input + s.tokens_output) as tokens,
            SUM(s.cost) as cost
        FROM session s
        GROUP BY mid, pid, var
        HAVING mid IS NOT NULL
        ORDER BY tokens DESC
    """).fetchall()

    if models:
        max_tok = models[0]["tokens"] or 1
        for m in models[:10]:
            name = m["mid"] or "?"
            name = truncate(name, 22)
            provider = m["pid"] or ""
            if m["var"]:
                provider += f":{m['var']}"
            provider = truncate(provider, 18)
            pct_share = (m["tokens"] or 0) / max(io_total, 1) * 100
            print_bar_row(f"{name}  {provider}", m["tokens"] or 0, max_tok, pct_share)

    # ── Provider Breakdown ────────────────────────────────────────────
    print_section("PROVIDER BREAKDOWN")
    providers = conn.execute("""
        SELECT
            COALESCE(
                json_extract(s.model, '$.providerID'),
                (SELECT json_extract(m.data, '$.providerID')
                 FROM message m
                 WHERE m.session_id = s.id
                   AND json_extract(m.data, '$.role') = 'assistant'
                   AND json_extract(m.data, '$.providerID') IS NOT NULL
                 LIMIT 1)
            ) as pid,
            COUNT(*) as sess,
            SUM(s.tokens_input + s.tokens_output) as tokens
        FROM session s
        GROUP BY pid
        HAVING pid IS NOT NULL
        ORDER BY tokens DESC
    """).fetchall()

    if providers:
        max_prov = providers[0]["tokens"] or 1
        for p in providers:
            pct_share = (p["tokens"] or 0) / max(io_total, 1) * 100
            print_bar_row(p["pid"] or "?", p["tokens"] or 0, max_prov, pct_share)

    # ── Agent Breakdown ───────────────────────────────────────────────
    print_section("AGENT BREAKDOWN  (from assistant messages)")
    agents = conn.execute("""
        SELECT
            json_extract(data, '$.agent') as agent,
            COUNT(*) as msgs,
            SUM(json_extract(data, '$.tokens.input')) as inp,
            SUM(json_extract(data, '$.tokens.output')) as out
        FROM message
        WHERE json_extract(data, '$.role') = 'assistant'
        GROUP BY agent
        ORDER BY (SUM(json_extract(data, '$.tokens.input')) + SUM(json_extract(data, '$.tokens.output'))) DESC
    """).fetchall()

    if agents:
        max_ag = (agents[0]["inp"] or 0) + (agents[0]["out"] or 0)
        for a in agents:
            total = (a["inp"] or 0) + (a["out"] or 0)
            out_r = (a["out"] or 0) / max(total, 1) * 100
            b = bar(total, max_ag, 16)
            label = f"{a['agent'] or '?'}  ({a['msgs']:,} msgs)"
            print_bar_row(label, total, max_ag)

    # ── Tool Usage ─────────────────────────────────────────────────────
    print_section("TOOL USAGE")
    tool_rows = conn.execute("""
        SELECT
            json_extract(data, '$.tool') as tool,
            COUNT(*) as cnt
        FROM part
        WHERE json_extract(data, '$.type') = 'tool'
        GROUP BY tool
        ORDER BY cnt DESC
    """).fetchall()

    if tool_rows:
        total_tools = sum(r["cnt"] for r in tool_rows)
        max_tool = tool_rows[0]["cnt"]
        for t in tool_rows[:25]:
            tool_name = t["tool"] or "?"
            tool_name = tool_name.replace("chrome-devtools_", "cdt_")
            tool_name = truncate(tool_name, 20)
            pct = t["cnt"] / max(total_tools, 1) * 100
            print_bar_row(tool_name, t["cnt"], max_tool, pct)

    # ── Top Projects ──────────────────────────────────────────────────
    print_section("TOP PROJECTS BY TOKENS")
    projects = conn.execute("""
        SELECT
            p.worktree,
            COUNT(s.id) as sess,
            SUM(s.tokens_input + s.tokens_output) as tokens
        FROM project p
        JOIN session s ON s.project_id = p.id
        GROUP BY p.id
        ORDER BY tokens DESC
        LIMIT 8
    """).fetchall()

    if projects:
        max_proj = projects[0]["tokens"] or 1
        for p in projects:
            path = p["worktree"] or "?"
            parts = path.split("/")
            name = "/".join(parts[-2:]) if len(parts) >= 2 else path
            name = truncate(name, 30)
            b = bar(p["tokens"] or 0, max_proj, 16)
            label = f"{name}  ({p['sess']} sess)"
            print_bar_row(label, p["tokens"] or 0, max_proj)

    # ── Weekly Trend ──────────────────────────────────────────────────
    print_section("WEEKLY TREND")
    weekly = conn.execute("""
        SELECT
            strftime('%Y-W%W', time_created / 1000, 'unixepoch', 'localtime') as week,
            SUM(tokens_input + tokens_output) as tokens,
            COUNT(*) as sess
        FROM session
        GROUP BY week
        ORDER BY week
    """).fetchall()

    if weekly:
        max_week = max(r["tokens"] or 1 for r in weekly)
        for w in weekly:
            wk = w["week"][-5:]
            print_bar_row(wk, w["tokens"] or 0, max_week)

    # ── Daily Heatmap (last 30 days) ─────────────────────────────────
    print_section("DAILY USAGE (last 30d)")
    daily = conn.execute("""
        SELECT
            date(time_created / 1000, 'unixepoch', 'localtime') as day,
            SUM(tokens_input + tokens_output) as tokens
        FROM session
        WHERE time_created > (strftime('%s', 'now') - 30 * 86400) * 1000
        GROUP BY day
        ORDER BY day DESC
    """).fetchall()

    if daily:
        max_day = max(r["tokens"] or 1 for r in daily)
        daily_sorted = list(reversed(daily))
        vals = [r["tokens"] or 0 for r in daily_sorted]
        print_kv("Trend", spark(vals))
        for d in daily[:10]:
            label = d["day"]
            print_bar_row(label, d["tokens"] or 0, max_day)

    # ── Top Sessions ──────────────────────────────────────────────────
    print_section("TOP 10 SESSIONS")
    top = conn.execute("""
        SELECT
            s.title,
            s.model,
            COALESCE(
                json_extract(s.model, '$.id'),
                (SELECT json_extract(m.data, '$.modelID')
                 FROM message m
                 WHERE m.session_id = s.id
                   AND json_extract(m.data, '$.role') = 'assistant'
                   AND json_extract(m.data, '$.modelID') IS NOT NULL
                 LIMIT 1)
            ) as model_name,
            s.tokens_input + s.tokens_output as tokens,
            s.time_created
        FROM session s
        ORDER BY tokens DESC
        LIMIT 10
    """).fetchall()

    for i, s in enumerate(top, 1):
        model_name = s["model_name"] or "?"
        model_name = truncate(model_name, 16)
        dt = datetime.fromtimestamp(s["time_created"] / 1000, tz=timezone.utc)
        title = truncate(s["title"] or "(untitled)", 28)
        tokens = fmt_tokens(s["tokens"] or 0)
        line = f"  {i:2}. {tokens:>7}  {model_name:16}  {dt:%m-%d}  {title}"
        w = 66
        inner = w - 4  # ║ ... ║
        padding = inner - display_len(line)
        if padding < 0:
            # truncate title further
            overflow = -padding
            title = truncate(title, max(0, display_len(title) - overflow))
            line = f"  {i:2}. {tokens:>7}  {model_name:16}  {dt:%m-%d}  {title}"
            padding = inner - display_len(line)
            if padding < 0:
                padding = 0
        print(f"  ║{line}{' ' * padding}║")

    print_footer()
    conn.close()
    print()


if __name__ == "__main__":
    main()