#!/usr/bin/env python3 """Home Assistant REST client for the Clawdmeter daemon (Phase 6). The watch stays "dumb": it sends a short command over BLE (which entity, which parameter, what value) and this module turns that into a Home Assistant service call over the local network. Everything HA-specific — the base URL, the long-lived access token, the controllable entities — lives on the PC, never on the device. Config lives OUTSIDE the repo at %LOCALAPPDATA%\\Clawdmeter\\ha_config.json so the token is never committed. Copy daemon/ha_config.example.json there and fill in `url` + `token`. The token is a secret: it is sent only as the HTTP Authorization header and is NEVER written to any log (mirrors how the OAuth secret is handled in the main daemon). Live self-test against your real light (after the config is filled in), run from the repo root with the venv python: .venv\\Scripts\\python.exe -m daemon.ha_client """ import asyncio import json import os from pathlib import Path import httpx # Reuse the daemon's logger so HA lines land in the same daemon.log. Importing # the daemon module is safe (the tray already imports it; it has no import-time # side effects beyond setting up the file logger). Fall back to print() when run # in isolation before packaging. try: from daemon.claude_usage_daemon_windows import log except Exception: # pragma: no cover - standalone/fallback def log(msg: str) -> None: print(msg, flush=True) # GL-C-006P advertises feature 32 (transition), so every change can EASE instead # of snapping. Keep it short: the joystick sends ~5x/sec, so a long transition # would lag behind the wrist. 0.3s reads as "smooth" without feeling laggy. DEFAULT_TRANSITION = 0.3 DEFAULT_MIN_KELVIN = 2000 DEFAULT_MAX_KELVIN = 6500 def _config_path() -> Path: if override := os.environ.get("CLAWDMETER_HA_CONFIG"): return Path(override) base = Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData" / "Local")) return base / "Clawdmeter" / "ha_config.json" def load_ha_config() -> dict | None: """Load {url, token, entities} from the local HA config file. Returns None (with a helpful log) when the file is missing or still holds the placeholder token, so the daemon runs fine with HA simply disabled. """ path = _config_path() try: # utf-8-sig: PowerShell's `Out-File -Encoding utf8` writes a BOM, which # would otherwise make json.loads choke on the first character. raw = path.read_text(encoding="utf-8-sig") except OSError: log(f"HA: no config at {path} - HA control disabled " f"(copy daemon/ha_config.example.json there)") return None try: cfg = json.loads(raw) except json.JSONDecodeError as e: log(f"HA: config at {path} is not valid JSON ({e}) — HA control disabled") return None url = (cfg.get("url") or "").strip().rstrip("/") token = (cfg.get("token") or "").strip() entities = cfg.get("entities") or [] if not url or not token or token.startswith("PASTE_"): log("HA: config present but `url`/`token` not filled in - HA control disabled") return None # Confirm load WITHOUT ever logging the token itself. log(f"HA: config loaded - url={url}, entities={entities}, token len={len(token)}") return {"url": url, "token": token, "entities": list(entities)} class HAClient: """Thin async wrapper over the Home Assistant REST API. One long-lived httpx client; calls are local and quick. Methods return True/False (or parsed state) and never raise on an HTTP/Zigbee failure — a dead HA or a flaky mesh must not crash the daemon loop. """ def __init__(self, url: str, token: str, *, timeout: float = 10.0) -> None: self._url = url.rstrip("/") self._http = httpx.AsyncClient( timeout=timeout, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, ) async def aclose(self) -> None: await self._http.aclose() async def _service(self, domain: str, service: str, data: dict) -> bool: url = f"{self._url}/api/services/{domain}/{service}" try: resp = await self._http.post(url, json=data) except httpx.HTTPError as e: log(f"HA: {domain}.{service} failed: {e}") return False if resp.status_code >= 400: # 401 (token) and 400 (bad entity) both land here. Log status + short # body only — never the request headers, which carry the token. log(f"HA: {domain}.{service} HTTP {resp.status_code}: {resp.text[:160]}") return False return True async def get_state(self, entity_id: str) -> dict | None: url = f"{self._url}/api/states/{entity_id}" try: resp = await self._http.get(url) except httpx.HTTPError as e: log(f"HA: get_state({entity_id}) failed: {e}") return None if resp.status_code >= 400: log(f"HA: get_state({entity_id}) HTTP {resp.status_code}") return None try: return resp.json() except ValueError: return None # --- light helpers -------------------------------------------------------- async def turn_off(self, entity_id: str, transition: float = DEFAULT_TRANSITION) -> bool: return await self._service("light", "turn_off", {"entity_id": entity_id, "transition": transition}) async def toggle(self, entity_id: str) -> bool: return await self._service("light", "toggle", {"entity_id": entity_id}) async def set_brightness(self, entity_id: str, pct: float, transition: float = DEFAULT_TRANSITION) -> bool: """Set brightness 0-100 %. 0 (or below) turns the light off.""" pct = max(0, min(100, round(pct))) if pct <= 0: return await self.turn_off(entity_id, transition) return await self._service("light", "turn_on", { "entity_id": entity_id, "brightness_pct": pct, "transition": transition}) async def set_color_temp(self, entity_id: str, kelvin: float, min_k: int = DEFAULT_MIN_KELVIN, max_k: int = DEFAULT_MAX_KELVIN, transition: float = DEFAULT_TRANSITION) -> bool: kelvin = int(max(min_k, min(max_k, round(kelvin)))) return await self._service("light", "turn_on", { "entity_id": entity_id, "color_temp_kelvin": kelvin, "transition": transition}) async def light_snapshot(self, entity_id: str) -> dict | None: """Read the light's current values so the watch dial starts from the real state instead of 0. Returns normalized fields, or None on failure.""" st = await self.get_state(entity_id) if not st: return None attrs = st.get("attributes", {}) bri_255 = attrs.get("brightness") # 0-255, or None when off bri_pct = round(bri_255 / 255 * 100) if isinstance(bri_255, (int, float)) else None return { "on": st.get("state") == "on", "brightness_pct": bri_pct, "color_temp_kelvin": attrs.get("color_temp_kelvin"), "min_kelvin": attrs.get("min_color_temp_kelvin", DEFAULT_MIN_KELVIN), "max_kelvin": attrs.get("max_color_temp_kelvin", DEFAULT_MAX_KELVIN), } async def _self_test() -> int: """Live smoke test: drive the first configured light so you can SEE it react, then restore its original state.""" cfg = load_ha_config() if not cfg: return 1 if not cfg["entities"]: log("HA self-test: no entities in config") return 1 eid = cfg["entities"][0] ha = HAClient(cfg["url"], cfg["token"]) try: snap = await ha.light_snapshot(eid) if snap is None: log(f"HA self-test: could not read {eid} - check url / token / entity_id") return 1 log(f"HA self-test: {eid} now -> {snap}") min_k, max_k = snap["min_kelvin"], snap["max_kelvin"] sequence = [ ("brightness 30%", lambda: ha.set_brightness(eid, 30)), ("brightness 80%", lambda: ha.set_brightness(eid, 80)), (f"warm {min_k}K", lambda: ha.set_color_temp(eid, min_k, min_k, max_k)), (f"cool {max_k}K", lambda: ha.set_color_temp(eid, max_k, min_k, max_k)), ] for label, action in sequence: ok = await action() log(f"HA self-test: {label} -> {'ok' if ok else 'FAIL'}") await asyncio.sleep(1.3) # Restore the original state. if snap["on"]: if snap["brightness_pct"]: await ha.set_brightness(eid, snap["brightness_pct"]) if snap["color_temp_kelvin"]: await ha.set_color_temp(eid, snap["color_temp_kelvin"], min_k, max_k) else: await ha.turn_off(eid) log("HA self-test: done (original state restored)") return 0 finally: await ha.aclose() if __name__ == "__main__": raise SystemExit(asyncio.run(_self_test()))