diff --git a/app/ha/client.py b/app/ha/client.py new file mode 100644 index 0000000..5f64ef9 --- /dev/null +++ b/app/ha/client.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import logging +from dataclasses import dataclass + +import requests + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class HaClientSettings: + url: str + token: str + timeout_seconds: int = 10 + + +class HaClient: + def __init__(self, settings: HaClientSettings) -> None: + self._settings = settings + self._session = requests.Session() + self._session.headers.update({ + "Authorization": f"Bearer {settings.token}", + "Content-Type": "application/json", + }) + + def list_entities(self) -> list[dict[str, object]]: + response = self._session.get( + f"{self._settings.url}/api/states", + timeout=self._settings.timeout_seconds, + ) + response.raise_for_status() + return response.json() diff --git a/app/ha/models.py b/app/ha/models.py new file mode 100644 index 0000000..bfb1c04 --- /dev/null +++ b/app/ha/models.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from pydantic import BaseModel + + +class HaState(BaseModel): + entity_id: str + state: str + attributes: dict[str, object] | None = None + last_changed: str | None = None + last_updated: str | None = None + + +class HaEntitySummary(BaseModel): + entity_id: str + domain: str + state_class: str | None = None + device_class: str | None = None + unit_of_measurement: str | None = None diff --git a/app/ha/reader.py b/app/ha/reader.py new file mode 100644 index 0000000..ccb72c0 --- /dev/null +++ b/app/ha/reader.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from collections.abc import Sequence + +from app.ha.client import HaClient +from app.ha.models import HaEntitySummary, HaState + + +class HaReader: + def __init__(self, client: HaClient) -> None: + self._client = client + + def read_entities(self) -> Sequence[HaEntitySummary]: + entities = self._client.list_entities() + summaries: list[HaEntitySummary] = [] + for item in entities: + entity_id = item.get("entity_id", "") + if "." not in entity_id: + continue + domain = entity_id.split(".", 1)[0] + attributes = item.get("attributes") or {} + summaries.append( + HaEntitySummary( + entity_id=entity_id, + domain=domain, + state_class=str(attributes.get("state_class") or ""), + device_class=str(attributes.get("device_class") or ""), + unit_of_measurement=str(attributes.get("unit_of_measurement") or ""), + ) + ) + return summaries diff --git a/tests/ha/test_ha_reader.py b/tests/ha/test_ha_reader.py new file mode 100644 index 0000000..02bbc40 --- /dev/null +++ b/tests/ha/test_ha_reader.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from app.ha.client import HaClient, HaClientSettings +from app.ha.models import HaEntitySummary, HaState +from app.ha.reader import HaReader + + +class FakeHaClient(HaClient): + def __init__(self) -> None: + super().__init__(HaClientSettings(url="http://test", token="token")) + + def list_entities(self) -> list[dict[str, object]]: + return [ + { + "entity_id": "sensor.temperature", + "state": "21.5", + "attributes": { + "state_class": "measurement", + "device_class": "temperature", + "unit_of_measurement": "°C", + }, + }, + { + "entity_id": "light.living_room", + "state": "on", + "attributes": {}, + }, + ] + + +def test_ha_reader_returns_summaries() -> None: + reader = HaReader(FakeHaClient()) + summaries = reader.read_entities() + assert len(summaries) == 2 + domains = {summary.domain for summary in summaries} + assert domains == {"sensor", "light"} + sensor = next(item for item in summaries if item.entity_id == "sensor.temperature") + assert sensor.unit_of_measurement == "°C"