72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Client Home Assistant REST API (read-only)."""
|
|
|
|
import os
|
|
from typing import Any, Optional
|
|
from urllib.parse import urljoin
|
|
|
|
import httpx
|
|
|
|
PUBLIC_URL = os.environ.get("HA_URL", "https://ha.loogle.it").rstrip("/")
|
|
API_URL = os.environ.get("HA_API_URL", PUBLIC_URL).rstrip("/")
|
|
HA_TOKEN = os.environ.get("HA_TOKEN", "").strip()
|
|
|
|
|
|
def is_configured() -> bool:
|
|
return bool(HA_TOKEN)
|
|
|
|
|
|
def _headers() -> dict:
|
|
if not HA_TOKEN:
|
|
raise RuntimeError(
|
|
"HA_TOKEN non configurato in .env — crea un long-lived token in Home Assistant"
|
|
)
|
|
return {"Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json"}
|
|
|
|
|
|
def _request(method: str, path: str, *, params: Optional[dict] = None) -> Any:
|
|
verify = os.environ.get("HA_VERIFY_SSL", "true").strip().lower() not in (
|
|
"0", "false", "no", "off",
|
|
)
|
|
url = path if path.startswith("http") else urljoin(API_URL + "/", path.lstrip("/"))
|
|
with httpx.Client(timeout=30.0, verify=verify) as client:
|
|
resp = client.request(method, url, headers=_headers(), params=params)
|
|
if resp.status_code >= 400:
|
|
raise RuntimeError(f"Home Assistant {path}: HTTP {resp.status_code} {resp.text[:200]}")
|
|
return resp.json()
|
|
|
|
|
|
def get_config() -> dict:
|
|
return _request("GET", "/api/config")
|
|
|
|
|
|
def get_entity(entity_id: str) -> dict:
|
|
return _request("GET", f"/api/states/{entity_id}")
|
|
|
|
|
|
def list_entities(domain: Optional[str] = None, limit: int = 100) -> list:
|
|
states = _request("GET", "/api/states")
|
|
if domain:
|
|
prefix = domain if domain.endswith(".") else f"{domain}."
|
|
states = [s for s in states if s.get("entity_id", "").startswith(prefix)]
|
|
return states[:limit]
|
|
|
|
|
|
def search_entities(query: str, limit: int = 30) -> list:
|
|
q = query.lower()
|
|
matches = []
|
|
for state in _request("GET", "/api/states"):
|
|
eid = state.get("entity_id", "")
|
|
name = (state.get("attributes") or {}).get("friendly_name", "")
|
|
blob = f"{eid} {name}".lower()
|
|
if q in blob:
|
|
matches.append({
|
|
"entity_id": eid,
|
|
"state": state.get("state"),
|
|
"friendly_name": name,
|
|
"last_changed": state.get("last_changed"),
|
|
})
|
|
if len(matches) >= limit:
|
|
break
|
|
return matches
|