232 lines
8.6 KiB
Python
232 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from datetime import datetime, timedelta, timezone
|
|
from threading import RLock
|
|
from typing import Any
|
|
import logging
|
|
|
|
from app.ha.exceptions import HaClientError, HaHttpError
|
|
|
|
from app.ha.client import HaClient
|
|
from app.ha.discovery import DiscoveredEntity, discover_entities
|
|
from app.ha.history import (
|
|
EntityHistorySeries,
|
|
LogbookEntry,
|
|
StateHistorySeries,
|
|
normalize_history_payload,
|
|
normalize_logbook_payload,
|
|
normalize_state_history_payload,
|
|
)
|
|
from app.ha.models import HaAutomationSummary, HaEntitySummary
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HaReader:
|
|
def __init__(self, client: HaClient) -> None:
|
|
self._client = client
|
|
self._automation_cache: list[
|
|
tuple[HaAutomationSummary, dict[str, object]]
|
|
] = []
|
|
self._automation_cache_at: datetime | None = None
|
|
self._automation_cache_lock = RLock()
|
|
|
|
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=_optional_str(item.get("state")),
|
|
last_changed=_optional_datetime(item.get("last_changed")),
|
|
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 read_state_history(
|
|
self,
|
|
entity_ids: list[str],
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> Sequence[StateHistorySeries]:
|
|
payload = self._client.get_history(entity_ids, start_time, end_time)
|
|
return normalize_state_history_payload(payload)
|
|
|
|
def read_logbook(
|
|
self,
|
|
entity_id: str,
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> Sequence[LogbookEntry]:
|
|
payload = self._client.get_logbook(entity_id, start_time, end_time)
|
|
return normalize_logbook_payload(payload, entity_id)
|
|
|
|
def call_service(
|
|
self,
|
|
domain: str,
|
|
service: str,
|
|
service_data: dict[str, object],
|
|
) -> Sequence[object]:
|
|
return self._client.call_service(domain, service, service_data)
|
|
|
|
def find_automations_for_entity(
|
|
self,
|
|
entity_id: str,
|
|
) -> list[HaAutomationSummary]:
|
|
current_states = {
|
|
raw_entity_id: item.get("state") == "on"
|
|
for item in self._client.list_entities()
|
|
if isinstance((raw_entity_id := item.get("entity_id")), str)
|
|
and raw_entity_id.startswith("automation.")
|
|
}
|
|
matches = [
|
|
summary.model_copy(
|
|
update={
|
|
"enabled": current_states.get(
|
|
summary.entity_id,
|
|
summary.enabled,
|
|
)
|
|
}
|
|
)
|
|
for summary, config in self._read_automation_configs()
|
|
if _contains_exact_value(config, entity_id)
|
|
]
|
|
return sorted(matches, key=lambda item: item.entity_id)
|
|
|
|
def _read_automation_configs(
|
|
self,
|
|
) -> list[tuple[HaAutomationSummary, dict[str, object]]]:
|
|
now = datetime.now(timezone.utc)
|
|
with self._automation_cache_lock:
|
|
if (
|
|
self._automation_cache_at is not None
|
|
and now - self._automation_cache_at < timedelta(minutes=10)
|
|
):
|
|
return list(self._automation_cache)
|
|
configs: list[tuple[HaAutomationSummary, dict[str, object]]] = []
|
|
for item in self._client.list_entities():
|
|
raw_entity_id = item.get("entity_id")
|
|
if not isinstance(raw_entity_id, str) or not raw_entity_id.startswith(
|
|
"automation."
|
|
):
|
|
continue
|
|
attributes = item.get("attributes")
|
|
if not isinstance(attributes, dict):
|
|
continue
|
|
config_id = attributes.get("id")
|
|
if not isinstance(config_id, str) or not config_id:
|
|
continue
|
|
try:
|
|
config = self._client.get_automation_config(config_id)
|
|
except HaHttpError as exc:
|
|
if exc.status_code == 404:
|
|
logger.info(
|
|
"Automation config not exposed by Home Assistant for %s.",
|
|
raw_entity_id,
|
|
)
|
|
continue
|
|
logger.warning(
|
|
"Automation config unavailable for %s: %s",
|
|
raw_entity_id,
|
|
exc,
|
|
)
|
|
continue
|
|
except (HaClientError, ValueError) as exc:
|
|
logger.warning(
|
|
"Automation config unavailable for %s: %s",
|
|
raw_entity_id,
|
|
exc,
|
|
)
|
|
continue
|
|
configs.append(
|
|
(
|
|
HaAutomationSummary(
|
|
entity_id=raw_entity_id,
|
|
config_id=config_id,
|
|
friendly_name=str(
|
|
attributes.get("friendly_name") or raw_entity_id
|
|
),
|
|
enabled=item.get("state") == "on",
|
|
),
|
|
config,
|
|
)
|
|
)
|
|
self._automation_cache = configs
|
|
self._automation_cache_at = now
|
|
return list(configs)
|
|
|
|
|
|
def _optional_str(value: object) -> str | None:
|
|
if value is None or value == "":
|
|
return None
|
|
return str(value)
|
|
|
|
|
|
def _optional_datetime(value: object) -> datetime | None:
|
|
if not isinstance(value, str) or not value:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
return parsed if parsed.tzinfo is not None else None
|
|
|
|
|
|
def _contains_exact_value(value: object, expected: str) -> bool:
|
|
if value == expected:
|
|
return True
|
|
if isinstance(value, dict):
|
|
return any(_contains_exact_value(item, expected) for item in value.values())
|
|
if isinstance(value, list):
|
|
return any(_contains_exact_value(item, expected) for item in value)
|
|
return False
|