40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from typing import Any
|
|
|
|
from app.ha.client import HaClient
|
|
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:
|
|
entity_id = item.get("entity_id", "")
|
|
if "." not in entity_id:
|
|
continue
|
|
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 _optional_str(value: object) -> str | None:
|
|
if value is None or value == "":
|
|
return None
|
|
return str(value)
|