56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
"""Statuts Home Assistant via l'API REST (`GET /api/states/<entity_id>`)."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import httpx
|
|
|
|
from config import HAEntity, config
|
|
|
|
|
|
@dataclass
|
|
class HAState:
|
|
label: str
|
|
state: str
|
|
unit: str = ""
|
|
ok: bool = True
|
|
|
|
@property
|
|
def display(self) -> str:
|
|
if not self.ok:
|
|
return "—"
|
|
s = self.state
|
|
if s in ("on", "home", "open", "unlocked", "playing"):
|
|
s = "ON"
|
|
elif s in ("off", "away", "not_home", "closed", "locked", "idle", "paused"):
|
|
s = "OFF"
|
|
elif s in ("unavailable", "unknown"):
|
|
s = "n/d"
|
|
return f"{s}{(' ' + self.unit) if self.unit and s not in ('ON', 'OFF', 'n/d') else ''}"
|
|
|
|
|
|
async def fetch_states() -> list[HAState]:
|
|
entities = config.ha_entities
|
|
if not config.ha_enabled or not entities:
|
|
return []
|
|
|
|
headers = {"Authorization": f"Bearer {config.ha_token}", "Content-Type": "application/json"}
|
|
results: list[HAState] = []
|
|
async with httpx.AsyncClient(timeout=15, headers=headers) as client:
|
|
for ent in entities:
|
|
results.append(await _fetch_one(client, ent))
|
|
return results
|
|
|
|
|
|
async def _fetch_one(client: httpx.AsyncClient, ent: HAEntity) -> HAState:
|
|
url = f"{config.ha_base_url}/api/states/{ent.entity_id}"
|
|
try:
|
|
resp = await client.get(url)
|
|
if resp.status_code != 200:
|
|
return HAState(label=ent.label, state=f"HTTP {resp.status_code}", ok=False)
|
|
data = resp.json()
|
|
unit = ent.unit or data.get("attributes", {}).get("unit_of_measurement", "")
|
|
return HAState(label=ent.label, state=str(data.get("state", "")), unit=unit)
|
|
except httpx.HTTPError:
|
|
return HAState(label=ent.label, state="erreur", ok=False)
|