vigilar/vigilar/sensors/bridge.py

"""Sensor Bridge — subprocess that bridges Zigbee2MQTT to the Vigilar internal bus."""

from __future__ import annotations

import logging
import signal
import time
from typing import Any

from vigilar.bus import MessageBus
from vigilar.config import MQTTConfig, VigilarConfig
from vigilar.constants import EventType, SensorProtocol, Topics
from vigilar.sensors.models import SensorEvent
from vigilar.sensors.registry import SensorRegistry
from vigilar.storage.db import get_db_path, get_engine
from vigilar.storage.queries import upsert_sensor_state

log = logging.getLogger(__name__)


def normalize_zigbee_payload(
    raw: dict[str, Any],
) -> list[dict[str, Any]]:
    """Normalize a Zigbee2MQTT payload into a list of internal event dicts.

    Each dict has keys: event_type, and one of (state, value, pct, unit).
    """
    results: list[dict[str, Any]] = []
    ts = int(time.time() * 1000)

    # Contact sensor
    if "contact" in raw:
        contact = raw["contact"]
        state = "CLOSED" if contact else "OPEN"
        event_type = EventType.CONTACT_CLOSED if contact else EventType.CONTACT_OPEN
        results.append({
            "event_type": event_type,
            "state": state,
            "ts": ts,
            "source": "zigbee",
        })

    # Motion / occupancy sensor
    if "occupancy" in raw:
        occupancy = raw["occupancy"]
        state = "ACTIVE" if occupancy else "IDLE"
        event_type = EventType.MOTION_START if occupancy else EventType.MOTION_END
        results.append({
            "event_type": event_type,
            "state": state,
            "ts": ts,
            "source": "zigbee",
        })

    # Temperature
    if "temperature" in raw:
        results.append({
            "event_type": "temperature",
            "value": float(raw["temperature"]),
            "unit": "C",
            "ts": ts,
            "source": "zigbee",
        })

    # Humidity
    if "humidity" in raw:
        results.append({
            "event_type": "humidity",
            "value": float(raw["humidity"]),
            "ts": ts,
            "source": "zigbee",
        })

    # Battery
    if "battery" in raw:
        results.append({
            "event_type": "battery",
            "pct": int(raw["battery"]),
            "ts": ts,
            "source": "zigbee",
        })

    # Link quality / signal strength
    if "linkquality" in raw:
        results.append({
            "event_type": "linkquality",
            "value": int(raw["linkquality"]),
            "ts": ts,
            "source": "zigbee",
        })

    return results


class SensorBridge:
    """Bridges Zigbee2MQTT messages to the internal Vigilar MQTT bus."""

    def __init__(self, config: VigilarConfig) -> None:
        self._config = config
        self._registry = SensorRegistry(config)
        self._z2m_prefix = config.zigbee2mqtt.mqtt_topic_prefix
        self._bus = MessageBus(config.mqtt, client_id="vigilar-sensor-bridge")
        self._engine = get_engine(get_db_path(config.system.data_dir))
        self._running = False

    def start(self) -> None:
        """Connect to MQTT and start processing sensor messages."""
        self._bus.connect()
        self._running = True

        # Subscribe to zigbee2mqtt topics
        z2m_wildcard = f"{self._z2m_prefix}/#"
        self._bus.subscribe(z2m_wildcard, self._handle_z2m_message)

        log.info("Sensor bridge started, subscribed to %s", z2m_wildcard)

    def stop(self) -> None:
        self._running = False
        self._bus.disconnect()
        log.info("Sensor bridge stopped")

    def _handle_z2m_message(self, topic: str, payload: dict[str, Any]) -> None:
        """Handle a Zigbee2MQTT message, normalize and republish."""
        # Topic format: zigbee2mqtt/<friendly_name>
        # Ignore bridge status messages like zigbee2mqtt/bridge/...
        parts = topic.split("/")
        if len(parts) < 2:
            return
        # Skip bridge info topics
        if parts[1] == "bridge":
            return

        friendly_name = parts[1]
        sensor_cfg = self._registry.get_sensor_by_zigbee_name(friendly_name)
        if sensor_cfg is None:
            log.debug("Unknown zigbee device: %s", friendly_name)
            return

        normalized = normalize_zigbee_payload(payload)
        for event_data in normalized:
            event_type = event_data["event_type"]

            # Create SensorEvent
            event = SensorEvent(
                sensor_id=sensor_cfg.id,
                event_type=event_type,
                state=event_data.get("state"),
                value=event_data.get("value"),
                timestamp=event_data["ts"],
                source_protocol=SensorProtocol.ZIGBEE,
            )

            # Publish to internal bus
            topic_out = Topics.sensor_event(sensor_cfg.id, event_type)
            self._bus.publish(topic_out, event_data)

            # Update DB state
            self._persist_state(sensor_cfg.id, event_data)

    def _persist_state(self, sensor_id: str, event_data: dict[str, Any]) -> None:
        """Write normalized event data to sensor_states table."""
        try:
            event_type = event_data["event_type"]

            if "state" in event_data:
                upsert_sensor_state(self._engine, sensor_id, event_type, event_data["state"])
            elif "value" in event_data:
                upsert_sensor_state(self._engine, sensor_id, event_type, event_data["value"])
            elif "pct" in event_data:
                upsert_sensor_state(self._engine, sensor_id, "battery", event_data["pct"])

            # Always update last_seen
            upsert_sensor_state(self._engine, sensor_id, "last_seen", event_data["ts"])
        except Exception:
            log.exception("Failed to persist state for %s", sensor_id)


def run_sensor_bridge(cfg: VigilarConfig) -> None:
    """Subprocess entry point for the sensor bridge."""
    logging.basicConfig(level=getattr(logging, cfg.system.log_level, logging.INFO))

    bridge = SensorBridge(cfg)
    bridge.start()

    # Block until signal
    shutdown = False

    def handle_signal(signum: int, frame: Any) -> None:
        nonlocal shutdown
        shutdown = True

    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)

    try:
        while not shutdown:
            time.sleep(1)
    except KeyboardInterrupt:
        pass
    finally:
        bridge.stop()