musicmouse/python-backend/musicmouse/services/mqtt/triggers.py

"""Button, touch and RFID events, published for Home Assistant to automate on.

This is what replaces the old direct ``hass-client`` calls. The backend no longer
knows that pressing the rotary encoder toggles ``light.kinderzimmer_fluter`` or that
the left ear means pink - it reports what happened, and the automation lives in Home
Assistant where it can be changed without a deploy.
"""

from __future__ import annotations

from typing import Any, ClassVar

from musicmouse.bus import EventBus
from musicmouse.config import MqttConfig
from musicmouse.events import ButtonEvent, RfidTokenRead, TouchButtonPressed, TouchButtonReleased
from musicmouse.hardware import Button, ButtonAction, TouchButton
from musicmouse.services.mqtt.entity import Entity

__all__ = ["ButtonTrigger", "TagScanner", "TouchTrigger", "trigger_entities"]

#: Which button actions are worth automating on, and the HA trigger type for each.
BUTTON_ACTIONS: dict[ButtonAction, str] = {
    ButtonAction.PRESSED: "button_short_press",
    ButtonAction.DOUBLE_CLICKED: "button_double_press",
    ButtonAction.LONG_PRESSED: "button_long_press",
}


class _Trigger(Entity):
    component = "device_automation"

    @property
    def discovery_topic(self) -> str:
        # Device triggers are addressed by node id + object id, not by unique id.
        return (
            f"{self.config.discovery_prefix}/device_automation/"
            f"{self.config.device_id}/{self.object_id}/config"
        )

    @property
    def trigger_topic(self) -> str:
        return f"{self.config.base_topic}/trigger/{self.object_id}"

    def _payload(self, trigger_type: str, subtype: str) -> dict[str, Any]:
        return {
            "automation_type": "trigger",
            "topic": self.trigger_topic,
            "type": trigger_type,
            "subtype": subtype,
            "device": self.device_block(),
        }


class ButtonTrigger(_Trigger):
    def __init__(
        self, bus: EventBus, config: MqttConfig, button: Button, action: ButtonAction
    ) -> None:
        self.button = button
        self.action = action
        super().__init__(
            bus,
            config,
            object_id=f"{button.slug}_{action.slug}",
            name=f"{button.slug} {action.slug}",
        )

    def discovery_payload(self) -> dict[str, Any]:
        return self._payload(BUTTON_ACTIONS[self.action], self.button.slug)

    def subscribe(self) -> None:
        self.bus.subscribe(ButtonEvent, self._on_button)

    async def _on_button(self, event: ButtonEvent) -> None:
        if event.button is self.button and event.action is self.action:
            await self.publish(self.trigger_topic, self.action.slug)


class TouchTrigger(_Trigger):
    TYPES: ClassVar[dict[bool, str]] = {
        True: "button_short_press",
        False: "button_short_release",
    }

    def __init__(
        self, bus: EventBus, config: MqttConfig, button: TouchButton, *, pressed: bool
    ) -> None:
        self.button = button
        self.pressed = pressed
        suffix = "touched" if pressed else "released"
        super().__init__(
            bus,
            config,
            object_id=f"{button.slug}_{suffix}",
            name=f"{button.slug} {suffix}",
        )

    def discovery_payload(self) -> dict[str, Any]:
        return self._payload(self.TYPES[self.pressed], self.button.slug)

    def subscribe(self) -> None:
        if self.pressed:
            self.bus.subscribe(TouchButtonPressed, self._on_touch)
        else:
            self.bus.subscribe(TouchButtonReleased, self._on_touch)

    async def _on_touch(self, event: TouchButtonPressed | TouchButtonReleased) -> None:
        if event.button is self.button:
            await self.publish(self.trigger_topic, self.button.slug)


class TagScanner(Entity):
    """The RFID reader, as an HA tag scanner - the natural fit for "a tag was read"."""

    component = "tag"

    def __init__(self, bus: EventBus, config: MqttConfig) -> None:
        super().__init__(bus, config, object_id="tag", name="Music Mouse Reader")

    @property
    def discovery_topic(self) -> str:
        return f"{self.config.discovery_prefix}/tag/{self.config.device_id}/config"

    @property
    def scan_topic(self) -> str:
        return f"{self.config.base_topic}/tag"

    def discovery_payload(self) -> dict[str, Any]:
        return {
            "topic": self.scan_topic,
            "value_template": "{{ value_json.tag_id }}",
            "device": self.device_block(),
        }

    def subscribe(self) -> None:
        self.bus.subscribe(RfidTokenRead, self._on_tag)

    async def _on_tag(self, event: RfidTokenRead) -> None:
        await self.publish(
            self.scan_topic,
            {"tag_id": event.tag_id.hex(), "figure": event.figure, "known": event.known},
        )


def trigger_entities(bus: EventBus, config: MqttConfig) -> list[Entity]:
    return [
        *(
            ButtonTrigger(bus, config, button, action)
            for button in Button
            for action in BUTTON_ACTIONS
        ),
        *(
            TouchTrigger(bus, config, button, pressed=pressed)
            for button in TouchButton
            for pressed in (True, False)
        ),
        TagScanner(bus, config),
    ]