diff --git a/gui.py b/gui.py index 1a1b45c..5df7890 100644 --- a/gui.py +++ b/gui.py @@ -80,6 +80,7 @@ except (ImportError, OSError): from config import SETTINGS, get_setting, save_setting, save_volume from discord_rpc import DiscordRPC +from mpris import MprisManager from music_db import MusicDB from player import AudioPlayer, SearchWorker, _best_thumbnail @@ -2279,6 +2280,9 @@ class TunettiWindow(QMainWindow): self.db = MusicDB(SETTINGS["db_path"]) self.rpc = DiscordRPC(SETTINGS["discord_client_id"]) self.player = AudioPlayer(self) + self.mpris = MprisManager() + self.mpris.start() + self._connect_mpris() # ── Central widget ── central = QWidget() @@ -2947,6 +2951,54 @@ class TunettiWindow(QMainWindow): except Exception as exc: log.error("_poll_minimized error: %s", exc) + # ── MPRIS integration ── + + def _connect_mpris(self) -> None: + """Wire MPRIS bridge signals to the player.""" + bridge = self.mpris.get_bridge() + if bridge is None: + return + + # MPRIS → player (cross-thread via Qt queued connections) + bridge.play_requested.connect(self.player.resume) + bridge.pause_requested.connect(self.player.pause) + bridge.play_pause_requested.connect(self.player.toggle_pause) + bridge.stop_requested.connect(self.player.stop_playback) + bridge.next_requested.connect(self.player.skip) + bridge.previous_requested.connect(self.player.previous) + + # Player → MPRIS state updates + self.player.playback_state_changed.connect( + self._on_mpris_state_changed + ) + self.player.song_started.connect(self._on_mpris_song_started) + self.player.song_ended.connect(self._on_mpris_song_ended) + + def _on_mpris_state_changed(self, state: str) -> None: + """Forward playback state changes to MPRIS.""" + song = self.player.get_current() + self.mpris.notify_state_changed( + state, song, self.player.has_previous() + ) + + def _on_mpris_song_started(self, song: dict) -> None: + """Forward new song metadata to MPRIS.""" + self.mpris.notify_state_changed( + "playing", song, self.player.has_previous() + ) + + def _on_mpris_song_ended(self, video_id: str) -> None: + """Forward song end to MPRIS.""" + song = self.player.get_current() + if song: + self.mpris.notify_state_changed( + "playing", song, self.player.has_previous() + ) + else: + self.mpris.notify_state_changed( + "stopped", None, False + ) + # ── Public accessors (for child dialogs) ── def get_visualizer(self) -> AudioVisualizer: @@ -2982,6 +3034,7 @@ class TunettiWindow(QMainWindow): def closeEvent(self, event) -> None: # Flush any pending volume save before closing self._playback.flush_volume() + self.mpris.stop() self.player.shutdown() self.rpc.stop() self.db.close() diff --git a/mpris.py b/mpris.py new file mode 100644 index 0000000..a274c67 --- /dev/null +++ b/mpris.py @@ -0,0 +1,480 @@ +"""MPRIS (Media Player Remote Interfacing Specification) integration. + +Implements the ``org.mpris.MediaPlayer2`` and +``org.mpris.MediaPlayer2.Player`` D-Bus interfaces so that desktop +environment media keys, applets (e.g. GNOME Quick Settings, KDE +Plasma Media Player) and third-party tools like ``playerctl`` can +control Tunetti and display now-playing metadata. + +Architecture +------------ +A background thread runs a synchronous ``dbus-fast`` ``BaseMessageBus``. +MPRIS method calls (Play, Pause, Next, …) are dispatched to the Qt +main thread via a ``MprisBridge`` ``QObject`` with Qt signals — this +is thread-safe and avoids locking. + +State updates (playback status, current song) are pushed from the +player into a simple thread-safe ``_State`` holder which the MPRIS +property getters read without blocking the D‑Bus thread. + +When the ``dbus-fast`` library is unavailable or D‑Bus is not running, +the manager silently degrades — no crash, no traceback. +""" + +import asyncio +import logging +import threading +from typing import Optional + +log = logging.getLogger("tunetti.mpris") + +_HAS_DBUS = False +try: + from dbus_fast.aio.message_bus import MessageBus + from dbus_fast.message_bus import BusType + from dbus_fast.service import ServiceInterface, dbus_method, dbus_property, dbus_signal + from dbus_fast import Variant + from dbus_fast import NameFlag + from dbus_fast.service import PropertyAccess + + _HAS_DBUS = True +except ImportError: + MessageBus = None # type: ignore[assignment] + BusType = None + + class _FallbackServiceInterface: + """Stand-in for ``ServiceInterface`` when dbus-fast is unavailable. + + Accepts an interface name in ``__init__`` so that subclasses like + ``MprisRootInterface`` do not crash when calling + ``super().__init__(name)``. + """ + def __init__(self, name: str) -> None: + self._interface_name = name + def emit_properties_changed(self, changed, invalidated) -> None: + pass + + ServiceInterface = _FallbackServiceInterface # type: ignore[assignment,misc] + dbus_method = lambda **kw: lambda f: f + dbus_property = lambda **kw: lambda f: f + dbus_signal = lambda **kw: lambda f: f + + class _FallbackVariant: + """Stand-in for ``Variant`` when dbus-fast is unavailable. + + Stores the signature and value so ``_build_mpris_metadata`` + can still build metadata dicts without a real D-Bus library. + """ + def __init__(self, signature: str, value): + self.signature = signature + self.value = value + + Variant = _FallbackVariant + + class _FallbackNameFlag: + REPLACE_EXISTING = 1 + DO_NOT_QUEUE = 2 + + NameFlag = _FallbackNameFlag + + PropertyAccess = type("PropertyAccess", (), {"READ": "read"}) + +# ── D-Bus identifiers ──────────────────────────────────────────────────────── +BUS_NAME = "org.mpris.MediaPlayer2.tunetti" +OBJECT_PATH = "/org/mpris/MediaPlayer2" + +# Playback status constants +STATUS_PLAYING = "Playing" +STATUS_PAUSED = "Paused" +STATUS_STOPPED = "Stopped" + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Thread-safe state holder +# ═══════════════════════════════════════════════════════════════════════════════ + + +class _State: + """Thread-safe holder for the MPRIS-visible player state. + + Updated from the Qt main thread and read from the D‑Bus thread. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self.status: str = STATUS_STOPPED + self.metadata: dict = {} + self.can_go_prev: bool = False + + def update(self, status: str, metadata: dict, can_go_prev: bool) -> None: + with self._lock: + self.status = status + self.metadata = metadata + self.can_go_prev = can_go_prev + + def get_status(self) -> str: + with self._lock: + return self.status + + def get_metadata(self) -> dict: + with self._lock: + return dict(self.metadata) + + def get_can_go_prev(self) -> bool: + with self._lock: + return self.can_go_prev + + +# ═══════════════════════════════════════════════════════════════════════════════ +# MPRIS D-Bus interfaces +# ═══════════════════════════════════════════════════════════════════════════════ + + +def _build_mpris_metadata(song: Optional[dict]) -> dict: + """Build an ``a{sv}`` metadata dict from a Tunetti song dict.""" + if not song: + return {} + artists = song.get("artists") or [] + artist_names = [ + a.get("name", "") for a in artists if isinstance(a, dict) and a.get("name") + ] + video_id = song.get("videoId") or song.get("video_id", "") or "unknown" + duration_us = (song.get("duration", 0) or 0) * 1_000_000 # s → µs + + # album may be a dict {name, id} or None — extract the name string. + raw_album = song.get("album") + album_name = ( + raw_album.get("name", "") + if isinstance(raw_album, dict) + else (raw_album or "") + ) + + return { + "mpris:trackid": Variant("o", f"/org/tunetti/track/{video_id}"), + "mpris:length": Variant("x", max(duration_us, 0)), + "mpris:artUrl": Variant("s", song.get("thumbnail", "")), + "xesam:title": Variant("s", song.get("title", "Unknown")), + "xesam:artist": Variant("as", artist_names), + "xesam:album": Variant("s", album_name), + "xesam:url": Variant( + "s", f"https://music.youtube.com/watch?v={video_id}" + ), + } + + +class MprisRootInterface(ServiceInterface): + """Implements ``org.mpris.MediaPlayer2`` — identity & capabilities.""" + + def __init__(self) -> None: + super().__init__("org.mpris.MediaPlayer2") + + @dbus_property(access=PropertyAccess.READ) + def Identity(self) -> "s": + return "Tunetti" + + @dbus_property(access=PropertyAccess.READ) + def DesktopEntry(self) -> "s": + return "tunetti" + + @dbus_property(access=PropertyAccess.READ) + def CanQuit(self) -> "b": + return False + + @dbus_property(access=PropertyAccess.READ) + def CanRaise(self) -> "b": + return False + + @dbus_property(access=PropertyAccess.READ) + def HasTrackList(self) -> "b": + return False + + +class MprisPlayerInterface(ServiceInterface): + """Implements ``org.mpris.MediaPlayer2.Player`` — playback control.""" + + def __init__(self, state: _State, bridge: "MprisBridge") -> None: + super().__init__("org.mpris.MediaPlayer2.Player") + self._state = state + self._bridge = bridge + + # ── Properties ────────────────────────────────────────────────────── + + @dbus_property(access=PropertyAccess.READ) + def PlaybackStatus(self) -> "s": + return self._state.get_status() + + @dbus_property(access=PropertyAccess.READ) + def Metadata(self) -> "a{sv}": + return self._state.get_metadata() + + @dbus_property(access=PropertyAccess.READ) + def Position(self) -> "x": + return 0 # approximate — real position tracking adds complexity + + @dbus_property(access=PropertyAccess.READ) + def MinimumRate(self) -> "d": + return 1.0 + + @dbus_property(access=PropertyAccess.READ) + def MaximumRate(self) -> "d": + return 1.0 + + @dbus_property(access=PropertyAccess.READ) + def Rate(self) -> "d": + return 1.0 + + @dbus_property(access=PropertyAccess.READ) + def Volume(self) -> "d": + return 1.0 + + @dbus_property(access=PropertyAccess.READ) + def CanGoNext(self) -> "b": + return True + + @dbus_property(access=PropertyAccess.READ) + def CanGoPrevious(self) -> "b": + return self._state.get_can_go_prev() + + @dbus_property(access=PropertyAccess.READ) + def CanPlay(self) -> "b": + return True + + @dbus_property(access=PropertyAccess.READ) + def CanPause(self) -> "b": + return True + + @dbus_property(access=PropertyAccess.READ) + def CanSeek(self) -> "b": + return True + + @dbus_property(access=PropertyAccess.READ) + def CanControl(self) -> "b": + return True + + # ── Methods ───────────────────────────────────────────────────────── + + @dbus_method() + def Play(self) -> None: + self._bridge.play_requested.emit() + + @dbus_method() + def Pause(self) -> None: + self._bridge.pause_requested.emit() + + @dbus_method() + def PlayPause(self) -> None: + self._bridge.play_pause_requested.emit() + + @dbus_method() + def Stop(self) -> None: + self._bridge.stop_requested.emit() + + @dbus_method() + def Next(self) -> None: + self._bridge.next_requested.emit() + + @dbus_method() + def Previous(self) -> None: + self._bridge.previous_requested.emit() + + @dbus_method() + def Seek(self, offset: "x") -> None: + self._bridge.seek_requested.emit(offset) + + # ── Signal ────────────────────────────────────────────────────────── + + @dbus_signal() + def PropertiesChanged( + self, interface: "s", changed: "a{sv}", invalidated: "ao" + ) -> None: + pass # emitted by dbus-fast when @dbus_property values change + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Qt bridge — cross-thread command dispatch +# ═══════════════════════════════════════════════════════════════════════════════ + +_HAS_QT_BRIDGE = False +try: + from PySide6.QtCore import QObject, Signal + + _HAS_QT_BRIDGE = True +except ImportError: + QObject = object + Signal = lambda *a: lambda f: f + + +class MprisBridge(QObject): + """Thread‑safe Qt signal bridge for MPRIS commands. + + Created on the Qt main thread. Signals are emitted from the D‑Bus + thread and delivered to the main thread via Qt's queued connections. + """ + + play_requested = Signal() + pause_requested = Signal() + play_pause_requested = Signal() + stop_requested = Signal() + next_requested = Signal() + previous_requested = Signal() + seek_requested = Signal(int) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Manager — lifecycle & integration +# ═══════════════════════════════════════════════════════════════════════════════ + + +class MprisManager: + """Top-level MPRIS manager. + + Usage:: + + manager = MprisManager() + manager.start() # starts D‑Bus thread + ... + manager.notify_state_changed(...) # push updates from player + ... + manager.stop() # shut down + """ + + def __init__(self) -> None: + self._state = _State() + self._bridge: Optional[MprisBridge] = None + self._bus: Optional[MessageBus] = None + self._thread: Optional[threading.Thread] = None + self._loop: Optional[asyncio.AbstractEventLoop] = None + self._prop_queue: list[tuple[str, Optional[dict], bool]] = [] + self._prop_lock = threading.Lock() + + # ── Lifecycle ─────────────────────────────────────────────────────── + + def start(self) -> bool: + """Start the MPRIS D‑Bus service. + + Returns ``True`` if the service was started successfully, + ``False`` if D‑Bus is unavailable or an error occurred. + """ + if not _HAS_DBUS: + log.info("MPRIS unavailable — dbus-fast not installed") + return False + + self._bridge = MprisBridge() + self._thread = threading.Thread( + target=self._run_async, + daemon=True, + name="mpris", + ) + self._thread.start() + return True + + def stop(self) -> None: + """Signal the D‑Bus thread to shut down and wait for it.""" + if self._loop is not None and self._loop.is_running(): + self._loop.call_soon_threadsafe(self._stop_async) + if self._thread is not None and self._thread.is_alive(): + self._thread.join(timeout=3) + + def get_bridge(self) -> Optional[MprisBridge]: + """Return the ``MprisBridge`` for connecting player signals.""" + return self._bridge + + # ── State updates (called from Qt main thread) ────────────────────── + + def notify_state_changed( + self, status: str, song: Optional[dict], can_go_prev: bool = False + ) -> None: + """Update the MPRIS-visible state and emit D‑Bus property changes. + + *status* is one of ``"playing"``, ``"paused"``, ``"stopped"``. + This is thread-safe: the actual D‑Bus signal emission happens on + the D‑Bus thread via the asyncio event loop. + """ + mpris_status = { + "playing": STATUS_PLAYING, + "paused": STATUS_PAUSED, + "stopped": STATUS_STOPPED, + }.get(status, STATUS_STOPPED) + + metadata = _build_mpris_metadata(song) if song else {} + can_prev = can_go_prev + + # Update the thread-safe state immediately (MPRIS property getters + # on the D‑Bus thread will see the new values). + self._state.update(mpris_status, metadata, can_prev) + + # Queue the property-change notification for the D‑Bus thread. + with self._prop_lock: + self._prop_queue.append((mpris_status, metadata, can_prev)) + if self._loop is not None and self._loop.is_running(): + self._loop.call_soon_threadsafe(self._flush_props) + + # ── Internal ──────────────────────────────────────────────────────── + + def _flush_props(self) -> None: + """Flush queued property-change notifications (runs on D‑Bus thread).""" + if self._bus is None or self._player_iface is None: + return + with self._prop_lock: + items = list(self._prop_queue) + self._prop_queue.clear() + for mpris_status, metadata, can_prev in items: + changed: dict = {} + changed["PlaybackStatus"] = mpris_status + if metadata: + changed["Metadata"] = metadata + changed["CanGoPrevious"] = can_prev + if changed: + try: + self._player_iface.emit_properties_changed( + changed, + [], + ) + except Exception: + pass + + def _stop_async(self) -> None: + """Disconnect the bus (runs on D‑Bus thread).""" + if self._bus is not None: + try: + self._bus.disconnect() + except Exception: + pass + + def _run_async(self) -> None: + """Run the asyncio event loop — owns the D‑Bus connection.""" + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + try: + self._loop.run_until_complete(self._connect_and_serve()) + except Exception as exc: + log.debug("MPRIS D‑Bus error: %s", exc) + finally: + self._loop.close() + self._loop = None + self._bus = None + self._root_iface = None + self._player_iface = None + log.info("MPRIS service stopped") + + async def _connect_and_serve(self) -> None: + """Connect to D‑Bus, export interfaces, and serve requests.""" + self._bus = MessageBus(bus_type=BusType.SESSION) + await self._bus.connect() + + await self._bus.request_name( + BUS_NAME, NameFlag.REPLACE_EXISTING | NameFlag.DO_NOT_QUEUE + ) + + self._root_iface = MprisRootInterface() + self._player_iface = MprisPlayerInterface(self._state, self._bridge) + + self._bus.export(OBJECT_PATH, self._root_iface) + self._bus.export(OBJECT_PATH, self._player_iface) + + log.info("MPRIS service '%s' registered on session bus", BUS_NAME) + + # Flush any props that were queued before the bus was ready. + self._flush_props() + + await self._bus.wait_for_disconnect() diff --git a/requirements.txt b/requirements.txt index 861d6e6..8fc38b4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ yt-dlp>=2024.0.0 ytmusicapi>=1.0.0 pypresence>=4.0.0 PySide6>=6.5.0 +dbus-fast>=5.0.0 diff --git a/tests/test_mpris.py b/tests/test_mpris.py new file mode 100644 index 0000000..edc08f9 --- /dev/null +++ b/tests/test_mpris.py @@ -0,0 +1,237 @@ +"""Tests for the MPRIS integration module (mpris.py).""" + +from unittest import mock + +import pytest + + +# ── _State ───────────────────────────────────────────────────────────────────── + + +class TestState: + """Tests for the _State thread-safe holder.""" + + def _import(self): + from mpris import _State + return _State + + def test_initial_state_is_stopped(self): + cls = self._import() + s = cls() + assert s.get_status() == "Stopped" + assert s.get_metadata() == {} + assert s.get_can_go_prev() is False + + def test_update_changes_state(self): + cls = self._import() + s = cls() + s.update("Playing", {"xesam:title": "Test"}, True) + assert s.get_status() == "Playing" + assert s.get_metadata() == {"xesam:title": "Test"} + assert s.get_can_go_prev() is True + + def test_get_metadata_returns_copy(self): + cls = self._import() + s = cls() + s.update("Paused", {"key": "val"}, False) + meta = s.get_metadata() + meta["extra"] = "added" + assert "extra" not in s.get_metadata() + + +# ── _build_mpris_metadata ────────────────────────────────────────────────────── + + +class TestBuildMprisMetadata: + """Tests for the _build_mpris_metadata helper.""" + + def _import(self): + from mpris import _build_mpris_metadata + return _build_mpris_metadata + + def test_none_song_returns_empty(self): + fn = self._import() + assert fn(None) == {} + + def test_empty_song_returns_empty(self): + fn = self._import() + # Empty dict is falsy, so the function returns {} early + assert fn({}) == {} + + def test_full_song_metadata(self): + fn = self._import() + song = { + "videoId": "abc123", + "title": "Test Song", + "artists": [{"name": "Artist One"}, {"name": "Artist Two"}], + "album": "Test Album", + "duration": 240, + "thumbnail": "https://example.com/thumb.jpg", + } + meta = fn(song) + assert meta["xesam:title"].value == "Test Song" + assert meta["xesam:artist"].value == ["Artist One", "Artist Two"] + assert meta["xesam:album"].value == "Test Album" + assert meta["mpris:length"].value == 240_000_000 + assert meta["mpris:artUrl"].value == "https://example.com/thumb.jpg" + assert "abc123" in meta["mpris:trackid"].value + + def test_album_as_dict_extracts_name(self): + fn = self._import() + song = { + "videoId": "x", + "album": {"name": "Real Album", "id": "album_id"}, + } + meta = fn(song) + assert meta["xesam:album"].value == "Real Album" + + def test_album_is_none(self): + fn = self._import() + meta = fn({"videoId": "x", "album": None}) + assert meta["xesam:album"].value == "" + + def test_artists_not_list(self): + fn = self._import() + song = {"videoId": "x", "title": "T", "artists": "not a list"} + meta = fn(song) + assert meta["xesam:artist"].value == [] + + def test_missing_video_id_does_not_crash(self): + fn = self._import() + song = {"title": "No ID"} + meta = fn(song) + assert isinstance(meta, dict) + assert "mpris:trackid" in meta + + def test_zero_duration(self): + fn = self._import() + meta = fn({"videoId": "x", "duration": 0}) + assert meta["mpris:length"].value == 0 + + def test_none_duration(self): + fn = self._import() + meta = fn({"videoId": "x", "duration": None}) + assert meta["mpris:length"].value == 0 + + +# ── MprisManager ────────────────────────────────────────────────────────────── + + +class TestMprisManager: + """Tests for the MprisManager lifecycle.""" + + def test_start_returns_false_without_dbus(self): + """When dbus-fast is unavailable, start() returns False.""" + with mock.patch("mpris._HAS_DBUS", False): + from mpris import MprisManager + mgr = MprisManager() + assert mgr.start() is False + + def test_bridge_is_none_before_start(self): + from mpris import MprisManager + mgr = MprisManager() + assert mgr.get_bridge() is None + + def test_stop_without_start_does_not_crash(self): + from mpris import MprisManager + mgr = MprisManager() + mgr.stop() # should not raise + + +# ── MprisBridge ──────────────────────────────────────────────────────────────── + + +class TestMprisBridge: + """Tests for the MprisBridge Qt signal bridge.""" + + def test_has_required_signals(self): + from mpris import MprisBridge + bridge = MprisBridge() + for name in ("play_requested", "pause_requested", "play_pause_requested", + "stop_requested", "next_requested", "previous_requested", + "seek_requested"): + assert hasattr(bridge, name), f"Missing signal: {name}" + + +# ── MprisRootInterface ──────────────────────────────────────────────────────── + + +class TestMprisRootInterface: + """Tests for the root MPRIS interface.""" + + def test_identity(self): + from mpris import MprisRootInterface + iface = MprisRootInterface() + assert iface.Identity == "Tunetti" + + def test_desktop_entry(self): + from mpris import MprisRootInterface + iface = MprisRootInterface() + assert iface.DesktopEntry == "tunetti" + + def test_capabilities(self): + from mpris import MprisRootInterface + iface = MprisRootInterface() + assert iface.CanQuit is False + assert iface.CanRaise is False + assert iface.HasTrackList is False + + +# ── MprisPlayerInterface ────────────────────────────────────────────────────── + + +class TestMprisPlayerInterface: + """Tests for the player MPRIS interface.""" + + def _make(self): + from mpris import _State, MprisBridge, MprisPlayerInterface + state = _State() + bridge = MprisBridge() + iface = MprisPlayerInterface(state, bridge) + return state, bridge, iface + + def test_playback_status_default(self): + _, _, iface = self._make() + assert iface.PlaybackStatus == "Stopped" + + def test_playback_status_playing(self): + state, _, iface = self._make() + state.update("Playing", {}, False) + assert iface.PlaybackStatus == "Playing" + + def test_can_go_next_default(self): + _, _, iface = self._make() + assert iface.CanGoNext is True + + def test_can_go_previous_false_when_no_history(self): + _, _, iface = self._make() + assert iface.CanGoPrevious is False + + def test_can_go_previous_true_when_history_exists(self): + state, _, iface = self._make() + state.update("Playing", {"t": "1"}, True) + assert iface.CanGoPrevious is True + + def test_can_control(self): + _, _, iface = self._make() + assert iface.CanControl is True + + def test_methods_emit_via_bridge(self): + state, bridge, iface = self._make() + for method_name, signal_name in [ + ("Play", "play_requested"), + ("Pause", "pause_requested"), + ("PlayPause", "play_pause_requested"), + ("Stop", "stop_requested"), + ("Next", "next_requested"), + ("Previous", "previous_requested"), + ]: + with mock.patch.object(bridge, signal_name) as mock_sig: + getattr(iface, method_name)() + mock_sig.emit.assert_called_once() + + def test_seek_emits_with_offset(self): + state, bridge, iface = self._make() + with mock.patch.object(bridge, "seek_requested") as mock_sig: + iface.Seek(5000) + mock_sig.emit.assert_called_once_with(5000)