from __future__ import annotations import json import logging from time import monotonic from dataclasses import dataclass, field from typing import Any from urllib import error, parse, request from websockets.exceptions import WebSocketException from websockets.sync.client import connect from app.config import Settings logger = logging.getLogger(__name__) SUCCESS_STATUS_CODES = {200, 201} class HomeAssistantConfigError(RuntimeError): """Raised when required Home Assistant outbound configuration is missing.""" class HomeAssistantRequestError(RuntimeError): """Raised when a Home Assistant outbound HTTP request fails.""" @dataclass(slots=True) class HomeAssistantClient: settings: Settings timeout_seconds: float | None = field(default=None) def __post_init__(self) -> None: if self.timeout_seconds is None: self.timeout_seconds = self.settings.home_assistant_timeout_seconds def is_configured(self) -> bool: return bool(self.settings.home_assistant_base_url and self.settings.home_assistant_auth_token) def publish_sensor( self, *, entity_id: str, state: str, attributes: dict[str, Any] | None = None, ) -> None: self._require_config() if not entity_id: raise ValueError("entity_id must not be empty") payload = { "entity_id": entity_id, "state": state, "attributes": attributes or {}, } self._post_json(f"/api/states/{entity_id}", payload, operation="publish_sensor") def trigger_webhook(self, *, webhook_id: str, body: Any) -> None: self._require_config() if not webhook_id: raise ValueError("webhook_id must not be empty") self._post_json(f"/api/webhook/{webhook_id}", body, operation="trigger_webhook") def discovery_registry_bindings(self, unique_ids: set[str]) -> dict[str, set[str]]: """Return HA device identifiers currently bound to MQTT unique IDs. This is deliberately a read-only WebSocket query. MQTT only confirms broker receipt; the entity/device registries are the authoritative HA observation that a discovery unload/re-add was actually processed. """ self._require_config() if not unique_ids: return {} try: deadline = monotonic() + self.timeout_seconds with connect(self._websocket_url(), open_timeout=self.timeout_seconds, close_timeout=self.timeout_seconds) as websocket: greeting = json.loads(self._websocket_recv(websocket, deadline)) if greeting.get("type") != "auth_required": raise HomeAssistantRequestError("Unexpected Home Assistant WebSocket greeting") websocket.send(json.dumps({"type": "auth", "access_token": self.settings.home_assistant_auth_token})) auth = json.loads(self._websocket_recv(websocket, deadline)) if auth.get("type") != "auth_ok": raise HomeAssistantRequestError("Home Assistant WebSocket authentication failed") entities = self._websocket_command(websocket, 1, "config/entity_registry/list", deadline) devices = self._websocket_command(websocket, 2, "config/device_registry/list", deadline) except (OSError, WebSocketException, TimeoutError, ValueError, KeyError, TypeError) as exc: raise HomeAssistantRequestError("Home Assistant registry query failed") from exc devices_by_id = { device["id"]: { identifier[1] for identifier in device.get("identifiers", []) if ( isinstance(identifier, (list, tuple)) and len(identifier) == 2 and identifier[0] == "mqtt" and isinstance(identifier[1], str) and identifier[1] ) } for device in devices if isinstance(device, dict) and isinstance(device.get("id"), str) } return { entity["unique_id"]: devices_by_id.get(entity.get("device_id"), set()) for entity in entities if entity.get("platform") == "mqtt" and entity.get("unique_id") in unique_ids } @staticmethod def _websocket_recv(websocket: Any, deadline: float) -> str: remaining = deadline - monotonic() if remaining <= 0: raise TimeoutError("Home Assistant WebSocket registry query timed out") return websocket.recv(timeout=remaining) @classmethod def _websocket_command( cls, websocket: Any, message_id: int, command: str, deadline: float ) -> list[dict[str, Any]]: websocket.send(json.dumps({"id": message_id, "type": command})) while True: response = json.loads(cls._websocket_recv(websocket, deadline)) if response.get("id") != message_id: continue if not response.get("success"): raise HomeAssistantRequestError(f"Home Assistant WebSocket {command} failed") return response.get("result", []) def _websocket_url(self) -> str: parsed = parse.urlsplit(self.settings.home_assistant_base_url) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise HomeAssistantConfigError("HOME_ASSISTANT_BASE_URL must be an HTTP(S) URL") scheme = "wss" if parsed.scheme == "https" else "ws" path = f"{parsed.path.rstrip('/')}/api/websocket" return parse.urlunsplit((scheme, parsed.netloc, path, "", "")) def _require_config(self) -> None: if self.is_configured(): return raise HomeAssistantConfigError( "Home Assistant outbound integration is not configured. " "Set HOME_ASSISTANT_BASE_URL and HOME_ASSISTANT_AUTH_TOKEN." ) def _post_json(self, path: str, payload: Any, *, operation: str) -> None: url = self._build_url(path) body = json.dumps(payload).encode("utf-8") req = request.Request(url, data=body, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {self.settings.home_assistant_auth_token}") try: with request.urlopen(req, timeout=self.timeout_seconds) as response: status_code = response.getcode() except error.HTTPError as exc: logger.warning( "Home Assistant outbound %s failed with HTTP %s for %s", operation, exc.code, url, ) raise HomeAssistantRequestError( f"Home Assistant outbound {operation} failed with HTTP {exc.code}" ) from exc except error.URLError as exc: logger.warning("Home Assistant outbound %s failed for %s: %s", operation, url, exc) raise HomeAssistantRequestError( f"Home Assistant outbound {operation} failed to reach Home Assistant" ) from exc if status_code not in SUCCESS_STATUS_CODES: logger.warning( "Home Assistant outbound %s returned unexpected status %s for %s", operation, status_code, url, ) raise HomeAssistantRequestError( f"Home Assistant outbound {operation} returned unexpected status {status_code}" ) def _build_url(self, path: str) -> str: base_url = self.settings.home_assistant_base_url.rstrip("/") quoted_path = parse.quote(path.lstrip("/"), safe="/") return f"{base_url}/{quoted_path}"