beefy-player/somafm_tui/mediasession.py

"""Desktop media integration, with a platform-neutral face.

The rest of the app only ever talks to this small surface::

    session = create_media_session(actions, enabled=True)
    session.start()
    session.set_track(station=…, title=…, artist=…, album=…, art_url=…)
    session.set_status("Playing")
    session.set_volume(0.7)

which keeps the platform differences in one place:

* **Linux / BSD / any D-Bus desktop** — :class:`somafm_tui.mpris.MprisService`
  exports ``org.mpris.MediaPlayer2.somafmtui.instance<pid>`` on the session
  bus, so GNOME/KDE widgets, lock screens and ``playerctl`` show the station
  and can drive playback.
* **macOS** — mpv *is* the app the system knows about: its Cocoa backend
  registers with ``MPRemoteCommandCenter``/``MPNowPlayingInfoCenter``, which
  is what Control Center, the menu bar and the media keys talk to.  So there
  is nothing to publish here, and the media keys are routed back into the TUI
  by mpv (see :mod:`somafm_tui.player`).  :class:`NowPlayingSession` only
  keeps the state the status bar renders and stays out of the way.

Anything else (or a disabled/absent backend) degrades to an inert session
whose status label is ``off`` — the TUI carries on regardless.
"""

from __future__ import annotations

import sys
from queue import Queue
from typing import Any

__all__ = [
    "MediaSession",
    "NowPlayingSession",
    "create_media_session",
    "media_actions_queue",
]


def media_actions_queue() -> Queue[str]:
    """Queue of commands coming back from the desktop media integration."""
    return Queue()


class MediaSession:
    """Inert media session: keeps state, publishes nothing.

    Used as the base for backends that have no IPC service of their own and as
    the fallback when desktop integration is unavailable.
    """

    #: Short name for the status bar ("mpris", "nowplaying", …).
    label = "media"

    def __init__(self, actions: Queue[str], *, enabled: bool = True) -> None:
        self.actions = actions
        self.enabled = bool(enabled)
        self.available = False
        self.error = ""
        self._status = "Stopped"
        self._volume = 0.7
        self._metadata: dict[str, Any] = {}

    # ----- lifecycle ---------------------------------------------------- #

    def start(self) -> None:
        return None

    def stop(self) -> None:
        return None

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

    @property
    def status_label(self) -> str:
        if not self.enabled:
            return "off"
        if self.available:
            return "on"
        return "error" if self.error else "off"

    def set_status(self, status: str) -> None:
        if status in ("Playing", "Paused", "Stopped"):
            self._status = status

    def set_volume(self, ratio: float) -> None:
        self._volume = max(0.0, min(1.0, float(ratio)))

    def set_track(
        self,
        *,
        station: str,
        title: str,
        artist: str,
        album: str = "",
        art_url: str = "",
        stream_url: str = "",
        track_key: str = "",
    ) -> None:
        self._metadata = {
            "station": station,
            "title": title,
            "artist": artist,
            "album": album,
            "art_url": art_url,
            "stream_url": stream_url,
            "track_key": track_key,
        }


class NowPlayingSession(MediaSession):
    """macOS: mpv owns the system Now Playing session, we just mirror state.

    mpv's Cocoa backend publishes the live ICY title (and play/pause state) to
    Control Center and receives the media keys; :mod:`somafm_tui.player` routes
    those key presses back into this app.  ``available`` is false when the audio
    backend is not mpv, because then nobody publishes anything.
    """

    label = "nowplaying"

    def __init__(
        self,
        actions: Queue[str],
        *,
        enabled: bool = True,
        backend: str | None = "mpv",
    ) -> None:
        super().__init__(actions, enabled=enabled)
        self.backend = backend
        self.available = bool(enabled) and backend == "mpv"

    @property
    def status_label(self) -> str:
        if not self.enabled:
            return "off"
        return "on" if self.available else "off"


def create_media_session(
    actions: Queue[str],
    *,
    enabled: bool = True,
    backend: str | None = None,
    platform: str | None = None,
) -> MediaSession:
    """Pick the right desktop media backend for this platform.

    ``platform``/``backend`` exist so the choice is testable without touching
    the real machine.
    """
    system = sys.platform if platform is None else platform
    if system == "darwin":
        return NowPlayingSession(actions, enabled=enabled, backend=backend)

    from .mpris import MprisService

    return MprisService(actions, enabled=enabled)