feat(macos): drive the system Now Playing session through mpv

macOS has no D-Bus/MPRIS: the audio engine is what the OS media UI talks
to. mpv's Cocoa backend registers with MPRemoteCommandCenter and
MPNowPlayingInfoCenter, so it already publishes the live ICY title and
receives the media keys.

Add a platform-neutral media session (mediasession.py) so the app keeps a
single interface: MprisService on D-Bus platforms, NowPlayingSession on
macOS. Map mpv's NEXT/PREV/PLAY/PLAYONLY/PAUSEONLY/STOP keys onto
script-messages and read them back as client-message IPC events, so the
hardware media keys and Control Center drive the TUI (station changes,
play/pause/stop) instead of fighting it.

--no-mpris (now also --no-media) disables the media keys too. macOS-only
mpv options are gated on sys.platform so the Linux command line is
unchanged; mpv aborts on unknown flags.
This commit is contained in:
beefcarver committed 2026-10-30 18:43:03 -04:00
1 parent d83a9ef03d
commit 9c722c5f61
8 files changed
+415 -37

No files matched your search

+7 -3
View File
@@ -34,8 +34,11 @@ def _build_parser() -> argparse.ArgumentParser:
parser.add_argument("--volume", type=int, metavar="0-100", help="initial volume") parser.add_argument("--volume", type=int, metavar="0-100", help="initial volume")
parser.add_argument( parser.add_argument(
"--no-mpris", "--no-mpris",
"--no-media",
dest="no_mpris",
action="store_true", action="store_true",
help="do not announce playback over MPRIS", help="do not publish to the desktop media session (MPRIS on Linux, "
"Now Playing on macOS) and ignore the media keys",
) )
parser.add_argument( parser.add_argument(
"--offline", "--offline",
@@ -51,7 +54,8 @@ def _build_parser() -> argparse.ArgumentParser:
"--spectrum-source", "--spectrum-source",
choices=SOURCE_MODES, choices=SOURCE_MODES,
default="auto", default="auto",
help="spectrum audio source: auto (monitor→stream→sim), monitor, " help="spectrum audio source: auto (monitor→stream→sim), monitor "
"(parec on Linux, an avfoundation loopback device on macOS), "
"stream (muted second ffmpeg decode), or sim", "stream (muted second ffmpeg decode), or sim",
) )
parser.add_argument( parser.add_argument(
@@ -109,7 +113,7 @@ def main(argv: Sequence[str] | None = None) -> int:
autoplay=args.autoplay, autoplay=args.autoplay,
station_id=args.station, station_id=args.station,
volume=args.volume, volume=args.volume,
mpris=not args.no_mpris, media=not args.no_mpris,
offline=args.offline, offline=args.offline,
audio=not args.no_audio, audio=not args.no_audio,
spectrum_source=args.spectrum_source, spectrum_source=args.spectrum_source,
+43 -28
View File
@@ -20,7 +20,7 @@ from textual.widgets import Input
from . import api, artwork, bandcamp, config from . import api, artwork, bandcamp, config
from .api import Station from .api import Station
from .fallback import fallback_channels from .fallback import fallback_channels
from .mpris import MprisService from .mediasession import create_media_session, media_actions_queue
from .player import Player from .player import Player
from .spectrum import SpectrumEngine from .spectrum import SpectrumEngine
from .widgets.header import HeaderBar from .widgets.header import HeaderBar
@@ -92,7 +92,8 @@ class SomaFMApp(App[None]):
autoplay: bool | None = None, autoplay: bool | None = None,
station_id: str | None = None, station_id: str | None = None,
volume: int | None = None, volume: int | None = None,
mpris: bool = True, media: bool = True,
mpris: bool | None = None,
offline: bool = False, offline: bool = False,
audio: bool = True, audio: bool = True,
spectrum_source: str = "auto", spectrum_source: str = "auto",
@@ -106,7 +107,11 @@ class SomaFMApp(App[None]):
# One-off command line overrides are deliberately *not* written back # One-off command line overrides are deliberately *not* written back
# into the config file (only an explicit --volume is remembered). # into the config file (only an explicit --volume is remembered).
self.autoplay = self.config.autoplay if autoplay is None else autoplay self.autoplay = self.config.autoplay if autoplay is None else autoplay
self.mpris_enabled = self.config.mpris and mpris # ``mpris`` is the historical keyword; desktop media integration is
# MPRIS on Linux and Now Playing (via mpv) on macOS.
if mpris is not None:
media = mpris
self.media_enabled = self.config.mpris and media
if volume is not None: if volume is not None:
self.config.volume = max(0, min(100, int(volume))) self.config.volume = max(0, min(100, int(volume)))
@@ -138,10 +143,15 @@ class SomaFMApp(App[None]):
self._closed = False self._closed = False
self._tick_timer = None self._tick_timer = None
self.player = Player() self.player = Player(media_keys=self.media_enabled)
self.spectrum = SpectrumEngine(source_mode=spectrum_source) self.spectrum = SpectrumEngine(source_mode=spectrum_source)
self.mpris_actions: queue.Queue[str] = queue.Queue() self.media_actions: queue.Queue[str] = media_actions_queue()
self.mpris = MprisService(self.mpris_actions, enabled=self.mpris_enabled) self.media = create_media_session(
self.media_actions,
enabled=self.media_enabled,
# No audio engine means nothing to publish from.
backend=Player.detect_backend() if audio else None,
)
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
# layout # layout
@@ -164,7 +174,7 @@ class SomaFMApp(App[None]):
self.player.start() self.player.start()
self.player.set_volume(self.volume) self.player.set_volume(self.volume)
self.spectrum.start() self.spectrum.start()
self.mpris.start() self.media.start()
self._http = api.make_client() self._http = api.make_client()
@@ -179,7 +189,7 @@ class SomaFMApp(App[None]):
if cached: if cached:
self.current.history = cached self.current.history = cached
self.current.now_playing = cached[0] self.current.now_playing = cached[0]
self._update_mpris() self._update_media()
self._tick_timer = self.set_interval(0.05, self._tick) self._tick_timer = self.set_interval(0.05, self._tick)
@@ -304,7 +314,7 @@ class SomaFMApp(App[None]):
self.playing_state = "loading" self.playing_state = "loading"
self._set_notice(f"tuning in to {station.title}…", 5) self._set_notice(f"tuning in to {station.title}…", 5)
self._save_config() self._save_config()
self._update_mpris() self._update_media()
self.query_one(StagePanel).refresh_stage() self.query_one(StagePanel).refresh_stage()
self.query_one(HeaderBar).refresh() self.query_one(HeaderBar).refresh()
@@ -330,7 +340,7 @@ class SomaFMApp(App[None]):
except Exception as exc: except Exception as exc:
self.playing_state = "error" self.playing_state = "error"
self._set_notice(f"could not resolve stream: {exc}", 8) self._set_notice(f"could not resolve stream: {exc}", 8)
self._update_mpris() self._update_media()
self.query_one(StagePanel).refresh_stage() self.query_one(StagePanel).refresh_stage()
return return
@@ -339,7 +349,7 @@ class SomaFMApp(App[None]):
if not urls: if not urls:
self.playing_state = "error" self.playing_state = "error"
self._set_notice("no stream mirrors available", 8) self._set_notice("no stream mirrors available", 8)
self._update_mpris() self._update_media()
self.query_one(StagePanel).refresh_stage() self.query_one(StagePanel).refresh_stage()
return return
@@ -372,7 +382,7 @@ class SomaFMApp(App[None]):
self._set_notice( self._set_notice(
f"♪ {songs[0].display_artist} — {songs[0].display_title}", 5 f"♪ {songs[0].display_artist} — {songs[0].display_title}", 5
) )
self._update_mpris() self._update_media()
self.query_one(StagePanel).refresh_stage() self.query_one(StagePanel).refresh_stage()
self.query_one(HeaderBar).refresh() self.query_one(HeaderBar).refresh()
@@ -384,7 +394,7 @@ class SomaFMApp(App[None]):
if time.monotonic() - self._last_history_fetch >= self._history_refresh_seconds: if time.monotonic() - self._last_history_fetch >= self._history_refresh_seconds:
self._fetch_history(self.current.id) self._fetch_history(self.current.id)
def _update_mpris(self) -> None: def _update_media(self) -> None:
station = self.current station = self.current
if station is None: if station is None:
return return
@@ -395,7 +405,7 @@ class SomaFMApp(App[None]):
else "SomaFM" else "SomaFM"
) )
if song is not None: if song is not None:
self.mpris.set_track( self.media.set_track(
station=station.title, station=station.title,
title=song.display_title, title=song.display_title,
artist=song.display_artist, artist=song.display_artist,
@@ -405,7 +415,7 @@ class SomaFMApp(App[None]):
track_key=f"{station.id}|{song.key()}", track_key=f"{station.id}|{song.key()}",
) )
else: else:
self.mpris.set_track( self.media.set_track(
station=station.title, station=station.title,
title=station.last_playing or station.title, title=station.last_playing or station.title,
artist=station.dj or "SomaFM", artist=station.dj or "SomaFM",
@@ -414,8 +424,8 @@ class SomaFMApp(App[None]):
stream_url=self._stream_url, stream_url=self._stream_url,
track_key=station.id, track_key=station.id,
) )
self.mpris.set_status(self._mpris_status()) self.media.set_status(self._media_status())
self.mpris.set_volume(self.volume / 100.0) self.media.set_volume(self.volume / 100.0)
self._update_bandcamp() self._update_bandcamp()
self._update_artwork() self._update_artwork()
@@ -518,7 +528,7 @@ class SomaFMApp(App[None]):
self._set_notice("artwork unavailable — using mosaic", 5) self._set_notice("artwork unavailable — using mosaic", 5)
self._refresh_art() self._refresh_art()
def _mpris_status(self) -> str: def _media_status(self) -> str:
if self.playing_state == "paused": if self.playing_state == "paused":
return "Paused" return "Paused"
if self.playing_state in ("playing", "loading", "buffering"): if self.playing_state in ("playing", "loading", "buffering"):
@@ -534,7 +544,7 @@ class SomaFMApp(App[None]):
if value > 0: if value > 0:
self._volume_before_mute = value self._volume_before_mute = value
self.player.set_volume(value) self.player.set_volume(value)
self.mpris.set_volume(value / 100.0) self.media.set_volume(value / 100.0)
self._save_config() self._save_config()
self.query_one(StatusBar).refresh() self.query_one(StatusBar).refresh()
self.query_one(HeaderBar).refresh() self.query_one(HeaderBar).refresh()
@@ -556,7 +566,7 @@ class SomaFMApp(App[None]):
def _tick_body(self) -> None: def _tick_body(self) -> None:
self.tick_count += 1 self.tick_count += 1
self._drain_player() self._drain_player()
self._drain_mpris() self._drain_media()
if self.notice and time.monotonic() >= self._notice_until: if self.notice and time.monotonic() >= self._notice_until:
self.notice = "" self.notice = ""
@@ -585,37 +595,42 @@ class SomaFMApp(App[None]):
if event.volume is not None and event.volume != self.volume: if event.volume is not None and event.volume != self.volume:
self.volume = event.volume self.volume = event.volume
self.config.volume = event.volume self.config.volume = event.volume
self.mpris.set_volume(self.volume / 100.0) self.media.set_volume(self.volume / 100.0)
self._save_config() self._save_config()
continue continue
if event.kind == "notice": if event.kind == "notice":
if event.detail: if event.detail:
self._set_notice(event.detail, 6) self._set_notice(event.detail, 6)
continue continue
if event.kind == "media":
# Media keys / Control Center on macOS, routed via mpv.
if event.detail:
self.media_actions.put(event.detail)
continue
self.playing_state = event.state or self.playing_state self.playing_state = event.state or self.playing_state
if event.url: if event.url:
self._stream_url = event.url self._stream_url = event.url
if event.state in ("playing", "loading", "buffering"): if event.state in ("playing", "loading", "buffering"):
self.mpris.set_status("Playing") self.media.set_status("Playing")
if event.url: if event.url:
self.spectrum.set_source(event.url) self.spectrum.set_source(event.url)
elif event.state == "paused": elif event.state == "paused":
self.mpris.set_status("Paused") self.media.set_status("Paused")
self.spectrum.set_source(None) self.spectrum.set_source(None)
elif event.state in ("stopped", "error"): elif event.state in ("stopped", "error"):
self.mpris.set_status("Stopped") self.media.set_status("Stopped")
self.spectrum.set_source(None) self.spectrum.set_source(None)
if event.detail: if event.detail:
self._set_notice(event.detail, 6) self._set_notice(event.detail, 6)
self.query_one(StagePanel).refresh_stage() self.query_one(StagePanel).refresh_stage()
def _drain_mpris(self) -> None: def _drain_media(self) -> None:
while True: while True:
try: try:
action = self.mpris_actions.get_nowait() action = self.media_actions.get_nowait()
except queue.Empty: except queue.Empty:
return return
if action == "play": if action == "play":
@@ -719,7 +734,7 @@ class SomaFMApp(App[None]):
self.player.stop() self.player.stop()
self.spectrum.set_source(None) self.spectrum.set_source(None)
self.playing_state = "stopped" self.playing_state = "stopped"
self.mpris.set_status("Stopped") self.media.set_status("Stopped")
self._set_notice("stopped", 2) self._set_notice("stopped", 2)
self.query_one(StagePanel).refresh_stage() self.query_one(StagePanel).refresh_stage()
self.query_one(HeaderBar).refresh() self.query_one(HeaderBar).refresh()
@@ -853,7 +868,7 @@ class SomaFMApp(App[None]):
self._tick_timer.stop() self._tick_timer.stop()
self._save_config() self._save_config()
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
self.mpris.stop() self.media.stop()
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
self.spectrum.stop() self.spectrum.stop()
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
+160
View File
@@ -0,0 +1,160 @@
"""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)
+73 -3
View File
@@ -20,6 +20,7 @@ import shlex
import shutil import shutil
import signal import signal
import subprocess import subprocess
import sys
import threading import threading
import time import time
from collections.abc import Callable from collections.abc import Callable
@@ -28,12 +29,35 @@ from pathlib import Path
from typing import Any from typing import Any
from .api import USER_AGENT from .api import USER_AGENT
from .config import RUNTIME_DIR from .config import RUNTIME_DIR, atomic_write_text
#: mpv input bindings that turn the system media keys into TUI commands.
#:
#: On macOS mpv's Cocoa backend implements MPRemoteCommandCenter, so the
#: hardware/keyboard media keys arrive as these special key names. Instead of
#: letting mpv handle them (which would fight the TUI and desync the station
#: list, the spectrum and the desktop "now playing" state) we bounce them back
#: over the IPC socket as ``client-message`` events.
MEDIA_KEY_BINDINGS = """\
# Generated by somafm-tui -- do not edit; it is rewritten on every launch.
PLAY script-message somafm toggle
PLAYONLY script-message somafm play
PAUSEONLY script-message somafm pause
STOP script-message somafm stop
NEXT script-message somafm next
PREV script-message somafm prev
"""
#: Commands a media key may ask for; everything else is ignored.
MEDIA_COMMANDS = frozenset({"play", "pause", "toggle", "stop", "next", "prev"})
#: Prefix of the ``client-message`` namespace we listen on.
MEDIA_MESSAGE_PREFIX = "somafm"
@dataclass @dataclass
class PlayerEvent: class PlayerEvent:
kind: str = "state" # "state" | "volume" | "notice" kind: str = "state" # "state" | "volume" | "notice" | "media"
state: str = "" state: str = ""
detail: str = "" detail: str = ""
volume: int | None = None volume: int | None = None
@@ -41,8 +65,9 @@ class PlayerEvent:
class Player: class Player:
def __init__(self) -> None: def __init__(self, media_keys: bool = True) -> None:
self.events: queue.Queue[PlayerEvent] = queue.Queue() self.events: queue.Queue[PlayerEvent] = queue.Queue()
self.media_keys = media_keys
self.backend: str | None = None self.backend: str | None = None
self._loop: asyncio.AbstractEventLoop | None = None self._loop: asyncio.AbstractEventLoop | None = None
self._thread: threading.Thread | None = None self._thread: threading.Thread | None = None
@@ -295,6 +320,7 @@ class Player:
f"--user-agent={USER_AGENT}", f"--user-agent={USER_AGENT}",
"--network-timeout=20", "--network-timeout=20",
] ]
command.extend(self._platform_options())
extra = os.environ.get("SOMAFM_MPV_ARGS") extra = os.environ.get("SOMAFM_MPV_ARGS")
if extra: if extra:
command.extend(shlex.split(extra)) command.extend(shlex.split(extra))
@@ -330,6 +356,32 @@ class Player:
if self._urls: if self._urls:
await self._mpv_load() await self._mpv_load()
def _platform_options(self, platform: str | None = None) -> list[str]:
"""mpv options that only exist (or only make sense) on some platforms.
Keeping them here means the Linux command line stays exactly what it
always was -- mpv aborts on unknown options, so never pass a macOS-only
flag unconditionally.
"""
system = sys.platform if platform is None else platform
if system != "darwin":
return []
return self._macos_media_key_options()
def _macos_media_key_options(self) -> list[str]:
"""Let mpv publish Now Playing, and route its media keys into the TUI."""
if not self.media_keys:
return ["--input-media-keys=no"]
options = ["--input-media-keys=yes"]
try:
RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
conf = RUNTIME_DIR / "mpv-input.conf"
atomic_write_text(conf, MEDIA_KEY_BINDINGS)
except OSError:
# Now Playing still works; only next/prev/stop stay inside mpv.
return options
return [*options, f"--input-conf={conf}"]
async def _mpv_send(self, command: list[Any]) -> None: async def _mpv_send(self, command: list[Any]) -> None:
writer = self._writer writer = self._writer
if writer is None: if writer is None:
@@ -367,6 +419,10 @@ class Player:
def _handle_mpv_message(self, message: dict[str, Any]) -> None: def _handle_mpv_message(self, message: dict[str, Any]) -> None:
event = message.get("event") event = message.get("event")
if event == "client-message":
self._handle_client_message(message.get("args"))
return
if event == "property-change": if event == "property-change":
name = message.get("name") name = message.get("name")
value = message.get("data") value = message.get("data")
@@ -409,6 +465,20 @@ class Player:
if error and error != "success": if error and error != "success":
self.events.put(PlayerEvent(kind="notice", detail=str(error))) self.events.put(PlayerEvent(kind="notice", detail=str(error)))
def _handle_client_message(self, args: Any) -> None:
"""Turn ``script-message somafm <command>`` into a player event.
mpv's macOS backend feeds the system media keys (and Control Center
buttons) into its input system; ``MEDIA_KEY_BINDINGS`` maps them onto
these messages, which arrive over the JSON IPC socket.
"""
if not isinstance(args, list) or len(args) < 2:
return
namespace, command = str(args[0]), str(args[1])
if namespace != MEDIA_MESSAGE_PREFIX or command not in MEDIA_COMMANDS:
return
self.events.put(PlayerEvent(kind="media", detail=command))
def _schedule_retry(self, reason: str) -> None: def _schedule_retry(self, reason: str) -> None:
if self._retry_scheduled or not self._urls or self._stopping.is_set(): if self._retry_scheduled or not self._urls or self._stopping.is_set():
return return
+6 -3
View File
@@ -32,7 +32,8 @@ class StatusBar(Widget):
width = max(20, self.size.width) width = max(20, self.size.width)
glyph, colour = state_glyph(app.playing_state) glyph, colour = state_glyph(app.playing_state)
frame = app.spectrum.snapshot() frame = app.spectrum.snapshot()
mpris = app.mpris.status_label session = app.media.label
status = app.media.status_label
backend = app.player.backend or "no audio" backend = app.player.backend or "no audio"
analyser = frame.label analyser = frame.label
@@ -44,19 +45,21 @@ class StatusBar(Widget):
# From most to least informative; pick the first combination that fits. # From most to least informative; pick the first combination that fits.
right_variants = ( right_variants = (
f"{state} {listeners}{volume} {backend} mpris {mpris} {analyser}", f"{state} {listeners}{volume} {backend} {session} {status} {analyser}",
f"{state} {volume} {session} {status} {analyser}",
f"{state} {volume} {analyser}", f"{state} {volume} {analyser}",
f"{state} {volume}", f"{state} {volume}",
volume, volume,
) )
options = ( options = (
(HINTS_LONG, right_variants[0]), (HINTS_LONG, right_variants[0]),
(HINTS_LONG, right_variants[1]),
(HINTS_MEDIUM, right_variants[0]), (HINTS_MEDIUM, right_variants[0]),
(HINTS_LONG, right_variants[1]),
(HINTS_MEDIUM, right_variants[1]), (HINTS_MEDIUM, right_variants[1]),
(HINTS_SHORT, right_variants[1]), (HINTS_SHORT, right_variants[1]),
(HINTS_SHORT, right_variants[2]), (HINTS_SHORT, right_variants[2]),
(HINTS_SHORT, right_variants[3]), (HINTS_SHORT, right_variants[3]),
(HINTS_SHORT, right_variants[4]),
) )
chosen: tuple[int, str, str] | None = None chosen: tuple[int, str, str] | None = None
+20
View File
@@ -157,3 +157,23 @@ async def test_stage_shows_bandcamp_link_only_when_present() -> None:
app.query_one(StagePanel).refresh_stage() app.query_one(StagePanel).refresh_stage()
await pilot.pause() await pilot.pause()
assert "Bandcamp" not in rendered_text(app) assert "Bandcamp" not in rendered_text(app)
@pytest.mark.asyncio
async def test_media_session_commands_drive_the_app() -> None:
app = SomaFMApp(autoplay=False, offline=True, audio=False)
async with app.run_test(size=(100, 30)) as pilot:
await pilot.pause()
# The status bar names whichever backend this platform uses.
assert app.media.label in {"mpris", "nowplaying"}
assert app.media.status_label == "off" # audio disabled
# The ticker drains the queue, so give it a couple of ticks.
before = app.current.id
app.media_actions.put("next")
await pilot.pause(0.2)
assert app.current.id != before
app.media_actions.put("volume:25")
await pilot.pause(0.2)
assert app.volume == 25
+56
View File
@@ -0,0 +1,56 @@
from __future__ import annotations
import queue
import sys
from somafm_tui import mediasession
from somafm_tui.mediasession import (
MediaSession,
NowPlayingSession,
create_media_session,
)
def test_linux_uses_mpris() -> None:
from somafm_tui.mpris import MprisService
session = create_media_session(queue.Queue(), enabled=False, platform="linux")
assert isinstance(session, MprisService)
assert session.label == "mpris"
def test_macos_uses_the_system_now_playing_session() -> None:
session = create_media_session(
queue.Queue(), enabled=True, backend="mpv", platform="darwin"
)
assert isinstance(session, NowPlayingSession)
assert session.label == "nowplaying"
assert session.status_label == "on"
def test_macos_without_mpv_reports_off() -> None:
session = create_media_session(
queue.Queue(), enabled=True, backend="ffplay", platform="darwin"
)
assert session.status_label == "off"
def test_disabled_session_is_inert_but_keeps_state() -> None:
actions: queue.Queue[str] = queue.Queue()
session = MediaSession(actions, enabled=False)
session.start()
session.set_status("Playing")
session.set_volume(1.7)
session.set_track(station="Groove Salad", title="T", artist="A", album="B")
session.stop()
assert session.status_label == "off"
assert session._status == "Playing"
assert session._volume == 1.0
assert session._metadata["album"] == "B"
assert actions.empty()
def test_platform_detection_is_available_for_tests() -> None:
# The factory must never import D-Bus code on a platform without it.
assert "linux" in sys.platform or sys.platform == "darwin"
assert mediasession.media_actions_queue().empty()
+50
View File
@@ -166,3 +166,53 @@ def test_backend_detection_without_tools(tmp_path, monkeypatch) -> None:
event, _ = wait_for(player, lambda ev: ev.kind == "state", timeout=1) event, _ = wait_for(player, lambda ev: ev.kind == "state", timeout=1)
assert event.state == "error" assert event.state == "error"
player.shutdown() player.shutdown()
def test_media_key_messages_become_player_events() -> None:
player = Player()
player._handle_mpv_message(
{"event": "client-message", "args": ["somafm", "next"]}
)
event = player.events.get_nowait()
assert event.kind == "media"
assert event.detail == "next"
player._handle_mpv_message(
{"event": "client-message", "args": ["somafm", "toggle"]}
)
assert player.events.get_nowait().detail == "toggle"
# Foreign namespaces, unknown commands and malformed payloads are ignored.
player._handle_mpv_message({"event": "client-message", "args": ["other", "next"]})
player._handle_mpv_message({"event": "client-message", "args": ["somafm", "seek"]})
player._handle_mpv_message({"event": "client-message", "args": None})
player._handle_mpv_message({"event": "client-message"})
assert player.events.empty()
def test_platform_options_never_leak_to_linux(tmp_path, monkeypatch) -> None:
import somafm_tui.player as player_module
monkeypatch.setattr(player_module, "RUNTIME_DIR", tmp_path)
assert Player()._platform_options("linux") == []
assert Player()._platform_options("win32") == []
options = Player()._platform_options("darwin")
assert "--input-media-keys=yes" in options
conf = next(option for option in options if option.startswith("--input-conf="))
assert conf.endswith("mpv-input.conf")
bindings = (tmp_path / "mpv-input.conf").read_text()
for line in (
"PLAY script-message somafm toggle",
"PLAYONLY script-message somafm play",
"PAUSEONLY script-message somafm pause",
"STOP script-message somafm stop",
"NEXT script-message somafm next",
"PREV script-message somafm prev",
):
assert line in bindings
# --no-mpris / --no-media must silence the media keys as well.
assert Player(media_keys=False)._platform_options("darwin") == [
"--input-media-keys=no"
]