60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from app.ha.client import HaClient
|
|
from app.ha.discovery import DiscoveredEntity, discover_entities
|
|
from app.ha.history import EntityHistorySeries, normalize_history_payload
|
|
from app.ha.models import HaEntitySummary
|
|
|
|
|
|
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:
|
|
raw_entity_id = item.get("entity_id")
|
|
if not isinstance(raw_entity_id, str) or "." not in raw_entity_id:
|
|
continue
|
|
entity_id = raw_entity_id
|
|
domain = entity_id.split(".", 1)[0]
|
|
raw_attributes = item.get("attributes") or {}
|
|
attributes: dict[str, Any] = raw_attributes if isinstance(raw_attributes, dict) else {}
|
|
summaries.append(
|
|
HaEntitySummary(
|
|
entity_id=entity_id,
|
|
domain=domain,
|
|
state_class=_optional_str(attributes.get("state_class")),
|
|
device_class=_optional_str(attributes.get("device_class")),
|
|
unit_of_measurement=_optional_str(attributes.get("unit_of_measurement")),
|
|
)
|
|
)
|
|
return summaries
|
|
|
|
def discover(
|
|
self,
|
|
domains: set[str] | None = None,
|
|
learnable: bool | None = None,
|
|
) -> Sequence[DiscoveredEntity]:
|
|
return discover_entities(list(self.read_entities()), domains=domains, learnable=learnable)
|
|
|
|
def read_history(
|
|
self,
|
|
entity_ids: list[str],
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> Sequence[EntityHistorySeries]:
|
|
payload = self._client.get_history(entity_ids, start_time, end_time)
|
|
return normalize_history_payload(payload)
|
|
|
|
|
|
def _optional_str(value: object) -> str | None:
|
|
if value is None or value == "":
|
|
return None
|
|
return str(value)
|