mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
135b8fd93b
Reporter MartinNYHC opened the Add Smart Plug dialog in HA mode, typed
a search prefix matching a multi-entity device (one switch.* plus
several sensor.*/binary_sensor.* siblings under the same friendly
name), clicked one of the non-switch siblings, and got a 422 on Save:
String should match pattern
'^(switch|light|input_boolean|script)\.[a-z0-9_]+$'
The screenshot confirms the bug shape — the X button next to the
"empty-looking" Select Entity field only renders when haEntityId is
truthy. So haEntityId was set, but selectedEntity (haEntities.find by
that id) returned undefined, so the input rendered the placeholder
text instead of the friendly-name display. That can only happen when
the user had earlier picked an entity whose domain is NOT in the
schema's allowed list, then the search cleared, the entity-list
refetched without a search param, and the refreshed list (filtered to
the default domains) no longer contained the user's pick.
Root cause was in HomeAssistantService.list_entities: when a search
query was present, the function bypassed the domain filter entirely
and returned matches across every HA domain. Offering a clickable
choice the schema can't accept is broken UX, and the cryptic Pydantic
pattern echo on save made it look like a backend/schema problem
rather than a search-permissiveness problem. Confirmed via git diff
that the smart-plug code path is unchanged between v0.2.4 and
0.2.4.1 — this has been latent since the script-domain commit in
February 2026, only noticed now because the reporter hadn't reopened
the modal in months.
Fix: always apply the allowed-domains filter ({switch, light,
input_boolean, script} — kept in sync with the regex in
backend/app/schemas/smart_plug.py:17). Search composes on top as a
substring match against entity_id or friendly_name, instead of
replacing the domain filter. Whitespace-only search strings now
fall back to the no-search behavior.
347 lines
14 KiB
Python
347 lines
14 KiB
Python
"""Service for communicating with Home Assistant via REST API."""
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
if TYPE_CHECKING:
|
|
from backend.app.models.smart_plug import SmartPlug
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HomeAssistantService:
|
|
"""Service for controlling Home Assistant entities via REST API."""
|
|
|
|
def __init__(self, timeout: float = 10.0):
|
|
self.timeout = timeout
|
|
self.base_url: str = ""
|
|
self.token: str = ""
|
|
|
|
def configure(self, url: str, token: str):
|
|
"""Configure HA connection settings."""
|
|
self.base_url = url.rstrip("/") if url else ""
|
|
self.token = token or ""
|
|
|
|
def _headers(self) -> dict:
|
|
return {
|
|
"Authorization": f"Bearer {self.token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def get_status(self, plug: "SmartPlug") -> dict:
|
|
"""Get current state of HA entity.
|
|
|
|
Returns dict with:
|
|
- state: "ON" or "OFF" or None if unreachable
|
|
- reachable: bool
|
|
- device_name: str or None
|
|
"""
|
|
if not self.base_url or not self.token:
|
|
return {"state": None, "reachable": False, "device_name": None}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/states/{plug.ha_entity_id}",
|
|
headers=self._headers(),
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
state_value = data.get("state", "").lower()
|
|
# Normalize to ON/OFF
|
|
if state_value == "on":
|
|
state = "ON"
|
|
elif state_value == "off":
|
|
state = "OFF"
|
|
else:
|
|
state = None
|
|
|
|
return {
|
|
"state": state,
|
|
"reachable": True,
|
|
"device_name": data.get("attributes", {}).get("friendly_name"),
|
|
}
|
|
except Exception as e:
|
|
logger.warning("Failed to get HA entity state for %s: %s", plug.ha_entity_id, e)
|
|
return {"state": None, "reachable": False, "device_name": None}
|
|
|
|
async def turn_on(self, plug: "SmartPlug") -> bool:
|
|
"""Turn on HA entity. Returns True if successful."""
|
|
success = await self._call_service(plug, "turn_on")
|
|
if success:
|
|
logger.info("Turned ON HA entity '%s' (%s)", plug.name, plug.ha_entity_id)
|
|
return success
|
|
|
|
async def turn_off(self, plug: "SmartPlug") -> bool:
|
|
"""Turn off HA entity. Returns True if successful."""
|
|
success = await self._call_service(plug, "turn_off")
|
|
if success:
|
|
logger.info("Turned OFF HA entity '%s' (%s)", plug.name, plug.ha_entity_id)
|
|
return success
|
|
|
|
async def toggle(self, plug: "SmartPlug") -> bool:
|
|
"""Toggle HA entity. Returns True if successful."""
|
|
success = await self._call_service(plug, "toggle")
|
|
if success:
|
|
logger.info("Toggled HA entity '%s' (%s)", plug.name, plug.ha_entity_id)
|
|
return success
|
|
|
|
async def _call_service(self, plug: "SmartPlug", action: str) -> bool:
|
|
"""Call HA service on entity."""
|
|
if not self.base_url or not self.token or not plug.ha_entity_id:
|
|
return False
|
|
|
|
domain = plug.ha_entity_id.split(".")[0] # "switch", "light", etc.
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/api/services/{domain}/{action}",
|
|
headers=self._headers(),
|
|
json={"entity_id": plug.ha_entity_id},
|
|
)
|
|
response.raise_for_status()
|
|
return True
|
|
except Exception as e:
|
|
logger.warning("Failed to %s HA entity %s: %s", action, plug.ha_entity_id, e)
|
|
return False
|
|
|
|
async def get_energy(self, plug: "SmartPlug") -> dict | None:
|
|
"""Get energy data from HA sensor entities or switch attributes.
|
|
|
|
First tries dedicated sensor entities if configured, then falls back
|
|
to checking the switch entity's attributes.
|
|
Returns dict with energy data or None if not available.
|
|
"""
|
|
if not self.base_url or not self.token:
|
|
return None
|
|
|
|
power = None
|
|
today = None
|
|
total = None
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
# Fetch power from dedicated sensor entity if configured
|
|
if plug.ha_power_entity:
|
|
power = await self._get_sensor_value(client, plug.ha_power_entity)
|
|
|
|
# Fetch today's energy from dedicated sensor entity if configured
|
|
if plug.ha_energy_today_entity:
|
|
today = await self._get_sensor_value(client, plug.ha_energy_today_entity)
|
|
|
|
# Fetch total energy from dedicated sensor entity if configured
|
|
if plug.ha_energy_total_entity:
|
|
total = await self._get_sensor_value(client, plug.ha_energy_total_entity)
|
|
|
|
# Fallback: try switch entity attributes (original behavior)
|
|
if power is None:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/states/{plug.ha_entity_id}",
|
|
headers=self._headers(),
|
|
)
|
|
response.raise_for_status()
|
|
attrs = response.json().get("attributes", {})
|
|
power = attrs.get("current_power_w") or attrs.get("power")
|
|
if today is None:
|
|
today = attrs.get("today_energy_kwh")
|
|
if total is None:
|
|
total = attrs.get("total_energy_kwh")
|
|
|
|
if power is None:
|
|
return None
|
|
|
|
return {
|
|
"power": power,
|
|
"voltage": None,
|
|
"current": None,
|
|
"today": today,
|
|
"total": total,
|
|
"yesterday": None,
|
|
"factor": None,
|
|
"apparent_power": None,
|
|
"reactive_power": None,
|
|
}
|
|
except Exception as e:
|
|
logger.debug("Failed to get HA energy data: %s", e)
|
|
return None
|
|
|
|
async def _get_sensor_value(self, client: httpx.AsyncClient, entity_id: str) -> float | None:
|
|
"""Fetch numeric value from a HA sensor entity."""
|
|
try:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/states/{entity_id}",
|
|
headers=self._headers(),
|
|
)
|
|
response.raise_for_status()
|
|
state = response.json().get("state")
|
|
if state and state not in ("unknown", "unavailable"):
|
|
return float(state)
|
|
except Exception:
|
|
pass # Sensor read is best-effort; caller handles None
|
|
return None
|
|
|
|
@staticmethod
|
|
def _validate_url(url: str) -> str | None:
|
|
"""Validate HA URL scheme and block dangerous destinations."""
|
|
try:
|
|
parsed = urlparse(url)
|
|
except ValueError:
|
|
return None
|
|
if parsed.scheme not in ("http", "https") or not parsed.hostname:
|
|
return None
|
|
blocked = ("169.254.169.254", "metadata.google.internal", "0.0.0.0") # nosec B104
|
|
if parsed.hostname.lower() in blocked or (parsed.hostname or "").startswith("169.254."):
|
|
return None
|
|
return f"{parsed.scheme}://{parsed.hostname}" + (f":{parsed.port}" if parsed.port else "") + (parsed.path or "")
|
|
|
|
async def test_connection(self, url: str, token: str) -> dict:
|
|
"""Test connection to Home Assistant.
|
|
|
|
Returns dict with:
|
|
- success: bool
|
|
- message: str or None (HA message on success)
|
|
- error: str or None (error message on failure)
|
|
"""
|
|
safe_url = self._validate_url(url)
|
|
if not safe_url:
|
|
return {"success": False, "message": None, "error": "Invalid Home Assistant URL"}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{safe_url.rstrip('/')}/api/",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return {
|
|
"success": True,
|
|
"message": data.get("message", "Connected"),
|
|
"error": None,
|
|
}
|
|
except httpx.HTTPStatusError as e:
|
|
if e.response.status_code == 401:
|
|
return {"success": False, "message": None, "error": "Invalid access token"}
|
|
return {"success": False, "message": None, "error": f"HTTP {e.response.status_code}"}
|
|
except httpx.TimeoutException:
|
|
return {"success": False, "message": None, "error": "Connection timeout"}
|
|
except httpx.ConnectError:
|
|
return {"success": False, "message": None, "error": "Could not connect to Home Assistant"}
|
|
except Exception as e:
|
|
return {"success": False, "message": None, "error": str(e)}
|
|
|
|
async def list_entities(self, url: str, token: str, search: str | None = None) -> list[dict]:
|
|
"""List available entities from HA.
|
|
|
|
Always filters to switch/light/input_boolean/script — the only domains
|
|
the SmartPlugBase.ha_entity_id pattern accepts. When a search query is
|
|
provided it narrows the same domain-filtered list by entity_id or
|
|
friendly_name substring (case-insensitive).
|
|
|
|
Previously search bypassed the domain filter, which let users pick a
|
|
sensor.* or binary_sensor.* entity from the dropdown that the backend
|
|
schema would then reject with the cryptic Pydantic pattern error
|
|
(#1388). Picking what you can't save isn't a useful UX.
|
|
|
|
Returns list of entity dicts with:
|
|
- entity_id: str
|
|
- friendly_name: str
|
|
- state: str
|
|
- domain: str
|
|
"""
|
|
# Allowed domains for smart plug control — must mirror the regex in
|
|
# backend/app/schemas/smart_plug.py:17 (SmartPlugBase.ha_entity_id).
|
|
allowed_domains = {"switch", "light", "input_boolean", "script"}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{url.rstrip('/')}/api/states",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
entities = []
|
|
search_lower = search.lower().strip() if search else None
|
|
|
|
for entity in response.json():
|
|
entity_id = entity.get("entity_id", "")
|
|
domain = entity_id.split(".")[0] if "." in entity_id else ""
|
|
friendly_name = entity.get("attributes", {}).get("friendly_name", entity_id)
|
|
|
|
if domain not in allowed_domains:
|
|
continue
|
|
|
|
if search_lower and (
|
|
search_lower not in entity_id.lower() and search_lower not in friendly_name.lower()
|
|
):
|
|
continue
|
|
|
|
entities.append(
|
|
{
|
|
"entity_id": entity_id,
|
|
"friendly_name": friendly_name,
|
|
"state": entity.get("state"),
|
|
"domain": domain,
|
|
}
|
|
)
|
|
|
|
return sorted(entities, key=lambda x: x["friendly_name"].lower())
|
|
except Exception as e:
|
|
logger.warning("Failed to list HA entities: %s", e)
|
|
return []
|
|
|
|
async def list_sensor_entities(self, url: str, token: str) -> list[dict]:
|
|
"""List available sensor entities for energy monitoring.
|
|
|
|
Returns list of sensor entities with power/energy units.
|
|
"""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{url.rstrip('/')}/api/states",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
# Valid units for energy monitoring sensors (lowercase for case-insensitive matching)
|
|
power_units = {"w", "kw", "mw"}
|
|
energy_units = {"kwh", "wh", "mwh"}
|
|
valid_units = power_units | energy_units
|
|
|
|
entities = []
|
|
for entity in response.json():
|
|
entity_id = entity.get("entity_id", "")
|
|
domain = entity_id.split(".")[0] if "." in entity_id else ""
|
|
|
|
# Filter to sensor domain only
|
|
if domain != "sensor":
|
|
continue
|
|
|
|
attrs = entity.get("attributes", {})
|
|
unit = attrs.get("unit_of_measurement", "")
|
|
|
|
# Only include sensors with power/energy units (case-insensitive)
|
|
if unit.lower() in valid_units:
|
|
entities.append(
|
|
{
|
|
"entity_id": entity_id,
|
|
"friendly_name": attrs.get("friendly_name", entity_id),
|
|
"state": entity.get("state"),
|
|
"unit_of_measurement": unit,
|
|
}
|
|
)
|
|
|
|
return sorted(entities, key=lambda x: x["friendly_name"].lower())
|
|
except Exception as e:
|
|
logger.warning("Failed to list HA sensor entities: %s", e)
|
|
return []
|
|
|
|
|
|
# Singleton instance
|
|
homeassistant_service = HomeAssistantService()
|