#!/usr/bin/env python3 """Read-only import for the HA "api" adapter (see DESIGN.md, "Deploy path: adapter per instance", and instances.yaml's `ken` entry). For instances with no filesystem/SSH access (HAOS), this pulls whatever the HA REST + WebSocket API exposes: - automations, scripts, scenes — REST, one GET per object (`/api/config//config/`), split one-file-per-object exactly like the docker-exec adapter's automations.yaml/scripts.yaml/scenes.yaml handling (scripts/ha/lib/split.py), so both adapters produce the same config//{automations,scripts,scenes}/ layout. - dashboards, area/entity registries, input_* helpers — WebSocket, list/get commands only. Never a mutating command. Full /config import is out of scope here (HAOS has no SSH) — see DESIGN.md. Entities without a storage-backed id (YAML-configured automations/scenes) can't be fetched via `/api/config/.../config/` — they are skipped and noted in the report, never silently dropped. """ import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import ha_api # noqa: E402 from split import write_split # noqa: E402 INPUT_HELPER_DOMAINS = ( "input_boolean", "input_number", "input_text", "input_datetime", "input_select", "input_button", ) def entity_domain(entity_id): return entity_id.split(".", 1)[0] def entity_object_id(entity_id): return entity_id.split(".", 1)[1] def new_report(): return { "automations_imported": 0, "automations_skipped_yaml": [], "automations_skipped_missing": [], "scripts_imported": 0, "scripts_skipped_missing": [], "scenes_imported": 0, "scenes_skipped_yaml": [], "scenes_skipped_missing": [], "dashboards_imported": 0, "dashboards_skipped": [], "helpers_imported": {}, "websocket_error": None, } def import_automations(client, states, config_out_dir, report): entities = [s for s in states if entity_domain(s["entity_id"]) == "automation"] objects_by_id = {} for ent in entities: auto_id = ent.get("attributes", {}).get("id") if auto_id is None: report["automations_skipped_yaml"].append(ent["entity_id"]) continue cfg = client.get(f"/api/config/automation/config/{auto_id}") if cfg is None: report["automations_skipped_missing"].append(str(auto_id)) continue objects_by_id[str(auto_id)] = cfg write_split(objects_by_id, os.path.join(config_out_dir, "automations")) report["automations_imported"] = len(objects_by_id) def import_scripts(client, states, config_out_dir, report): entities = [s for s in states if entity_domain(s["entity_id"]) == "script"] objects_by_key = {} for ent in entities: object_id = entity_object_id(ent["entity_id"]) cfg = client.get(f"/api/config/script/config/{object_id}") if cfg is None: report["scripts_skipped_missing"].append(object_id) continue objects_by_key[object_id] = cfg write_split(objects_by_key, os.path.join(config_out_dir, "scripts")) report["scripts_imported"] = len(objects_by_key) def import_scenes(client, states, config_out_dir, report): entities = [s for s in states if entity_domain(s["entity_id"]) == "scene"] objects_by_id = {} for ent in entities: scene_id = ent.get("attributes", {}).get("id") if scene_id is None: report["scenes_skipped_yaml"].append(ent["entity_id"]) continue cfg = client.get(f"/api/config/scene/config/{scene_id}") if cfg is None: report["scenes_skipped_missing"].append(str(scene_id)) continue objects_by_id[str(scene_id)] = cfg write_split(objects_by_id, os.path.join(config_out_dir, "scenes")) report["scenes_imported"] = len(objects_by_id) def import_dashboards_and_registries(base_url, token, states, storage_out_dir, report): """WebSocket-only exports: dashboards, area/entity registries, input_* helpers. Written through split.write_split() like the config-side objects above, so a domain/dashboard that disappears between runs has its stale export file cleaned up too (idempotent: unchanged instance -> zero diff). """ try: import ha_ws except ImportError as exc: report["websocket_error"] = str(exc) return documents = {} try: with ha_ws.HaWsClient(base_url, token) as ws: try: default_config = ws.command("lovelace/config") except ha_ws.HaWsError as exc: report["dashboards_skipped"].append(("", str(exc))) else: documents["lovelace"] = {"key": "lovelace", "data": {"config": default_config}} report["dashboards_imported"] += 1 dashboards = ws.command("lovelace/dashboards/list") documents["lovelace_dashboards"] = { "key": "lovelace_dashboards", "data": {"items": dashboards}, } for dash in dashboards: dash_id = dash["id"] if dash.get("mode") != "storage": report["dashboards_skipped"].append( (dash_id, f"mode={dash.get('mode')!r} is file-based, not reachable via api") ) continue try: cfg = ws.command("lovelace/config", url_path=dash["url_path"]) except ha_ws.HaWsError as exc: report["dashboards_skipped"].append((dash_id, str(exc))) continue documents[f"lovelace.{dash_id}"] = { "key": f"lovelace.{dash_id}", "data": {"config": cfg}, } report["dashboards_imported"] += 1 resources = ws.command("lovelace/resources") documents["lovelace_resources"] = { "key": "lovelace_resources", "data": {"items": resources}, } areas = ws.command("config/area_registry/list") documents["core.area_registry"] = { "key": "core.area_registry", "data": {"areas": areas}, } entity_registry = ws.command("config/entity_registry/list") documents["core.entity_registry"] = { "key": "core.entity_registry", "data": {"entities": entity_registry}, } present_domains = {entity_domain(s["entity_id"]) for s in states} for domain in INPUT_HELPER_DOMAINS: if domain not in present_domains: continue items = ws.command(f"{domain}/list") documents[domain] = {"key": domain, "data": {"items": items}} report["helpers_imported"][domain] = len(items) except Exception as exc: # noqa: BLE001 - surface any transport/handshake error verbatim report["websocket_error"] = str(exc) return write_split(documents, storage_out_dir) def print_report(report): print("-> api import report:", file=sys.stderr) print( f" automations: {report['automations_imported']} imported, " f"{len(report['automations_skipped_yaml'])} skipped (yaml, no storage id), " f"{len(report['automations_skipped_missing'])} skipped (config not found)", file=sys.stderr, ) for eid in report["automations_skipped_yaml"]: print(f" yaml automation (not importable via api): {eid}", file=sys.stderr) for eid in report["automations_skipped_missing"]: print(f" automation id {eid}: config not found (404)", file=sys.stderr) print( f" scripts: {report['scripts_imported']} imported, " f"{len(report['scripts_skipped_missing'])} skipped (config not found)", file=sys.stderr, ) for object_id in report["scripts_skipped_missing"]: print(f" script {object_id}: config not found (404)", file=sys.stderr) print( f" scenes: {report['scenes_imported']} imported, " f"{len(report['scenes_skipped_yaml'])} skipped (yaml, no storage id), " f"{len(report['scenes_skipped_missing'])} skipped (config not found)", file=sys.stderr, ) for eid in report["scenes_skipped_yaml"]: print(f" yaml scene (not importable via api): {eid}", file=sys.stderr) for eid in report["scenes_skipped_missing"]: print(f" scene id {eid}: config not found (404)", file=sys.stderr) if report["websocket_error"]: print( f" dashboards/registries/helpers: SKIPPED — {report['websocket_error']}", file=sys.stderr, ) return print(f" dashboards: {report['dashboards_imported']} imported", file=sys.stderr) for dash_id, reason in report["dashboards_skipped"]: print(f" skipped dashboard '{dash_id}': {reason}", file=sys.stderr) for domain, count in report["helpers_imported"].items(): print(f" {domain}: {count} items", file=sys.stderr) def main(argv): if len(argv) != 5: print( "usage: import_api.py ", file=sys.stderr, ) return 2 _, base_url, token_path, config_out_dir, storage_out_dir = argv try: token = ha_api.read_token(token_path) except ha_api.HaApiError as exc: print(f"error: {exc}", file=sys.stderr) return 1 client = ha_api.Client(base_url, token) try: states = client.get("/api/states") except Exception as exc: # noqa: BLE001 - surface any transport error verbatim print(f"error: fetching /api/states failed: {exc}", file=sys.stderr) return 1 report = new_report() try: import_automations(client, states, config_out_dir, report) import_scripts(client, states, config_out_dir, report) import_scenes(client, states, config_out_dir, report) except Exception as exc: # noqa: BLE001 - surface any transport error verbatim print(f"error: importing automations/scripts/scenes failed: {exc}", file=sys.stderr) return 1 import_dashboards_and_registries(base_url, token, states, storage_out_dir, report) print_report(report) return 1 if report["websocket_error"] else 0 if __name__ == "__main__": raise SystemExit(main(sys.argv))