mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
cd004df817
Binds binary_sensor and reading-carrying sensor entities to a printer and renders their state on its card, worded by Home Assistant's device_class. Optional per-sensor alert condition drives a notification on the transition into the alert state and an opt-in interlock that holds queued prints while alerting -- a hold with a readable waiting_reason, never a failure, and only ever on a sensor that was read successfully. Sensors get their own table rather than a wider entity pattern on SmartPlug: get_smart_plug_by_printer would otherwise hand the card's power button a door contact to switch. The hold is passed to the model matcher directly rather than merged into busy_printers: _check_auto_drying reads that set as "is currently printing" and would put an idle-but-held printer down the mid-print drying path. The notification_providers migration spells its default FALSE, not 0 -- Postgres rejects an integer default for a boolean and _safe_execute swallows the error.
472 lines
19 KiB
Python
472 lines
19 KiB
Python
"""Service for communicating with Home Assistant via REST API."""
|
|
|
|
import asyncio
|
|
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:
|
|
"""Normalise a caller-supplied HA URL, or return None if it is unsafe.
|
|
|
|
The stored ``ha_url`` setting is already validated at the schema layer
|
|
(``LAN_SERVICE_URL_SETTINGS`` in schemas/settings.py), but
|
|
``test_connection`` takes its URL straight from the request body, so
|
|
the same policy has to be applied here.
|
|
|
|
Delegates to ``_url_safety.assert_safe_lan_service_url`` rather than
|
|
the string blocklist this replaces. That blocklist only knew three
|
|
literal hostnames plus a ``169.254.`` prefix and never parsed the
|
|
hostname as an IP, so it let through the Alibaba (100.100.100.200)
|
|
and AWS-IPv6 (fd00:ec2::254) metadata endpoints, numeric-encoded
|
|
loopback, multicast, and IPv4-mapped IPv6 encodings of the IMDS
|
|
address it did know about.
|
|
|
|
Loopback and RFC-1918 remain permitted — Home Assistant is a
|
|
LAN-resident service by design, and the shared guard is documented
|
|
that way.
|
|
"""
|
|
from backend.app.api.routes._url_safety import assert_safe_lan_service_url
|
|
|
|
try:
|
|
assert_safe_lan_service_url(url, label="Home Assistant URL")
|
|
except ValueError:
|
|
return None
|
|
# Guard passed, so the scheme is http/https and a hostname is present;
|
|
# re-parse only to drop query/fragment and normalise the authority.
|
|
parsed = urlparse(url)
|
|
if not parsed.hostname:
|
|
return None
|
|
# urlparse strips the brackets off an IPv6 literal, so they have to go
|
|
# back on or the rebuilt URL is unparseable ("http://fd00::1:8123").
|
|
host = f"[{parsed.hostname}]" if ":" in parsed.hostname else parsed.hostname
|
|
return f"{parsed.scheme.lower()}://{host}" + (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 []
|
|
|
|
async def list_display_entities(self, url: str, token: str, search: str | None = None) -> list[dict]:
|
|
"""List entities that can be bound to a printer for display (#1148, #448).
|
|
|
|
Covers every ``binary_sensor.*`` plus the ``sensor.*`` entities that
|
|
carry a reading. Distinct from ``list_sensor_entities``, which exists
|
|
for a plug's energy monitoring and therefore only admits power/energy
|
|
units — an enclosure thermometer is exactly what that one filters out.
|
|
|
|
A ``sensor.*`` qualifies when it has a unit or its state parses as a
|
|
number. That drops the text sensors (``sensor.washing_machine_status``)
|
|
that the card has no way to render as a value.
|
|
"""
|
|
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 ""
|
|
if domain not in ("binary_sensor", "sensor"):
|
|
continue
|
|
|
|
attrs = entity.get("attributes", {})
|
|
unit = attrs.get("unit_of_measurement")
|
|
state = entity.get("state")
|
|
|
|
if domain == "sensor" and not unit and as_float(state) is None:
|
|
continue
|
|
|
|
friendly_name = attrs.get("friendly_name") or entity_id
|
|
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": state,
|
|
"domain": domain,
|
|
"device_class": attrs.get("device_class"),
|
|
"unit_of_measurement": unit,
|
|
}
|
|
)
|
|
|
|
return sorted(entities, key=lambda x: x["friendly_name"].lower())
|
|
except Exception as e:
|
|
logger.warning("Failed to list HA display entities: %s", e)
|
|
return []
|
|
|
|
async def fetch_states(self, entity_ids: list[str]) -> dict[str, dict | None]:
|
|
"""Read several entities in one pass, keyed by entity_id.
|
|
|
|
One GET per entity over a shared client rather than a single
|
|
``/api/states`` sweep: the poller only ever wants a handful of bound
|
|
entities, and pulling every state in the user's Home Assistant on a
|
|
15-second cadence is a lot of payload to throw away.
|
|
|
|
A ``None`` value means that entity could not be read — the callers
|
|
treat that as "no opinion" rather than as a state, so an unreachable
|
|
Home Assistant never trips an alert or holds a print.
|
|
"""
|
|
if not entity_ids:
|
|
return {}
|
|
if not self.base_url or not self.token:
|
|
return dict.fromkeys(entity_ids)
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
|
|
async def _one(entity_id: str) -> tuple[str, dict | None]:
|
|
try:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/states/{entity_id}",
|
|
headers=self._headers(),
|
|
)
|
|
response.raise_for_status()
|
|
return entity_id, response.json()
|
|
except Exception as e:
|
|
logger.debug("Failed to read HA entity %s: %s", entity_id, e)
|
|
return entity_id, None
|
|
|
|
results = await asyncio.gather(*(_one(e) for e in entity_ids))
|
|
|
|
return dict(results)
|
|
|
|
|
|
def as_float(value) -> float | None:
|
|
"""Parse a HA state to a number, or None for "unknown"/"unavailable"/text."""
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
# Singleton instance
|
|
homeassistant_service = HomeAssistantService()
|