SmartestHome/hosts/touch-panel/agent/touchpanel_agent/mqtt_discovery.py

"""Home Assistant MQTT Discovery payloads and command dispatch.

SECURITY BOUNDARY — this module is the entire remote-control API of the touch panel.

Same principle as hosts/thin-client/agent/thinclient_agent/mqtt_discovery.py (Phase
11.4), applied to this device: the local LLM never gets a network path to this
machine. The only chain is: LLM tool call -> Home Assistant service call -> MQTT ->
this dispatcher. That property holds only as long as this stays the sole inbound
control surface: no HTTP listener, no websocket server, no exposed Sway IPC socket, no
shell endpoint. A new feature belongs as another entity below, not as another
listener.

Every command handler here is a fixed, enumerated action. A payload never becomes an
argv element, a shell string, or a URL host — see the launch table in main.py, which
builds every command from local constants and uses the payload only to pick between
known values.
"""

from __future__ import annotations

import json
import logging
from typing import Callable

from . import __version__

log = logging.getLogger(__name__)

DISCOVERY_PREFIX = "homeassistant"


class Discovery:
    def __init__(self, client, node_id: str, friendly_name: str):
        self.client = client
        self.node_id = node_id
        self.friendly_name = friendly_name
        self.base = f"touchpanel/{node_id}"
        self.availability_topic = f"{self.base}/availability"
        self.media_state_topic = f"{self.base}/media/state"
        self._handlers: dict[str, Callable[[str], None]] = {}

        self.device = {
            "identifiers": [f"touchpanel_{node_id}"],
            "name": friendly_name,
            "manufacturer": "SmartestHome",
            "model": "Sway touch panel",
            "sw_version": __version__,
        }

    # --- plumbing -----------------------------------------------------------
    def _publish_config(self, component: str, object_id: str, payload: dict) -> None:
        payload = {
            "availability_topic": self.availability_topic,
            "device": self.device,
            "unique_id": f"{self.node_id}_{object_id}",
            **payload,
        }
        topic = f"{DISCOVERY_PREFIX}/{component}/{self.node_id}/{object_id}/config"
        self.client.publish(topic, json.dumps(payload), qos=1, retain=True)

    def _command_topic(self, suffix: str, handler) -> str:
        topic = f"{self.base}/{suffix}"
        self._handlers[topic] = handler
        return topic

    def subscribe_all(self) -> None:
        for topic in self._handlers:
            self.client.subscribe(topic, qos=1)

    def dispatch(self, topic: str, payload: str) -> None:
        handler = self._handlers.get(topic)
        if handler is None:
            log.warning("no handler for %s", topic)
            return
        try:
            handler(payload)
        except Exception:
            log.exception("handler for %s failed", topic)

    def publish_available(self, available: bool = True) -> None:
        self.client.publish(
            self.availability_topic,
            "online" if available else "offline",
            qos=1,
            retain=True,
        )

    def publish_media_state(self, state: dict) -> None:
        self.client.publish(self.media_state_topic, json.dumps(state), qos=0, retain=True)

    # --- entities -----------------------------------------------------------
    def register_media_player(self, on_command, on_volume) -> None:
        command_topic = self._command_topic("media/command", on_command)
        volume_topic = self._command_topic("media/volume/set", lambda p: on_volume(float(p)))

        # Core Home Assistant's MQTT integration has NO media_player platform — see
        # hosts/thin-client/README.md's identical caveat. The button/number entities
        # below give the same transport control with stock HA.
        self._publish_config(
            "media_player",
            "media",
            {
                "name": "Media",
                "state_topic": self.media_state_topic,
                "state_template": "{{ value_json.state }}",
                "command_topic": command_topic,
                "volume_command_topic": volume_topic,
                "volume_state_topic": self.media_state_topic,
                "volume_template": "{{ value_json.volume }}",
                "title_template": "{{ value_json.title }}",
                "artist_template": "{{ value_json.artist }}",
                "album_template": "{{ value_json.album }}",
            },
        )

        for object_id, name, payload, icon in (
            ("media_play_pause", "Play/pause", "PLAY_PAUSE", "mdi:play-pause"),
            ("media_next", "Next track", "NEXT", "mdi:skip-next"),
            ("media_previous", "Previous track", "PREVIOUS", "mdi:skip-previous"),
            ("media_stop", "Stop", "STOP", "mdi:stop"),
        ):
            self._publish_config(
                "button",
                object_id,
                {
                    "name": name,
                    "command_topic": command_topic,
                    "payload_press": payload,
                    "icon": icon,
                },
            )

        self._publish_config(
            "sensor",
            "media_state",
            {
                "name": "Playback state",
                "state_topic": self.media_state_topic,
                "value_template": "{{ value_json.state }}",
                "json_attributes_topic": self.media_state_topic,
                "icon": "mdi:spotify",
            },
        )

        self._publish_config(
            "number",
            "media_volume",
            {
                "name": "Volume",
                "command_topic": volume_topic,
                "state_topic": self.media_state_topic,
                "value_template": "{{ value_json.volume }}",
                "min": 0,
                "max": 1,
                "step": 0.05,
                "mode": "slider",
                "icon": "mdi:volume-high",
            },
        )

    def register_app_launchers(self, apps, on_launch) -> None:
        for key, app in apps.items():
            self._publish_config(
                "button",
                f"launch_{key}",
                {
                    "name": f"Show {app.name}",
                    "command_topic": self._command_topic(
                        f"app/{key}/launch",
                        lambda _payload, key=key: on_launch(key),
                    ),
                    "icon": app.icon,
                },
            )

    def register_workspace_select(self, workspaces, on_workspace, state_topic_value) -> None:
        self._publish_config(
            "select",
            "workspace",
            {
                "name": "Screen",
                "command_topic": self._command_topic("workspace/set", on_workspace),
                "state_topic": f"{self.base}/workspace/state",
                "options": list(workspaces),
                "icon": "mdi:view-dashboard",
            },
        )
        self.client.publish(
            f"{self.base}/workspace/state", state_topic_value, qos=1, retain=True
        )

    def publish_workspace(self, name: str) -> None:
        self.client.publish(f"{self.base}/workspace/state", name, qos=1, retain=True)