fix: override MprisService to properly handle dbus errors

Author: abmantis

aiosendspin_mpris/adapter.py

@@ -24,12 +24,15 @@
 if sys.platform == "linux":
     try:
         from mpris_api.adapter.IMprisAdapterPlayer import IMprisAdapterPlayer
+        from mpris_api.adapter.IMprisAdapterPlayLists import IMprisAdapterPlayLists
         from mpris_api.adapter.IMprisAdapterRoot import IMprisAdapterRoot
         from mpris_api.adapter.IMprisAdapterTrackList import IMprisAdapterTrackList
         from mpris_api.common.DbusObject import DbusObject
         from mpris_api.model.MprisLoopStatus import MprisLoopStatus
         from mpris_api.model.MprisMetaData import MprisMetaData
         from mpris_api.model.MprisPlaybackStatus import MprisPlaybackStatus
+        from mpris_api.model.MprisPlaylist import MprisPlaylist
+        from mpris_api.model.MprisPlaylistOrdering import MprisPlaylistOrdering
         from tunit.unit import Microseconds
 
         MPRIS_AVAILABLE = True  # pyright: ignore[reportConstantRedefinition]
@@ -56,10 +59,17 @@ class _DummyIMprisAdapterTrackList:
         def __init__(self) -> None:
             pass
 
+    class _DummyIMprisAdapterPlayLists:
+        """Dummy adapter base class when mpris_api is not installed."""
+
+        def __init__(self) -> None:
+            pass
+
     if not TYPE_CHECKING:  # otherwise pyright complains too much
         IMprisAdapterRoot = _DummyIMprisAdapterRoot
         IMprisAdapterPlayer = _DummyIMprisAdapterPlayer
         IMprisAdapterTrackList = _DummyIMprisAdapterTrackList
+        IMprisAdapterPlayLists = _DummyIMprisAdapterPlayLists
 
 
 @dataclass
@@ -350,3 +360,33 @@ def canEditTracks(self) -> bool:
     def getTracks(self) -> list[DbusObject]:
         """Return list of tracks (empty - not supported)."""
         return []
+
+
+class SendspinMprisAdapterPlaylists(IMprisAdapterPlayLists):
+    """Stub adapter for the MPRIS PlayLists interface.
+
+    This provides an empty implementation to prevent D-Bus errors when
+    clients query for the PlayLists interface.
+    """
+
+    @override
+    def getPlaylistCount(self) -> int:
+        return 0
+
+    @override
+    def getAvailableOrderings(self) -> list[MprisPlaylistOrdering]:
+        return []
+
+    @override
+    def getActivePlaylist(self) -> MprisPlaylist | None:
+        return None
+
+    @override
+    def activatePlaylist(self, playlistId: str) -> None:
+        pass
+
+    @override
+    def getPlaylists(
+        self, index: int, maxCount: int, order: MprisPlaylistOrdering, reverseOrder: bool
+    ) -> list[MprisPlaylist]:
+        return []

aiosendspin_mpris/mpris.py

@@ -5,12 +5,13 @@
 import asyncio
 import logging
 from collections.abc import Callable
-from typing import TYPE_CHECKING, final, override
+from typing import TYPE_CHECKING
 
 from .adapter import (
     MPRIS_AVAILABLE,
     MprisState,
     SendspinMprisAdapterPlayer,
+    SendspinMprisAdapterPlaylists,
     SendspinMprisAdapterRoot,
     SendspinMprisAdapterTrackList,
 )
@@ -22,34 +23,8 @@
     from mpris_api.MprisService import MprisService
     from mpris_api.MprisUpdateNotifier import MprisUpdateNotifier
 
-_LOGGER = logging.getLogger(__name__)
-
-if MPRIS_AVAILABLE:
-    from mpris_api.MprisService import MprisService
-
-
-@final
-class _MprisErrorFilter(logging.Filter):
-    """Filter to suppress expected MPRIS interface errors.
-
-    Some MPRIS clients query for optional interfaces (like Playlists) which is not
-    properly supported by the mpris-api lib. This filter suppresses the
-    resulting D-Bus errors to avoid cluttering the output.
-    """
 
-    _SUPPRESSED_INTERFACES = ("org.mpris.MediaPlayer2.Playlists",)
-
-    @override
-    def filter(self, record: logging.LogRecord) -> bool:
-        """Return False to suppress the log record, True to allow it."""
-        if record.levelno != logging.ERROR:
-            return True
-
-        msg = record.getMessage()
-        if "could not find an interface" not in msg:
-            return True
-
-        return not any(iface in msg for iface in self._SUPPRESSED_INTERFACES)
+_LOGGER = logging.getLogger(__name__)
 
 
 class SendspinMpris:
@@ -83,11 +58,11 @@ class SendspinMpris:
     _adapter_root: SendspinMprisAdapterRoot | None
     _adapter_player: SendspinMprisAdapterPlayer | None
     _adapter_tracklist: SendspinMprisAdapterTrackList | None
+    _adapter_playlists: SendspinMprisAdapterPlaylists | None
     _service: MprisService | None
     _update_notifier: MprisUpdateNotifier | None
     _running: bool
     _listener_removers: list[Callable[[], None]]
-    _error_filter: _MprisErrorFilter | None
 
     def __init__(
         self, client: SendspinClient, name: str = "Sendspin", desktop_entry: str | None = None
@@ -110,11 +85,11 @@ def __init__(
         self._adapter_root = None
         self._adapter_player = None
         self._adapter_tracklist = None
+        self._adapter_playlists = None
         self._service = None
         self._update_notifier = None
         self._running = False
         self._listener_removers = []
-        self._error_filter = None
 
     def start(self) -> None:
         """Start the MPRIS D-Bus service and attach listeners to the client.
@@ -130,6 +105,8 @@ def start(self) -> None:
             _LOGGER.debug("MPRIS not available: mpris_api package not installed or not on Linux")
             return
 
+        from .mpris_service import PatchedMprisService
+
         if self._running:
             _LOGGER.debug("MPRIS interface already running")
             return
@@ -139,21 +116,19 @@ def start(self) -> None:
         except RuntimeError as err:
             raise RuntimeError("MPRIS must be started from within a running event loop") from err
 
-        # Add filter to suppress expected D-Bus errors for unimplemented interfaces
-        self._error_filter = _MprisErrorFilter()
-        logging.getLogger().addFilter(self._error_filter)
-
         self._adapter_root = SendspinMprisAdapterRoot(
             identity=self._name, desktop_entry=self._desktop_entry
         )
         self._adapter_player = SendspinMprisAdapterPlayer(self._client, self._loop, self._state)
         self._adapter_tracklist = SendspinMprisAdapterTrackList()
+        self._adapter_playlists = SendspinMprisAdapterPlaylists()
 
-        self._service = MprisService(
+        self._service = PatchedMprisService(
             name=self._name,
             adapterRoot=self._adapter_root,
             adapterPlayer=self._adapter_player,
             adapterTrackList=self._adapter_tracklist,
+            adapterPlayLists=self._adapter_playlists,
         )
         self._service.start()
         self._update_notifier = self._service.updateNotifier
@@ -189,10 +164,7 @@ def stop(self) -> None:
         self._adapter_root = None
         self._adapter_player = None
         self._adapter_tracklist = None
-
-        if self._error_filter is not None:
-            logging.getLogger().removeFilter(self._error_filter)
-            self._error_filter = None
+        self._adapter_playlists = None
 
         _LOGGER.info("MPRIS interface stopped")
 

aiosendspin_mpris/mpris_service.py

@@ -0,0 +1,74 @@
+"""Patched MPRIS service."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, override
+
+from dbus_next.aio.message_bus import MessageBus
+from dbus_next.errors import InvalidAddressError
+from mpris_api.adapter.IMprisAdapterPlayer import IMprisAdapterPlayer
+from mpris_api.adapter.IMprisAdapterPlayLists import IMprisAdapterPlayLists
+from mpris_api.adapter.IMprisAdapterRoot import IMprisAdapterRoot
+from mpris_api.adapter.IMprisAdapterTrackList import IMprisAdapterTrackList
+from mpris_api.interface.MprisInterfacePlayLists import MprisInterfacePlayLists
+from mpris_api.model.MprisConstant import MprisConstant
+from mpris_api.MprisService import MprisService
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class PatchedMprisInterfacePlayLists(MprisInterfacePlayLists):
+    """Patched MprisInterfacePlayLists to return lists instead of tuples for D-Bus STRUCT types."""
+
+    @override
+    def activePlaylist(self) -> list[Any]:  # pyright: ignore[reportExplicitAny, reportIncompatibleMethodOverride]
+        """Return active playlist as a list.
+
+        The upstream definition seems to be wrong since DBus requires a list, not tuple.
+        """
+        return []
+
+
+class PatchedMprisService(MprisService):
+    """Patched MprisService to handle InvalidAddressError gracefully."""
+
+    def __init__(  # noqa: D107
+        self,
+        name: str,
+        adapterRoot: IMprisAdapterRoot,  # noqa: N803
+        adapterPlayer: IMprisAdapterPlayer,  # noqa: N803
+        adapterTrackList: IMprisAdapterTrackList,  # noqa: N803
+        adapterPlayLists: IMprisAdapterPlayLists,  # noqa: N803
+    ) -> None:
+        super().__init__(
+            name=name,
+            adapterRoot=adapterRoot,
+            adapterPlayer=adapterPlayer,
+            adapterTrackList=adapterTrackList,
+            adapterPlayLists=adapterPlayLists,
+        )
+        self._interfacePlayLists: MprisInterfacePlayLists | None = PatchedMprisInterfacePlayLists(
+            adapter=adapterPlayLists
+        )
+
+    @override
+    async def _loop(self) -> None:
+        try:
+            self._messageBus = messageBus = await MessageBus().connect()  # noqa: N806  # pyright: ignore[reportUnannotatedClassAttribute]
+
+            messageBus.export(MprisConstant.PATH, self._interfaceRoot)
+            messageBus.export(MprisConstant.PATH, self._interfacePlayer)
+            if self._interfaceTrackList:
+                messageBus.export(MprisConstant.PATH, self._interfaceTrackList)
+            if self._interfacePlayLists:
+                messageBus.export(MprisConstant.PATH, self._interfacePlayLists)
+
+            _ = await messageBus.request_name(self._name)
+
+            await messageBus.wait_for_disconnect()
+
+        except InvalidAddressError:
+            _LOGGER.warning("MPRIS not available: DBus address error")
+        finally:
+            self._disconnect()