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 and a future web request all emit ``NextTrackRequested`` and a
single reaction acts on it.
"""

from __future__ import annotations

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

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

__all__ = [
    "ActiveFigureChanged",
    "ButtonEvent",
    "ConnectionChanged",
    "DeviceConnected",
    "DeviceDisconnected",
    "Event",
    "EventSource",
    "InputEvent",
    "IntentEvent",
    "LedEffectChanged",
    "LedEffectRequested",
    "NextTrackRequested",
    "PauseRequested",
    "PlayAlbumRequested",
    "PlayFigureRequested",
    "PlayRequested",
    "PlaySeriesLatestRequested",
    "PlaybackChanged",
    "PlaylistFinished",
    "PreviousTrackRequested",
    "RfidTokenRead",
    "RotaryTurned",
    "SetVolumeRequested",
    "StateEvent",
    "TouchButtonPressed",
    "TouchButtonReleased",
    "TrackChanged",
    "VolumeChangeRequested",
    "VolumeChanged",
]

type EventSource = 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 PlaylistFinished(InputEvent):
    """The player reached the end of the playlist."""


@dataclass(frozen=True, slots=True, kw_only=True)
class DeviceConnected(InputEvent):
    port: str


@dataclass(frozen=True, slots=True, kw_only=True)
class DeviceDisconnected(InputEvent):
    port: str
    reason: str | None = None


# -------------------------------------------------------------------------- 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 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 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 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 ConnectionChanged(StateEvent):
    target: Literal["firmware", "mqtt", "lirc"]
    connected: bool