musicmouse/python-backend/musicmouse/devices/mouse.py

"""The mouse itself: the object that talks to the firmware.

It owns the physical state - which figure is on the reader, how bright the button
backlights are, which effect each LED zone is showing - and it is the single writer to
all three LED zones. "Last write wins" is therefore a well-defined rule: whoever sets a
zone last, whether a figure animation or an MQTT command, is what the strip shows.

Every write emits :class:`~musicmouse.events.LedEffectChanged`, so front-ends can
publish the strip's real state instead of echoing their own commands back.
"""

from __future__ import annotations

import logging
from dataclasses import replace

from musicmouse.bus import EventBus
from musicmouse.devices.transport import Transport
from musicmouse.devices.wire import (
    FirmwareLog,
    FrameDecoder,
    ProtocolError,
    UnsupportedEffectError,
    encode_button_brightness,
    encode_effect,
)
from musicmouse.effects import OFF, LedEffect
from musicmouse.events import (
    ActiveFigureChanged,
    ButtonEvent,
    ConnectionChanged,
    EventSource,
    InputEvent,
    LedEffectChanged,
    RfidTokenRead,
    RotaryTurned,
    SelectorPressed,
    SelectorTurned,
)
from musicmouse.hardware import NO_FIGURE_TAG, Button, ButtonAction, LedZone, RotaryDirection

_log = logging.getLogger(__name__)

__all__ = ["MusicMouseDevice"]

_BACKLIT_BUTTONS = (Button.LEFT, Button.RIGHT)


class MusicMouseDevice:
    def __init__(
        self,
        bus: EventBus,
        transport: Transport,
        tag_map: dict[bytes, str],
        *,
        port: str = "",
    ) -> None:
        self._bus = bus
        self._transport = transport
        self._tag_map = dict(tag_map)
        self._decoder = FrameDecoder()
        self.port = port

        self._active_figure: str | None = None
        self._button_brightness: float = 0.0
        self._effects: dict[LedZone, LedEffect] = {}
        self._knob_captured = False

    # -------------------------------------------------------------------- state

    @property
    def active_figure(self) -> str | None:
        """The figure currently on the reader, or ``None`` if there is none."""
        return self._active_figure

    @property
    def button_led_brightness(self) -> float:
        return self._button_brightness

    @property
    def connected(self) -> bool:
        return self._transport.connected

    def effect(self, zone: LedZone) -> LedEffect | None:
        return self._effects.get(zone)

    @property
    def knob_captured(self) -> bool:
        return self._knob_captured

    # ------------------------------------------------------------------ actions

    def set_effect(
        self, zone: LedZone, effect: LedEffect, *, origin: EventSource = "system"
    ) -> None:
        """Show ``effect`` on ``zone``. The most recent call wins."""
        try:
            frame = encode_effect(zone, effect)
        except UnsupportedEffectError as exc:
            _log.error("%s", exc)
            return

        self._effects[zone] = effect
        self._transport.write(frame)
        self._bus.emit(
            LedEffectChanged(zone=zone, effect=effect, origin=origin, source="device")
        )

    def set_button_brightness(self, brightness: float, *, origin: EventSource = "system") -> None:
        """Set both prev/next button backlights (``0..1``)."""
        brightness = min(1.0, max(0.0, brightness))
        self._button_brightness = brightness
        for button in _BACKLIT_BUTTONS:
            self._transport.write(encode_button_brightness(button, brightness))
        _log.debug("Button backlights -> %.2f (%s)", brightness, origin)

    def capture_knob(self, captured: bool) -> None:
        """Hand the rotary encoder to the ring's album picker, or give it back.

        While captured, turns and presses come out as :class:`SelectorTurned` and
        :class:`SelectorPressed` instead of :class:`RotaryTurned` and
        :class:`ButtonEvent`, so the volume and Home Assistant never see them. Done here,
        as the frames are decoded, rather than by each listener checking a flag: the
        press that closes the picker would otherwise race whoever handles it after.
        """
        self._knob_captured = captured

    def all_leds_off(self, *, origin: EventSource = "system") -> None:
        for zone in LedZone:
            self.set_effect(zone, OFF(), origin=origin)
        self.set_button_brightness(0.0, origin=origin)

    # ------------------------------------------------------------- link callbacks

    def on_connected(self) -> None:
        """Re-apply memorized state, so a reconnect is invisible from the outside."""
        self._bus.emit(ConnectionChanged(target="firmware", connected=True, source="device"))

        for zone, effect in self._effects.items():
            self._transport.write(encode_effect(zone, effect))
        for button in _BACKLIT_BUTTONS:
            self._transport.write(encode_button_brightness(button, self._button_brightness))
        if self._effects:
            _log.info("Restored %d LED zone(s) after reconnect", len(self._effects))

    def on_disconnected(self, _reason: str = "") -> None:
        # SerialLink already logs the reason; the event only says the link is down.
        self._bus.emit(ConnectionChanged(target="firmware", connected=False, source="device"))

    def feed(self, data: bytes) -> None:
        """Hand bytes from the link to the decoder and publish what comes out."""
        self._decoder.push(data)
        while True:
            try:
                item = self._decoder.take()
            except ProtocolError as exc:
                _log.warning("Discarding bad frame from firmware: %s", exc)
                continue
            if item is None:
                return

            if isinstance(item, FirmwareLog):
                if item.text:
                    _log.info("[firmware] %s", item.text)
            else:
                self._publish(item)

    # ---------------------------------------------------------------- internals

    def _publish(self, event: InputEvent) -> None:
        if isinstance(event, RfidTokenRead):
            self._publish_tag_read(event)
        elif self._knob_captured and isinstance(event, RotaryTurned):
            if event.direction is not RotaryDirection.NONE:
                # The first turn after boot reports no increment, but it was a detent.
                steps = abs(event.increment) or 1
                if event.direction is RotaryDirection.DOWN:
                    steps = -steps
                self._bus.emit(SelectorTurned(steps=steps, source="device"))
        elif (
            self._knob_captured
            and isinstance(event, ButtonEvent)
            and event.button is Button.ROTARY
        ):
            if event.action is ButtonAction.PRESSED:
                self._bus.emit(SelectorPressed(source="device"))
        else:
            self._bus.emit(event)

    def _publish_tag_read(self, event: RfidTokenRead) -> None:
        if event.tag_id == NO_FIGURE_TAG:
            figure, known = None, True
        elif (name := self._tag_map.get(event.tag_id)) is not None:
            figure, known = name, True
        else:
            figure, known = None, False
            _log.warning("Unknown RFID tag %s - not configured as a figure", event.tag_id.hex())

        self._bus.emit(replace(event, figure=figure, known=known))
        if not known:
            # Leave the active figure alone: an unreadable tag is not a removal.
            return

        previous, self._active_figure = self._active_figure, figure
        if previous != figure:
            self._bus.emit(
                ActiveFigureChanged(figure=figure, previous=previous, source="device")
            )