"""MPRIS (D-Bus) интеграция: система видит плеер как источник звука. Благодаря этому плеером можно управлять из панели GNOME (расширение media-controls), из `playerctl` и т.п.: play/pause, следующий/предыдущий, перемотка, метаданные и обложка. Модуль необязательный: если нет `dbus-next` или сессионного D-Bus, всё превращается в no-op, и приложения работают как обычно. """ import asyncio import logging import threading from typing import Any, Callable log = logging.getLogger("mpris") try: # pragma: no cover - зависит от окружения from dbus_next import Variant from dbus_next.aio import MessageBus from dbus_next.constants import PropertyAccess from dbus_next.service import ServiceInterface, dbus_property, method, signal HAVE_DBUS = True except Exception: # pragma: no cover HAVE_DBUS = False BUS_NAME = "org.mpris.MediaPlayer2.mtsmusic" OBJECT_PATH = "/org/mpris/MediaPlayer2" def track_object_path(track_id: Any) -> str: """D-Bus object path для mpris:trackid (только [A-Za-z0-9_]).""" safe = "".join(ch if ch.isalnum() else "_" for ch in str(track_id)) or "unknown" return f"/mtsmusic/track/{safe}" if HAVE_DBUS: class _RootInterface(ServiceInterface): def __init__(self, service: "MprisService"): super().__init__("org.mpris.MediaPlayer2") self._svc = service @method() def Raise(self): self._svc.dispatch("raise") @method() def Quit(self): self._svc.dispatch("quit") @dbus_property(access=PropertyAccess.READ) def Identity(self) -> "s": return self._svc.identity @dbus_property(access=PropertyAccess.READ) def DesktopEntry(self) -> "s": return self._svc.desktop_entry @dbus_property(access=PropertyAccess.READ) def CanQuit(self) -> "b": return True @dbus_property(access=PropertyAccess.READ) def CanRaise(self) -> "b": return True @dbus_property(access=PropertyAccess.READ) def HasTrackList(self) -> "b": return False @dbus_property(access=PropertyAccess.READ) def SupportedUriSchemes(self) -> "as": return [] @dbus_property(access=PropertyAccess.READ) def SupportedMimeTypes(self) -> "as": return [] class _PlayerInterface(ServiceInterface): def __init__(self, service: "MprisService"): super().__init__("org.mpris.MediaPlayer2.Player") self._svc = service # -- методы ------------------------------------------------------ @method() def Play(self): self._svc.dispatch("play") @method() def Pause(self): self._svc.dispatch("pause") @method() def PlayPause(self): self._svc.dispatch("play_pause") @method() def Stop(self): self._svc.dispatch("stop") @method() def Next(self): self._svc.dispatch("next") @method() def Previous(self): self._svc.dispatch("previous") @method() def Seek(self, offset: "x"): self._svc.dispatch("seek", int(offset)) @method() def SetPosition(self, track_id: "o", position: "x"): self._svc.dispatch("set_position", str(track_id), int(position)) @method() def OpenUri(self, uri: "s"): self._svc.dispatch("open_uri", str(uri)) # -- сигналы ----------------------------------------------------- @signal() def Seeked(self, position: "x") -> "x": return position # -- свойства ---------------------------------------------------- @dbus_property(access=PropertyAccess.READ) def PlaybackStatus(self) -> "s": return self._svc.playback_status @dbus_property(access=PropertyAccess.READ) def Metadata(self) -> "a{sv}": return self._svc.metadata @dbus_property(access=PropertyAccess.READ) def Position(self) -> "x": return int(self._svc.position) @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 CanGoNext(self) -> "b": return self._svc.can_next @dbus_property(access=PropertyAccess.READ) def CanGoPrevious(self) -> "b": return self._svc.can_previous @dbus_property(access=PropertyAccess.READ) def CanPlay(self) -> "b": return self._svc.can_play @dbus_property(access=PropertyAccess.READ) def CanPause(self) -> "b": return self._svc.can_pause @dbus_property(access=PropertyAccess.READ) def CanSeek(self) -> "b": return self._svc.can_seek @dbus_property(access=PropertyAccess.READ) def CanControl(self) -> "b": return True @dbus_property(access=PropertyAccess.READWRITE) def Volume(self) -> "d": return float(self._svc.volume) @Volume.setter def Volume(self, value: "d"): self._svc.dispatch("volume", float(value)) @dbus_property(access=PropertyAccess.READWRITE) def LoopStatus(self) -> "s": return self._svc.loop_status @LoopStatus.setter def LoopStatus(self, value: "s"): self._svc.loop_status = str(value) @dbus_property(access=PropertyAccess.READWRITE) def Shuffle(self) -> "b": return self._svc.shuffle @Shuffle.setter def Shuffle(self, value: "b"): self._svc.shuffle = bool(value) @dbus_property(access=PropertyAccess.READWRITE) def Rate(self) -> "d": return float(self._svc.rate) @Rate.setter def Rate(self, value: "d"): self._svc.rate = float(value) class MprisService: """Публикует MPRIS на сессионном D-Bus; состояние задаётся снаружи. ``handler`` — объект с необязательными методами ``on_play``, ``on_pause``, ``on_play_pause``, ``on_next``, ``on_previous``, ``on_stop``, ``on_seek(offset_us)``, ``on_set_position(track_path, us)``, ``on_volume(value)``, ``on_raise()``, ``on_quit()``. """ def __init__( self, handler: Any, identity: str = "MTS Music", desktop_entry: str = "mts-music", ): self.handler = handler self.identity = identity self.desktop_entry = desktop_entry self.enabled = False self.playback_status = "Stopped" self.metadata: dict[str, Any] = {} self.position = 0 self.volume = 1.0 self.can_next = False self.can_previous = False self.can_play = True self.can_pause = True self.can_seek = False self.loop_status = "None" self.shuffle = False self.rate = 1.0 self._loop: asyncio.AbstractEventLoop | None = None self._bus: Any = None self._player: Any = None self._root: Any = None self._thread: threading.Thread | None = None if HAVE_DBUS: self._player = _PlayerInterface(self) self._root = _RootInterface(self) self._thread = threading.Thread( target=self._run, name="mpris-dbus", daemon=True ) self._thread.start() # -- жизненный цикл --------------------------------------------------- def _run(self): loop = asyncio.new_event_loop() self._loop = loop asyncio.set_event_loop(loop) try: loop.run_until_complete(self._connect()) loop.run_forever() except Exception: # pragma: no cover - окружение без D-Bus log.debug("MPRIS недоступен", exc_info=True) async def _connect(self): self._bus = await MessageBus().connect() self._bus.export(OBJECT_PATH, self._player) self._bus.export(OBJECT_PATH, self._root) await self._bus.request_name(BUS_NAME) self.enabled = True log.debug("MPRIS зарегистрирован: %s", BUS_NAME) def close(self): if self._loop and self._loop.is_running(): self._loop.call_soon_threadsafe(self._loop.stop) # -- диспетчеризация команд из D-Bus --------------------------------- def dispatch(self, action: str, *args: Any): callback: Callable[..., Any] | None = getattr( self.handler, "on_" + action, None ) if callback is None: return try: callback(*args) except Exception: # pragma: no cover log.debug("ошибка обработчика MPRIS: %s", action, exc_info=True) def _emit(self, changed: dict[str, Any]): if not (self.enabled and self._loop): return self._loop.call_soon_threadsafe(self._player.emit_properties_changed, changed) # -- обновление состояния (из основного потока) ---------------------- def update_playback_status(self, status: str): if status == self.playback_status: return self.playback_status = status self.can_seek = status != "Stopped" or self.can_seek self._emit({"PlaybackStatus": status}) def update_metadata( self, track_id: Any, title: str = "", artist: str = "", album: str = "", duration_ms: int = 0, art_url: str = "", ): metadata: dict[str, Any] = { "mpris:trackid": Variant("o", track_object_path(track_id)), "mpris:length": Variant("x", int(duration_ms) * 1000), "xesam:title": Variant("s", title or ""), "xesam:artist": Variant("as", [artist] if artist else []), "xesam:album": Variant("s", album or ""), } if art_url: metadata["mpris:artUrl"] = Variant("s", art_url) self.metadata = metadata self._emit({"Metadata": metadata}) def clear_metadata(self): self.metadata = {} self._emit({"Metadata": self.metadata}) def update_position(self, position_us: int): # Position в MPRIS не рассылается через PropertiesChanged — клиенты # читают его сами (или ориентируются на сигнал Seeked). self.position = max(0, int(position_us)) def notify_seeked(self, position_us: int): self.update_position(position_us) if self.enabled and self._loop: self._loop.call_soon_threadsafe(self._player.Seeked, int(position_us)) def update_volume(self, volume: float): volume = max(0.0, min(1.0, float(volume))) if abs(volume - self.volume) < 1e-4: return self.volume = volume self._emit({"Volume": volume}) def update_capabilities( self, can_next: bool | None = None, can_previous: bool | None = None, can_play: bool | None = None, can_pause: bool | None = None, can_seek: bool | None = None, ): changed: dict[str, Any] = {} if can_next is not None and can_next != self.can_next: self.can_next = can_next changed["CanGoNext"] = can_next if can_previous is not None and can_previous != self.can_previous: self.can_previous = can_previous changed["CanGoPrevious"] = can_previous if can_play is not None and can_play != self.can_play: self.can_play = can_play changed["CanPlay"] = can_play if can_pause is not None and can_pause != self.can_pause: self.can_pause = can_pause changed["CanPause"] = can_pause if can_seek is not None and can_seek != self.can_seek: self.can_seek = can_seek changed["CanSeek"] = can_seek if changed: self._emit(changed)