From 00224510cbd8db9c9f16f295f869feef0eb42b51 Mon Sep 17 00:00:00 2001 From: Jonathan Marien Date: Mon, 15 Jun 2026 17:37:48 -0400 Subject: [PATCH] feat(claude_usage): token history, API status, model breakdown and pinnable popup Builds on the existing claude_usage widget; everything is off by default and closes #967, #970 and #971. - refresh action plus a header button that force an immediate fetch and skip cache_ttl, with in-flight dedup; a {stale} indicator that shows once the OAuth token has expired; and per-window reset wording (relative or absolute, with an optional date) that fixes the broken "Resets in Sat 6:00 AM" line - local token-usage history: a Session/Today/Week/Month/Year section read from Claude Code's own transcripts under ~/.claude/projects, scanned off-thread and incrementally, with the cache bounded (hourly past 15 days and daily past ~13 months are pruned on scan). No API key, no network call, content never read - Claude API status dot driven by the public status.claude.com page (no auth), keeping the last known status on a failed poll - per-model token breakdown in the Tokens section, following the same period toggle. Model names are derived from the id so new models need no upkeep, and fast-mode usage (usage.speed) is split into its own row - pinnable, draggable popup matching the cpu/memory/gpu widgets, via a shared create_pin_button helper in stat_popup - the popup resizes to fit the selected period instead of cramming or stretching - docs and example CSS under docs/ Co-Authored-By: ManaphatDev <106569727+Gaer12TH@users.noreply.github.com> --- README.md | 1 + docs/widgets/(Widget)-Claude-Usage.md | 136 ++++- src/core/utils/stat_popup.py | 46 +- .../validation/widgets/yasb/claude_usage.py | 29 ++ .../services/claude_usage/claude_api.py | 40 +- .../widgets/services/claude_usage/status.py | 129 +++++ .../services/claude_usage/token_history.py | 477 ++++++++++++++++++ src/core/widgets/yasb/claude_usage.py | 466 +++++++++++++++-- 8 files changed, 1270 insertions(+), 54 deletions(-) create mode 100644 src/core/widgets/services/claude_usage/status.py create mode 100644 src/core/widgets/services/claude_usage/token_history.py diff --git a/README.md b/README.md index c2dd9953d..23f396027 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,7 @@ choco install yasb | [Bluetooth](https://github.com/amnweb/yasb/wiki/(Widget)-Bluetooth) | Shows the current Bluetooth status and connected devices. | | [Brightness](https://github.com/amnweb/yasb/wiki/(Widget)-Brightness) | Displays and change the current brightness level. | | [Cava](https://github.com/amnweb/yasb/wiki/(Widget)-Cava) | Displays audio visualizer using Cava. | +| [Claude Usage](https://github.com/amnweb/yasb/wiki/(Widget)-Claude-Usage) | Shows your Claude (Claude Code) subscription usage with a popup of the 5-hour and 7-day limits. | | [Copilot](https://github.com/amnweb/yasb/wiki/(Widget)-Copilot) | GitHub Copilot usage with a detailed menu showing statistics | | [CPU](https://github.com/amnweb/yasb/wiki/(Widget)-CPU) | Shows the current CPU usage and information. | | [Clock](https://github.com/amnweb/yasb/wiki/(Widget)-Clock) | Displays the current time and date, with customizable formats. | diff --git a/docs/widgets/(Widget)-Claude-Usage.md b/docs/widgets/(Widget)-Claude-Usage.md index 5a0c08169..eac925e36 100644 --- a/docs/widgets/(Widget)-Claude-Usage.md +++ b/docs/widgets/(Widget)-Claude-Usage.md @@ -14,21 +14,35 @@ extra configuration is required as long as you are signed in to Claude Code. | `label_alt` | string | `'Claude {seven_day}%'` | The alternative format string, toggled by the `toggle_label` callback. | | `update_interval` | integer | `60` | How often the label and reset countdown are refreshed, in seconds. Must be between 30 and 3600. | | `cache_ttl` | integer | `120` | How long (seconds) a fetched result is cached on disk before the endpoint is queried again. The endpoint is rate-limited, so keep this at a sane value. | +| `five_hour_reset_format` | string | `'relative'` | How the 5-hour window's reset line is phrased in the popup: `relative` (`Resets in 4h 11m`) or `absolute` (`Resets on Sat @ 6:00 AM`). | +| `seven_day_reset_format` | string | `'absolute'` | How the 7-day window's reset line is phrased in the popup: `relative` or `absolute`. | +| `reset_show_date` | boolean | `true` | In `absolute` mode, include the month/day (`Resets on Sat, Jun 13 @ 6:00 AM`) so two windows resetting on the same weekday stay distinguishable. | +| `token_history` | dict | `{'enabled': false, ...}` | Optional local token-usage history. See [Token history](#token-history). | +| `status` | dict | `{'enabled': false, ...}` | Optional Claude API status indicator. See [API status](#api-status). | | `tooltip` | boolean | `true` | Whether to show a summary tooltip on hover. | | `callbacks` | dict | `{'on_left': 'toggle_menu', 'on_middle': 'do_nothing', 'on_right': 'toggle_label'}` | Mouse-click callbacks. | -| `menu` | dict | `{'blur': true, 'round_corners': true, 'round_corners_type': 'normal', 'border_color': 'System', 'alignment': 'right', 'direction': 'down', 'offset_top': 6, 'offset_left': 0}` | Popup menu settings. | +| `menu` | dict | `{'blur': true, 'round_corners': true, 'round_corners_type': 'normal', 'border_color': 'System', 'alignment': 'right', 'direction': 'down', 'offset_top': 6, 'offset_left': 0, 'pin_icon': '', 'unpin_icon': ''}` | Popup menu settings. | ## Placeholders The label is plain text by default. You can prepend a Nerd Font glyph in a `` if you -want an icon (e.g. `\U000f06a9 {five_hour}%`). The following placeholders can be -used in `label` / `label_alt`: +want an icon (e.g. `\U000f06a9 {five_hour}%`), or embed your own image with an +`` tag (e.g. ` {five_hour}%`). The +following placeholders can be used in `label` / `label_alt`: - `{five_hour}` — 5-hour window utilization (percent, `--` when unavailable). - `{seven_day}` — 7-day window utilization (percent, `--` when unavailable). - `{five_hour_reset}` — time until the 5-hour window resets. Shown as a countdown when under a day away (e.g. `4h 27m`), otherwise as a local weekday + time (e.g. `Sat 6:00 AM`). - `{seven_day_reset}` — time until the 7-day window resets (e.g. `Sat 6:00 AM`). +- `{stale}` — a warning glyph shown only while Claude Code's OAuth token has expired, empty + otherwise. Place it in its own `` (e.g. `{five_hour}% {stale}`). +- `{session_tokens}` `{today_tokens}` `{week_tokens}` `{month_tokens}` `{year_tokens}` — compact + token totals (e.g. `1.2M`) for each period. Require `token_history.enabled`; `--` otherwise. +- `{status}` — a status dot, coloured by the current Claude API status level via + `.status.` classes. Place it in its own ``. Requires + `status.enabled`; empty otherwise. +- `{status_text}` — the status description (e.g. `All Systems Operational`). ```yaml claude_usage: @@ -40,7 +54,7 @@ claude_usage: cache_ttl: 120 callbacks: on_left: "toggle_menu" # open the usage menu - on_middle: "do_nothing" + on_middle: "refresh" # force an immediate re-fetch, bypassing cache_ttl on_right: "toggle_label" # switch the bar text between 5h and 7d menu: blur: true @@ -59,8 +73,10 @@ claude_usage: - **label_alt:** The alternative format string, toggled with the `toggle_label` callback. - **update_interval:** How often the bar label and reset countdown are refreshed, in seconds (30–3600). - **cache_ttl:** How long a fetched result is cached on disk before the usage endpoint is queried again. Because the endpoint is rate-limited (HTTP 429), the widget serves the last cached value on any error instead of going blank. +- **five_hour_reset_format / seven_day_reset_format:** How each window's reset line is phrased in the popup. `relative` shows a countdown (`Resets in 4h 11m`); `absolute` shows a local weekday and time (`Resets on Sat @ 6:00 AM`). The exact reset timestamp is always shown on the line below. +- **reset_show_date:** In `absolute` mode, include the month/day in the reset line so the 5-hour and 7-day windows can be told apart when they fall on the same weekday. No effect in `relative` mode. - **tooltip:** Whether to show a summary tooltip on hover. -- **callbacks:** Mouse-click callbacks. Built-in actions: `toggle_menu` (open/close the popup menu), `toggle_label` (swap between `label` and `label_alt`), `do_nothing`, and `exec`. +- **callbacks:** Mouse-click callbacks. Built-in actions: `toggle_menu` (open/close the popup menu), `toggle_label` (swap between `label` and `label_alt`), `refresh` (force an immediate re-fetch, bypassing `cache_ttl`), `do_nothing`, and `exec`. - **menu:** A dictionary specifying the popup menu settings: - **blur:** Enable blur effect for the menu. - **round_corners:** Enable round corners (not supported on Windows 10). @@ -69,6 +85,7 @@ claude_usage: - **alignment:** Horizontal alignment of the menu (`left`, `right`, `center`). - **direction:** Whether the menu opens `down` or `up`. - **offset_top / offset_left:** Pixel offsets for fine positioning. + - **pin_icon / unpin_icon:** Nerd Font glyphs for the pin button in the popup header. The button keeps the popup open and lets it be dragged when pinned. ## Authentication @@ -77,15 +94,91 @@ The widget reuses Claude Code's existing OAuth session. It reads the access toke that environment variable is set) and never logs or stores it elsewhere. If you are not signed in to Claude Code, the widget shows `--` until you sign in. +Only Claude Code itself renews the OAuth token. If it has expired (e.g. you have not used +Claude Code in a while), the usage endpoint rejects the request and the widget keeps serving +the last cached values; the `{stale}` placeholder shows a warning glyph until the token is +refreshed by running any Claude Code command. The `refresh` action forces a re-fetch but +cannot renew an expired token. + +## Refresh + +The popup header has a refresh button that forces an immediate re-fetch, bypassing `cache_ttl`. +The same action is available as the `refresh` callback for any mouse button. While the menu is +open, its sections redraw in place when fresh data arrives. A refresh is ignored while a fetch +is already in flight. + +## Token history + +When `token_history.enabled` is `true`, the popup gains a **Tokens** section with a +Session / Today / Week / Month / Year toggle, the selected period's total, and an optional +usage graph. The same totals are available on the bar via the `{*_tokens}` placeholders. + +The data comes from Claude Code's own session transcripts (`~/.claude/projects/**/*.jsonl`): +no API key and no network. Only numeric token counts, timestamps, the model name and the +session id are read; message content is never touched. The scan is incremental (a file is +re-parsed only when its size or mtime changes) and runs off the UI thread. + +```yaml + token_history: + enabled: true + default_period: "today" # session | today | week | month | year + show_graph: true + show_graph_grid: false + week_starts_on: "monday" # monday | sunday + count_cache_read: true # false counts only new input/output/cache-creation + scan_interval: 120 # seconds between transcript scans (30–3600) +``` + +- **enabled:** Turn the Tokens section and `{*_tokens}` placeholders on. +- **default_period:** Which period is selected when the menu first opens. +- **show_graph / show_graph_grid:** Show a usage graph for the selected period, with an optional grid. +- **show_models:** Show a per-model token breakdown in the Tokens section, following the selected period. Top 5 models, computed from local transcripts. +- **week_starts_on:** First day of the week for the Week total. +- **count_cache_read:** Whether cache-read tokens count toward the totals. They dominate for heavy users; set `false` for "new work only". +- **scan_interval:** Seconds between transcript scans (30–3600). + +> Session is the most recently active session's whole lifetime, so it can span days and may exceed Today. + +## API status + +When `status.enabled` is `true`, the widget can show a coloured dot reflecting the public +Claude API status (`status.claude.com`, no authentication). Use the `{status}` placeholder on +the bar, and/or an optional status line in the popup header (`show_in_menu`). + +```yaml + status: + enabled: true + show_in_menu: true + icon: "●" # any glyph; coloured by .status. + poll_interval: 300 # seconds between status checks (60–3600) +``` + +- **enabled:** Turn the `{status}`/`{status_text}` placeholders and the menu status line on. +- **show_in_menu:** Show a dot + description line in the popup header. +- **icon:** The glyph used for the dot. Its colour comes from the `.status.` class. +- **poll_interval:** Seconds between status checks (60–3600). + ## Widget Style ```css .claude-usage {} .claude-usage .widget-container {} .claude-usage .icon {} .claude-usage .label {} +.claude-usage .stale {} /* warning glyph while the OAuth token is expired */ +.claude-usage .status {} /* {status} dot on the bar */ +.claude-usage .status.none {} /* green / minor / major / critical / unknown */ +.claude-usage .status.minor {} +.claude-usage .status.major {} +.claude-usage .status.critical {} /* Popup menu */ .claude-usage-menu {} -.claude-usage-menu .header {} /* "Claude Usage" title */ +.claude-usage-menu .header {} /* header row (title + refresh button) */ +.claude-usage-menu .header .text {} /* "Claude Usage" title */ +.claude-usage-menu .header .refresh {} /* refresh button */ +.claude-usage-menu .header .refresh:hover {} +.claude-usage-menu .status-row {} /* status line below the header (show_in_menu) */ +.claude-usage-menu .status-row .dot {} /* coloured via .dot. */ +.claude-usage-menu .status-row .status-text {} .claude-usage-menu .section {} .claude-usage-menu .section .title {} .claude-usage-menu .section .progress {} /* progress-bar track */ @@ -99,6 +192,25 @@ signed in to Claude Code, the widget shows `--` until you sign in. .claude-usage-menu .section .footer .percent.medium {} .claude-usage-menu .section .footer .percent.high {} .claude-usage-menu .section .date {} /* absolute reset timestamp */ +/* Token history section (token_history.enabled) */ +.claude-usage-menu .section.tokens {} +.claude-usage-menu .section .period-toggle {} +.claude-usage-menu .section .period-btn {} +.claude-usage-menu .section .period-btn.active {} +.claude-usage-menu .section .token-total {} +.claude-usage-menu .section.tokens .model-usage {} /* per-model breakdown container */ +.claude-usage-menu .section.tokens .model-usage .title {} /* "Models" header */ +.claude-usage-menu .section.tokens .model-rows {} /* the per-model bar rows */ +.claude-usage-menu .section.tokens .model-name {} +.claude-usage-menu .section.tokens .model-total {} +.claude-usage-menu .section.tokens .model-rows .progress.model-0 .fill {} /* bar accent 0..4 */ +.claude-usage-menu .section.tokens .model-rows .progress.model-1 .fill {} +.claude-usage-menu .section.tokens .model-rows .progress.model-2 .fill {} +.claude-usage-menu .section.tokens .model-rows .progress.model-3 .fill {} +.claude-usage-menu .section.tokens .model-rows .progress.model-4 .fill {} +.claude-usage-menu .header .pin-btn {} /* pin button (use font-family "Segoe Fluent Icons" for the glyphs) */ +.claude-usage-menu .header .pin-btn.pinned {} /* while pinned */ +.claude-usage-menu .section .graph-container {} ``` ## Example Style @@ -117,10 +229,20 @@ signed in to Claude Code, the widget shows `--` until you sign in. min-width: 260px; } .claude-usage-menu .header { + padding: 14px 16px 10px 16px; +} +.claude-usage-menu .header .text { color: #cdd6f4; font-size: 15px; font-weight: bold; - padding: 14px 16px 10px 16px; +} +.claude-usage-menu .header .refresh { + color: #6c7086; + font-size: 15px; + padding: 0 2px; +} +.claude-usage-menu .header .refresh:hover { + color: #fab387; } .claude-usage-menu .section { padding: 4px 16px 12px 16px; diff --git a/src/core/utils/stat_popup.py b/src/core/utils/stat_popup.py index 565c5b890..6ab4ea128 100644 --- a/src/core/utils/stat_popup.py +++ b/src/core/utils/stat_popup.py @@ -1,6 +1,6 @@ from PyQt6.QtCore import QEvent, QPointF, Qt from PyQt6.QtGui import QBrush, QColor, QLinearGradient, QPainter, QPainterPath, QPen -from PyQt6.QtWidgets import QFrame, QGridLayout, QHBoxLayout, QLabel, QPushButton, QVBoxLayout +from PyQt6.QtWidgets import QFrame, QGridLayout, QHBoxLayout, QLabel, QPushButton, QVBoxLayout, QWidget from core.utils.tooltip import set_tooltip from core.utils.utilities import PopupWidget, refresh_widget_style @@ -60,6 +60,34 @@ def closeEvent(self, event): return super().closeEvent(event) + def resizeEvent(self, event): + if self._is_pinned: + # A pinned popup keeps the position the user dragged it to. PopupWidget.resizeEvent + # re-anchors to the bar on every resize, which is what snaps a pinned popup back on a + # refresh; resize the content to fill the new size but skip the re-anchor. + self._popup_content.setGeometry(0, 0, self.width(), self.height()) + QWidget.resizeEvent(self, event) + return + super().resizeEvent(event) + + +def create_pin_button(popup: PopupWidget, pin_icon: str, unpin_icon: str) -> QPushButton: + """A checkable header button that keeps ``popup`` open and draggable while checked.""" + pin_btn = QPushButton(pin_icon) + pin_btn.setCheckable(True) + pin_btn.setProperty("class", "pin-btn") + set_tooltip(pin_btn, "Pin this window") + + def on_toggled(checked: bool): + pin_btn.setText(unpin_icon if checked else pin_icon) + pin_btn.setProperty("class", "pin-btn pinned" if checked else "pin-btn") + set_tooltip(pin_btn, "Unpin this window" if checked else "Pin this window") + refresh_widget_style(pin_btn) + popup._is_pinned = checked + + pin_btn.toggled.connect(on_toggled) + return pin_btn + class GraphWidget(QFrame): """Rolling area chart for percentage-based history data (0-100).""" @@ -236,21 +264,7 @@ def build_stat_popup( header_layout.addWidget(title_label) header_layout.addStretch() - pin_icon = menu_config.pin_icon - unpin_icon = menu_config.unpin_icon - pin_btn = QPushButton(pin_icon) - pin_btn.setCheckable(True) - pin_btn.setProperty("class", "pin-btn") - set_tooltip(pin_btn, "Pin this window") - - def on_pin_toggled(checked: bool): - pin_btn.setText(unpin_icon if checked else pin_icon) - pin_btn.setProperty("class", "pin-btn pinned" if checked else "pin-btn") - set_tooltip(pin_btn, "Pin this window" if not checked else "Unpin this window") - refresh_widget_style(pin_btn) - popup._is_pinned = checked - - pin_btn.toggled.connect(on_pin_toggled) + pin_btn = create_pin_button(popup, menu_config.pin_icon, menu_config.unpin_icon) header_layout.addWidget(pin_btn) layout.addWidget(header) diff --git a/src/core/validation/widgets/yasb/claude_usage.py b/src/core/validation/widgets/yasb/claude_usage.py index 87cf34c11..0b5ba4ee0 100644 --- a/src/core/validation/widgets/yasb/claude_usage.py +++ b/src/core/validation/widgets/yasb/claude_usage.py @@ -1,3 +1,5 @@ +from typing import Literal + from pydantic import Field from core.validation.widgets.base_model import ( @@ -21,6 +23,27 @@ class ClaudeUsageMenuConfig(CustomBaseModel): direction: str = "down" offset_top: int = 6 offset_left: int = 0 + pin_icon: str = "\ue718" + unpin_icon: str = "\ue77a" + + +class ClaudeTokenHistoryConfig(CustomBaseModel): + enabled: bool = False + default_period: Literal["session", "today", "week", "month", "year"] = "today" + show_graph: bool = False + show_graph_grid: bool = False + show_models: bool = False + week_starts_on: Literal["monday", "sunday"] = "monday" + # Cache-read tokens dominate the totals for heavy users; set false for "new work only". + count_cache_read: bool = True + scan_interval: int = Field(default=120, ge=30, le=3600) + + +class ClaudeStatusConfig(CustomBaseModel): + enabled: bool = False + show_in_menu: bool = True + icon: str = "●" # coloured via .status. CSS classes + poll_interval: int = Field(default=300, ge=60, le=3600) class ClaudeUsageConfig(CustomBaseModel): @@ -28,6 +51,12 @@ class ClaudeUsageConfig(CustomBaseModel): label_alt: str = "Claude {seven_day}%" update_interval: int = Field(default=60, ge=30, le=3600) cache_ttl: int = Field(default=120, ge=0, le=3600) + token_history: ClaudeTokenHistoryConfig = ClaudeTokenHistoryConfig() + status: ClaudeStatusConfig = ClaudeStatusConfig() + # Popup reset line per window: "relative" -> "Resets in 4h 11m", "absolute" -> "Resets on Sat @ 6:00 AM". + five_hour_reset_format: Literal["relative", "absolute"] = "relative" + seven_day_reset_format: Literal["relative", "absolute"] = "absolute" + reset_show_date: bool = True tooltip: bool = True callbacks: ClaudeUsageCallbacksConfig = ClaudeUsageCallbacksConfig() menu: ClaudeUsageMenuConfig = ClaudeUsageMenuConfig() diff --git a/src/core/widgets/services/claude_usage/claude_api.py b/src/core/widgets/services/claude_usage/claude_api.py index 5c7e12db4..b46b74bf6 100644 --- a/src/core/widgets/services/claude_usage/claude_api.py +++ b/src/core/widgets/services/claude_usage/claude_api.py @@ -23,6 +23,7 @@ "seven_raw": None, "seven_reset_iso": None, "fetched_at": 0, + "token_expired": False, } @@ -30,6 +31,24 @@ def _claude_config_dir() -> str: return os.environ.get("CLAUDE_CONFIG_DIR") or os.path.join(os.path.expanduser("~"), ".claude") +def _token_expired() -> bool: + """True when Claude Code's OAuth access token has expired (so a refresh is needed). + + Reads only the ``expiresAt`` timestamp (ms epoch); the token itself is never touched + here. Returns False when the value is missing or unreadable so a parsing quirk never + produces a false 'expired' warning. + """ + try: + cred_path = os.path.join(_claude_config_dir(), ".credentials.json") + with open(cred_path, encoding="utf-8") as f: + expires_at = json.load(f)["claudeAiOauth"].get("expiresAt") + if not expires_at: + return False + return (expires_at / 1000) < time.time() + except Exception: + return False + + def _cache_path() -> str: return str(app_data_path("claude_usage_cache.json")) @@ -57,9 +76,14 @@ def fetch_usage(cache_path: str, cache_ttl: int) -> dict[str, Any]: returned so the widget keeps showing the most recent known values. The OAuth token is read from Claude Code's credentials store and is never logged. """ + # Recomputed live on every call (independent of the usage cache) so the expired-token + # warning stays accurate even when serving a stale record. + token_expired = _token_expired() + cache = _read_cache(cache_path) now = int(time.time()) if cache and (now - int(cache.get("fetched_at", 0))) < cache_ttl: + cache["token_expired"] = token_expired return cache try: @@ -84,14 +108,19 @@ def fetch_usage(cache_path: str, cache_ttl: int) -> dict[str, Any]: "seven_raw": seven_raw, "seven_reset_iso": payload["seven_day"].get("resets_at"), "fetched_at": now, + # A successful fetch proves the token is currently valid. + "token_expired": False, } _write_cache(cache_path, record) return record except Exception as e: logger.debug("usage fetch failed: %s", e) if cache: + cache["token_expired"] = token_expired return cache - return dict(EMPTY_RECORD) + record = dict(EMPTY_RECORD) + record["token_expired"] = token_expired + return record class _UsageWorker(QThread): @@ -165,9 +194,16 @@ def release(self) -> None: self.deleteLater() def _tick(self) -> None: + self._start_worker(self._cache_ttl) + + def refresh_now(self) -> None: + """Force an immediate fetch, bypassing the cache TTL.""" + self._start_worker(0) + + def _start_worker(self, cache_ttl: int) -> None: if self._worker is not None: return # a fetch is already in flight - worker = _UsageWorker(self._cache_path, self._cache_ttl, self) + worker = _UsageWorker(self._cache_path, cache_ttl, self) worker.data_ready.connect(self._on_data) worker.finished.connect(self._on_finished) self._worker = worker diff --git a/src/core/widgets/services/claude_usage/status.py b/src/core/widgets/services/claude_usage/status.py new file mode 100644 index 000000000..a9cfdf111 --- /dev/null +++ b/src/core/widgets/services/claude_usage/status.py @@ -0,0 +1,129 @@ +"""Claude API status polling. + +Reads the public Anthropic/Claude status page (an Atlassian Statuspage) at +``https://status.claude.com/api/v2/status.json``. The ``status.indicator`` field +is one of ``none`` / ``minor`` / ``major`` / ``critical`` and maps to a coloured +dot in the widget. No authentication is required. +""" + +import json +import logging +import time +import urllib.request +from typing import Any, ClassVar + +from PyQt6.QtCore import QObject, QThread, QTimer, pyqtSignal + +logger = logging.getLogger("claude_usage") + +STATUS_URL = "https://status.claude.com/api/v2/status.json" +# Indicator values an Atlassian Statuspage can report, worst-first. +STATUS_LEVELS = ("critical", "major", "minor", "none", "unknown") + +EMPTY_STATUS: dict[str, Any] = {"indicator": "unknown", "description": "Status unavailable", "fetched_at": 0} + + +def fetch_status() -> dict[str, Any]: + """Return the current Claude API status, or an 'unknown' record on any error.""" + try: + request = urllib.request.Request(STATUS_URL, headers={"User-Agent": "yasb-claude-usage-widget"}) + with urllib.request.urlopen(request, timeout=10) as response: + payload = json.loads(response.read().decode("utf-8")) + status = payload.get("status") or {} + indicator = status.get("indicator") or "none" + if indicator not in STATUS_LEVELS: + indicator = "unknown" + return { + "indicator": indicator, + "description": status.get("description") or "", + "fetched_at": int(time.time()), + } + except Exception as e: + logger.debug("status fetch failed: %s", e) + return dict(EMPTY_STATUS, fetched_at=int(time.time())) + + +class _StatusWorker(QThread): + """Runs the (blocking) status request off the UI thread.""" + + data_ready = pyqtSignal(dict) + + def run(self) -> None: + self.data_ready.emit(fetch_status()) + + +class ClaudeStatusService(QObject): + """Shared Claude API status poller (mirrors ClaudeUsageService). + + One instance per ``poll_interval`` fetches the status page on a timer and + shares it with every widget on the same interval. The last known non-error + status is kept when a fetch fails, so a transient network blip does not flip + the dot to grey. + """ + + data_ready = pyqtSignal(dict) + + _instances: ClassVar[dict[int, ClaudeStatusService]] = {} + + @classmethod + def get_instance(cls, poll_interval_s: int) -> ClaudeStatusService: + key = int(poll_interval_s) + inst = cls._instances.get(key) + if inst is None: + inst = cls(poll_interval_s=key, _key=key) + cls._instances[key] = inst + inst._refcount += 1 + return inst + + def __init__(self, poll_interval_s: int, _key: int): + super().__init__() + self._key = _key + self._refcount = 0 + self._worker: _StatusWorker | None = None + self._data: dict[str, Any] = dict(EMPTY_STATUS) + + self._timer = QTimer(self) + self._timer.setInterval(max(int(poll_interval_s), 1) * 1000) + self._timer.timeout.connect(self._tick) + self._timer.start() + self._tick() + + def latest(self) -> dict[str, Any]: + return self._data + + def refresh_now(self) -> None: + """Force an immediate status fetch; an in-flight fetch is still deduplicated.""" + self._tick() + + def release(self) -> None: + self._refcount -= 1 + if self._refcount > 0: + return + self._timer.stop() + ClaudeStatusService._instances.pop(self._key, None) + if self._worker is not None and self._worker.isRunning(): + self._worker.finished.connect(self.deleteLater) + else: + self.deleteLater() + + def _tick(self) -> None: + if self._worker is not None: + return # a fetch is already in flight + worker = _StatusWorker(self) + worker.data_ready.connect(self._on_data) + worker.finished.connect(self._on_finished) + self._worker = worker + worker.start() + + def _on_data(self, data: dict[str, Any]) -> None: + # Keep the last known good status if this fetch failed. + if data.get("indicator") == "unknown" and self._data.get("indicator") not in (None, "unknown"): + data = dict(self._data, fetched_at=data.get("fetched_at", 0)) + self._data = data + self.data_ready.emit(data) + + def _on_finished(self) -> None: + worker = self._worker + self._worker = None + if worker is not None: + worker.deleteLater() diff --git a/src/core/widgets/services/claude_usage/token_history.py b/src/core/widgets/services/claude_usage/token_history.py new file mode 100644 index 000000000..e93808684 --- /dev/null +++ b/src/core/widgets/services/claude_usage/token_history.py @@ -0,0 +1,477 @@ +"""Local Claude Code token-usage history. + +Claude Code writes a JSONL transcript per session under +``~/.claude/projects/**/*.jsonl``. Every assistant message line carries a +``message.usage`` block with token counts. This module scans those files and +aggregates the token counts into per-day (local time), per-hour and per-session +buckets, which the widget turns into Session / Today / Week / Month / Year totals. + +Only numeric token counts, timestamps, the model name, the request speed and the +session id are read; message content is never touched. The scan is incremental: each file's +parsed contribution is cached keyed by (mtime, size), so a steady-state poll +only re-parses the session file(s) that actually changed. +""" + +import glob +import json +import logging +import os +import time +from datetime import datetime, timedelta +from typing import Any, ClassVar + +from PyQt6.QtCore import QObject, QThread, QTimer, pyqtSignal + +from core.utils.system import app_data_path +from core.widgets.services.claude_usage.claude_api import _claude_config_dir + +logger = logging.getLogger("claude_usage") + +CACHE_VERSION = 4 +# Token count order used throughout: input, output, cache_creation, cache_read. +_TOKEN_SLOTS = 4 +# Retention caps applied on every scan so the cache and per-tick merge stay bounded over time. +# Hourly only feeds the Today/Session graphs (Session spans at most 14 days); daily feeds up to +# the Year view (one calendar year). +HOURLY_RETENTION_DAYS = 15 +DAILY_RETENTION_DAYS = 400 + + +def _projects_dir() -> str: + return os.path.join(_claude_config_dir(), "projects") + + +def _history_cache_path() -> str: + return str(app_data_path("claude_token_history.json")) + + +def _read_json(path: str) -> dict[str, Any] | None: + try: + with open(path, encoding="utf-8") as f: + return json.load(f) + except Exception: + return None + + +def _write_json(path: str, data: dict[str, Any]) -> None: + try: + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f) + except Exception as e: + logger.debug("failed to write token history cache: %s", e) + + +def _add4(dst: list[int], src: list[int]) -> None: + for i in range(_TOKEN_SLOTS): + dst[i] += src[i] + + +def _merge_daily(into: dict[str, dict[str, list[int]]], src: dict[str, dict[str, list[int]]]) -> None: + for date, models in src.items(): + bucket = into.setdefault(date, {}) + for model, counts in models.items(): + slot = bucket.setdefault(model, [0, 0, 0, 0]) + _add4(slot, counts) + + +def _merge_hourly(into: dict[str, list[int]], src: dict[str, list[int]]) -> None: + for hour, counts in src.items(): + _add4(into.setdefault(hour, [0, 0, 0, 0]), counts) + + +def _merge_sessions(into: dict[str, dict[str, Any]], src: dict[str, dict[str, Any]]) -> None: + for sid, info in src.items(): + cur = into.get(sid) + if cur is None: + into[sid] = { + "t": list(info["t"]), + "models": {m: list(c) for m, c in info.get("models", {}).items()}, + "first": info["first"], + "last": info["last"], + } + else: + _add4(cur["t"], info["t"]) + models = cur.setdefault("models", {}) + for m, c in info.get("models", {}).items(): + _add4(models.setdefault(m, [0, 0, 0, 0]), c) + cur["first"] = min(cur["first"], info["first"]) + cur["last"] = max(cur["last"], info["last"]) + + +def _parse_file(path: str) -> dict[str, Any]: + """Parse one JSONL transcript into its daily + hourly + session token contribution.""" + daily: dict[str, dict[str, list[int]]] = {} + hourly: dict[str, list[int]] = {} + sessions: dict[str, dict[str, Any]] = {} + try: + with open(path, encoding="utf-8") as f: + for line in f: + try: + obj = json.loads(line) + except Exception: + continue + message = obj.get("message") + if not isinstance(message, dict): + continue + usage = message.get("usage") + if not isinstance(usage, dict): + continue + counts = [ + int(usage.get("input_tokens") or 0), + int(usage.get("output_tokens") or 0), + int(usage.get("cache_creation_input_tokens") or 0), + int(usage.get("cache_read_input_tokens") or 0), + ] + if not any(counts): + continue + iso = obj.get("timestamp") + if not iso: + continue + try: + local = datetime.fromisoformat(iso.replace("Z", "+00:00")).astimezone() + except Exception: + continue + date_key = f"{local:%Y-%m-%d}" + hour_key = f"{local:%Y-%m-%dT%H}" + tsec = local.timestamp() + model = message.get("model") or "unknown" + if usage.get("speed") == "fast": + # Fast mode keeps the same model id but bills at a premium, so track it apart. + model = f"{model}[fast]" + sid = obj.get("sessionId") or "unknown" + + slot = daily.setdefault(date_key, {}).setdefault(model, [0, 0, 0, 0]) + _add4(slot, counts) + _add4(hourly.setdefault(hour_key, [0, 0, 0, 0]), counts) + + sess = sessions.get(sid) + if sess is None: + sessions[sid] = { + "t": list(counts), + "models": {model: list(counts)}, + "first": tsec, + "last": tsec, + } + else: + _add4(sess["t"], counts) + _add4(sess["models"].setdefault(model, [0, 0, 0, 0]), counts) + sess["first"] = min(sess["first"], tsec) + sess["last"] = max(sess["last"], tsec) + except Exception as e: + logger.debug("failed to parse %s: %s", path, e) + return {"daily": daily, "hourly": hourly, "sessions": sessions} + + +def _prune(daily: dict[str, Any], hourly: dict[str, Any], now: datetime) -> None: + """Drop daily buckets older than DAILY_RETENTION_DAYS and hourly older than HOURLY_RETENTION_DAYS. + + Bucket keys are zero-padded local timestamps, so they sort chronologically and compare against + a cutoff key directly. + """ + day_cutoff = f"{now - timedelta(days=DAILY_RETENTION_DAYS):%Y-%m-%d}" + hour_cutoff = f"{now - timedelta(days=HOURLY_RETENTION_DAYS):%Y-%m-%dT%H}" + for key in [k for k in daily if k < day_cutoff]: + del daily[key] + for key in [k for k in hourly if k < hour_cutoff]: + del hourly[key] + + +def scan(cache_path: str) -> dict[str, Any]: + """Incrementally scan all session transcripts and return the merged aggregate. + + Unchanged files reuse their cached contribution; only new or modified files + (by mtime/size) are re-parsed. + """ + cache = _read_json(cache_path) or {} + prev_files: dict[str, Any] = cache.get("files", {}) if cache.get("version") == CACHE_VERSION else {} + + now = datetime.now().astimezone() + file_cutoff = now.timestamp() - DAILY_RETENTION_DAYS * 86400 + current_files: dict[str, Any] = {} + pattern = os.path.join(_projects_dir(), "**", "*.jsonl") + for path in glob.glob(pattern, recursive=True): + try: + st = os.stat(path) + except OSError: + continue + if st.st_mtime < file_cutoff: + # The whole transcript predates the daily window; skipping it keeps `files` bounded. + continue + prev = prev_files.get(path) + if prev and prev.get("mtime") == st.st_mtime and prev.get("size") == st.st_size: + current_files[path] = prev + else: + contrib = _parse_file(path) + current_files[path] = { + "mtime": st.st_mtime, + "size": st.st_size, + "daily": contrib["daily"], + "hourly": contrib["hourly"], + "sessions": contrib["sessions"], + } + + daily: dict[str, dict[str, list[int]]] = {} + hourly: dict[str, list[int]] = {} + sessions: dict[str, dict[str, Any]] = {} + for fc in current_files.values(): + _merge_daily(daily, fc["daily"]) + _merge_hourly(hourly, fc.get("hourly", {})) + _merge_sessions(sessions, fc["sessions"]) + + _prune(daily, hourly, now) + _write_json( + cache_path, + { + "version": CACHE_VERSION, + "files": current_files, + "daily": daily, + "hourly": hourly, + "sessions": sessions, + "updated_at": int(time.time()), + }, + ) + return {"daily": daily, "hourly": hourly, "sessions": sessions, "scanned_at": int(time.time())} + + +def _date_keys_in_range(start: datetime, end: datetime) -> list[str]: + """Local YYYY-MM-DD keys from start.date() through end.date(), inclusive.""" + keys = [] + day = start.date() + last = end.date() + while day <= last: + keys.append(f"{day:%Y-%m-%d}") + day += timedelta(days=1) + return keys + + +def _sum_daily(daily: dict[str, dict[str, list[int]]], date_key: str, count_cache_read: bool) -> int: + bucket = daily.get(date_key) + if not bucket: + return 0 + total = 0 + for counts in bucket.values(): + total += counts[0] + counts[1] + counts[2] + if count_cache_read: + total += counts[3] + return total + + +def _sum_hourly(hourly: dict[str, list[int]], hour_key: str, count_cache_read: bool) -> int: + counts = hourly.get(hour_key) + if not counts: + return 0 + return counts[0] + counts[1] + counts[2] + (counts[3] if count_cache_read else 0) + + +def _hour_keys_in_range(start: datetime, end: datetime) -> list[str]: + """Local YYYY-MM-DDTHH keys from start's hour through end's hour, inclusive.""" + keys = [] + cur = start.replace(minute=0, second=0, microsecond=0) + last = end.replace(minute=0, second=0, microsecond=0) + while cur <= last: + keys.append(f"{cur:%Y-%m-%dT%H}") + cur += timedelta(hours=1) + return keys + + +def _month_keys_in_range(start: datetime, end: datetime) -> list[str]: + """Local YYYY-MM keys from start's month through end's month, inclusive.""" + keys = [] + year, month = start.year, start.month + while (year, month) <= (end.year, end.month): + keys.append(f"{year:04d}-{month:02d}") + month += 1 + if month > 12: + month = 1 + year += 1 + return keys + + +def _sum_month(daily: dict[str, dict[str, list[int]]], month_key: str, count_cache_read: bool) -> int: + prefix = f"{month_key}-" + return sum(_sum_daily(daily, dk, count_cache_read) for dk in daily if dk.startswith(prefix)) + + +def _sorted_models(totals: dict[str, int]) -> list[tuple[str, int]]: + """Per-model (model_id, tokens) pairs sorted descending; drops zero totals and .""" + items = [(m, n) for m, n in totals.items() if n > 0 and m != ""] + items.sort(key=lambda kv: kv[1], reverse=True) + return items + + +def _models_in_range( + daily: dict[str, dict[str, list[int]]], start: datetime, end: datetime, count_cache_read: bool +) -> list[tuple[str, int]]: + totals: dict[str, int] = {} + for date_key in _date_keys_in_range(start, end): + for model, counts in daily.get(date_key, {}).items(): + total = counts[0] + counts[1] + counts[2] + (counts[3] if count_cache_read else 0) + totals[model] = totals.get(model, 0) + total + return _sorted_models(totals) + + +def summarize( + agg: dict[str, Any], + *, + count_cache_read: bool = True, + week_starts_on: str = "monday", + now: datetime | None = None, +) -> dict[str, Any]: + """Derive Session/Today/Week/Month/Year totals and a per-period graph series. + + Each period's series matches its window: Today is hourly (midnight to now), Session + is hourly across the session's span (capped to 14 days), Week and Month are daily, + and Year is monthly. ``now`` (local, tz-aware) is injectable for deterministic tests. + """ + now = now or datetime.now().astimezone() + daily = agg.get("daily", {}) + hourly = agg.get("hourly", {}) + sessions = agg.get("sessions", {}) + ccr = count_cache_read + + def window_total(start: datetime) -> int: + return sum(_sum_daily(daily, k, ccr) for k in _date_keys_in_range(start, now)) + + today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) + weekday = now.weekday() # Mon=0 + back = (weekday + 1) % 7 if week_starts_on == "sunday" else weekday + week_start = today_start - timedelta(days=back) + month_start = today_start.replace(day=1) + year_start = today_start.replace(month=1, day=1) + + # Session = the most recently active session id. + session_total = 0 + session_id = None + session_series: list[float] = [] + session_models: list[tuple[str, int]] = [] + if sessions: + session_id, info = max(sessions.items(), key=lambda kv: kv[1]["last"]) + t = info["t"] + session_total = t[0] + t[1] + t[2] + (t[3] if ccr else 0) + first = datetime.fromtimestamp(info["first"]).astimezone() + last = datetime.fromtimestamp(info["last"]).astimezone() + s_start = max(first, now - timedelta(days=14)) + session_series = [_sum_hourly(hourly, k, ccr) for k in _hour_keys_in_range(s_start, last)] + session_models = _sorted_models( + {m: c[0] + c[1] + c[2] + (c[3] if ccr else 0) for m, c in info.get("models", {}).items()} + ) + + series_by_period = { + "session": session_series, + "today": [_sum_hourly(hourly, k, ccr) for k in _hour_keys_in_range(today_start, now)], + "week": [_sum_daily(daily, k, ccr) for k in _date_keys_in_range(week_start, now)], + "month": [_sum_daily(daily, k, ccr) for k in _date_keys_in_range(month_start, now)], + "year": [_sum_month(daily, k, ccr) for k in _month_keys_in_range(year_start, now)], + } + models_by_period = { + "session": session_models, + "today": _models_in_range(daily, today_start, now, ccr), + "week": _models_in_range(daily, week_start, now, ccr), + "month": _models_in_range(daily, month_start, now, ccr), + "year": _models_in_range(daily, year_start, now, ccr), + } + + return { + "totals": { + "session": session_total, + "today": window_total(today_start), + "week": window_total(week_start), + "month": window_total(month_start), + "year": window_total(year_start), + }, + "series_by_period": series_by_period, + "models_by_period": models_by_period, + "session_id": session_id, + } + + +class _ScanWorker(QThread): + """Runs the (blocking) transcript scan off the UI thread.""" + + data_ready = pyqtSignal(dict) + + def __init__(self, cache_path: str, parent: Any = None): + super().__init__(parent) + self._cache_path = cache_path + + def run(self) -> None: + try: + self.data_ready.emit(scan(self._cache_path)) + except Exception as e: + logger.debug("token history scan failed: %s", e) + self.data_ready.emit({"daily": {}, "hourly": {}, "sessions": {}, "scanned_at": int(time.time())}) + + +class TokenHistoryService(QObject): + """Shared local token-history poller (mirrors ClaudeUsageService). + + One instance per ``scan_interval`` scans the transcripts on a timer and shares + the merged aggregate with every widget on the same interval. + """ + + data_ready = pyqtSignal(dict) + + _instances: ClassVar[dict[int, TokenHistoryService]] = {} + + @classmethod + def get_instance(cls, scan_interval_s: int) -> TokenHistoryService: + key = int(scan_interval_s) + inst = cls._instances.get(key) + if inst is None: + inst = cls(scan_interval_s=key, _key=key) + cls._instances[key] = inst + inst._refcount += 1 + return inst + + def __init__(self, scan_interval_s: int, _key: int): + super().__init__() + self._key = _key + self._refcount = 0 + self._cache_path = _history_cache_path() + self._worker: _ScanWorker | None = None + cached = _read_json(self._cache_path) or {} + self._data: dict[str, Any] = { + "daily": cached.get("daily", {}), + "hourly": cached.get("hourly", {}), + "sessions": cached.get("sessions", {}), + "scanned_at": cached.get("updated_at", 0), + } + + self._timer = QTimer(self) + self._timer.setInterval(max(int(scan_interval_s), 1) * 1000) + self._timer.timeout.connect(self._tick) + self._timer.start() + self._tick() + + def latest(self) -> dict[str, Any]: + return self._data + + def release(self) -> None: + self._refcount -= 1 + if self._refcount > 0: + return + self._timer.stop() + TokenHistoryService._instances.pop(self._key, None) + if self._worker is not None and self._worker.isRunning(): + self._worker.finished.connect(self.deleteLater) + else: + self.deleteLater() + + def _tick(self) -> None: + if self._worker is not None: + return # a scan is already in flight + worker = _ScanWorker(self._cache_path, self) + worker.data_ready.connect(self._on_data) + worker.finished.connect(self._on_finished) + self._worker = worker + worker.start() + + def _on_data(self, data: dict[str, Any]) -> None: + self._data = data + self.data_ready.emit(data) + + def _on_finished(self) -> None: + worker = self._worker + self._worker = None + if worker is not None: + worker.deleteLater() diff --git a/src/core/widgets/yasb/claude_usage.py b/src/core/widgets/yasb/claude_usage.py index bb9269633..4159a2c55 100644 --- a/src/core/widgets/yasb/claude_usage.py +++ b/src/core/widgets/yasb/claude_usage.py @@ -2,13 +2,31 @@ from datetime import UTC, datetime from typing import Any -from PyQt6.QtWidgets import QFrame, QHBoxLayout, QLabel, QVBoxLayout +from PyQt6.QtCore import Qt +from PyQt6.QtWidgets import QFrame, QGridLayout, QHBoxLayout, QLabel, QPushButton, QVBoxLayout +from core.utils.stat_popup import GraphWidget, PinnablePopup, create_pin_button from core.utils.tooltip import set_tooltip -from core.utils.utilities import PopupWidget, refresh_widget_style +from core.utils.utilities import refresh_widget_style from core.validation.widgets.yasb.claude_usage import ClaudeUsageConfig from core.widgets.base import BaseWidget from core.widgets.services.claude_usage.claude_api import ClaudeUsageService +from core.widgets.services.claude_usage.status import STATUS_LEVELS, ClaudeStatusService +from core.widgets.services.claude_usage.token_history import TokenHistoryService, summarize + +_TOKEN_PERIODS: list[tuple[str, str]] = [ + ("session", "Session"), + ("today", "Today"), + ("week", "Week"), + ("month", "Month"), + ("year", "Year"), +] +_EMPTY_TOKEN_SUMMARY: dict[str, Any] = { + "totals": {}, + "series_by_period": {}, + "models_by_period": {}, + "session_id": None, +} class UsageBar(QFrame): @@ -18,10 +36,10 @@ class UsageBar(QFrame): low values (the QProgressBar::chunk square-fill issue), and stays fully CSS-styleable. """ - def __init__(self, value: int, level: str, parent: QFrame | None = None): + def __init__(self, value: int, level: str, accent: str = "", parent: QFrame | None = None): super().__init__(parent) self._value = max(0, min(100, value)) - self.setProperty("class", f"progress {level}") + self.setProperty("class", " ".join(c for c in ("progress", level, accent) if c)) self._fill = QFrame(self) self._fill.setProperty("class", "fill") @@ -39,27 +57,58 @@ def resizeEvent(self, event): class ClaudeUsageWidget(BaseWidget): validation_schema = ClaudeUsageConfig + # Shown via {stale} when Claude Code's OAuth token has expired (nf-fa-warning). + STALE_ICON = "" + def __init__(self, config: ClaudeUsageConfig): super().__init__(class_name="claude-usage") self.config = config self._show_alt_label = False - self._menu: PopupWidget | None = None + self._menu: PinnablePopup | None = None + self._usage_frames: list[QFrame] = [] self._service_released = False self._service = ClaudeUsageService.get_instance(self.config.update_interval, self.config.cache_ttl) self._data: dict[str, Any] = self._service.latest() + # Local token-history (optional): scans Claude Code's session transcripts off-thread. + self._token_service: TokenHistoryService | None = None + self._token_summary: dict[str, Any] = dict(_EMPTY_TOKEN_SUMMARY) + self._selected_period = self.config.token_history.default_period + self._period_buttons: dict[str, QPushButton] = {} + self._token_total_label: QLabel | None = None + self._token_graph: GraphWidget | None = None + self._model_container: QFrame | None = None + self._model_layout: QGridLayout | None = None + if self.config.token_history.enabled: + self._token_service = TokenHistoryService.get_instance(self.config.token_history.scan_interval) + self._token_summary = self._summarize_tokens(self._token_service.latest()) + + # Claude API status (optional): polls the public status page off-thread. + self._status_service: ClaudeStatusService | None = None + self._status: dict[str, Any] = {"indicator": "unknown", "description": ""} + self._status_dot: QLabel | None = None + self._status_text_label: QLabel | None = None + if self.config.status.enabled: + self._status_service = ClaudeStatusService.get_instance(self.config.status.poll_interval) + self._status = self._status_service.latest() + self._init_container() self.build_widget_label(self.config.label, self.config.label_alt) self.register_callback("toggle_label", self._toggle_label) self.register_callback("toggle_menu", self._toggle_menu) + self.register_callback("refresh", self._refresh) self.callback_left = self.config.callbacks.on_left self.callback_middle = self.config.callbacks.on_middle self.callback_right = self.config.callbacks.on_right self._service.data_ready.connect(self._on_data) + if self._token_service is not None: + self._token_service.data_ready.connect(self._on_token_data) + if self._status_service is not None: + self._status_service.data_ready.connect(self._on_status_data) self.destroyed.connect(lambda *_: self._release_service()) self._update_label() @@ -71,6 +120,16 @@ def _release_service(self) -> None: self._service.release() except RuntimeError: pass + if self._token_service is not None: + try: + self._token_service.release() + except RuntimeError: + pass + if self._status_service is not None: + try: + self._status_service.release() + except RuntimeError: + pass def closeEvent(self, event): self._release_service() @@ -79,15 +138,111 @@ def closeEvent(self, event): def _on_data(self, data: dict[str, Any]) -> None: self._data = data self._update_label() + self._refresh_usage_sections() + + def _refresh(self) -> None: + self._service.refresh_now() + if self._status_service is not None: + self._status_service.refresh_now() + + def _summarize_tokens(self, agg: dict[str, Any]) -> dict[str, Any]: + th = self.config.token_history + return summarize(agg, count_cache_read=th.count_cache_read, week_starts_on=th.week_starts_on) + + def _on_token_data(self, agg: dict[str, Any]) -> None: + self._token_summary = self._summarize_tokens(agg) + self._update_label() + self._sync_token_section() + + def _on_status_data(self, data: dict[str, Any]) -> None: + self._status = data + self._update_label() + self._sync_status() + + def _status_level(self) -> str: + level = self._status.get("indicator") + return level if level in STATUS_LEVELS else "unknown" + + def _build_status_row(self) -> QFrame: + row = QFrame() + row.setProperty("class", "status-row") + row_layout = QHBoxLayout(row) + row_layout.setContentsMargins(0, 0, 0, 0) + row_layout.setSpacing(0) + self._status_dot = QLabel(self.config.status.icon) + self._status_dot.setProperty("class", "dot") + row_layout.addWidget(self._status_dot) + self._status_text_label = QLabel("") + self._status_text_label.setProperty("class", "status-text") + row_layout.addWidget(self._status_text_label) + row_layout.addStretch() + self._sync_status() + return row + + def _sync_status(self) -> None: + """Update the menu status dot colour and description in place.""" + if self._status_dot is None: + return + try: + self._status_dot.setProperty("class", f"dot {self._status_level()}") + self._status_text_label.setText(self._status.get("description", "") or "Status unavailable") + refresh_widget_style(self._status_dot, self._status_text_label) + except RuntimeError: + self._status_dot = None + self._status_text_label = None + + @staticmethod + def _fmt_tokens(value: Any) -> str: + """Compact token count, e.g. '15K', '1.2M', '218M', '3.4B', '1T'; '--' when unknown.""" + if not isinstance(value, (int, float)): + return "--" + n = int(value) + if n < 1000: + return str(n) + units = ("K", "M", "B", "T") + magnitude = 0 + scaled = n / 1000.0 + # Rounding can lift a value to 1000 of its unit (999_999 -> "1000K"); carry it up instead. + while round(scaled, 1) >= 1000 and magnitude < len(units) - 1: + scaled /= 1000.0 + magnitude += 1 + return f"{scaled:.1f}".rstrip("0").rstrip(".") + units[magnitude] def _format_values(self) -> dict[str, str]: + totals = self._token_summary.get("totals", {}) return { "five_hour": self._pct(self._data.get("five")), "seven_day": self._pct(self._data.get("seven")), "five_hour_reset": self._fmt_reset(self._data.get("five_reset_iso")), "seven_day_reset": self._fmt_reset(self._data.get("seven_reset_iso")), + "stale": self.STALE_ICON if self._data.get("token_expired") else "", + "session_tokens": self._fmt_tokens(totals.get("session")), + "today_tokens": self._fmt_tokens(totals.get("today")), + "week_tokens": self._fmt_tokens(totals.get("week")), + "month_tokens": self._fmt_tokens(totals.get("month")), + "year_tokens": self._fmt_tokens(totals.get("year")), + "status": self.config.status.icon if self.config.status.enabled else "", + "status_text": self._status.get("description", "") if self.config.status.enabled else "", } + @staticmethod + def _pretty_model(model_id: str) -> str: + """Display name derived from the id so new models need no upkeep: the name is the first + word after the 'claude-' prefix and the version is the first two short (<=2 digit) groups + (date stamps and a bracketed suffix are dropped). A '[fast]' tag, added when the request + ran in fast mode, renders as a trailing 'Fast'. Non-Claude ids are returned unchanged.""" + lowered = model_id.lower() + if not lowered.startswith("claude-"): + return model_id + is_fast = "[fast]" in lowered + parts = lowered.split("[", 1)[0].removeprefix("claude-").split("-") + name = next((p for p in parts if p.isalpha()), None) + if name is None: + return model_id + nums = [p for p in parts if p.isdigit() and len(p) <= 2] + label = f"{name.capitalize()} {'.'.join(nums[:2])}".rstrip() + return f"{label} Fast" if is_fast else label + @staticmethod def _pct(value: Any) -> str: return "--" if value is None else str(value) @@ -119,6 +274,53 @@ def _fmt_reset(iso: str | None) -> str: except Exception: return "--" + @staticmethod + def _fmt_duration(iso: str | None) -> str: + """Relative time-until-reset, e.g. '6d 21h', '4h 14m', '10m'; '--' when unknown.""" + if not iso: + return "--" + try: + target = datetime.fromisoformat(iso.replace("Z", "+00:00")) + seconds = int((target - datetime.now(UTC)).total_seconds()) + if seconds <= 0: + return "0m" + minutes = seconds // 60 + days, rem = divmod(minutes, 1440) + hours, mins = divmod(rem, 60) + if days: + return f"{days}d {hours}h" + if hours: + return f"{hours}h {mins}m" + return f"{mins}m" + except Exception: + return "--" + + @staticmethod + def _fmt_weekday(iso: str | None, with_date: bool = False) -> str: + """Absolute reset as a local weekday + time, e.g. 'Sat @ 6:00 AM'; '--' when unknown. + + With ``with_date`` the month/day is included ('Sat, Jun 13 @ 6:00 AM') so two windows + resetting on the same weekday stay distinguishable. + """ + if not iso: + return "--" + try: + local = datetime.fromisoformat(iso.replace("Z", "+00:00")).astimezone() + hour12 = local.hour % 12 or 12 + ampm = "AM" if local.hour < 12 else "PM" + day = f"{local:%a, %b} {local.day}" if with_date else f"{local:%a}" + return f"{day} @ {hour12}:{local.minute:02d} {ampm}" + except Exception: + return "--" + + def _reset_phrase(self, iso: str | None, reset_format: str) -> str: + """Reset line for the popup footer, phrased per the window's reset_format.""" + if reset_format == "absolute": + value = self._fmt_weekday(iso, with_date=self.config.reset_show_date) + return f"Resets on {value}" if value != "--" else "Reset time unknown" + value = self._fmt_duration(iso) + return f"Resets in {value}" if value != "--" else "Reset time unknown" + @staticmethod def _fmt_reset_at(iso: str | None) -> str: """Absolute local reset timestamp, e.g. '6/7/2026, 5:50:00 AM'.""" @@ -163,23 +365,32 @@ def _update_label(self) -> None: continue current_widget = active_widgets[index] if "" in part: - current_widget.setText(re.sub(r"|", "", part).strip()) + text = re.sub(r"|", "", part).strip() else: - try: - current_widget.setText(part.strip().format(**values)) - except Exception: - current_widget.setText(part.strip()) + text = part.strip() + try: + rendered = text.format(**values) + except Exception: + rendered = text + current_widget.setText(rendered) + # Hide the label when its placeholder renders empty (e.g. {stale} on a valid + # token) so it does not leave a constant gap from its own margin/spacing. + current_widget.setVisible(bool(rendered)) + if "{status}" in part: + base = current_widget.property("class") or "status" + base = " ".join(t for t in base.split() if t not in STATUS_LEVELS) + current_widget.setProperty("class", f"{base} {self._status_level()}") if self.config.tooltip: - set_tooltip( - current_widget, - f"Claude usage — 5h: {values['five_hour']}% · 7d: {values['seven_day']}%", - ) + tip = f"Claude usage — 5h: {values['five_hour']}% · 7d: {values['seven_day']}%" + if self._data.get("token_expired"): + tip += "\nToken expired — run `claude -p` to refresh" + set_tooltip(current_widget, tip) refresh_widget_style(*active_widgets) def _toggle_menu(self) -> None: self._build_menu() - def _build_section(self, title: str, value: Any, raw: Any, reset_iso: str | None) -> QFrame: + def _build_section(self, title: str, value: Any, raw: Any, reset_iso: str | None, reset_format: str) -> QFrame: level = self._level_class(value) frame = QFrame() frame.setProperty("class", "section") @@ -200,7 +411,7 @@ def _build_section(self, title: str, value: Any, raw: Any, reset_iso: str | None footer_layout.setContentsMargins(0, 0, 0, 0) footer_layout.setSpacing(0) - reset_label = QLabel(f"Resets in {self._fmt_reset(reset_iso)}") + reset_label = QLabel(self._reset_phrase(reset_iso, reset_format)) reset_label.setProperty("class", "reset") footer_layout.addWidget(reset_label) footer_layout.addStretch() @@ -217,8 +428,194 @@ def _build_section(self, title: str, value: Any, raw: Any, reset_iso: str | None return frame + def _build_token_section(self) -> QFrame: + """Tokens section: a Session/Today/Week/Month/Year toggle, the selected total, + and an optional usage graph for the selected period.""" + th = self.config.token_history + frame = QFrame() + frame.setProperty("class", "section tokens") + layout = QVBoxLayout(frame) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(0) + + if th.show_models: + self._model_container = QFrame() + self._model_container.setProperty("class", "model-usage") + container_layout = QVBoxLayout(self._model_container) + container_layout.setContentsMargins(0, 0, 0, 0) + container_layout.setSpacing(0) + model_title = QLabel("Models") + model_title.setProperty("class", "title") + container_layout.addWidget(model_title) + rows = QFrame() + rows.setProperty("class", "model-rows") + self._model_layout = QGridLayout(rows) + self._model_layout.setContentsMargins(0, 0, 0, 0) + self._model_layout.setHorizontalSpacing(8) + self._model_layout.setVerticalSpacing(4) + self._model_layout.setColumnStretch(1, 1) + container_layout.addWidget(rows) + layout.addWidget(self._model_container) + else: + self._model_container = None + self._model_layout = None + + title_label = QLabel("Tokens") + title_label.setProperty("class", "title") + layout.addWidget(title_label) + + toggle = QFrame() + toggle.setProperty("class", "period-toggle") + toggle_layout = QHBoxLayout(toggle) + toggle_layout.setContentsMargins(0, 0, 0, 0) + toggle_layout.setSpacing(0) + self._period_buttons = {} + for key, text in _TOKEN_PERIODS: + btn = QPushButton(text) + btn.setProperty("class", "period-btn") + btn.clicked.connect(lambda _=False, k=key: self._select_period(k)) + toggle_layout.addWidget(btn) + self._period_buttons[key] = btn + layout.addWidget(toggle) + + self._token_total_label = QLabel("--") + self._token_total_label.setProperty("class", "token-total") + layout.addWidget(self._token_total_label) + + if th.show_graph: + graph_container = QFrame() + graph_container.setProperty("class", "graph-container") + graph_layout = QVBoxLayout(graph_container) + graph_layout.setContentsMargins(0, 0, 0, 0) + graph_layout.setSpacing(0) + self._token_graph = GraphWidget("token-graph", show_grid=th.show_graph_grid) + graph_layout.addWidget(self._token_graph) + layout.addWidget(graph_container) + else: + self._token_graph = None + + self._sync_token_section() + return frame + + def _select_period(self, period: str) -> None: + self._selected_period = period + self._sync_token_section() + + def _sync_token_section(self) -> None: + """Refresh the token total, active toggle button, and graph in place.""" + if self._token_total_label is None: + return + totals = self._token_summary.get("totals", {}) + try: + self._token_total_label.setText(self._fmt_tokens(totals.get(self._selected_period))) + for key, btn in self._period_buttons.items(): + active = key == self._selected_period + btn.setProperty("class", "period-btn active" if active else "period-btn") + refresh_widget_style(btn) + if self._token_graph is not None: + series = self._token_summary.get("series_by_period", {}).get(self._selected_period, []) + peak = max(series) if series else 0 + normalized = [(v / peak * 100.0) if peak else 0.0 for v in series] + if len(normalized) == 1: + # Duplicate a lone sample so the graph draws a flat line rather than nothing. + normalized.append(normalized[0]) + self._token_graph.set_data(normalized) + self._sync_model_rows() + if self._menu is not None and self._menu.isVisible(): + # A period switch changes the model-row count. activate() refreshes the cached + # size hint after the rebuild, and resize() (not adjustSize, which only grows a + # visible window) makes the popup track its content both ways, so it neither + # crams the taller periods nor stretches the shorter ones. + self._menu_layout.activate() + self._menu.resize(self._menu.sizeHint()) + except RuntimeError: + # Popup (and its labels) was destroyed; references are stale until reopened. + self._token_total_label = None + self._token_graph = None + self._period_buttons = {} + + def _sync_model_rows(self) -> None: + """Rebuild the per-model bars for the selected period; hide the container when empty.""" + if self._model_layout is None: + return + try: + while self._model_layout.count(): + item = self._model_layout.takeAt(0) + widget = item.widget() + if widget is not None: + # Detach now (not just deleteLater) so the popup's size hint reflects the new + # row count synchronously, letting the caller resize the popup correctly. + widget.setParent(None) + widget.deleteLater() + models = self._token_summary.get("models_by_period", {}).get(self._selected_period, [])[:5] + self._model_container.setVisible(bool(models)) + peak = models[0][1] if models else 0 + # A grid keeps the name and total columns the same width across rows (sized to their + # widest cell at layout time), so the bar column lines up without measuring fonts. + for index, (model_id, total) in enumerate(models): + name = QLabel(self._pretty_model(model_id)) + name.setProperty("class", "model-name") + bar = UsageBar(int(total / peak * 100) if peak else 0, "", accent=f"model-{index % 5}") + total_label = QLabel(self._fmt_tokens(total)) + total_label.setProperty("class", "model-total") + total_label.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) + self._model_layout.addWidget(name, index, 0) + self._model_layout.addWidget(bar, index, 1) + self._model_layout.addWidget(total_label, index, 2) + except RuntimeError: + self._model_container = None + self._model_layout = None + + def _build_usage_frames(self) -> list[QFrame]: + return [ + self._build_section( + "5-Hour", + self._data.get("five"), + self._data.get("five_raw"), + self._data.get("five_reset_iso"), + self.config.five_hour_reset_format, + ), + self._build_section( + "7-Day", + self._data.get("seven"), + self._data.get("seven_raw"), + self._data.get("seven_reset_iso"), + self.config.seven_day_reset_format, + ), + ] + + def _add_menu_sections(self, layout: QVBoxLayout) -> None: + self._usage_frames = self._build_usage_frames() + for frame in self._usage_frames: + layout.addWidget(frame) + if self.config.token_history.enabled: + layout.addWidget(self._build_token_section()) + + def _refresh_usage_sections(self) -> None: + """Rebuild the 5h/7d sections in place when fresh usage data arrives while the popup is open. + + The token section keeps its own state and refreshes separately, so it stays put rather than + being rebuilt, which would otherwise discard the graph and per-model rows on every poll. + """ + menu = self._menu + try: + if menu is None or not menu.isVisible(): + return + layout = self._menu_layout + new_frames = self._build_usage_frames() + for old, new in zip(self._usage_frames, new_frames): + index = layout.indexOf(old) + layout.removeWidget(old) + old.hide() + old.deleteLater() + layout.insertWidget(index, new) + self._usage_frames = new_frames + menu.adjustSize() + except RuntimeError: + self._menu = None # popup was already destroyed + def _build_menu(self) -> None: - self._menu = PopupWidget( + self._menu = PinnablePopup( self, self.config.menu.blur, self.config.menu.round_corners, @@ -230,21 +627,32 @@ def _build_menu(self) -> None: layout = QVBoxLayout(self._menu) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) + self._menu_layout = layout - header = QLabel("Claude Usage") + header = QFrame() header.setProperty("class", "header") - layout.addWidget(header) + header_layout = QHBoxLayout(header) + header_layout.setContentsMargins(0, 0, 0, 0) + header_layout.setSpacing(0) - layout.addWidget( - self._build_section( - "5-Hour", self._data.get("five"), self._data.get("five_raw"), self._data.get("five_reset_iso") - ) - ) - layout.addWidget( - self._build_section( - "7-Day", self._data.get("seven"), self._data.get("seven_raw"), self._data.get("seven_reset_iso") - ) - ) + title_label = QLabel("Claude Usage") + title_label.setProperty("class", "text") + header_layout.addWidget(title_label) + header_layout.addStretch() + + refresh_btn = QPushButton("\U000f0450") + refresh_btn.setProperty("class", "refresh") + set_tooltip(refresh_btn, "Refresh now") + refresh_btn.clicked.connect(self._refresh) + header_layout.addWidget(refresh_btn) + + pin_btn = create_pin_button(self._menu, self.config.menu.pin_icon, self.config.menu.unpin_icon) + header_layout.addWidget(pin_btn) + + layout.addWidget(header) + if self.config.status.enabled and self.config.status.show_in_menu: + layout.addWidget(self._build_status_row()) + self._add_menu_sections(layout) self._menu.adjustSize() self._menu.setPosition(