"""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 (v2):: { "version": 2, "ha": {"url": str, "token": str, "entities": [str, ...]}, "buttons": [{"id": str, "label": str, "icon": str, "action": "toggle"|"bri"|"ct", "entity": str, "value": int|None}], "providers": { # v3: multi-provider usage "anthropic": {"enabled": bool}, "openai": {"enabled": bool}, "zai": {"enabled": bool, "base_url": str, "api_key": str}, }, "active_provider": "anthropic", # which provider the watch displays "display_order": ["anthropic", "openai", "zai"], # on-watch cycle order "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. Provider API keys (e.g. z.ai) are secrets too and follow the same rule. """ import json import os from pathlib import Path CONFIG_VERSION = 2 DEFAULT_PORT = 8723 # FastAPI control panel — bound to 127.0.0.1 only VALID_ACTIONS = ("toggle", "bri", "ct") # v3 providers. Order here is the default on-watch cycle order. Anthropic needs # no extra config (it reads Claude Code's OAuth creds); z.ai needs a base URL + # API key; OpenAI reads the local Codex install. Keep this list and the per- # provider default shapes in sync with default_config(). PROVIDER_IDS = ("anthropic", "openai", "zai") 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_providers() -> dict: """Per-provider config defaults. Anthropic is enabled+active out of the box so a v1 config (no providers section) keeps behaving exactly as before.""" return { "anthropic": {"enabled": True}, "openai": {"enabled": False}, "zai": {"enabled": False, "base_url": "", "api_key": ""}, } def default_config() -> dict: return { "version": CONFIG_VERSION, "ha": {"url": "", "token": "", "entities": []}, "buttons": [], "providers": default_providers(), "active_provider": "anthropic", "display_order": list(PROVIDER_IDS), "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"] # v3 providers — deep-merge per provider so new keys added in later # versions always resolve, and a v1 config (no providers) keeps the # defaults (anthropic enabled + active) → behaviour unchanged. if isinstance(loaded.get("providers"), dict): for pid, pconf in loaded["providers"].items(): if pid in cfg["providers"] and isinstance(pconf, dict): cfg["providers"][pid].update(pconf) if isinstance(loaded.get("active_provider"), str): cfg["active_provider"] = loaded["active_provider"] if isinstance(loaded.get("display_order"), list): cfg["display_order"] = [p for p in loaded["display_order"] if p in PROVIDER_IDS] # active provider must be a known id; fall back to anthropic otherwise. if cfg["active_provider"] not in PROVIDER_IDS: cfg["active_provider"] = "anthropic" # display order must cover every provider (append any missing at the end). for pid in PROVIDER_IDS: if pid not in cfg["display_order"]: cfg["display_order"].append(pid) 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 [])} # --- v3 provider accessors ------------------------------------------------- def active_provider_id(cfg: dict | None = None) -> str: """The provider id the watch should display right now (always a valid id).""" if cfg is None: cfg = load_config() pid = cfg.get("active_provider", "anthropic") return pid if pid in PROVIDER_IDS else "anthropic" def provider_conf(cfg: dict, pid: str) -> dict: """Per-provider config dict (empty dict if the id is unknown).""" return (cfg.get("providers") or {}).get(pid, {}) def enabled_providers(cfg: dict | None = None) -> list: """Enabled provider ids in the configured on-watch cycle order.""" if cfg is None: cfg = load_config() order = cfg.get("display_order") or list(PROVIDER_IDS) return [p for p in order if p in PROVIDER_IDS and provider_conf(cfg, p).get("enabled")]