162 lines
7.0 KiB
Python
Executable File
162 lines
7.0 KiB
Python
Executable File
#!/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())
|