310 lines
11 KiB
Python
310 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
import requests
|
|
|
|
from app.ha.exceptions import (
|
|
HaAuthError,
|
|
HaHttpError,
|
|
HaTimeoutError,
|
|
HaUnexpectedPayloadError,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_ENTITY_ID_PATTERN = re.compile(r"^[a-z0-9_]+\.[a-z0-9_]+$")
|
|
_SERVICE_PART_PATTERN = re.compile(r"^[a-z0-9_]+$")
|
|
_MAX_HISTORY_SECONDS = 31 * 24 * 60 * 60
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HaClientSettings:
|
|
url: str
|
|
token: str
|
|
timeout_seconds: int = 10
|
|
|
|
|
|
class HaClient:
|
|
def __init__(self, settings: HaClientSettings) -> None:
|
|
self._settings = settings
|
|
self._session = requests.Session()
|
|
self._session.headers.update({
|
|
"Authorization": f"Bearer {settings.token}",
|
|
"Content-Type": "application/json",
|
|
})
|
|
|
|
def close(self) -> None:
|
|
self._session.close()
|
|
|
|
def list_entities(self) -> list[dict[str, object]]:
|
|
payload = self._get_json("/api/states")
|
|
if not isinstance(payload, list):
|
|
raise HaUnexpectedPayloadError(
|
|
"Antwort von Home Assistant hat unerwartetes Format."
|
|
)
|
|
return payload
|
|
|
|
def get_history(
|
|
self,
|
|
entity_ids: list[str],
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> list[object]:
|
|
if not entity_ids:
|
|
raise ValueError("Mindestens eine entity_id ist erforderlich.")
|
|
if len(entity_ids) > 100:
|
|
raise ValueError("Es können höchstens 100 Entities abgefragt werden.")
|
|
if any(not _ENTITY_ID_PATTERN.fullmatch(entity_id) for entity_id in entity_ids):
|
|
raise ValueError("entity_id enthält ein ungültiges Format.")
|
|
if start_time.tzinfo is None or end_time.tzinfo is None:
|
|
raise ValueError("start_time und end_time müssen eine Zeitzone enthalten.")
|
|
if end_time <= start_time:
|
|
raise ValueError("end_time muss nach start_time liegen.")
|
|
if (end_time - start_time).total_seconds() > _MAX_HISTORY_SECONDS:
|
|
raise ValueError("History-Abfragen sind auf 31 Tage begrenzt.")
|
|
|
|
start = quote(start_time.isoformat(), safe=":+")
|
|
payload = self._get_json(
|
|
f"/api/history/period/{start}",
|
|
params={
|
|
"filter_entity_id": ",".join(entity_ids),
|
|
"end_time": end_time.isoformat(),
|
|
"minimal_response": "1",
|
|
"no_attributes": "1",
|
|
},
|
|
)
|
|
if not isinstance(payload, list):
|
|
raise HaUnexpectedPayloadError(
|
|
"History-Antwort von Home Assistant hat unerwartetes Format."
|
|
)
|
|
return payload
|
|
|
|
def get_logbook(
|
|
self,
|
|
entity_id: str,
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> list[object]:
|
|
self._validate_period([entity_id], start_time, end_time)
|
|
start = quote(start_time.isoformat(), safe=":+")
|
|
payload = self._get_json(
|
|
f"/api/logbook/{start}",
|
|
params={
|
|
"entity": entity_id,
|
|
"end_time": end_time.isoformat(),
|
|
},
|
|
)
|
|
if not isinstance(payload, list):
|
|
raise HaUnexpectedPayloadError(
|
|
"Logbook-Antwort von Home Assistant hat unerwartetes Format."
|
|
)
|
|
return payload
|
|
|
|
def get_automation_config(self, automation_id: str) -> dict[str, object]:
|
|
if not automation_id or len(automation_id) > 120:
|
|
raise ValueError("Ungültige Automation-ID.")
|
|
payload = self._get_json(
|
|
f"/api/config/automation/config/{quote(automation_id, safe='')}"
|
|
)
|
|
if not isinstance(payload, dict):
|
|
raise HaUnexpectedPayloadError(
|
|
"Automation-Konfiguration hat ein unerwartetes Format."
|
|
)
|
|
return payload
|
|
|
|
def call_service(
|
|
self,
|
|
domain: str,
|
|
service: str,
|
|
service_data: dict[str, object],
|
|
) -> list[object]:
|
|
if not _SERVICE_PART_PATTERN.fullmatch(domain):
|
|
raise ValueError("Ungültige Service-Domain.")
|
|
if not _SERVICE_PART_PATTERN.fullmatch(service):
|
|
raise ValueError("Ungültiger Service-Name.")
|
|
payload = self._post_json(f"/api/services/{domain}/{service}", service_data)
|
|
if not isinstance(payload, list):
|
|
raise HaUnexpectedPayloadError(
|
|
"Service-Antwort von Home Assistant hat unerwartetes Format."
|
|
)
|
|
return payload
|
|
|
|
def list_entity_metadata(self, entity_ids: list[str]) -> dict[str, dict[str, str | None]]:
|
|
if not entity_ids:
|
|
return {}
|
|
if any(not _ENTITY_ID_PATTERN.fullmatch(entity_id) for entity_id in entity_ids):
|
|
raise ValueError("entity_id enthält ein ungültiges Format.")
|
|
template = _metadata_template(entity_ids)
|
|
rendered = self._post_text("/api/template", {"template": template})
|
|
try:
|
|
payload = json.loads(rendered)
|
|
except json.JSONDecodeError as exc:
|
|
raise HaUnexpectedPayloadError("Entity-Metadaten konnten nicht gelesen werden.") from exc
|
|
if not isinstance(payload, list):
|
|
raise HaUnexpectedPayloadError("Entity-Metadaten haben ein unerwartetes Format.")
|
|
result: dict[str, dict[str, str | None]] = {}
|
|
for item in payload:
|
|
if not isinstance(item, dict):
|
|
raise HaUnexpectedPayloadError("Entity-Metadaten haben ein unerwartetes Format.")
|
|
entity_id = item.get("entity_id")
|
|
if not isinstance(entity_id, str) or "." not in entity_id:
|
|
raise HaUnexpectedPayloadError("Entity-Metadaten enthalten ungültige entity_id.")
|
|
result[entity_id] = {
|
|
key: _optional_string(item.get(key))
|
|
for key in ("area_id", "area_name", "device_id", "device_name")
|
|
}
|
|
return result
|
|
|
|
def _get_json(
|
|
self,
|
|
path: str,
|
|
*,
|
|
params: dict[str, str] | None = None,
|
|
) -> object:
|
|
try:
|
|
response = self._session.get(
|
|
f"{self._settings.url.rstrip('/')}{path}",
|
|
params=params,
|
|
timeout=self._settings.timeout_seconds,
|
|
)
|
|
except requests.Timeout as exc:
|
|
raise HaTimeoutError("Zeitüberschreitung beim Zugriff auf Home Assistant.") from exc
|
|
except requests.RequestException as exc:
|
|
raise HaHttpError(
|
|
getattr(getattr(exc, "response", None), "status_code", 502),
|
|
"Netzwerkfehler beim Zugriff auf Home Assistant.",
|
|
) from exc
|
|
|
|
if response.status_code in (401, 403):
|
|
raise HaAuthError(
|
|
response.status_code,
|
|
"Authentifizierung bei Home Assistant fehlgeschlagen.",
|
|
)
|
|
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.HTTPError as exc:
|
|
raise HaHttpError(
|
|
response.status_code,
|
|
"Home Assistant meldet einen Fehler.",
|
|
) from exc
|
|
|
|
try:
|
|
payload = response.json()
|
|
except ValueError as exc:
|
|
raise HaUnexpectedPayloadError(
|
|
"Antwort von Home Assistant ist kein gültiges JSON."
|
|
) from exc
|
|
|
|
return payload
|
|
|
|
def _post_json(self, path: str, payload: Any) -> object:
|
|
try:
|
|
response = self._session.post(
|
|
f"{self._settings.url.rstrip('/')}{path}",
|
|
json=payload,
|
|
timeout=self._settings.timeout_seconds,
|
|
)
|
|
except requests.Timeout as exc:
|
|
raise HaTimeoutError("Zeitüberschreitung beim Zugriff auf Home Assistant.") from exc
|
|
except requests.RequestException as exc:
|
|
raise HaHttpError(
|
|
getattr(getattr(exc, "response", None), "status_code", 502),
|
|
"Netzwerkfehler beim Zugriff auf Home Assistant.",
|
|
) from exc
|
|
if response.status_code in (401, 403):
|
|
raise HaAuthError(
|
|
response.status_code,
|
|
"Authentifizierung bei Home Assistant fehlgeschlagen.",
|
|
)
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.HTTPError as exc:
|
|
raise HaHttpError(response.status_code, "Home Assistant meldet einen Fehler.") from exc
|
|
try:
|
|
return response.json()
|
|
except ValueError as exc:
|
|
raise HaUnexpectedPayloadError(
|
|
"Antwort von Home Assistant ist kein gültiges JSON."
|
|
) from exc
|
|
|
|
def _post_text(self, path: str, payload: dict[str, str]) -> str:
|
|
try:
|
|
response = self._session.post(
|
|
f"{self._settings.url.rstrip('/')}{path}",
|
|
json=payload,
|
|
timeout=self._settings.timeout_seconds,
|
|
)
|
|
except requests.Timeout as exc:
|
|
raise HaTimeoutError("Zeitüberschreitung beim Zugriff auf Home Assistant.") from exc
|
|
except requests.RequestException as exc:
|
|
raise HaHttpError(
|
|
getattr(getattr(exc, "response", None), "status_code", 502),
|
|
"Netzwerkfehler beim Zugriff auf Home Assistant.",
|
|
) from exc
|
|
|
|
if response.status_code in (401, 403):
|
|
raise HaAuthError(
|
|
response.status_code,
|
|
"Authentifizierung bei Home Assistant fehlgeschlagen.",
|
|
)
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.HTTPError as exc:
|
|
raise HaHttpError(response.status_code, "Home Assistant meldet einen Fehler.") from exc
|
|
return response.text
|
|
|
|
@staticmethod
|
|
def _validate_period(
|
|
entity_ids: list[str],
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
) -> None:
|
|
if not entity_ids:
|
|
raise ValueError("Mindestens eine entity_id ist erforderlich.")
|
|
if len(entity_ids) > 100:
|
|
raise ValueError("Es können höchstens 100 Entities abgefragt werden.")
|
|
if any(not _ENTITY_ID_PATTERN.fullmatch(entity_id) for entity_id in entity_ids):
|
|
raise ValueError("entity_id enthält ein ungültiges Format.")
|
|
if start_time.tzinfo is None or end_time.tzinfo is None:
|
|
raise ValueError("start_time und end_time müssen eine Zeitzone enthalten.")
|
|
if end_time <= start_time:
|
|
raise ValueError("end_time muss nach start_time liegen.")
|
|
if (end_time - start_time).total_seconds() > _MAX_HISTORY_SECONDS:
|
|
raise ValueError("History-Abfragen sind auf 31 Tage begrenzt.")
|
|
|
|
|
|
def _metadata_template(entity_ids: list[str]) -> str:
|
|
ids = json.dumps(entity_ids, ensure_ascii=True)
|
|
return (
|
|
"{% set ids = "
|
|
f"{ids}"
|
|
" %}["
|
|
"{% for entity_id in ids %}"
|
|
"{% set device = device_id(entity_id) %}"
|
|
"{{ "
|
|
"{"
|
|
"\"entity_id\": entity_id,"
|
|
"\"area_id\": area_id(entity_id),"
|
|
"\"area_name\": area_name(entity_id),"
|
|
"\"device_id\": device,"
|
|
"\"device_name\": device_attr(device, 'name') if device else none"
|
|
"}"
|
|
" | tojson }}"
|
|
"{% if not loop.last %},{% endif %}"
|
|
"{% endfor %}]"
|
|
)
|
|
|
|
|
|
def _optional_string(value: object) -> str | None:
|
|
if value is None or value == "":
|
|
return None
|
|
return str(value)
|