from __future__ import annotations import json from collections.abc import AsyncIterator from dataclasses import dataclass from datetime import datetime, timezone import httpx import websockets from app.core.models import EntityState, StateEvent @dataclass(frozen=True) class HaClientConfig: core_url: str token: str websocket_url: str | None = None timeout_seconds: float = 15.0 class FutureHaClient: def __init__(self, config: HaClientConfig) -> None: self._config = config self._core_url = config.core_url.rstrip("/") def read_states(self) -> list[EntityState]: with httpx.Client(timeout=self._config.timeout_seconds) as client: response = client.get( f"{self._core_url}/api/states", headers=self._headers, ) response.raise_for_status() payload = response.json() result: list[EntityState] = [] for item in payload if isinstance(payload, list) else []: if not isinstance(item, dict): continue entity_id = item.get("entity_id") if not isinstance(entity_id, str) or "." not in entity_id: continue attributes = item.get("attributes") attr = attributes if isinstance(attributes, dict) else {} result.append( EntityState( entity_id=entity_id, domain=entity_id.split(".", 1)[0], state=item.get("state") if isinstance(item.get("state"), str) else None, changed_at=_parse_datetime(item.get("last_changed")), area_name=attr.get("area_name") if isinstance(attr.get("area_name"), str) else None, device_id=attr.get("device_id") if isinstance(attr.get("device_id"), str) else None, friendly_name=( attr.get("friendly_name") if isinstance(attr.get("friendly_name"), str) else None ), ) ) return result def call_service( self, domain: str, service: str, service_data: dict[str, object], ) -> None: with httpx.Client(timeout=self._config.timeout_seconds) as client: response = client.post( f"{self._core_url}/api/services/{domain}/{service}", headers=self._headers, json=service_data, ) response.raise_for_status() def read_history( self, entity_ids: list[str], start_time: datetime, end_time: datetime, ) -> dict[str, list[StateEvent]]: params = { "filter_entity_id": ",".join(entity_ids), "end_time": end_time.isoformat(), "minimal_response": "1", } with httpx.Client(timeout=self._config.timeout_seconds) as client: response = client.get( f"{self._core_url}/api/history/period/{start_time.isoformat()}", headers=self._headers, params=params, ) response.raise_for_status() payload = response.json() result: dict[str, list[StateEvent]] = {entity_id: [] for entity_id in entity_ids} for series in payload if isinstance(payload, list) else []: if not isinstance(series, list): continue for item in series: if not isinstance(item, dict): continue entity_id = item.get("entity_id") if not isinstance(entity_id, str): continue result.setdefault(entity_id, []).append( StateEvent( entity_id=entity_id, new_state=item.get("state") if isinstance(item.get("state"), str) else None, changed_at=_parse_datetime( item.get("last_changed") or item.get("last_updated") ), ) ) return result async def listen_state_events(self) -> AsyncIterator[StateEvent]: websocket_url = self._config.websocket_url or _default_websocket_url(self._core_url) async with websockets.connect(websocket_url, ping_interval=None) as websocket: auth_required = json.loads(await websocket.recv()) if auth_required.get("type") != "auth_required": raise RuntimeError("Home Assistant websocket did not request authentication") await websocket.send(json.dumps({"type": "auth", "access_token": self._config.token})) auth_result = json.loads(await websocket.recv()) if auth_result.get("type") != "auth_ok": raise RuntimeError("Home Assistant websocket authentication failed") await websocket.send( json.dumps({"id": 1, "type": "subscribe_events", "event_type": "state_changed"}) ) async for raw_message in websocket: data = json.loads(raw_message) if data.get("type") != "event": continue event = data.get("event") if not isinstance(event, dict) or event.get("event_type") != "state_changed": continue event_data = event.get("data") if not isinstance(event_data, dict): continue entity_id = event_data.get("entity_id") new_state = event_data.get("new_state") if not isinstance(entity_id, str) or not isinstance(new_state, dict): continue state = new_state.get("state") attributes = new_state.get("attributes") attr = attributes if isinstance(attributes, dict) else {} yield StateEvent( entity_id=entity_id, new_state=state if isinstance(state, str) else None, changed_at=_parse_datetime( new_state.get("last_changed") or new_state.get("last_updated") ), attributes={ key: value for key, value in { "area_name": attr.get("area_name"), "device_id": attr.get("device_id"), "friendly_name": attr.get("friendly_name"), }.items() if isinstance(value, str) }, ) @property def _headers(self) -> dict[str, str]: return { "Authorization": f"Bearer {self._config.token}", "Content-Type": "application/json", } def service_for_state(domain: str, target_state: str | None) -> tuple[str, str] | None: if target_state is None: return None if domain in {"light", "switch", "fan", "humidifier"}: if target_state == "on": return domain, "turn_on" if target_state == "off": return domain, "turn_off" if domain == "cover": if target_state in {"open", "opening"}: return domain, "open_cover" if target_state in {"closed", "closing"}: return domain, "close_cover" return None def _default_websocket_url(core_url: str) -> str: stripped = core_url.rstrip("/") if stripped.endswith("/core"): return stripped.replace("http://", "ws://").replace("https://", "wss://") + "/websocket" return stripped.replace("http://", "ws://").replace("https://", "wss://") + "/api/websocket" def _parse_datetime(value: object) -> datetime: if not isinstance(value, str): return datetime.now(timezone.utc) try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return datetime.now(timezone.utc) if parsed.tzinfo is None: return parsed.replace(tzinfo=timezone.utc) return parsed