0d68933b85
Removes child lock toggles, motion sensor LED indicators, and Zigbee2MQTT permit join from the smart-home dashboard view. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""Home Assistant integration – fetch entity states & call services."""
|
||
|
||
from __future__ import annotations
|
||
|
||
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."""
|
||
attrs = entity.get("attributes", {})
|
||
return attrs.get("friendly_name", entity.get("entity_id", "unknown"))
|
||
|
||
|
||
def _parse_light(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||
attrs = entity.get("attributes", {})
|
||
brightness_raw = attrs.get("brightness")
|
||
brightness_pct: Optional[int] = None
|
||
if brightness_raw is not None:
|
||
try:
|
||
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": 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]:
|
||
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]:
|
||
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),
|
||
"state": state_value,
|
||
"unit": attrs.get("unit_of_measurement", ""),
|
||
"device_class": attrs.get("device_class", ""),
|
||
}
|
||
|
||
|
||
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"}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Filters — exclude Unraid infrastructure entities from the smart-home view.
|
||
# Unraid integration exposes Docker containers, system services, VMs,
|
||
# disk temps, CPU/GPU temps etc. as HA entities. They clutter the dashboard.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Switches whose friendly_name starts with these are hidden
|
||
_SWITCH_EXCLUDE_PREFIXES = (
|
||
"Daddelolymp Docker:",
|
||
"Daddelolymp Service:",
|
||
"Daddelolymp VM:",
|
||
"Daddelolymp Array:",
|
||
"Moneyboy Docker:",
|
||
"Moneyboy Service:",
|
||
"Moneyboy VM:",
|
||
"Moneyboy Array:",
|
||
)
|
||
|
||
# Also exclude exact switch names that are server-level, not smart-home
|
||
_SWITCH_EXCLUDE_EXACT = {
|
||
"Daddelolymp",
|
||
"Moneyboy",
|
||
}
|
||
|
||
# Switches whose friendly_name contains these substrings are technical settings
|
||
_SWITCH_EXCLUDE_SUBSTRINGS = (
|
||
"Child lock",
|
||
"Led indication",
|
||
"Indicator",
|
||
"Permit join",
|
||
)
|
||
|
||
# Sensor friendly_names containing these substrings are hardware, not room sensors
|
||
_SENSOR_EXCLUDE_SUBSTRINGS = (
|
||
"System: CPU",
|
||
"System: Motherboard",
|
||
"Disk:",
|
||
"GPU:",
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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."""
|
||
|
||
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,
|
||
}
|
||
|
||
if not url or not token:
|
||
result["error"] = "Missing Home Assistant URL or token"
|
||
return result
|
||
|
||
base_url = url.rstrip("/")
|
||
headers = {
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15, verify=False) as client:
|
||
resp = await client.get(f"{base_url}/api/states", headers=headers)
|
||
resp.raise_for_status()
|
||
entities: List[Dict[str, Any]] = resp.json()
|
||
except httpx.HTTPStatusError as exc:
|
||
result["error"] = f"HTTP {exc.response.status_code}"
|
||
return result
|
||
except httpx.RequestError as exc:
|
||
result["error"] = f"Connection failed: {exc}"
|
||
return result
|
||
except Exception as exc:
|
||
result["error"] = str(exc)
|
||
return result
|
||
|
||
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", "")
|
||
domain = entity_id.split(".")[0] if "." in entity_id else ""
|
||
attrs = entity.get("attributes", {})
|
||
state = entity.get("state", "")
|
||
|
||
if state in ("unavailable", "unknown"):
|
||
continue
|
||
|
||
friendly = _friendly_name(entity)
|
||
|
||
if domain == "light":
|
||
lights.append(_parse_light(entity))
|
||
|
||
elif domain == "switch":
|
||
# Skip Unraid infrastructure switches + technical settings
|
||
if friendly in _SWITCH_EXCLUDE_EXACT:
|
||
continue
|
||
if any(friendly.startswith(p) for p in _SWITCH_EXCLUDE_PREFIXES):
|
||
continue
|
||
if any(sub in friendly for sub in _SWITCH_EXCLUDE_SUBSTRINGS):
|
||
continue
|
||
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 in _SENSOR_CLASSES:
|
||
# Skip hardware monitoring sensors (disk/CPU/GPU temps)
|
||
if any(sub in friendly for sub in _SENSOR_EXCLUDE_SUBSTRINGS):
|
||
continue
|
||
sensors.append(_parse_sensor(entity))
|
||
|
||
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["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}
|