"""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: """Stub — no-op; real implementation emits D-Bus PropertiesChanged.""" 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()