Backend Monitorink: serveur PNG (Claude usage + météo + HA)

backend/integrations/claude_usage.py

@@ -0,0 +1,133 @@
+"""Récupération de l'usage de l'abonnement Claude via l'endpoint OAuth `/usage`.
+
+Validé empiriquement (compte Max 5x) :
+  GET https://api.anthropic.com/api/oauth/usage
+  headers: Authorization: Bearer <token>, anthropic-beta: oauth-2025-04-20,
+           User-Agent: claude-code/<version>, Content-Type: application/json
+  réponse: { five_hour:{utilization,resets_at}, seven_day:{...},
+             seven_day_opus, seven_day_sonnet, extra_usage }
+
+Le token utilisé est un token longue durée dédié généré par `claude setup-token`
+(env MONITORINK_CLAUDE_TOKEN). Aucun refresh/écriture n'est effectué ici : en cas de 401,
+on remonte un état d'erreur pour affichage (« token à régénérer »).
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+import httpx
+
+from config import config
+
+USAGE_URL = "https://api.anthropic.com/api/oauth/usage"
+
+
+@dataclass
+class Window:
+    """Une fenêtre glissante d'usage (5h ou 7j)."""
+
+    utilization: float  # 0..100
+    resets_at: datetime | None
+
+    @property
+    def remaining_pct(self) -> float:
+        return max(0.0, 100.0 - self.utilization)
+
+    @property
+    def resets_in_human(self) -> str:
+        if not self.resets_at:
+            return ""
+        delta = self.resets_at - datetime.now(timezone.utc)
+        secs = int(delta.total_seconds())
+        if secs <= 0:
+            return "bientôt"
+        h, m = divmod(secs // 60, 60)
+        if h >= 24:
+            return f"{h // 24}j {h % 24}h"
+        if h:
+            return f"{h}h{m:02d}"
+        return f"{m}min"
+
+
+@dataclass
+class ClaudeUsage:
+    ok: bool
+    error: str | None = None
+    five_hour: Window | None = None
+    seven_day: Window | None = None
+    seven_day_opus: Window | None = None
+    seven_day_sonnet: Window | None = None
+    burn_rate: float | None = None  # tokens/min (ccusage, optionnel)
+
+
+def _parse_window(raw: dict | None) -> Window | None:
+    if not raw:
+        return None
+    util = float(raw.get("utilization", 0) or 0)
+    resets = raw.get("resets_at")
+    dt = None
+    if resets:
+        try:
+            dt = datetime.fromisoformat(str(resets).replace("Z", "+00:00"))
+        except ValueError:
+            dt = None
+    return Window(utilization=util, resets_at=dt)
+
+
+def _burn_rate_from_ccusage() -> float | None:
+    """Burn rate du bloc 5h actif via `ccusage blocks --json` (best-effort)."""
+    try:
+        proc = subprocess.run(
+            ["ccusage", "blocks", "--active", "--json"],
+            capture_output=True,
+            text=True,
+            timeout=20,
+        )
+        if proc.returncode != 0:
+            return None
+        data = json.loads(proc.stdout)
+        blocks = data.get("blocks") or []
+        for b in blocks:
+            if b.get("isActive"):
+                bp = b.get("burnRate") or {}
+                rate = bp.get("tokensPerMinute") or bp.get("tokensPerMinuteForIndicator")
+                return float(rate) if rate is not None else None
+    except (subprocess.SubprocessError, json.JSONDecodeError, ValueError, FileNotFoundError):
+        return None
+    return None
+
+
+async def fetch_usage() -> ClaudeUsage:
+    if not config.claude_token:
+        return ClaudeUsage(ok=False, error="MONITORINK_CLAUDE_TOKEN manquant")
+
+    headers = {
+        "Authorization": f"Bearer {config.claude_token}",
+        "anthropic-beta": "oauth-2025-04-20",
+        "User-Agent": config.claude_ua,
+        "Content-Type": "application/json",
+    }
+    try:
+        async with httpx.AsyncClient(timeout=15) as client:
+            resp = await client.get(USAGE_URL, headers=headers)
+    except httpx.HTTPError as exc:
+        return ClaudeUsage(ok=False, error=f"réseau: {exc}")
+
+    if resp.status_code == 401:
+        return ClaudeUsage(ok=False, error="token expiré — relancer `claude setup-token`")
+    if resp.status_code != 200:
+        return ClaudeUsage(ok=False, error=f"HTTP {resp.status_code}")
+
+    data = resp.json()
+    burn = _burn_rate_from_ccusage() if config.ccusage_enabled else None
+    return ClaudeUsage(
+        ok=True,
+        five_hour=_parse_window(data.get("five_hour")),
+        seven_day=_parse_window(data.get("seven_day")),
+        seven_day_opus=_parse_window(data.get("seven_day_opus")),
+        seven_day_sonnet=_parse_window(data.get("seven_day_sonnet")),
+        burn_rate=burn,
+    )

backend/integrations/homeassistant.py

@@ -0,0 +1,55 @@
+"""Statuts Home Assistant via l'API REST (`GET /api/states/<entity_id>`)."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import httpx
+
+from config import HAEntity, config
+
+
+@dataclass
+class HAState:
+    label: str
+    state: str
+    unit: str = ""
+    ok: bool = True
+
+    @property
+    def display(self) -> str:
+        if not self.ok:
+            return "—"
+        s = self.state
+        if s in ("on", "home", "open", "unlocked", "playing"):
+            s = "ON"
+        elif s in ("off", "away", "not_home", "closed", "locked", "idle", "paused"):
+            s = "OFF"
+        elif s in ("unavailable", "unknown"):
+            s = "n/d"
+        return f"{s}{(' ' + self.unit) if self.unit and s not in ('ON', 'OFF', 'n/d') else ''}"
+
+
+async def fetch_states() -> list[HAState]:
+    entities = config.ha_entities
+    if not config.ha_enabled or not entities:
+        return []
+
+    headers = {"Authorization": f"Bearer {config.ha_token}", "Content-Type": "application/json"}
+    results: list[HAState] = []
+    async with httpx.AsyncClient(timeout=15, headers=headers) as client:
+        for ent in entities:
+            results.append(await _fetch_one(client, ent))
+    return results
+
+
+async def _fetch_one(client: httpx.AsyncClient, ent: HAEntity) -> HAState:
+    url = f"{config.ha_base_url}/api/states/{ent.entity_id}"
+    try:
+        resp = await client.get(url)
+        if resp.status_code != 200:
+            return HAState(label=ent.label, state=f"HTTP {resp.status_code}", ok=False)
+        data = resp.json()
+        unit = ent.unit or data.get("attributes", {}).get("unit_of_measurement", "")
+        return HAState(label=ent.label, state=str(data.get("state", "")), unit=unit)
+    except httpx.HTTPError:
+        return HAState(label=ent.label, state="erreur", ok=False)

backend/integrations/weather.py

@@ -0,0 +1,86 @@
+"""Météo via Open-Meteo (gratuit, sans clé API)."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import httpx
+
+from config import config
+
+API_URL = "https://api.open-meteo.com/v1/forecast"
+
+# Codes WMO -> (libellé court FR, emoji). Suffisant pour un dashboard e-ink.
+WMO = {
+    0: ("Dégagé", "☀"),
+    1: ("Peu nuageux", "🌤"),
+    2: ("Nuageux", "⛅"),
+    3: ("Couvert", "☁"),
+    45: ("Brouillard", "🌫"),
+    48: ("Brouillard givrant", "🌫"),
+    51: ("Bruine légère", "🌦"),
+    53: ("Bruine", "🌦"),
+    55: ("Bruine forte", "🌦"),
+    61: ("Pluie faible", "🌧"),
+    63: ("Pluie", "🌧"),
+    65: ("Pluie forte", "🌧"),
+    71: ("Neige faible", "🌨"),
+    73: ("Neige", "🌨"),
+    75: ("Neige forte", "🌨"),
+    80: ("Averses", "🌦"),
+    81: ("Averses", "🌧"),
+    82: ("Fortes averses", "⛈"),
+    95: ("Orage", "⛈"),
+    96: ("Orage + grêle", "⛈"),
+    99: ("Orage + grêle", "⛈"),
+}
+
+
+@dataclass
+class Weather:
+    ok: bool
+    error: str | None = None
+    temp: float | None = None
+    feels_like: float | None = None
+    label: str = ""
+    icon: str = ""
+    temp_min: float | None = None
+    temp_max: float | None = None
+    precip_prob: int | None = None
+
+
+async def fetch_weather() -> Weather:
+    params = {
+        "latitude": config.weather_lat,
+        "longitude": config.weather_lon,
+        "current": "temperature_2m,apparent_temperature,weather_code",
+        "daily": "temperature_2m_max,temperature_2m_min,precipitation_probability_max",
+        "timezone": config.timezone,
+        "forecast_days": 1,
+    }
+    try:
+        async with httpx.AsyncClient(timeout=15) as client:
+            resp = await client.get(API_URL, params=params)
+            resp.raise_for_status()
+            data = resp.json()
+    except httpx.HTTPError as exc:
+        return Weather(ok=False, error=f"réseau: {exc}")
+
+    cur = data.get("current", {})
+    daily = data.get("daily", {})
+    code = int(cur.get("weather_code", -1))
+    label, icon = WMO.get(code, ("—", "·"))
+
+    def _first(key: str):
+        vals = daily.get(key) or []
+        return vals[0] if vals else None
+
+    return Weather(
+        ok=True,
+        temp=cur.get("temperature_2m"),
+        feels_like=cur.get("apparent_temperature"),
+        label=label,
+        icon=icon,
+        temp_min=_first("temperature_2m_min"),
+        temp_max=_first("temperature_2m_max"),
+        precip_prob=_first("precipitation_probability_max"),
+    )