"""Provider abstraction (v3). A ``Provider`` polls one AI coding service (Claude Code, OpenAI Codex, z.ai GLM Coding Plan, ...) and returns a normalized ``ProviderStatus`` the daemon writes to the watch. Each provider owns its own auth + usage source; the daemon loop and the firmware stay provider-agnostic — the BLE payload just carries a ``pv`` id that selects which brand theme the watch wears. The metric model mirrors Claude's: a short (5h) and long (7d/weekly) rate-limit utilization percentage plus reset timers, with today's local token/cost as a bonus when the provider can compute it. """ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass @dataclass class ProviderStatus: # Rate-limit utilization (0-100) + reset minutes — the two bars on the watch. s: float = 0.0 # short window (5h) utilization % sr: int = -1 # short window reset, minutes (-1 = unknown) w: float = 0.0 # long window (7d) utilization % wr: int = -1 # long window reset, minutes st: str = "unknown" # status string ("allowed"/"limited"/...) ok: bool = False # rate-limit data is fresh/valid this cycle # Today's local usage (optional — providers that can read it fill these in). tokens_today: int = 0 output_today: int = 0 cost_cents_today: int = 0 messages_today: int = 0 # True when the provider is configured but auth failed (expired token etc.). # Distinct from a transient network blip, which leaves ok=False without this. auth_problem: bool = False def to_payload(self) -> dict: """Compact BLE field names the firmware parser already understands.""" return { "s": self.s, "sr": self.sr, "w": self.w, "wr": self.wr, "st": self.st, "ok": self.ok, "tk": self.tokens_today, "to": self.output_today, "tc": self.cost_cents_today, "tn": self.messages_today, } class Provider(ABC): """One usage source. Subclasses set the class attributes and implement poll().""" id: str = "" # payload `pv` id: anthropic / openai / zai label: str = "" # human label ("Claude Code") accent: str = "d97757" # brand accent hex — firmware theme hint @abstractmethod async def poll(self) -> ProviderStatus: """Return the current status. MUST NOT raise on a network/auth failure — return a status with ok=False (and auth_problem=True on a genuine auth error) so the daemon loop and the watch degrade gracefully.""" raise NotImplementedError async def aclose(self) -> None: """Release any held resources (override when the provider owns a client).""" return None class StubProvider(Provider): """Placeholder for a provider id that isn't implemented yet (OpenAI/z.ai before M2/M3). Reports nothing rather than crashing when selected.""" def __init__(self, pid: str, label: str = "", accent: str = "d97757") -> None: self.id = pid self.label = label or pid self.accent = accent async def poll(self) -> ProviderStatus: return ProviderStatus(ok=False, st="not configured")