32 lines
1.1 KiB
Python
32 lines
1.1 KiB
Python
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
|