"""Unified Clawdmeter configuration (Phase 7). Single source of truth at ``%LOCALAPPDATA%\\Clawdmeter\\config.json``, shared by the tray daemon and the FastAPI control panel. Replaces the Phase-6 ``ha_config.json`` (auto-migrated on first load). Shape:: { "version": 1, "ha": {"url": str, "token": str, "entities": [str, ...]}, "buttons": [{"id": str, "label": str, "icon": str, "action": "toggle"|"bri"|"ct", "entity": str, "value": int|None}], "settings": {"device_address": str, "autostart": bool} } The token is a secret: it lives only in this file (outside the repo, gitignored) and is NEVER logged (only its length) — same discipline as ha_client.py. The control panel masks it in API responses. """ import json import os from pathlib import Path CONFIG_VERSION = 1 DEFAULT_PORT = 8723 # FastAPI control panel — bound to 127.0.0.1 only VALID_ACTIONS = ("toggle", "bri", "ct") def _dir() -> Path: base = Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData" / "Local")) return base / "Clawdmeter" def config_path() -> Path: if override := os.environ.get("CLAWDMETER_CONFIG"): return Path(override) return _dir() / "config.json" def _legacy_ha_path() -> Path: return _dir() / "ha_config.json" def default_config() -> dict: return { "version": CONFIG_VERSION, "ha": {"url": "", "token": "", "entities": []}, "buttons": [], "settings": {"device_address": "", "autostart": False, "low_battery_pct": 15}, } def _migrate_legacy(cfg: dict) -> dict: """Import the Phase-6 ha_config.json into the unified config when the unified HA section is still empty. Non-destructive: the old file is left in place.""" legacy = _legacy_ha_path() if cfg["ha"]["url"] or not legacy.exists(): return cfg try: # utf-8-sig: PowerShell's Out-File writes a BOM that json.loads chokes on. old = json.loads(legacy.read_text(encoding="utf-8-sig")) except (OSError, json.JSONDecodeError): return cfg cfg["ha"]["url"] = (old.get("url") or "").strip().rstrip("/") cfg["ha"]["token"] = (old.get("token") or "").strip() cfg["ha"]["entities"] = list(old.get("entities") or []) return cfg def load_config() -> dict: """Load the unified config, filling defaults for any missing section so new keys added in later versions always resolve. Never raises.""" cfg = default_config() try: loaded = json.loads(config_path().read_text(encoding="utf-8-sig")) except (OSError, json.JSONDecodeError): loaded = None if isinstance(loaded, dict): for section in ("ha", "settings"): if isinstance(loaded.get(section), dict): cfg[section].update(loaded[section]) if isinstance(loaded.get("buttons"), list): cfg["buttons"] = loaded["buttons"] if not cfg["ha"]["url"] and not cfg["ha"]["token"]: cfg = _migrate_legacy(cfg) # Normalize the HA section the same way ha_client expects it. cfg["ha"]["url"] = (cfg["ha"].get("url") or "").strip().rstrip("/") cfg["ha"]["token"] = (cfg["ha"].get("token") or "").strip() cfg["ha"]["entities"] = list(cfg["ha"].get("entities") or []) return cfg def save_config(cfg: dict) -> None: """Persist atomically (temp file + os.replace) so a crash mid-write can't leave a truncated config.""" path = config_path() path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(path.name + ".tmp") tmp.write_text(json.dumps(cfg, ensure_ascii=False, indent=2), encoding="utf-8") os.replace(tmp, path) def ha_settings(cfg: dict | None = None) -> dict | None: """Return {url, token, entities} when HA is configured (url + real token), else None — the shape the daemon needs to build an HAClient. Logging of the token is the caller's responsibility (ha_client logs only the length).""" if cfg is None: cfg = load_config() ha = cfg.get("ha", {}) url = (ha.get("url") or "").strip().rstrip("/") token = (ha.get("token") or "").strip() if not url or not token or token.startswith("PASTE_"): return None return {"url": url, "token": token, "entities": list(ha.get("entities") or [])}