85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from datetime import datetime
|
|
from typing import Any
|
|
import logging
|
|
|
|
from app.ha.exceptions import HaClientError
|
|
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HaReader:
|
|
def __init__(self, client: HaClient) -> None:
|
|
self._client = client
|
|
|
|
def read_entities(self) -> Sequence[HaEntitySummary]:
|
|
entities = self._client.list_entities()
|
|
entity_ids = [
|
|
raw_entity_id
|
|
for item in entities
|
|
if isinstance((raw_entity_id := item.get("entity_id")), str) and "." in raw_entity_id
|
|
]
|
|
try:
|
|
metadata_by_entity = self._client.list_entity_metadata(entity_ids)
|
|
except (HaClientError, ValueError) as exc:
|
|
logger.warning("HA metadata enrichment skipped: %s", exc)
|
|
metadata_by_entity = {}
|
|
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 {}
|
|
metadata = metadata_by_entity.get(entity_id, {})
|
|
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")),
|
|
friendly_name=_optional_str(attributes.get("friendly_name")),
|
|
area_id=_optional_str(metadata.get("area_id") or attributes.get("area_id")),
|
|
area_name=_optional_str(metadata.get("area_name") or attributes.get("area_name")),
|
|
device_id=_optional_str(metadata.get("device_id") or attributes.get("device_id")),
|
|
device_name=_optional_str(
|
|
metadata.get("device_name")
|
|
or attributes.get("device_name")
|
|
or attributes.get("device")
|
|
),
|
|
)
|
|
)
|
|
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)
|