musicmouse/python-backend/musicmouse/reactions/playback.py

"""What the mouse plays, and when.

Physical inputs are turned into *intents*, and the intents are what actually drive the
player. That indirection is the point: an MQTT command or a future web request emits
the same intent and lands in the same handler, so there is one place per behaviour.
"""

from __future__ import annotations

import logging

from musicmouse.app import App
from musicmouse.events import (
    ActiveFigureChanged,
    ButtonEvent,
    NextTrackRequested,
    PauseRequested,
    PlayAlbumRequested,
    PlayFigureRequested,
    PlaylistFinished,
    PlayRemoteSlotRequested,
    PlayRequested,
    PlaySeriesLatestRequested,
    PreviousTrackRequested,
    RotaryTurned,
    SeekRequested,
    SetVolumeRequested,
    VolumeChangeRequested,
)
from musicmouse.hardware import Button, ButtonAction, RotaryDirection
from musicmouse.library.models import Album
from musicmouse.reactions.registry import on

_log = logging.getLogger(__name__)


# ------------------------------------------------------------------ figure on/off


@on(ActiveFigureChanged)
def figure_placed_or_removed(event: ActiveFigureChanged, app: App) -> None:
    if event.figure is None:
        _figure_removed(event.previous, app)
    else:
        app.bus.emit(
            PlayFigureRequested(
                figure=event.figure,
                restart=app.state.last_partially_played_figure != event.figure,
                source="device",
            )
        )


def _figure_removed(previous: str | None, app: App) -> None:
    if app.player.is_playing:
        app.player.pause()
        # Remember where we were, so putting the same figure back resumes.
        app.state.last_partially_played_figure = previous
        _log.info("Figure %r removed mid-playlist", previous)
    else:
        app.state.last_partially_played_figure = None


@on(PlayFigureRequested)
def play_figure(event: PlayFigureRequested, app: App) -> None:
    playlist = app.playlist(event.figure)
    if playlist is None:
        return

    if not event.restart and app.player.playlist is playlist:
        _log.info("Resuming %r", event.figure)
        app.player.play()
        return

    _log.info("Starting %r from the beginning", event.figure)
    app.player.set_playlist(playlist)
    app.player.play_from_start()


def _start_album(app: App, album: Album, track_index: int) -> None:
    # A figure album keeps the figure's own resume bookkeeping honest: it is the same
    # Playlist object either way, because both come from the library index.
    playlist = app.playlists.get(album.figure) if album.figure else album.to_playlist()
    app.player.set_playlist(playlist or album.to_playlist())
    app.player.play_track(track_index)


@on(PlayAlbumRequested)
def play_album(event: PlayAlbumRequested, app: App) -> None:
    """Play any album in the library. This is the web front-end's way in."""
    album = app.library.get(event.album_id)
    if album is None:
        _log.warning("No album %r in the library", event.album_id)
        return
    _start_album(app, album, event.track_index)


@on(PlaySeriesLatestRequested)
def play_series_latest(event: PlaySeriesLatestRequested, app: App) -> None:
    """Play the newest episode of a podcast show - the IR remote's number-key way in."""
    album = app.library.latest_episode(event.series)
    if album is None:
        _log.warning("No episodes for series %r", event.series)
        return
    _start_album(app, album, 0)


@on(PlayRemoteSlotRequested)
def play_remote_slot(event: PlayRemoteSlotRequested, app: App) -> None:
    """Play what a number key is assigned - pressed on the IR remote or a keyboard.

    A slot naming one song starts the album there; the rest of the album follows, as
    for any other start mid-album.
    """
    slot = app.config.remote.get(event.digit)
    if slot is None:
        _log.debug("Number key %s is not assigned", event.digit)
        return
    if slot.target_kind == "series":
        app.bus.emit(PlaySeriesLatestRequested(series=slot.target, source=event.source))
        return
    album = app.library.get(slot.target)
    if album is None:
        _log.warning("Number key %s: no album %r in the library", event.digit, slot.target)
        return
    _start_album(app, album, slot.track_index)


@on(PlaylistFinished)
def playlist_finished(_event: PlaylistFinished, app: App) -> None:
    # Nothing was left half-played, so the next placement starts from the top.
    app.state.last_partially_played_figure = None


# ---------------------------------------------------------------- physical inputs


@on(ButtonEvent)
def button_pressed(event: ButtonEvent, app: App) -> None:
    if event.action is not ButtonAction.PRESSED:
        return
    if event.button is Button.LEFT and app.player.is_playing:
        app.bus.emit(PreviousTrackRequested(source="device"))
    elif event.button is Button.RIGHT and app.player.is_playing:
        app.bus.emit(NextTrackRequested(source="device"))
    # The rotary press is published to Home Assistant by the MQTT service; what it
    # controls is an automation over there, not something this backend decides.


@on(RotaryTurned)
def rotary_turned(event: RotaryTurned, app: App) -> None:
    step = app.config.general.volume_increment * abs(event.increment)
    if event.direction is RotaryDirection.UP:
        app.bus.emit(VolumeChangeRequested(delta=step, source="device"))
    elif event.direction is RotaryDirection.DOWN:
        app.bus.emit(VolumeChangeRequested(delta=-step, source="device"))


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


@on(NextTrackRequested)
def next_track(event: NextTrackRequested, app: App) -> None:
    _log.debug("Next track (%s)", event.source)
    app.player.next_track()


@on(PreviousTrackRequested)
def previous_track(event: PreviousTrackRequested, app: App) -> None:
    _log.debug("Previous track (%s)", event.source)
    app.player.previous_track()


@on(PlayRequested)
def play(_event: PlayRequested, app: App) -> None:
    app.player.play()


@on(PauseRequested)
def pause(_event: PauseRequested, app: App) -> None:
    app.player.pause()


@on(SeekRequested)
def seek(event: SeekRequested, app: App) -> None:
    app.player.seek(event.position)


@on(VolumeChangeRequested)
def change_volume(event: VolumeChangeRequested, app: App) -> None:
    app.player.change_volume(event.delta, source=event.source)


@on(SetVolumeRequested)
def set_volume(event: SetVolumeRequested, app: App) -> None:
    app.player.set_volume(event.volume, source=event.source)