68 lines
2.5 KiB
Python
68 lines
2.5 KiB
Python
import logging
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HomeAssistantClient:
|
|
"""Thin wrapper around the Home Assistant REST API.
|
|
|
|
Instantiating with url=None or api_key=None produces a disabled client;
|
|
all methods become no-ops that return None.
|
|
"""
|
|
|
|
def __init__(self, url: str | None, api_key: str | None):
|
|
self._enabled = bool(url and api_key)
|
|
self._base_url = (url or "").rstrip("/")
|
|
self._headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return self._enabled
|
|
|
|
def get_state(self, entity_id: str, timeout: int = 10) -> int | None:
|
|
"""Fetch the current state of a HA entity and return it as an integer.
|
|
|
|
Returns None if HA is not configured, the entity is not found,
|
|
the state is non-numeric, or any network/HTTP error occurs.
|
|
"""
|
|
if not self._enabled:
|
|
return None
|
|
url = f"{self._base_url}/api/states/{entity_id}"
|
|
try:
|
|
resp = requests.get(url, headers=self._headers, timeout=timeout)
|
|
resp.raise_for_status()
|
|
state = resp.json().get("state")
|
|
return int(float(state))
|
|
except (requests.RequestException, ValueError, TypeError, KeyError) as exc:
|
|
logger.warning("HA API error for %s: %s", entity_id, exc)
|
|
return None
|
|
|
|
def list_battery_entities(self) -> list[dict]:
|
|
"""Return all HA entities that look like battery sensors.
|
|
|
|
Includes entities where device_class == 'battery' or 'battery' appears
|
|
in the entity_id. Returns [] when disabled or on any error.
|
|
"""
|
|
if not self._enabled:
|
|
return []
|
|
url = f"{self._base_url}/api/states"
|
|
try:
|
|
resp = requests.get(url, headers=self._headers, timeout=15)
|
|
resp.raise_for_status()
|
|
result = []
|
|
for s in resp.json():
|
|
eid = s.get("entity_id", "")
|
|
attrs = s.get("attributes", {})
|
|
if (attrs.get("device_class") == "battery"
|
|
or "battery" in eid.lower()):
|
|
result.append({
|
|
"entity_id": eid,
|
|
"friendly_name": attrs.get("friendly_name", eid),
|
|
})
|
|
return sorted(result, key=lambda x: x["entity_id"])
|
|
except (requests.RequestException, ValueError, TypeError) as exc:
|
|
logger.warning("HA list_battery_entities error: %s", exc)
|
|
return []
|