309 lines
12 KiB
Python
309 lines
12 KiB
Python
"""Home Assistant MQTT Discovery payloads and command dispatch.
|
|
|
|
SECURITY BOUNDARY — this module is the entire remote-control API of the Steam TV box.
|
|
|
|
Same principle as hosts/thin-client/agent/thinclient_agent/mqtt_discovery.py and the
|
|
touch panel's, 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.
|
|
|
|
It is worth restating on this host in particular, because it is the one machine here
|
|
with a real GPU, a Steam login and a games library on it — i.e. the one with something
|
|
worth taking. Nothing in this image listens on a port except sshd (key-only) and wayvnc
|
|
(password-mandatory, fails closed). Steam's own remote-play ports are not opened by
|
|
anything here; if you want them, that is a deliberate decision to make in the firewall,
|
|
not a default this image ships.
|
|
|
|
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, room: str = ""):
|
|
self.client = client
|
|
self.node_id = node_id
|
|
self.friendly_name = friendly_name
|
|
self.room = (room or "").strip()
|
|
self.base = f"steamtv/{node_id}"
|
|
self.availability_topic = f"{self.base}/availability"
|
|
self.media_state_topic = f"{self.base}/media/state"
|
|
self.session_state_topic = f"{self.base}/session/state"
|
|
self._handlers: dict[str, Callable[[str], None]] = {}
|
|
|
|
self.device = {
|
|
"identifiers": [f"steamtv_{node_id}"],
|
|
"name": friendly_name,
|
|
"manufacturer": "SmartestHome",
|
|
"model": "Steam TV box",
|
|
"sw_version": __version__,
|
|
}
|
|
# Which room this physically sits in, as an HA area_id. `suggested_area` is
|
|
# honoured by HA only when the device is FIRST discovered — moving a device
|
|
# later means moving it in HA too, this cannot un-file it. Omitted entirely
|
|
# when unset, because an empty suggested_area is not the same request as no
|
|
# suggestion. See docs/rooms-and-endpoints.md.
|
|
if self.room:
|
|
self.device["suggested_area"] = self.room
|
|
|
|
# --- 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)
|
|
|
|
# --- media --------------------------------------------------------------
|
|
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:play-circle",
|
|
},
|
|
)
|
|
|
|
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",
|
|
},
|
|
)
|
|
|
|
# --- apps / workspaces --------------------------------------------------
|
|
def register_app_launchers(self, apps, on_launch) -> None:
|
|
for key, app in apps.items():
|
|
self._publish_config(
|
|
"button",
|
|
f"launch_{key}",
|
|
{
|
|
"name": f"Launch {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)
|
|
|
|
# --- session mode -------------------------------------------------------
|
|
def register_session_mode(self, modes, on_select, on_stop_media) -> None:
|
|
"""The "what is this box doing" surface — see session_mode.py.
|
|
|
|
A sensor and a select rather than one entity, because the two are not the same
|
|
question. The sensor reports four states (gaming / steam / media / idle); the
|
|
select offers only the two that are meaningful to *ask for*. "idle" is not
|
|
something you can request, and "steam" (Steam up but not focused) is a
|
|
transitional state nobody sets on purpose.
|
|
"""
|
|
self._publish_config(
|
|
"sensor",
|
|
"session_mode",
|
|
{
|
|
"name": "Session",
|
|
"state_topic": self.session_state_topic,
|
|
"value_template": "{{ value_json.mode }}",
|
|
"json_attributes_topic": self.session_state_topic,
|
|
"icon": "mdi:gamepad-variant",
|
|
},
|
|
)
|
|
|
|
self._publish_config(
|
|
"select",
|
|
"session_mode_select",
|
|
{
|
|
"name": "Mode",
|
|
"command_topic": self._command_topic("session/mode/set", on_select),
|
|
"state_topic": self.session_state_topic,
|
|
"value_template": "{{ value_json.mode }}",
|
|
"options": list(modes),
|
|
"icon": "mdi:gamepad-variant",
|
|
},
|
|
)
|
|
|
|
# The only thing in this whole surface that shuts something down, and therefore
|
|
# its own explicit button rather than a side effect of switching mode. See
|
|
# SessionMode.select()'s docstring for why the mode switch is additive.
|
|
self._publish_config(
|
|
"button",
|
|
"stop_media",
|
|
{
|
|
"name": "Stop media apps",
|
|
"command_topic": self._command_topic(
|
|
"session/media/stop", lambda _payload: on_stop_media()
|
|
),
|
|
"icon": "mdi:close-circle-outline",
|
|
},
|
|
)
|
|
|
|
def publish_session_state(self, mode: str, attributes: dict | None = None) -> None:
|
|
payload = {"mode": mode}
|
|
if attributes:
|
|
payload.update(attributes)
|
|
self.client.publish(
|
|
self.session_state_topic, json.dumps(payload), qos=1, retain=True
|
|
)
|
|
|
|
# --- audio output -------------------------------------------------------
|
|
def register_audio_output(self, options, current, on_select) -> None:
|
|
self._publish_config(
|
|
"select",
|
|
"audio_output",
|
|
{
|
|
"name": "Audio output",
|
|
"command_topic": self._command_topic("audio/output/set", on_select),
|
|
"state_topic": f"{self.base}/audio/output/state",
|
|
"options": list(options),
|
|
"icon": "mdi:speaker",
|
|
},
|
|
)
|
|
self.publish_audio_output(current)
|
|
|
|
def publish_audio_output(self, option: str) -> None:
|
|
self.client.publish(
|
|
f"{self.base}/audio/output/state", option, qos=1, retain=True
|
|
)
|
|
|
|
# --- display power ------------------------------------------------------
|
|
def register_display_power(self, on_command, initial: bool = True) -> None:
|
|
self._publish_config(
|
|
"switch",
|
|
"display",
|
|
{
|
|
"name": "Display",
|
|
"command_topic": self._command_topic("display/set", on_command),
|
|
"state_topic": f"{self.base}/display/state",
|
|
"payload_on": "ON",
|
|
"payload_off": "OFF",
|
|
"icon": "mdi:television",
|
|
},
|
|
)
|
|
self.publish_display_power(initial)
|
|
|
|
def publish_display_power(self, on: bool) -> None:
|
|
self.client.publish(
|
|
f"{self.base}/display/state", "ON" if on else "OFF", qos=1, retain=True
|
|
)
|