sillyhome-future/app/core/ha_client.py

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()

    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