HA controls + wider layout: toggle lights/switches, cover controls, more sensors
- Backend: call_ha_service() for controlling entities via HA REST API - Backend: POST /api/ha/control with JWT auth + cache invalidation - Backend: Parse switches, binary_sensors, humidity, climate entities - Frontend: HA card now xl:col-span-2 (double width) - Frontend: Interactive toggles for lights/switches, cover up/stop/down - Frontend: Temperature + humidity sensors, climate display, binary sensors - Frontend: Two-column internal layout (controls left, sensors right) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Sam
parent
4e7c1909ee
commit
f5b7c53f18
6 changed files with 683 additions and 168 deletions
|
|
@ -1,8 +1,18 @@
|
|||
"""Home Assistant integration – fetch entity states & call services."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import logging
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _friendly_name(entity: Dict[str, Any]) -> str:
|
||||
"""Extract the friendly name from an entity's attributes, falling back to entity_id."""
|
||||
|
|
@ -11,9 +21,7 @@ def _friendly_name(entity: Dict[str, Any]) -> str:
|
|||
|
||||
|
||||
def _parse_light(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Parse a light entity into a normalised dictionary."""
|
||||
attrs = entity.get("attributes", {})
|
||||
state = entity.get("state", "unknown")
|
||||
brightness_raw = attrs.get("brightness")
|
||||
brightness_pct: Optional[int] = None
|
||||
if brightness_raw is not None:
|
||||
|
|
@ -21,36 +29,41 @@ def _parse_light(entity: Dict[str, Any]) -> Dict[str, Any]:
|
|||
brightness_pct = round(int(brightness_raw) / 255 * 100)
|
||||
except (ValueError, TypeError):
|
||||
brightness_pct = None
|
||||
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
"state": state,
|
||||
"state": entity.get("state", "unknown"),
|
||||
"brightness": brightness_pct,
|
||||
"color_mode": attrs.get("color_mode"),
|
||||
}
|
||||
|
||||
|
||||
def _parse_switch(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
"state": entity.get("state", "unknown"),
|
||||
}
|
||||
|
||||
|
||||
def _parse_cover(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Parse a cover entity into a normalised dictionary."""
|
||||
attrs = entity.get("attributes", {})
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
"state": entity.get("state", "unknown"),
|
||||
"current_position": attrs.get("current_position"),
|
||||
"device_class": attrs.get("device_class", ""),
|
||||
}
|
||||
|
||||
|
||||
def _parse_sensor(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Parse a temperature sensor entity into a normalised dictionary."""
|
||||
attrs = entity.get("attributes", {})
|
||||
state_value = entity.get("state", "unknown")
|
||||
try:
|
||||
state_value = round(float(state_value), 1)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
|
|
@ -60,30 +73,65 @@ def _parse_sensor(entity: Dict[str, Any]) -> Dict[str, Any]:
|
|||
}
|
||||
|
||||
|
||||
def _parse_binary_sensor(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
attrs = entity.get("attributes", {})
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
"state": entity.get("state", "unknown"),
|
||||
"device_class": attrs.get("device_class", ""),
|
||||
}
|
||||
|
||||
|
||||
def _parse_climate(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
attrs = entity.get("attributes", {})
|
||||
current_temp = attrs.get("current_temperature")
|
||||
target_temp = attrs.get("temperature")
|
||||
try:
|
||||
current_temp = round(float(current_temp), 1) if current_temp is not None else None
|
||||
except (ValueError, TypeError):
|
||||
current_temp = None
|
||||
try:
|
||||
target_temp = round(float(target_temp), 1) if target_temp is not None else None
|
||||
except (ValueError, TypeError):
|
||||
target_temp = None
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
"state": entity.get("state", "unknown"),
|
||||
"current_temperature": current_temp,
|
||||
"target_temperature": target_temp,
|
||||
"hvac_modes": attrs.get("hvac_modes", []),
|
||||
"humidity": attrs.get("current_humidity"),
|
||||
}
|
||||
|
||||
|
||||
# Sensor device_classes we care about
|
||||
_SENSOR_CLASSES = {"temperature", "humidity"}
|
||||
|
||||
# Binary sensor device_classes we display
|
||||
_BINARY_SENSOR_CLASSES = {"door", "window", "motion", "occupancy"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fetch all entity states
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def fetch_ha_data(url: str, token: str) -> Dict[str, Any]:
|
||||
"""Fetch and categorise entity states from a Home Assistant instance.
|
||||
"""Fetch and categorise entity states from a Home Assistant instance."""
|
||||
|
||||
Args:
|
||||
url: Base URL of the Home Assistant instance (e.g. ``http://192.168.1.100:8123``).
|
||||
token: Long-lived access token for authentication.
|
||||
|
||||
Returns:
|
||||
Dictionary containing:
|
||||
- ``online``: Whether the HA instance is reachable.
|
||||
- ``lights``: List of light entities with state and brightness.
|
||||
- ``covers``: List of cover entities with state and position.
|
||||
- ``sensors``: List of temperature sensor entities.
|
||||
- ``lights_on``: Count of lights currently in the ``on`` state.
|
||||
- ``lights_total``: Total number of light entities.
|
||||
- ``error``: Error message if the request failed, else ``None``.
|
||||
"""
|
||||
result: Dict[str, Any] = {
|
||||
"online": False,
|
||||
"lights": [],
|
||||
"switches": [],
|
||||
"covers": [],
|
||||
"sensors": [],
|
||||
"binary_sensors": [],
|
||||
"climate": [],
|
||||
"lights_on": 0,
|
||||
"lights_total": 0,
|
||||
"switches_on": 0,
|
||||
"switches_total": 0,
|
||||
"error": None,
|
||||
}
|
||||
|
||||
|
|
@ -115,8 +163,11 @@ async def fetch_ha_data(url: str, token: str) -> Dict[str, Any]:
|
|||
result["online"] = True
|
||||
|
||||
lights: List[Dict[str, Any]] = []
|
||||
switches: List[Dict[str, Any]] = []
|
||||
covers: List[Dict[str, Any]] = []
|
||||
sensors: List[Dict[str, Any]] = []
|
||||
binary_sensors: List[Dict[str, Any]] = []
|
||||
climate_list: List[Dict[str, Any]] = []
|
||||
|
||||
for entity in entities:
|
||||
entity_id: str = entity.get("entity_id", "")
|
||||
|
|
@ -130,20 +181,121 @@ async def fetch_ha_data(url: str, token: str) -> Dict[str, Any]:
|
|||
if domain == "light":
|
||||
lights.append(_parse_light(entity))
|
||||
|
||||
elif domain == "switch":
|
||||
switches.append(_parse_switch(entity))
|
||||
|
||||
elif domain == "cover":
|
||||
covers.append(_parse_cover(entity))
|
||||
|
||||
elif domain == "sensor":
|
||||
device_class = attrs.get("device_class", "")
|
||||
if device_class == "temperature":
|
||||
if device_class in _SENSOR_CLASSES:
|
||||
sensors.append(_parse_sensor(entity))
|
||||
|
||||
lights_on = sum(1 for light in lights if light["state"] == "on")
|
||||
elif domain == "binary_sensor":
|
||||
device_class = attrs.get("device_class", "")
|
||||
if device_class in _BINARY_SENSOR_CLASSES:
|
||||
binary_sensors.append(_parse_binary_sensor(entity))
|
||||
|
||||
elif domain == "climate":
|
||||
climate_list.append(_parse_climate(entity))
|
||||
|
||||
result["lights"] = lights
|
||||
result["switches"] = switches
|
||||
result["covers"] = covers
|
||||
result["sensors"] = sensors
|
||||
result["lights_on"] = lights_on
|
||||
result["binary_sensors"] = binary_sensors
|
||||
result["climate"] = climate_list
|
||||
result["lights_on"] = sum(1 for l in lights if l["state"] == "on")
|
||||
result["lights_total"] = len(lights)
|
||||
result["switches_on"] = sum(1 for s in switches if s["state"] == "on")
|
||||
result["switches_total"] = len(switches)
|
||||
|
||||
logger.info(
|
||||
"[HA] Fetched %d lights (%d on), %d switches (%d on), %d covers, "
|
||||
"%d sensors, %d binary, %d climate",
|
||||
len(lights), result["lights_on"],
|
||||
len(switches), result["switches_on"],
|
||||
len(covers), len(sensors), len(binary_sensors), len(climate_list),
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Call HA service (control entities)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Map of (domain, action) → HA service name
|
||||
_ACTION_MAP: Dict[tuple, str] = {
|
||||
("light", "toggle"): "toggle",
|
||||
("light", "turn_on"): "turn_on",
|
||||
("light", "turn_off"): "turn_off",
|
||||
("switch", "toggle"): "toggle",
|
||||
("switch", "turn_on"): "turn_on",
|
||||
("switch", "turn_off"): "turn_off",
|
||||
("cover", "open"): "open_cover",
|
||||
("cover", "close"): "close_cover",
|
||||
("cover", "stop"): "stop_cover",
|
||||
("climate", "turn_on"): "turn_on",
|
||||
("climate", "turn_off"): "turn_off",
|
||||
}
|
||||
|
||||
|
||||
async def call_ha_service(
|
||||
url: str,
|
||||
token: str,
|
||||
entity_id: str,
|
||||
action: str,
|
||||
service_data: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Call a Home Assistant service to control an entity.
|
||||
|
||||
Args:
|
||||
url: Base URL of the HA instance.
|
||||
token: Long-lived access token.
|
||||
entity_id: e.g. ``light.wohnzimmer_deckenlampe``
|
||||
action: e.g. ``toggle``, ``turn_on``, ``turn_off``, ``open``, ``close``, ``stop``
|
||||
service_data: Optional extra data (e.g. brightness).
|
||||
|
||||
Returns:
|
||||
``{"ok": True}`` on success, ``{"ok": False, "error": "..."}`` on failure.
|
||||
"""
|
||||
|
||||
if "." not in entity_id:
|
||||
return {"ok": False, "error": f"Invalid entity_id: {entity_id}"}
|
||||
|
||||
domain = entity_id.split(".")[0]
|
||||
service = _ACTION_MAP.get((domain, action))
|
||||
if service is None:
|
||||
return {"ok": False, "error": f"Unknown action '{action}' for domain '{domain}'"}
|
||||
|
||||
base_url = url.rstrip("/")
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
payload: Dict[str, Any] = {"entity_id": entity_id}
|
||||
if service_data:
|
||||
payload.update(service_data)
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10, verify=False) as client:
|
||||
resp = await client.post(
|
||||
f"{base_url}/api/services/{domain}/{service}",
|
||||
headers=headers,
|
||||
json=payload,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
logger.error("[HA] Service call failed: HTTP %s for %s/%s %s",
|
||||
exc.response.status_code, domain, service, entity_id)
|
||||
return {"ok": False, "error": f"HTTP {exc.response.status_code}"}
|
||||
except Exception as exc:
|
||||
logger.error("[HA] Service call error: %s for %s/%s %s",
|
||||
exc, domain, service, entity_id)
|
||||
return {"ok": False, "error": str(exc)}
|
||||
|
||||
logger.info("[HA] %s/%s → %s ✓", domain, service, entity_id)
|
||||
return {"ok": True}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue