bambuddy/backend/app/services/ha_sensor_manager.py

"""Polls the Home Assistant entities bound to printers (#1148, #448).

One background loop reads every configured entity on a fixed cadence and keeps
the result in memory. Three things consume it:

* the printer card, which reads the cache instead of hitting Home Assistant
  once per card per refresh;
* notifications, fired on a transition *into* the alert state, never on every
  poll while it persists;
* the print interlock, which holds queued jobs for a printer while one of its
  sensors is alerting.

Everything degrades to "no opinion" when Home Assistant cannot be reached: an
unreadable sensor never alerts, never notifies, and never holds a print. A
door contact that stops responding must not strand the queue.
"""

import asyncio
import logging
from dataclasses import dataclass
from typing import Protocol

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.models.printer import Printer
from backend.app.models.printer_ha_sensor import LAST_STATE_MAX_LENGTH, PrinterHASensor
from backend.app.services.homeassistant import as_float, homeassistant_service
from backend.app.utils.local_time import utcnow_naive

logger = logging.getLogger(__name__)

# Fast enough that an enclosure door reads as live, slow enough that a handful
# of tiny LAN requests stays background noise.
POLL_INTERVAL = 15


@dataclass
class SensorReading:
    """The last thing we managed to read for one sensor."""

    state: str | None  # raw HA state, None when unreadable
    value: float | None  # parsed number for numeric sensors
    alerting: bool
    reachable: bool


def persistable_state(state: str | None, max_length: int) -> str | None:
    """Fit a raw HA state into a last_state column.

    A numeric entity can start reporting free text (an enum, an error string)
    longer than the column. PostgreSQL rejects the oversized row, and since a
    poll pass commits every sensor at once, one such entity would sink every
    other sensor's update on every tick -- and for printer sensors that also
    freezes the print interlock's view of the world.

    The cached SensorReading keeps the full state; only what is persisted is
    cut, and the comparison against the stored value is done on the cut form so
    an unchanged-but-long state does not read as a change on every poll.

    Shared with the storage-location poller, which has the same column on its
    own table -- each caller passes its own model's width.
    """
    return state[:max_length] if state else state


class HASensorManager:
    def __init__(self):
        self._task: asyncio.Task | None = None
        # sensor id -> last reading. Sensors absent from this map have not been
        # polled yet; callers must not read that as "not alerting" without also
        # checking, which is why get_reading returns None rather than a default.
        self._readings: dict[int, SensorReading] = {}
        # sensor id -> alerting, from the last reading we could actually take.
        # Kept apart from _readings because a dropout must not read as the
        # alert clearing: on -> unavailable -> on is one continuous alert, and
        # notifying off _readings alone would ping the user on every reconnect
        # of a flaky contact. Absent means "never had a reachable reading".
        self._last_alerting: dict[int, bool] = {}

    # -- lifecycle ---------------------------------------------------------

    def start(self):
        if self._task is None:
            self._task = asyncio.create_task(self._poll_loop())
            logger.info("Home Assistant sensor poller started")

    def stop(self):
        if self._task:
            self._task.cancel()
            self._task = None
            logger.info("Home Assistant sensor poller stopped")

    # -- cache access ------------------------------------------------------

    def get_reading(self, sensor_id: int) -> SensorReading | None:
        return self._readings.get(sensor_id)

    def forget(self, sensor_id: int):
        """Drop a deleted sensor's cached reading so its id cannot be reused
        by a later row and answer with the old sensor's state."""
        self._readings.pop(sensor_id, None)
        self._last_alerting.pop(sensor_id, None)

    async def blocked_printers(self, db: AsyncSession) -> dict[int, str]:
        """Printers currently held by an interlock, mapped to the sensor names.

        A sensor counts only when it is configured to block, *and* was read
        successfully, *and* is in its alert state. Anything we could not read
        is omitted, so the queue keeps moving when Home Assistant is down.

        One query for the whole fleet — the scheduler calls this on every pass,
        and per-printer lookups would put a query per printer in that loop.
        """
        result = await db.execute(select(PrinterHASensor).where(PrinterHASensor.block_print.is_(True)))
        blocked: dict[int, list[str]] = {}
        for sensor in result.scalars().all():
            reading = self._readings.get(sensor.id)
            if reading and reading.reachable and reading.alerting:
                blocked.setdefault(sensor.printer_id, []).append(sensor.name)
        return {printer_id: ", ".join(names) for printer_id, names in blocked.items()}

    # -- polling -----------------------------------------------------------

    async def _poll_loop(self):
        while True:
            try:
                await asyncio.sleep(POLL_INTERVAL)
                await self.poll_once()
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.warning("Home Assistant sensor poll failed: %s", e)

    async def poll_once(self):
        """One pass over every configured sensor."""
        from backend.app.core.database import async_session

        async with async_session() as db:
            result = await db.execute(select(PrinterHASensor))
            sensors = list(result.scalars().all())

            # Drop readings for rows that no longer exist. The delete route
            # calls forget(), but a printer deleted with sensors attached takes
            # them out by cascade, and a restored backup can renumber them —
            # either way a stale id must not answer for a later sensor.
            live = {s.id for s in sensors}
            for stale in set(self._readings) - live:
                self.forget(stale)

            if not sensors:
                return

            if not await self._configure(db):
                # Not configured is not a failure to report every 15 seconds,
                # but the readings must not go stale-but-confident either.
                for sensor in sensors:
                    self._readings[sensor.id] = SensorReading(None, None, False, False)
                return

            states = await homeassistant_service.fetch_states(sorted({s.entity_id for s in sensors}))
            await self._apply(db, sensors, states)

    async def refresh_one(self, db: AsyncSession, sensor: PrinterHASensor):
        """Read a single sensor now, on the caller's session.

        Used after a create or an edit so the card shows a state straight away
        instead of blank until the next tick. Deliberately not a full
        ``poll_once``: a request handler must not wait on every configured
        entity, and must not fire another user's notification as a side effect
        of this one saving a form.
        """
        self.forget(sensor.id)
        if not await self._configure(db):
            self._readings[sensor.id] = SensorReading(None, None, False, False)
            return

        states = await homeassistant_service.fetch_states([sensor.entity_id])
        reading = evaluate(sensor, states.get(sensor.entity_id))
        self._readings[sensor.id] = reading
        if reading.reachable:
            self._last_alerting[sensor.id] = reading.alerting

        sensor.last_checked = utcnow_naive()
        persisted = persistable_state(reading.state, LAST_STATE_MAX_LENGTH)
        if reading.reachable and sensor.last_state != persisted:
            sensor.last_state = persisted
            sensor.last_changed = sensor.last_checked
        await db.commit()
        await db.refresh(sensor)

    async def _configure(self, db: AsyncSession) -> bool:
        from backend.app.api.routes.settings import get_homeassistant_settings

        try:
            ha_settings = await get_homeassistant_settings(db)
        except Exception as e:
            logger.warning("Failed to read Home Assistant settings: %s", e)
            return False
        if not ha_settings["ha_url"] or not ha_settings["ha_token"]:
            return False
        homeassistant_service.configure(ha_settings["ha_url"], ha_settings["ha_token"])
        return True

    async def _apply(self, db: AsyncSession, sensors: list[PrinterHASensor], states: dict[str, dict | None]):
        """Fold poll results into the cache, the DB and any notifications."""
        from backend.app.services.notification_service import notification_service

        now = utcnow_naive()
        alerts: list[tuple[PrinterHASensor, SensorReading]] = []

        for sensor in sensors:
            payload = states.get(sensor.entity_id)
            reading = evaluate(sensor, payload)
            was_alerting = self._last_alerting.get(sensor.id)
            self._readings[sensor.id] = reading

            sensor.last_checked = now
            if reading.reachable:
                persisted = persistable_state(reading.state, LAST_STATE_MAX_LENGTH)
                if sensor.last_state != persisted:
                    sensor.last_state = persisted
                    sensor.last_changed = now

            # Notify on the edge into alerting only. `was_alerting is None` is
            # a cold cache (first poll after a restart) — a door that was
            # already open then has not just been opened, and re-announcing it
            # on every restart would train users to ignore the alert.
            if sensor.notify_on_alert and reading.reachable and reading.alerting and was_alerting is False:
                alerts.append((sensor, reading))

            if reading.reachable:
                self._last_alerting[sensor.id] = reading.alerting

        await db.commit()

        for sensor, reading in alerts:
            # db.get, not sensor.printer: touching the lazy relationship from
            # an async session raises MissingGreenlet.
            printer = await db.get(Printer, sensor.printer_id)
            try:
                await notification_service.on_ha_sensor_alert(
                    printer_id=sensor.printer_id,
                    printer_name=printer.name if printer else "Unknown",
                    sensor_name=sensor.name,
                    state=describe_state(sensor, reading),
                    db=db,
                )
            except Exception as e:
                logger.warning("Failed to send HA sensor alert for '%s': %s", sensor.name, e)


class _AlertableSensor(Protocol):
    """Structural type for evaluate()/describe_state().

    PrinterHASensor and LocationHASensor are unrelated SQLAlchemy models —
    one has no base class in common with the other beyond ``Base`` — but both
    carry these five fields with the same meaning, and location_ha_sensor_
    manager.py imports these two functions to reuse the exact same alert
    logic rather than reimplementing it. A concrete PrinterHASensor
    annotation here would be a lie for half of the actual callers.
    """

    kind: str
    unit: str | None
    alert_state: str | None
    alert_above: float | None
    alert_below: float | None


def evaluate(sensor: _AlertableSensor, payload: dict | None) -> SensorReading:
    """Turn one HA state payload into a reading.

    Split out from the manager so the alert rules can be tested without a
    poller, a database or a Home Assistant.
    """
    if payload is None:
        return SensorReading(state=None, value=None, alerting=False, reachable=False)

    state = payload.get("state")
    # HA reports these two for entities whose integration is down. Treating
    # them as a state would make "unavailable" a value the card renders and
    # the thresholds compare against.
    if state in (None, "unknown", "unavailable"):
        return SensorReading(state=None, value=None, alerting=False, reachable=False)

    state = str(state)
    if sensor.kind == "numeric":
        value = as_float(state)
        if value is None:
            # A sensor that used to report numbers and now reports text is
            # not a reading we can place against a threshold.
            return SensorReading(state=state, value=None, alerting=False, reachable=True)
        alerting = (sensor.alert_above is not None and value > sensor.alert_above) or (
            sensor.alert_below is not None and value < sensor.alert_below
        )
        return SensorReading(state=state, value=value, alerting=alerting, reachable=True)

    normalized = state.lower()
    alerting = sensor.alert_state is not None and normalized == sensor.alert_state
    return SensorReading(state=normalized, value=None, alerting=alerting, reachable=True)


def describe_state(sensor: _AlertableSensor, reading: SensorReading) -> str:
    """Human-readable state for a notification body ("open", "31.4 °C")."""
    if sensor.kind == "numeric" and reading.value is not None:
        return f"{reading.value:g} {sensor.unit}".strip() if sensor.unit else f"{reading.value:g}"
    return reading.state or "unknown"


ha_sensor_manager = HASensorManager()