106 lines
3.9 KiB
Python
106 lines
3.9 KiB
Python
"""Home Assistant MQTT Discovery payloads and command dispatch.
|
|
|
|
SECURITY BOUNDARY — this module is the entire remote-control API of the kitchen
|
|
display. Same principle as every other host's agent in this project (Phase 11.4):
|
|
the local LLM never gets a network path to this machine directly. The only chain is
|
|
LLM tool call -> Home Assistant service call -> MQTT -> this dispatcher. There is no
|
|
HTTP listener, no websocket server, no exposed Sway IPC socket here.
|
|
|
|
This is deliberately NOT how the LLM reaches pantry-vision's actual inventory data —
|
|
that's a separate, already-published API (pantry-vision/server.py) that Home
|
|
Assistant or the LLM can call directly for reads/writes to Grocy. This agent only
|
|
ever controls what the physical screen is showing, exactly the same division of
|
|
labour as hosts/thin-client's "Show admin canvas" button vs. admin-canvas's own
|
|
write API.
|
|
"""
|
|
|
|
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"kitchendisplay/{node_id}"
|
|
self.availability_topic = f"{self.base}/availability"
|
|
self._handlers: dict[str, Callable[[str], None]] = {}
|
|
|
|
self.device = {
|
|
"identifiers": [f"kitchendisplay_{node_id}"],
|
|
"name": friendly_name,
|
|
"manufacturer": "SmartestHome",
|
|
"model": "Sway kitchen display",
|
|
"sw_version": __version__,
|
|
}
|
|
|
|
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 register_screens(self, on_show) -> None:
|
|
"""One button per screen — payload is the fixed page fragment (scan/
|
|
inventory/recipes), an enumerated constant handled in main.py, never passed
|
|
through as a URL/argv element unchanged (same rule as every other agent's
|
|
mqtt_discovery.py in this project).
|
|
"""
|
|
for object_id, name, fragment, icon in (
|
|
("show_scan", "Show scan", "scan", "mdi:camera"),
|
|
("show_inventory", "Show inventory", "inventory", "mdi:fridge-outline"),
|
|
("show_recipes", "Show recipes", "recipes", "mdi:chef-hat"),
|
|
):
|
|
self._publish_config(
|
|
"button",
|
|
object_id,
|
|
{
|
|
"name": name,
|
|
"command_topic": self._command_topic(
|
|
f"show/{fragment}", lambda _payload, fragment=fragment: on_show(fragment)
|
|
),
|
|
"icon": icon,
|
|
},
|
|
)
|