feature/ha-api-integration: lifespan und Settings für HA-Reader vorbereiten

app/ha/__init__.py

@@ -0,0 +1 @@
+# sillyhome-next.ha

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

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

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

app/main.py

@@ -1,12 +1,46 @@
-from fastapi import FastAPI
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, Depends
+
+from app.api.v1.entities import router as entities_router
+from app.ha.client import HaClient, HaClientSettings
+from app.ha.reader import HaReader
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    settings = HaClientSettings(
+        url=app.state.settings.ha_url,
+        token=app.state.settings.ha_token,
+    )
+    client = HaClient(settings=settings)
+    app.state.ha_reader = HaReader(client=client)
+    yield
+
 
 app = FastAPI(
     title="SillyHome Next API",
     description="Lokales Smart-Home-Intelligenzsystem für Home Assistant.",
     version="0.1.0",
+    lifespan=lifespan,
 )
 
 
+class Settings:
+    ha_url: str
+    ha_token: str
+
+
+app.state.settings = Settings()
+
+
+def get_ha_reader() -> HaReader:
+    return app.state.ha_reader
+
+
+app.include_router(entities_router, dependencies=[Depends(get_ha_reader)])
+
+
 @app.get("/health")
 def health() -> dict[str, str]:
     return {"status": "ok"}

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"