v2 Phase 6 (step 1): Home Assistant REST client + live self-test
daemon/ha_client.py: HAClient (brightness, color_temp, on/off, state snapshot) over the HA REST API with smooth `transition`; token is read from a local %LOCALAPPDATA%\Clawdmeter\ha_config.json and is never logged. Verified end-to-end against a real Gledopto GL-C-006P tunable-white dimmer. ha_config.example.json is the committed template; the real config (with the token) lives outside the repo and is .gitignored as a safety net. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wenil
@@ -0,0 +1,220 @@
|
||||
#!/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()))
|
||||
Reference in New Issue
Block a user