musicmouse/python-backend/musicmouse/events.py

"""The event vocabulary.

Three flavours, distinguished by base class:

* :class:`InputEvent`  - something happened (hardware, player).
* :class:`IntentEvent` - something was requested (button, MQTT, web, simulator).
* :class:`StateEvent`  - something changed.

The intent layer is what lets several front-ends drive the same behaviour: a button
press, an MQTT command, a web request and the IR remote all emit ``NextTrackRequested``
and a single reaction acts on it.
"""

from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Literal, TypeAlias

from musicmouse.effects import LedEffect
from musicmouse.hardware import Button, ButtonAction, LedZone, RotaryDirection, TouchButton
from musicmouse.media import Playlist, Track

if TYPE_CHECKING:
    from musicmouse.config import Digit

__all__ = [
    "ActiveFigureChanged",
    "ButtonEvent",
    "CancelTimerRequested",
    "ClockStateChanged",
    "ConnectionChanged",
    "Event",
    "EventSource",
    "FigureSelectionChanged",
    "FigureSelectorExpired",
    "InputEvent",
    "IntentEvent",
    "LedEffectChanged",
    "LedEffectRequested",
    "NextTrackRequested",
    "PauseRequested",
    "PlayAlbumRequested",
    "PlayFigureRequested",
    "PlayRemoteSlotRequested",
    "PlayRequested",
    "PlaySeriesLatestRequested",
    "PlaybackChanged",
    "PlaylistFinished",
    "PreviousTrackRequested",
    "RfidTokenRead",
    "RotaryTurned",
    "SelectorPressed",
    "SelectorTurned",
    "SetTimerRequested",
    "SetVolumeRequested",
    "StateEvent",
    "ToggleClockRequested",
    "ToggleScreenRequested",
    "TouchButtonPressed",
    "TouchButtonReleased",
    "TrackChanged",
    "UserActivity",
    "VolumeChangeRequested",
    "VolumeChanged",
]

EventSource: TypeAlias = Literal["device", "player", "mqtt", "web", "lirc", "simulator", "system"]


@dataclass(frozen=True, slots=True, kw_only=True)
class Event:
    """Base for every event. Keyword-only so subclasses can add required fields."""

    source: EventSource = "system"
    timestamp: float = field(default_factory=time.monotonic, compare=False)


@dataclass(frozen=True, slots=True, kw_only=True)
class InputEvent(Event):
    """Something happened out in the world."""


@dataclass(frozen=True, slots=True, kw_only=True)
class IntentEvent(Event):
    """Something was requested. May come from any front-end."""


@dataclass(frozen=True, slots=True, kw_only=True)
class StateEvent(Event):
    """Something changed. Front-ends mirror these outwards."""


# --------------------------------------------------------------------------- input


@dataclass(frozen=True, slots=True, kw_only=True)
class RfidTokenRead(InputEvent):
    """A tag was read. ``figure`` is ``None`` for the all-zero "removed" tag and for
    tags that match no configured figure (``known`` tells the two apart)."""

    tag_id: bytes
    figure: str | None = None
    known: bool = True

    def __repr__(self) -> str:
        tag = self.tag_id.hex()
        return f"RfidTokenRead({tag}, figure={self.figure!r})"


@dataclass(frozen=True, slots=True, kw_only=True)
class ButtonEvent(InputEvent):
    button: Button
    action: ButtonAction

    def __repr__(self) -> str:
        return f"ButtonEvent({self.button.slug}, {self.action.slug})"


@dataclass(frozen=True, slots=True, kw_only=True)
class TouchButtonPressed(InputEvent):
    button: TouchButton

    def __repr__(self) -> str:
        return f"TouchButtonPressed({self.button.slug})"


@dataclass(frozen=True, slots=True, kw_only=True)
class TouchButtonReleased(InputEvent):
    button: TouchButton

    def __repr__(self) -> str:
        return f"TouchButtonReleased({self.button.slug})"


@dataclass(frozen=True, slots=True, kw_only=True)
class RotaryTurned(InputEvent):
    position: int
    increment: int
    direction: RotaryDirection

    def __repr__(self) -> str:
        return f"RotaryTurned(pos={self.position}, incr={self.increment}, {self.direction.name})"


@dataclass(frozen=True, slots=True, kw_only=True)
class SelectorTurned(InputEvent):
    """The rotary encoder turned while the ring's album picker had it.

    Not a :class:`RotaryTurned`: while the picker is open the knob is not a volume
    knob, and nothing that listens for one should hear it.
    """

    #: Detents, positive for the direction that turns the volume up.
    steps: int


@dataclass(frozen=True, slots=True, kw_only=True)
class SelectorPressed(InputEvent):
    """The rotary encoder was pressed while the ring's album picker had it. Instead of,
    not as well as, a :class:`ButtonEvent` - so it is not published to Home Assistant."""


@dataclass(frozen=True, slots=True, kw_only=True)
class FigureSelectorExpired(InputEvent):
    """The ring's album picker went untouched for long enough to close by itself."""

    figure: str


@dataclass(frozen=True, slots=True, kw_only=True)
class PlaylistFinished(InputEvent):
    """The player reached the end of the playlist."""


# -------------------------------------------------------------------------- intents


@dataclass(frozen=True, slots=True, kw_only=True)
class PlayRequested(IntentEvent):
    pass


@dataclass(frozen=True, slots=True, kw_only=True)
class PauseRequested(IntentEvent):
    pass


@dataclass(frozen=True, slots=True, kw_only=True)
class NextTrackRequested(IntentEvent):
    pass


@dataclass(frozen=True, slots=True, kw_only=True)
class PreviousTrackRequested(IntentEvent):
    pass


@dataclass(frozen=True, slots=True, kw_only=True)
class PlayFigureRequested(IntentEvent):
    """Start a figure's playlist. ``restart=False`` resumes where it left off."""

    figure: str
    restart: bool = True


@dataclass(frozen=True, slots=True, kw_only=True)
class PlayAlbumRequested(IntentEvent):
    """Start any album from the library, figure or not."""

    album_id: str
    track_index: int = 0


@dataclass(frozen=True, slots=True, kw_only=True)
class PlaySeriesLatestRequested(IntentEvent):
    """Start the newest episode of a podcast show.

    The IR remote's number-key mapping assigns a whole show rather than one fixed
    episode, so this is resolved to an actual album fresh on every press.
    """

    series: str


@dataclass(frozen=True, slots=True, kw_only=True)
class PlayRemoteSlotRequested(IntentEvent):
    """Play whatever the top-level ``remote:`` mapping assigns to a number key.

    The IR remote's digits emit this, and so do the number keys of the web front-end's
    keyboard - one reaction resolves the slot for both.
    """

    digit: Digit


@dataclass(frozen=True, slots=True, kw_only=True)
class SeekRequested(IntentEvent):
    #: Seconds from the start of the current track.
    position: float


@dataclass(frozen=True, slots=True, kw_only=True)
class VolumeChangeRequested(IntentEvent):
    delta: int


@dataclass(frozen=True, slots=True, kw_only=True)
class SetVolumeRequested(IntentEvent):
    volume: int


@dataclass(frozen=True, slots=True, kw_only=True)
class ToggleScreenRequested(IntentEvent):
    """Switch the attached monitor off if it is on, on if it is off (``general.screen``)."""


@dataclass(frozen=True, slots=True, kw_only=True)
class ToggleClockRequested(IntentEvent):
    """Show or hide the clock overlay. Only the kiosk acts on it - the overlay is a
    front-end thing, the backend just passes the remote's key along."""


@dataclass(frozen=True, slots=True, kw_only=True)
class SetTimerRequested(IntentEvent):
    """Start the parents' countdown, replacing one already running."""

    #: Wall-clock end, as Unix seconds.
    ends_at: float


@dataclass(frozen=True, slots=True, kw_only=True)
class CancelTimerRequested(IntentEvent):
    """Stop the parents' countdown, or silence it if it is ringing."""


@dataclass(frozen=True, slots=True, kw_only=True)
class UserActivity(IntentEvent):
    """Someone is at the device - a key press or tap in the kiosk browser. Requests
    nothing by itself; it keeps the screen from switching off (``ScreenService``)."""


@dataclass(frozen=True, slots=True, kw_only=True)
class LedEffectRequested(IntentEvent):
    zone: LedZone
    effect: LedEffect

    def __repr__(self) -> str:
        return f"LedEffectRequested({self.zone}, {self.effect}, from={self.source})"


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


@dataclass(frozen=True, slots=True, kw_only=True)
class PlaybackChanged(StateEvent):
    playing: bool
    figure: str | None = None
    playlist: Playlist | None = None


@dataclass(frozen=True, slots=True, kw_only=True)
class TrackChanged(StateEvent):
    index: int
    track: Track | None


@dataclass(frozen=True, slots=True, kw_only=True)
class VolumeChanged(StateEvent):
    volume: int


@dataclass(frozen=True, slots=True, kw_only=True)
class ActiveFigureChanged(StateEvent):
    figure: str | None
    previous: str | None = None


@dataclass(frozen=True, slots=True, kw_only=True)
class FigureSelectionChanged(StateEvent):
    """The ring's album picker opened, moved or closed."""

    figure: str
    #: Index into the figure's albums of the highlighted one.
    selected: int
    count: int
    open: bool


@dataclass(frozen=True, slots=True, kw_only=True)
class LedEffectChanged(StateEvent):
    """Emitted on *every* write to an LED zone, whatever caused it.

    Front-ends publish zone state from this rather than echoing their own commands,
    so Home Assistant keeps showing the strip's real state when a figure animation
    overrides an MQTT-set colour.
    """

    zone: LedZone
    effect: LedEffect
    origin: EventSource

    def __repr__(self) -> str:
        return f"LedEffectChanged({self.zone}, {self.effect}, origin={self.origin})"


@dataclass(frozen=True, slots=True, kw_only=True)
class ClockStateChanged(StateEvent):
    """The parents' timer started, rang or stopped, or today's school day changed.
    The values themselves live on :class:`~musicmouse.app.AppState`."""


@dataclass(frozen=True, slots=True, kw_only=True)
class ConnectionChanged(StateEvent):
    target: Literal["firmware", "mqtt", "lirc"]
    connected: bool