NikkeDoy
7a70232239
All checks were successful
SonarQube Code Quality Scan / SonarQube Scan (push) Successful in 1m41s
- Add docstrings to all empty Qt stub methods (S1186) - Fix BLOCKER S1845: rename playbackState to get_playback_state - Rename enum fields to snake_case with CamelCase aliases (S116) - Rename stub methods to snake_case (set_source, set_position, etc.) (S100) - Rename unused timeout param to msecs (S1172) - Remove unused state variables in tests (S1481) - Replace 'is' with '==' for enum comparisons (S5795) - Remove unnecessary list() calls (S7504) - Rename QTh to qt_cls (S117) - Remove commented-out code blocks (S125) - Add docstring to FallbackServiceInterface.emit_properties_changed (S1186)
481 lines
17 KiB
Python
481 lines
17 KiB
Python
"""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()
|