first commit

scripts/generate_ha_migration.py

@@ -0,0 +1,161 @@
+#!/usr/bin/env python3
+"""Generate a Home Assistant entity-ID takeover mapping for GREE Controller."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+ENTITY_RE = re.compile(r"^climate\.[a-z0-9_]+$")
+
+
+def request_json(url: str, token: str = "") -> Any:
+    headers = {"Accept": "application/json"}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    request = urllib.request.Request(url, headers=headers)
+    try:
+        with urllib.request.urlopen(request, timeout=10) as response:
+            return json.load(response)
+    except urllib.error.HTTPError as err:
+        detail = err.read().decode("utf-8", errors="replace")
+        raise RuntimeError(f"HTTP {err.code}: {detail or err.reason}") from err
+    except urllib.error.URLError as err:
+        raise RuntimeError(f"Connection failed: {err.reason}") from err
+
+
+def normalize_url(value: str) -> str:
+    value = value.strip().rstrip("/")
+    parsed = urllib.parse.urlparse(value)
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        raise ValueError(f"Invalid URL: {value}")
+    return value
+
+
+def parse_mapping(value: str) -> tuple[str, str]:
+    if "=" not in value:
+        raise argparse.ArgumentTypeError("mapping must be ENTITY_ID=CONTROLLER_DEVICE_ID")
+    entity_id, device_id = (part.strip() for part in value.split("=", 1))
+    if not ENTITY_RE.fullmatch(entity_id):
+        raise argparse.ArgumentTypeError(f"invalid climate entity ID: {entity_id}")
+    if not device_id:
+        raise argparse.ArgumentTypeError("controller device ID cannot be empty")
+    return entity_id, device_id
+
+
+def validate_ha_entity(base_url: str, token: str, entity_id: str) -> dict[str, Any]:
+    encoded = urllib.parse.quote(entity_id, safe="")
+    state = request_json(f"{base_url}/api/states/{encoded}", token)
+    if not isinstance(state, dict) or state.get("entity_id") != entity_id:
+        raise RuntimeError(f"Home Assistant did not return {entity_id}")
+    return state
+
+
+def controller_devices(base_url: str, token: str) -> list[dict[str, Any]]:
+    data = request_json(f"{base_url}/api/integrations/home-assistant/devices", token)
+    if not isinstance(data, list):
+        raise RuntimeError("Controller returned an invalid devices response")
+    return [item for item in data if isinstance(item, dict) and item.get("id")]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Generate /config/gree_controller_entities.json so the custom Home Assistant "
+            "integration can claim an existing climate entity ID after the old integration is unloaded."
+        )
+    )
+    parser.add_argument("--map", action="append", default=[], type=parse_mapping, metavar="ENTITY=DEVICE", help="Repeatable mapping, e.g. climate.klima_salon=gree-aabbccddeeff")
+    parser.add_argument("--entity", help="Single existing HA climate entity, e.g. climate.klima_salon")
+    parser.add_argument("--device", help="Controller device ID for --entity")
+    parser.add_argument("--controller-url", help="Optional controller URL used to validate or auto-select a single device")
+    parser.add_argument("--controller-token", default="", help="GREE Controller Home Assistant access token")
+    parser.add_argument("--ha-url", help="Optional Home Assistant URL used to validate source entities")
+    parser.add_argument("--ha-token", default="", help="Optional Home Assistant Long-Lived Access Token")
+    parser.add_argument("--output", default="home-assistant/generated/gree_controller_entities.json", help="Output mapping file")
+    args = parser.parse_args()
+
+    mappings: list[tuple[str, str]] = list(args.map)
+    devices: list[dict[str, Any]] = []
+    controller_url = normalize_url(args.controller_url) if args.controller_url else ""
+    ha_url = normalize_url(args.ha_url) if args.ha_url else ""
+
+    if controller_url:
+        devices = controller_devices(controller_url, args.controller_token)
+
+    if args.entity:
+        if not ENTITY_RE.fullmatch(args.entity):
+            parser.error("--entity must be a climate.* entity ID using lowercase letters, digits and underscores")
+        device_id = (args.device or "").strip()
+        if not device_id:
+            if len(devices) == 1:
+                device_id = str(devices[0]["id"])
+            elif not controller_url:
+                parser.error("--device is required unless --controller-url identifies exactly one device")
+            else:
+                choices = ", ".join(f"{item['id']} ({item.get('name', 'unnamed')})" for item in devices) or "none"
+                parser.error(f"--device is required because the controller exposes {len(devices)} devices: {choices}")
+        mappings.append((args.entity, device_id))
+
+    if not mappings:
+        parser.error("provide at least one --map or --entity")
+
+    by_entity: dict[str, str] = {}
+    by_device: dict[str, str] = {}
+    for entity_id, device_id in mappings:
+        if entity_id in by_entity and by_entity[entity_id] != device_id:
+            parser.error(f"duplicate entity mapping: {entity_id}")
+        if device_id in by_device and by_device[device_id] != entity_id:
+            parser.error(f"one controller device cannot claim two climate entity IDs: {device_id}")
+        by_entity[entity_id] = device_id
+        by_device[device_id] = entity_id
+
+    if devices:
+        valid_ids = {str(item["id"]) for item in devices}
+        missing = [device_id for device_id in by_device if device_id not in valid_ids]
+        if missing:
+            parser.error(f"controller device ID not found: {', '.join(missing)}")
+
+    source_metadata: dict[str, dict[str, Any]] = {}
+    if ha_url:
+        if not args.ha_token:
+            parser.error("--ha-token is required when --ha-url is used")
+        for entity_id in by_entity:
+            state = validate_ha_entity(ha_url, args.ha_token, entity_id)
+            source_metadata[entity_id] = {
+                "friendly_name": state.get("attributes", {}).get("friendly_name", ""),
+                "state": state.get("state"),
+            }
+
+    payload = {
+        "version": 1,
+        "entities": [
+            {
+                "entity_id": entity_id,
+                "device_id": device_id,
+                **({"source": source_metadata[entity_id]} if entity_id in source_metadata else {}),
+            }
+            for entity_id, device_id in sorted(by_entity.items())
+        ],
+    }
+
+    output = Path(args.output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+
+    print(f"Generated: {output}")
+    print("Copy this file to Home Assistant as /config/gree_controller_entities.json.")
+    print("Before adding GREE Controller in HA, verify the Rust controller can control the AC, then disable/remove the old GREE integration so the requested entity IDs are free.")
+    print("Automations and dashboards that reference the same entity_id can then continue using it.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())