"""Render kernel for cctally reporting subcommands. Pure-fn layer (no I/O at import time): holds every ANSI-table renderer and JSON shaper used by the ``daily`` / ``monthly`` / ``weekly`` / ``session`` / ``blocks`` / ``project`` / ``codex-{daily,monthly,session}`` / ``five-hour-blocks`` subcommands. Two contiguous source regions collapse into one sibling here: * Region A (was bin/cctally L2175-L4661, ~2,486 LOC): block / bucket / weekly / codex-bucket / codex-session / claude-session / project renderers and their JSON-shape siblings, plus the ``_CODEX_MONTHS`` table and the project-row label disambiguator. * Region B (was bin/cctally L14350-L14507, ~158 LOC): the 5h-blocks-table render pair (``_five_hour_blocks_to_json`` + ``_render_five_hour_blocks_table``). Sibling dependencies (loaded at module-load time via ``_load_lib``): * ``_lib_blocks`` — ``Block`` (typing for ``_render_blocks_table``) and ``BLOCK_DURATION`` (block-duration fallback). * ``_lib_aggregators`` — the four bucket / session dataclasses consumed by the bucket / weekly / codex / claude-session renderers (``BucketUsage``, ``CodexBucketUsage``, ``CodexSessionUsage``, ``ClaudeSessionUsage``). * ``_lib_subscription_weeks`` — ``SubWeek`` (typing + runtime field access in ``_weekly_to_json`` / ``_render_weekly_table``). * ``_lib_pricing`` — ``_short_model_name`` (model-name shortener used across every breakdown-aware table). * ``_lib_display_tz`` — ``_resolve_tz`` (IANA tz resolution for the Codex session-table date columns). ``bin/cctally`` back-references via module-level callable shims (spec §5.5; same precedent as ``bin/_cctally_record.py``'s 34 shims): * ``_supports_color_stdout`` / ``_supports_unicode_stdout`` / ``_style_ansi`` — ANSI capability + style primitives. * ``_fmt_num`` / ``_truncate_num`` — numeric formatting helpers used by every render path. * ``_boxed_table`` — generic boxed-table renderer reused by ``_render_five_hour_blocks_table``. * ``_format_block_start`` — 5h-block Block-Start cell formatter (consumed only by the 5h-blocks renderer). Each shim resolves ``sys.modules['cctally'].X`` at CALL TIME (not bind time), so monkeypatches on cctally's namespace propagate into the moved code unchanged. ``bin/cctally`` eager-re-exports every public symbol below so the ~25 internal call sites + SourceFileLoader-based tests resolve unchanged. Spec: docs/superpowers/specs/2026-05-13-bin-cctally-split-design.md """ from __future__ import annotations import argparse import datetime as dt import json import math import os import pathlib import re import sys from typing import Any, Callable def _cctally(): """Resolve the current ``cctally`` module at call-time (spec §5.5).""" return sys.modules["cctally"] def _load_lib(name: str): cached = sys.modules.get(name) if cached is not None: return cached import importlib.util as _ilu p = pathlib.Path(__file__).resolve().parent / f"{name}.py" spec = _ilu.spec_from_file_location(name, p) mod = _ilu.module_from_spec(spec) sys.modules[name] = mod spec.loader.exec_module(mod) return mod _lib_blocks = _load_lib("_lib_blocks") Block = _lib_blocks.Block BLOCK_DURATION = _lib_blocks.BLOCK_DURATION _lib_aggregators = _load_lib("_lib_aggregators") BucketUsage = _lib_aggregators.BucketUsage CodexBucketUsage = _lib_aggregators.CodexBucketUsage CodexSessionUsage = _lib_aggregators.CodexSessionUsage ClaudeSessionUsage = _lib_aggregators.ClaudeSessionUsage _lib_subscription_weeks = _load_lib("_lib_subscription_weeks") SubWeek = _lib_subscription_weeks.SubWeek _lib_pricing = _load_lib("_lib_pricing") _short_model_name = _lib_pricing._short_model_name _lib_display_tz = _load_lib("_lib_display_tz") _resolve_tz = _lib_display_tz._resolve_tz # fmt/color/table primitives honest-imported from _lib_fmt (#126 C11) _lib_fmt = _load_lib("_lib_fmt") _supports_color_stdout = _lib_fmt._supports_color_stdout _supports_unicode_stdout = _lib_fmt._supports_unicode_stdout _style_ansi = _lib_fmt._style_ansi _fmt_num = _lib_fmt._fmt_num _truncate_num = _lib_fmt._truncate_num _boxed_table = _lib_fmt._boxed_table stable_sum = _lib_fmt.stable_sum # Module-level back-ref shims. Each shim resolves # ``sys.modules['cctally'].X`` at CALL TIME (not bind time), so # monkeypatches on cctally's namespace propagate into the moved code # unchanged. Mirrors the precedent established in # ``bin/_cctally_record.py`` / ``bin/_cctally_cache.py``. def _format_block_start(*args, **kwargs): return sys.modules["cctally"]._format_block_start(*args, **kwargs) def _ellipsize(content: str, width: int, unicode_ok: bool) -> str: """Truncate ``content`` to ``width`` cells, marking the cut with an ellipsis. Returns ``content`` unchanged when it already fits. Used for BOTH header and (text) data cells so a column scaled below its header-label width stays aligned with the border row (issue #102 (a)) — the header render previously padded without truncating, so a label wider than its scaled column overflowed the box.""" if len(content) <= width: return content ell = "…" if unicode_ok else "..." return content[: max(0, width - len(ell))] + ell def _scale_down_col_widths( nat_widths: list[int], aligns: list[str], data_widths: list[int], available: int, *, grow_idx: int, min_text: int = 4, ) -> list[int]: """Allocate the per-column widths for the ``--compact`` / auto-overflow scale-down branch of the project + Claude-session renderers. Policy (issue #102 — recommendation (a) + (b)): * **Numeric (right-aligned) columns are protected.** Their hard floor is the widest DATA value (``data_widths[i]``), so the full number always shows; the row render must never ellipsis-truncate a right-aligned cell. A silently-wrong figure (``12,345,…``) is worse than honest overflow. * **Text (left-aligned) columns absorb the squeeze** and may truncate — including their header label, which the header render ellipsizes the same way (so a column may drop below header width while staying box-aligned). * When the natural widths fit, they are used verbatim and the slack goes to ``grow_idx``. When shrinking is required, widths scale proportionally, clamped to ``[floor, natural]``; any residual after reclaiming all text slack is accepted as honest overflow rather than corrupting a number. ``data_widths[i]`` is the widest cell EXCLUDING the header, so a numeric column is sized to its number — its (possibly wider, truncatable) header label does not inflate the protected floor. """ num_cols = len(nat_widths) floors = [ min(max(data_widths[i], 1), nat_widths[i]) if aligns[i] == "right" else min(min_text, nat_widths[i]) for i in range(num_cols) ] total_nat = sum(nat_widths) if total_nat <= available: widths = list(nat_widths) slack = available - total_nat if slack > 0 and 0 <= grow_idx < num_cols: widths[grow_idx] += slack return widths scale = available / total_nat if total_nat > 0 else 1.0 widths = [ min(nat_widths[i], max(int(nat_widths[i] * scale), floors[i])) for i in range(num_cols) ] overflow = sum(widths) - available if overflow > 0: # Reclaim from text columns (most slack above floor first); numeric # columns are immovable. Residual overflow is honest (issue #102 (b)). text_idx = sorted( (i for i in range(num_cols) if aligns[i] != "right"), key=lambda i: widths[i] - floors[i], reverse=True, ) for i in text_idx: if overflow <= 0: break take = min(widths[i] - floors[i], overflow) widths[i] -= take overflow -= take else: slack = available - sum(widths) if slack > 0 and 0 <= grow_idx < num_cols: widths[grow_idx] += slack return widths # Optional dependency: zoneinfo.ZoneInfo is referenced only as a string # annotation in moved code; no runtime import needed. def _render_title_banner(title: str, *, unicode_ok: bool, color: bool) -> str: """ccusage-style rounded title banner (the box around a report title). Returns the multi-line banner WITHOUT a trailing blank line; callers add spacing. Shared by `_render_blocks_table` and `_render_active_block_box`. """ def _bold(s: str) -> str: return _style_ansi(s, "1", color) title_padded = f" {title} " tw = len(title_padded) dash = "─" if unicode_ok else "-" vb = "│" if unicode_ok else "|" if unicode_ok: top = f" ╭{dash * tw}╮" bot = f" ╰{dash * tw}╯" else: top = f" +{'-' * tw}+" bot = f" +{'-' * tw}+" return "\n".join([ top, f" {vb}" + " " * tw + vb, f" {vb}" + _bold(title_padded) + vb, f" {vb}" + " " * tw + vb, bot, ]) def _fmt_block_time_local(ts: dt.datetime, tz: "ZoneInfo | None") -> str: """Block-start timestamp in display tz, ccusage `toLocaleString`-style.""" local = ts.astimezone(tz) hour_12 = local.hour % 12 or 12 ampm = "a.m." if local.hour < 12 else "p.m." return ( f"{local.year}-{local.month:02d}-{local.day:02d}, " f"{hour_12}:{local.minute:02d}:{local.second:02d} {ampm}" ) def _fmt_block_duration_hm(total_seconds: float) -> str: total_minutes = int(total_seconds / 60) return f"{total_minutes // 60}h {total_minutes % 60:02d}m" def _render_active_block_box( block: "Block", *, now: dt.datetime, tz: "ZoneInfo | None", token_limit_explicit: int | None, color: bool, unicode_ok: bool, ) -> str: """ccusage `-a` 'Current Session Block Status' box (#86 Session F). Token Limit Status sub-block appears only when `-t` was explicitly passed (`token_limit_explicit` not None and > 0). Heuristic-anchored active blocks get a `~`-prefixed start + approximate-start legend (Codex F2). """ def _b(s: str) -> str: return _style_ansi(s, "1", color) lines: list[str] = [ _render_title_banner("Current Session Block Status", unicode_ok=unicode_ok, color=color), ] started = _fmt_block_time_local(block.start_time, tz) approx = block.anchor == "heuristic" if approx: started = f"~{started}" elapsed = max((now - block.start_time).total_seconds(), 0) remaining = max((block.end_time - now).total_seconds(), 0) lines.append(f"Block Started: {started} " f"({_fmt_block_duration_hm(elapsed)} ago)") lines.append(f"Time Remaining: {_fmt_block_duration_hm(remaining)}") if approx: # Keep this wording in sync with the table footer legend below. lines.append("~ = approximate start " "(no recorded Anthropic reset for this window)") lines += ["", _b("Current Usage:"), f" Input Tokens: {_fmt_num(block.input_tokens)}", f" Output Tokens: {_fmt_num(block.output_tokens)}", f" Total Cost: ${block.cost_usd:.2f}"] if block.burn_rate: lines += ["", _b("Burn Rate:"), f" Tokens/minute: " f"{_fmt_num(int(block.burn_rate['tokensPerMinute']))}", f" Cost/hour: ${block.burn_rate['costPerHour']:.2f}"] if block.projection: proj = block.projection lines += ["", _b("Projected Usage (if current rate continues):"), f" Total Tokens: {_fmt_num(proj['totalTokens'])}", f" Total Cost: ${proj['totalCost']:.2f}"] if token_limit_explicit is not None and token_limit_explicit > 0: limit = token_limit_explicit current = block.total_tokens remaining_tokens = max(limit - current, 0) pct_used = (proj["totalTokens"] / limit) * 100.0 cur_pct = (current / limit) * 100.0 # Keep the EXCEEDS/WARNING/OK thresholds (>100 / >80) in sync with # the JSON status ladder in _lib_blocks._blocks_to_json. if pct_used > 100: status = _style_ansi("EXCEEDS LIMIT", "31", color) elif pct_used > 80: status = _style_ansi("WARNING", "33", color) else: status = _style_ansi("OK", "32", color) lines += ["", _b("Token Limit Status:"), f" Limit: {_fmt_num(limit)} tokens", f" Current Usage: {_fmt_num(current)} ({cur_pct:.1f}%)", f" Remaining: {_fmt_num(remaining_tokens)} tokens", f" Projected Usage: {pct_used:.1f}% {status}"] return "\n".join(lines) def _render_blocks_table( blocks: list[Block], breakdown: bool = False, *, now: dt.datetime | None = None, tz: "ZoneInfo | None" = None, compact: bool = False, token_limit: int | None = None, ) -> str: """Render blocks as a ccusage-style ANSI table with box-drawing borders. Uses a two-pass approach matching upstream ccusage's ResponsiveTable: Pass 1 - Build all cell content as plain strings (no ANSI, no padding). Pass 2 - Compute column widths from content, then render with borders, padding, and ANSI colors. ``now`` pins the current instant for ACTIVE-row elapsed/remaining calculations (typically via ``_command_as_of()``). Defaults to wall- clock UTC so production behavior is unchanged; fixture-based tests pass a pinned value so goldens stay byte-stable. ``tz`` is the resolved display zone (``None`` means host local). Block-start cells are rendered in this zone. ``compact`` forces the scale-down code path regardless of the actual terminal width (Session A ``--compact`` flag; spec §7.6.1). Mirrors the same kwarg on ``_render_bucket_table``. Auto-detected width-overflow continues to trigger the same path as before. """ if not blocks: return "No session blocks found in the specified date range." color = _supports_color_stdout() unicode_ok = _supports_unicode_stdout() if now is None: now = dt.datetime.now(dt.timezone.utc) # ── ANSI helpers ──────────────────────────────────────────────────── def _dim(s: str) -> str: return _style_ansi(s, "90", color) def _cyan(s: str) -> str: return _style_ansi(s, "36", color) def _green(s: str) -> str: return _style_ansi(s, "32", color) def _blue(s: str) -> str: return _style_ansi(s, "34", color) def _yellow(s: str) -> str: return _style_ansi(s, "33", color) # ── time formatting ───────────────────────────────────────────────── def _fmt_time_local(ts: dt.datetime) -> str: return _fmt_block_time_local(ts, tz) def _fmt_duration_hm(total_seconds: float) -> str: return _fmt_block_duration_hm(total_seconds) def _fmt_gap_duration(total_seconds: float) -> str: total_minutes = int(total_seconds / 60) if total_minutes < 60: return f"{max(total_minutes, 1)}m gap" h, m = divmod(total_minutes, 60) return f"{h}h {m:02d}m gap" if m else f"{h}h gap" # ── determine if % column is needed ───────────────────────────────── max_completed_tokens = _lib_blocks._max_completed_block_tokens(blocks) active_block: Block | None = None for b in blocks: if b.is_active and not b.is_gap: active_block = b # ``token_limit`` param (#86 Session F): when caller passes an explicit # limit, honor it directly (forces the %/REMAINING/PROJECTED surface even # with no completed history, suppressed only when <= 0). When None, fall # back to the internal auto-max recompute so the renderer stays # standalone-callable and the default ``cmd_blocks`` path (which passes the # resolved auto-max) is byte-identical. if token_limit is not None: show_pct = token_limit > 0 eff_limit = token_limit if token_limit > 0 else 0 else: show_pct = max_completed_tokens > 0 eff_limit = max_completed_tokens if show_pct else 0 # ── column layout ─────────────────────────────────────────────────── headers = ["Block Start", "Duration/\u2026", "Models", "Tokens"] aligns = ["left", "left", "left", "right"] if show_pct: headers.append("%") aligns.append("right") headers.append("Cost") aligns.append("right") num_cols = len(headers) def _empty_cells() -> list[str]: return [""] * num_cols # ── Pass 1: build all row data as plain strings ───────────────────── # Each "row" is a list of display lines, each line is a list[str] of # cells. We also track per-row metadata for colorizing in pass 2. ROW_NORMAL = "normal" ROW_GAP = "gap" ROW_ACTIVE = "active" ROW_REMAINING = "remaining" ROW_PROJECTED = "projected" # (lines, row_type) all_rows: list[tuple[list[list[str]], str]] = [] for block in blocks: if block.is_gap: gap_seconds = (block.end_time - block.start_time).total_seconds() gap_dur = _fmt_gap_duration(gap_seconds) # Build gap text as single string; wrapping happens later if needed gap_text = ( f"{_fmt_time_local(block.start_time)} - " f"{_fmt_time_local(block.end_time)} ({gap_dur})" ) cells1 = _empty_cells() cells1[0] = gap_text cells1[1] = "(inactive)" cells1[2] = "-" cells1[3] = "-" if show_pct: cells1[4] = "-" cells1[-1] = "-" else: cells1[-1] = "-" all_rows.append(([cells1], ROW_GAP)) else: if block.is_active: elapsed_secs = (now - block.start_time).total_seconds() remaining_secs = max( (block.end_time - now).total_seconds(), 0 ) dur_str = ( f"{_fmt_duration_hm(elapsed_secs)} " f"elapsed, {_fmt_duration_hm(remaining_secs)} remaining)" ) duration_col = "ACTIVE" row_type = ROW_ACTIVE else: if block.actual_end_time: elapsed_secs = ( block.actual_end_time - block.start_time ).total_seconds() else: elapsed_secs = BLOCK_DURATION.total_seconds() dur_str = f"{_fmt_duration_hm(elapsed_secs)})" duration_col = "" row_type = ROW_NORMAL time_str = _fmt_time_local(block.start_time) if not block.is_gap and block.anchor == "heuristic": time_str = f"~{time_str}" start_text = f"{time_str} ({dur_str}" short_models = [ f"- {_short_model_name(m)}" for m in block.models ] if not short_models: short_models = [""] pct_str = "" if show_pct and eff_limit > 0: pct_val = (block.total_tokens / eff_limit) * 100.0 pct_str = f"{pct_val:.1f}%" tokens_str = _fmt_num(block.total_tokens) cost_str = f"${block.cost_usd:.2f}" # First line cells1 = _empty_cells() cells1[0] = start_text # may overflow; wrapping handled later cells1[1] = duration_col cells1[2] = short_models[0] cells1[3] = tokens_str ci = 4 if show_pct: cells1[ci] = pct_str ci += 1 cells1[ci] = cost_str display_lines: list[list[str]] = [cells1] # Continuation lines for remaining models for mi in range(1, len(short_models)): cont = _empty_cells() cont[2] = short_models[mi] display_lines.append(cont) all_rows.append((display_lines, row_type)) # Footer rows (REMAINING, PROJECTED) footer_rows: list[tuple[list[list[str]], str]] = [] if show_pct and eff_limit > 0: active_tokens = active_block.total_tokens if active_block else 0 remaining_tokens = max(eff_limit - active_tokens, 0) remaining_pct = (remaining_tokens / eff_limit) * 100.0 rem_label = f"(assuming {_fmt_num(eff_limit)} token limit)" rem_cells = _empty_cells() rem_cells[0] = rem_label rem_cells[1] = "REMAINING" rem_cells[3] = _fmt_num(remaining_tokens) ci = 4 if show_pct: rem_cells[ci] = f"{remaining_pct:.1f}%" ci += 1 footer_rows.append(([rem_cells], ROW_REMAINING)) if active_block and active_block.projection: proj = active_block.projection proj_tokens = proj.get("totalTokens", 0) proj_pct = ( (proj_tokens / eff_limit) * 100.0 if eff_limit > 0 else 0 ) proj_cost = proj.get("totalCost", 0.0) proj_cells = _empty_cells() proj_cells[0] = "(assuming current burn rate)" proj_cells[1] = "PROJECTED" proj_cells[3] = _fmt_num(proj_tokens) ci = 4 if show_pct: proj_cells[ci] = f"{proj_pct:.1f}%" ci += 1 proj_cells[ci] = f"${proj_cost:.2f}" footer_rows.append(([proj_cells], ROW_PROJECTED)) # ── Pass 2: compute column widths from content ────────────────────── # Measure max content width per column from headers + all cell data. content_widths = [len(h) for h in headers] def _measure_rows( rows: list[tuple[list[list[str]], str]], skip_col0: bool = False, ) -> None: for display_lines, _ in rows: for line_cells in display_lines: for i, cell in enumerate(line_cells): if skip_col0 and i == 0: continue content_widths[i] = max(content_widths[i], len(cell)) _measure_rows(all_rows) # Footer labels (col 0) are right-justified into whatever width col 0 # gets — they should not inflate it. _measure_rows(footer_rows, skip_col0=True) # Add padding matching upstream ccusage's ResponsiveTable. col_widths: list[int] = [] for i, cw in enumerate(content_widths): if aligns[i] == "right": col_widths.append(max(cw + 3, 11)) elif i == 1: # Duration column col_widths.append(max(cw + 2, 15)) else: col_widths.append(max(cw + 2, 10)) # Get terminal width (matches ccusage's ResponsiveTable.toString()). try: term_width = os.get_terminal_size().columns except (OSError, ValueError): term_width = int(os.environ.get("COLUMNS", "120")) # Scale down only when table exceeds terminal width. # ccusage does NOT expand columns when the table fits — it uses the # padded content widths as-is. Session A (spec §7.6.1): the # ``compact`` kwarg forces this branch regardless of terminal width # (Review-A P2-B; mirrors ``_render_bucket_table`` semantics). table_overhead = 3 * num_cols + 1 available_width = term_width - table_overhead if compact or (sum(col_widths) + table_overhead > term_width): scale_factor = available_width / sum(col_widths) col_widths = [ max( math.floor(w * scale_factor), 10 if aligns[i] == "right" else 10 if i == 0 else 12 if i == 1 else 8, ) for i, w in enumerate(col_widths) ] # ── box-drawing characters ────────────────────────────────────────── if unicode_ok: ch = { "tl": "\u250c", "tm": "\u252c", "tr": "\u2510", "ml": "\u251c", "mm": "\u253c", "mr": "\u2524", "bl": "\u2514", "bm": "\u2534", "br": "\u2518", "h": "\u2500", "v": "\u2502", } else: ch = {k: c for k, c in zip( ["tl", "tm", "tr", "ml", "mm", "mr", "bl", "bm", "br", "h", "v"], "+++++++++-|", )} def hline(left: str, mid: str, right: str) -> str: segs = [ch["h"] * (col_widths[i] + 2) for i in range(num_cols)] return _dim(left + mid.join(segs) + right) def padcell(text: str, width: int, align: str) -> str: """Pad cell content to width. Text may contain ANSI codes, so we compute visible length by stripping escape sequences.""" vis_len = len(re.sub(r"\033\[[0-9;]*m", "", text)) pad_needed = width - vis_len if pad_needed <= 0: return text if align == "right": return " " * pad_needed + text return text + " " * pad_needed def make_row(cells: list[str]) -> str: parts: list[str] = [] for i, cell_text in enumerate(cells): padded = padcell(cell_text, col_widths[i], aligns[i]) parts.append(f" {padded} ") v = _dim(ch["v"]) return v + v.join(parts) + v # ── Wrap + colorize helpers ───────────────────────────────────────── def _wrap_col0(text: str, width: int) -> list[str]: """Wrap column-0 text to fit *width*, breaking at word boundaries.""" if len(text) <= width: return [text] # Try to break at a word boundary. split_at = width space_idx = text.rfind(" ", 0, width + 1) if space_idx > width // 2: split_at = space_idx + 1 part1 = text[:split_at].rstrip() part2 = text[split_at:].lstrip() lines_out = [part1] if part2: lines_out.extend(_wrap_col0(part2, width)) return lines_out def _colorize_cell(text: str, col_idx: int, row_type: str) -> str: """Apply ANSI color to a cell based on row type and column.""" if row_type == ROW_GAP: return _dim(text) if text else text if col_idx == 1: if row_type == ROW_ACTIVE: return _green(text) if text == "ACTIVE" else text if row_type == ROW_REMAINING: return _blue(text) if text == "REMAINING" else text if row_type == ROW_PROJECTED: return _yellow(text) if text == "PROJECTED" else text return text # ── Render output ─────────────────────────────────────────────────── # Title banner title = "Claude Code Token Usage Report - Session Blocks" lines: list[str] = [] lines.append(_render_title_banner(title, unicode_ok=unicode_ok, color=color)) lines.append("") # Header lines.append(hline(ch["tl"], ch["tm"], ch["tr"])) header_cells = [_cyan(h) for h in headers] lines.append(make_row(header_cells)) lines.append(hline(ch["ml"], ch["mm"], ch["mr"])) # Data rows col0_w = col_widths[0] def _render_block_row( display_lines: list[list[str]], row_type: str ) -> None: """Render one block's display lines, wrapping col 0 and truncating the Tokens column as needed.""" for li, line_cells in enumerate(display_lines): # Wrap column 0 if it overflows. col0_text = line_cells[0] col0_parts = _wrap_col0(col0_text, col0_w) if col0_text else [""] for wi, c0_part in enumerate(col0_parts): cells = _empty_cells() cells[0] = _colorize_cell(c0_part, 0, row_type) if wi == 0: # First wrap-line carries the real cell data. for ci in range(1, num_cols): raw = line_cells[ci] # Truncate tokens column if needed. if ci == 3 and raw and raw != "-": raw = _truncate_num(raw, col_widths[ci]) cells[ci] = _colorize_cell(raw, ci, row_type) lines.append(make_row(cells)) for idx, (display_lines, row_type) in enumerate(all_rows): _render_block_row(display_lines, row_type) if idx < len(all_rows) - 1: lines.append(hline(ch["ml"], ch["mm"], ch["mr"])) # Footer rows for fi, (display_lines, row_type) in enumerate(footer_rows): lines.append(hline(ch["ml"], ch["mm"], ch["mr"])) # Right-align the label in column 0 for footer rows. for line_cells in display_lines: line_cells[0] = line_cells[0].rjust(col0_w) _render_block_row(display_lines, row_type) # Bottom border lines.append(hline(ch["bl"], ch["bm"], ch["br"])) rendered = "\n".join(lines) has_heuristic = any( (not b.is_gap) and b.anchor == "heuristic" for b in blocks ) if has_heuristic: legend = _dim( "~ = approximate start " "(no recorded Anthropic reset for this window)" ) rendered = f"{rendered}\n{legend}" return rendered def _daily_row_dict(d: BucketUsage, *, date_key: str) -> dict[str, Any]: """Single bucket → upstream-shaped row dict. Shared by `_bucket_to_json` and `_bucket_by_project_to_json` so the per-row shape (field set + order) can never drift. Key order matches ccusage: date_key, inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens, totalTokens, totalCost, modelsUsed, modelBreakdowns. """ return { date_key: d.bucket, "inputTokens": d.input_tokens, "outputTokens": d.output_tokens, "cacheCreationTokens": d.cache_creation_tokens, "cacheReadTokens": d.cache_read_tokens, "totalTokens": d.total_tokens, "totalCost": d.cost_usd, "modelsUsed": list(d.models), "modelBreakdowns": list(d.model_breakdowns), } def _bucket_totals_dict(buckets) -> dict[str, Any]: """Aggregate totals across buckets (key order matches ccusage; note totalCost BEFORE totalTokens).""" return { "inputTokens": sum(b.input_tokens for b in buckets), "outputTokens": sum(b.output_tokens for b in buckets), "cacheCreationTokens": sum(b.cache_creation_tokens for b in buckets), "cacheReadTokens": sum(b.cache_read_tokens for b in buckets), "totalCost": stable_sum(b.cost_usd for b in buckets), "totalTokens": sum(b.total_tokens for b in buckets), } def _bucket_to_json( buckets: list[BucketUsage], *, list_key: str, date_key: str, ) -> str: """Serialize bucket aggregates to JSON matching upstream ccusage's shape. `list_key` is the top-level array name ("daily" or "monthly"). `date_key` is the per-item bucket field name ("date" or "month"). Key order inside each item matches ccusage: date_key, inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens, totalTokens, totalCost, modelsUsed, modelBreakdowns. Totals key order (note: totalCost BEFORE totalTokens, per ccusage). """ bucket_list = [_daily_row_dict(d, date_key=date_key) for d in buckets] totals = _bucket_totals_dict(buckets) return json.dumps({list_key: bucket_list, "totals": totals}, indent=2) def _bucket_by_project_to_json(project_groups, *, date_key: str = "date") -> str: """Serialize ``[(label, [BucketUsage]), ...]`` to ``{projects:{label:[rows]}, totals}`` (upstream ccusage daily --instances shape). Caller passes the disambiguated (unique, non-aliased) label per group; insertion order is preserved as the JSON key order.""" projects: dict[str, Any] = {} all_buckets: list = [] for label, buckets in project_groups: projects[label] = [_daily_row_dict(b, date_key=date_key) for b in buckets] all_buckets.extend(buckets) return json.dumps( {"projects": projects, "totals": _bucket_totals_dict(all_buckets)}, indent=2, ) def _weekly_to_json( buckets: list[BucketUsage], weeks: list[SubWeek], week_pct_overlay: list[tuple[float | None, float | None]], ) -> str: """Serialize weekly rollup to JSON. Shape: { "weekly": [ { "week": "YYYY-MM-DD", # API-derived week_start_date (stable contract / lookup key) "displayWeek": "YYYY-MM-DD", # effective post-reset start; equals `week` for non-reset weeks "weekStartAt": "...ISO...", "weekEndAt": "...ISO...", "weekSource": "snapshot" | "extrapolated", "inputTokens": int, "cacheCreationTokens": int, "cacheReadTokens": int, "outputTokens": int, "totalTokens": int, "totalCost": float, "usedPct": float | null, "dollarsPerPercent": float | null, "modelsUsed": [...], "modelBreakdowns": [...] }, ... ], "totals": { inputTokens, cacheCreationTokens, cacheReadTokens, outputTokens, totalTokens, totalCost } } """ assert len(week_pct_overlay) == len(buckets), ( f"week_pct_overlay length {len(week_pct_overlay)} does not match " f"buckets length {len(buckets)} — caller contract violated" ) # Build dict lookup from week-start-date ISO → SubWeek for metadata. week_by_key = {w.start_date.isoformat(): w for w in weeks} weekly_list: list[dict[str, Any]] = [] tot_input = tot_cache_c = tot_cache_r = tot_output = tot_total = 0 tot_cost = 0.0 for i, bucket in enumerate(buckets): w = week_by_key.get(bucket.bucket) if w is None: # Defensive: bucket key should always match a SubWeek (_aggregate_weekly # only emits keys derived from the provided weeks list). Raise loud # rather than silently emit partial data. raise ValueError( f"bucket key {bucket.bucket!r} has no matching SubWeek in `weeks`" ) pct, dpc = week_pct_overlay[i] weekly_list.append({ "week": bucket.bucket, "displayWeek": w.display_start_date.isoformat(), "weekStartAt": w.start_ts, "weekEndAt": w.end_ts, "weekSource": w.source, "inputTokens": bucket.input_tokens, "cacheCreationTokens": bucket.cache_creation_tokens, "cacheReadTokens": bucket.cache_read_tokens, "outputTokens": bucket.output_tokens, "totalTokens": bucket.total_tokens, "totalCost": bucket.cost_usd, "usedPct": pct, "dollarsPerPercent": dpc, "modelsUsed": bucket.models, "modelBreakdowns": bucket.model_breakdowns, }) tot_input += bucket.input_tokens tot_cache_c += bucket.cache_creation_tokens tot_cache_r += bucket.cache_read_tokens tot_output += bucket.output_tokens tot_total += bucket.total_tokens tot_cost += bucket.cost_usd payload = { "weekly": weekly_list, "totals": { "inputTokens": tot_input, "cacheCreationTokens": tot_cache_c, "cacheReadTokens": tot_cache_r, "outputTokens": tot_output, "totalTokens": tot_total, "totalCost": tot_cost, }, } return json.dumps(payload, indent=2) def _daily_compact_split(bucket: str) -> str: """YYYY-MM-DD → "YYYY\\nMM-DD" for compact-mode Date column.""" m = re.match(r"^(\d{4})-(\d{2})-(\d{2})$", bucket) return f"{m.group(1)}\n{m.group(2)}-{m.group(3)}" if m else bucket def _monthly_compact_split(bucket: str) -> str: """YYYY-MM → "YYYY\\nMM" for compact-mode Month column. Deliberate deviation from ccusage: upstream renders a synthetic "-01" day component in compact mode because its formatter is daily-oriented. We omit it — same information, less visual noise. """ m = re.match(r"^(\d{4})-(\d{2})$", bucket) return f"{m.group(1)}\n{m.group(2)}" if m else bucket _CODEX_MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] def _codex_daily_bucket_display(bucket: str) -> str: """YYYY-MM-DD → "Mon DD, YYYY" (e.g. "Dec 25, 2025"). Upstream shape.""" m = re.match(r"^(\d{4})-(\d{2})-(\d{2})$", bucket) if not m: return bucket return f"{_CODEX_MONTHS[int(m.group(2)) - 1]} {int(m.group(3)):02d}, {m.group(1)}" def _codex_monthly_bucket_display(bucket: str) -> str: """YYYY-MM → "Mon YYYY" (e.g. "Dec 2025"). Upstream shape.""" m = re.match(r"^(\d{4})-(\d{2})$", bucket) if not m: return bucket return f"{_CODEX_MONTHS[int(m.group(2)) - 1]} {m.group(1)}" def _codex_last_activity_iso(ts: dt.datetime) -> str: """ISO-8601 UTC with milliseconds and Z suffix (e.g. "2025-12-25T10:03:52.375Z"). Matches upstream ccusage-codex's session `lastActivity` format byte-exactly. """ utc = ts.astimezone(dt.timezone.utc) # Python's .isoformat() defaults to microseconds (6 digits); upstream uses # milliseconds (3 digits). Truncate and append Z. return utc.strftime("%Y-%m-%dT%H:%M:%S") + f".{utc.microsecond // 1000:03d}Z" def _emit_codex_no_data(args: argparse.Namespace, list_key: str) -> None: """Print upstream's empty-result sentinel for codex-{daily,monthly,session}. Matches ccusage-codex byte-exactly: - JSON: ``{"":[],"totals":null}`` (compact separators, no whitespace — upstream uses ``JSON.stringify(...)`` with no indent argument for the empty case, even though the happy-path uses indent=2). - Text: ``"No Codex usage data found."`` when no filters are in effect, or ``"No Codex usage data found for provided filters."`` when --since or --until is set (matching upstream's filter-aware messaging). """ filter_applied = bool(getattr(args, "since", None) or getattr(args, "until", None)) if getattr(args, "json", False): # Compact separators to match Node's `JSON.stringify(obj)` output exactly. print(json.dumps({list_key: [], "totals": None}, separators=(",", ":"))) else: if filter_applied: print("No Codex usage data found for provided filters.") else: print("No Codex usage data found.") def _codex_models_dict( model_breakdowns: list[dict[str, Any]], ) -> dict[str, dict[str, Any]]: """Convert our internal list-of-breakdowns into upstream's models dict. Input: list of {modelName, inputTokens, cachedInputTokens, outputTokens, reasoningOutputTokens, totalTokens, cost, isFallback}. Output: {: {inputTokens, cachedInputTokens, outputTokens, reasoningOutputTokens, totalTokens, isFallback}} Insertion order: whatever the caller passed (aggregators sort by cost desc). Note: the per-model `cost` / `modelName` keys from the list are dropped — upstream's dict doesn't include them at the per-model level. """ out: dict[str, dict[str, Any]] = {} for mb in model_breakdowns: out[mb["modelName"]] = { "inputTokens": mb["inputTokens"], "cachedInputTokens": mb["cachedInputTokens"], "outputTokens": mb["outputTokens"], "reasoningOutputTokens": mb["reasoningOutputTokens"], "totalTokens": mb["totalTokens"], "isFallback": mb["isFallback"], } return out def _codex_bucket_to_json( buckets: list[CodexBucketUsage], *, list_key: str, # "daily" or "monthly" date_key: str, # "date" or "month" display_fn: Callable[[str], str], # maps bucket key → human display ) -> str: """Serialize Codex bucket aggregates to JSON matching upstream exactly. Per-entry shape: {, inputTokens, cachedInputTokens, outputTokens, reasoningOutputTokens, totalTokens, costUSD, models} Totals: {inputTokens, cachedInputTokens, outputTokens, reasoningOutputTokens, totalTokens, costUSD} """ bucket_list: list[dict[str, Any]] = [] tot_input = tot_cached = tot_output = tot_reasoning = tot_tokens = 0 tot_cost = 0.0 for b in buckets: bucket_total = b.input_tokens + b.output_tokens bucket_list.append({ date_key: display_fn(b.bucket), "inputTokens": b.input_tokens, "cachedInputTokens": b.cached_input_tokens, "outputTokens": b.output_tokens, "reasoningOutputTokens": b.reasoning_output_tokens, "totalTokens": bucket_total, "costUSD": b.cost_usd, "models": _codex_models_dict(b.model_breakdowns), }) tot_input += b.input_tokens tot_cached += b.cached_input_tokens tot_output += b.output_tokens tot_reasoning += b.reasoning_output_tokens tot_tokens += bucket_total tot_cost += b.cost_usd totals = { "inputTokens": tot_input, "cachedInputTokens": tot_cached, "outputTokens": tot_output, "reasoningOutputTokens": tot_reasoning, "totalTokens": tot_tokens, "costUSD": tot_cost, } return json.dumps({list_key: bucket_list, "totals": totals}, indent=2) def _codex_root_short_labels(roots: list[str]) -> list[str]: """Given >=2 distinct root paths, return a short distinguishing label per root: the first path segment after their longest common ancestor — e.g. {".../rootA/.codex", ".../rootB/.codex"} -> ["rootA", "rootB"]. If a single segment isn't unique, fall back to the full post-ancestor tail, then to the whole path, so the returned labels are always pairwise distinct. Order-preserving: labels[i] corresponds to roots[i]. """ parts = [pathlib.PurePosixPath(r).parts for r in roots] common = 0 for seg in zip(*parts): if len(set(seg)) == 1: common += 1 else: break singles = [pr[common] if common < len(pr) else (pr[-1] if pr else "") for pr in parts] if len(set(singles)) == len(singles): return singles tails = ["/".join(pr[common:]) or (pr[-1] if pr else "") for pr in parts] if len(set(tails)) == len(tails): return tails return list(roots) def _codex_session_disambiguate(sessions: list[CodexSessionUsage]) -> dict[int, str]: """Return ``{row_index: " (label)"}`` for codex-session rows whose ``session_id_path`` collides with another row's — a genuine cross-$CODEX_HOME-root collision (issue #110). Non-colliding rows are absent from the dict; callers render them byte-identically (single-root data never collides, so this returns ``{}`` and output is unchanged). ``label`` is a short distinguishing segment of the matched $CODEX_HOME root (see ``_codex_root_short_labels``). Mirrors ``_project_disambiguate_labels``. """ path_counts: dict[str, int] = {} for s in sessions: path_counts[s.session_id_path] = path_counts.get(s.session_id_path, 0) + 1 groups: dict[str, list[int]] = {} for idx, s in enumerate(sessions): if path_counts[s.session_id_path] > 1: groups.setdefault(s.session_id_path, []).append(idx) out: dict[int, str] = {} for idxs in groups.values(): labels = _codex_root_short_labels([sessions[i].codex_root for i in idxs]) for i, label in zip(idxs, labels): out[i] = f" ({label})" return out def _codex_sessions_to_json(sessions: list[CodexSessionUsage]) -> str: """Serialize Codex session aggregates to JSON matching upstream exactly. Per-session shape: {sessionId, lastActivity, sessionFile, directory, inputTokens, cachedInputTokens, outputTokens, reasoningOutputTokens, totalTokens, costUSD, models} """ session_list: list[dict[str, Any]] = [] tot_input = tot_cached = tot_output = tot_reasoning = tot_tokens = 0 tot_cost = 0.0 # Issue #110: only when two cross-root sessions share `sessionId` (the # relative path) does this map a colliding row to a label; otherwise empty, # so the per-session shape stays upstream-byte-identical (no `codexRoot`). disambig = _codex_session_disambiguate(sessions) for idx, s in enumerate(sessions): session_total = s.input_tokens + s.output_tokens entry = { "sessionId": s.session_id_path, "lastActivity": _codex_last_activity_iso(s.last_activity), "sessionFile": s.session_file, "directory": s.directory, "inputTokens": s.input_tokens, "cachedInputTokens": s.cached_input_tokens, "outputTokens": s.output_tokens, "reasoningOutputTokens": s.reasoning_output_tokens, "totalTokens": session_total, "costUSD": s.cost_usd, "models": _codex_models_dict(s.model_breakdowns), } if idx in disambig: # Additive disambiguator — `sessionId` keeps its upstream-compatible # relative-path value; consumers key off `codexRoot` to tell the # (correctly separate) colliding rows apart. entry["codexRoot"] = s.codex_root session_list.append(entry) tot_input += s.input_tokens tot_cached += s.cached_input_tokens tot_output += s.output_tokens tot_reasoning += s.reasoning_output_tokens tot_tokens += session_total tot_cost += s.cost_usd totals = { "inputTokens": tot_input, "cachedInputTokens": tot_cached, "outputTokens": tot_output, "reasoningOutputTokens": tot_reasoning, "totalTokens": tot_tokens, "costUSD": tot_cost, } return json.dumps({"sessions": session_list, "totals": totals}, indent=2) def _claude_sessions_to_json(sessions: list[ClaudeSessionUsage]) -> str: """Serialize Claude sessions to JSON (spec A2.8, amended by issue #104). Per-session: sessionId, projectPath, sourcePaths (list), firstActivity / lastActivity ISO strings, modelsUsed, token counts (input/cacheCreation/cacheRead/output/total), totalCost, modelBreakdowns (camelCased token field names, cost). `totalTokens` (per-session + totals) sums ALL four token components (input + output + cacheCreation + cacheRead) per issue #104 — matching `daily`/`monthly` and ccusage v20. (The field name/shape is unchanged; only the value definition widened to include cache.) totals: same 6 numeric fields aggregated across sessions. """ sess_list: list[dict[str, Any]] = [] tot_i = tot_cc = tot_cr = tot_o = tot_t = 0 tot_cost = 0.0 for s in sessions: sess_list.append({ "sessionId": s.session_id, "projectPath": s.project_path, "sourcePaths": list(s.source_paths), "firstActivity": s.first_activity.isoformat(), "lastActivity": s.last_activity.isoformat(), "modelsUsed": list(s.models), "inputTokens": s.input_tokens, "cacheCreationTokens": s.cache_creation_tokens, "cacheReadTokens": s.cache_read_tokens, "outputTokens": s.output_tokens, "totalTokens": s.total_tokens, "totalCost": s.cost_usd, "modelBreakdowns": [ { "model": mb["model"], "inputTokens": mb["input"], "cacheCreationTokens": mb["cache_create"], "cacheReadTokens": mb["cache_read"], "outputTokens": mb["output"], "cost": mb["cost"], } for mb in s.model_breakdowns ], }) tot_i += s.input_tokens tot_cc += s.cache_creation_tokens tot_cr += s.cache_read_tokens tot_o += s.output_tokens tot_t += s.total_tokens tot_cost += s.cost_usd payload = { "sessions": sess_list, "totals": { "inputTokens": tot_i, "cacheCreationTokens": tot_cc, "cacheReadTokens": tot_cr, "outputTokens": tot_o, "totalTokens": tot_t, "totalCost": tot_cost, }, } return json.dumps(payload, indent=2) def _render_bucket_table( buckets: list[BucketUsage], *, first_col_name: str, title_suffix: str, compact_split_fn: Callable[[str], str], breakdown: bool = False, compact: bool = False, project_groups=None, ) -> str: """Render bucket aggregates as a ccusage-style ANSI table. Shared between `daily` and `monthly` subcommands. Parameters: first_col_name — header for the bucket column ("Date" or "Month"). title_suffix — banner text suffix ("Daily" or "Monthly"). compact_split_fn — function that splits a bucket string into "YYYY\\n..." for compact-mode two-line display. compact — force compact layout regardless of terminal width (Session A `--compact` flag; spec §7.6.1). project_groups — when provided as ``[(label, [BucketUsage]), ...]``, render section layout: a ``Project: