info
343240fec1
Nach dem ersten fehlgeschlagenen Bus-Connect wird der Controller dauerhaft deaktiviert, statt bei jedem Hotkey-Druck einen neuen Connect-Versuch zu starten. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
"""Linux-MPRIS-Implementierung via dbus-next."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MPRIS_PREFIX = "org.mpris.MediaPlayer2."
|
|
MPRIS_PATH = "/org/mpris/MediaPlayer2"
|
|
PLAYER_IFACE = "org.mpris.MediaPlayer2.Player"
|
|
DBUS_SERVICE = "org.freedesktop.DBus"
|
|
DBUS_PATH = "/org/freedesktop/DBus"
|
|
DBUS_IFACE = "org.freedesktop.DBus"
|
|
|
|
|
|
class MprisController:
|
|
def __init__(self) -> None:
|
|
self._paused: list[str] = []
|
|
self._bus: Any = None
|
|
self._bus_broken: bool = False
|
|
|
|
async def _ensure_bus(self) -> Any:
|
|
if self._bus_broken:
|
|
raise RuntimeError("D-Bus session bus is unavailable")
|
|
if self._bus is None:
|
|
from dbus_next.aio import MessageBus
|
|
self._bus = await MessageBus().connect()
|
|
return self._bus
|
|
|
|
async def _list_player_names(self) -> list[str]:
|
|
bus = await self._ensure_bus()
|
|
intro = await bus.introspect(DBUS_SERVICE, DBUS_PATH)
|
|
obj = bus.get_proxy_object(DBUS_SERVICE, DBUS_PATH, intro)
|
|
iface = obj.get_interface(DBUS_IFACE)
|
|
names = await iface.call_list_names()
|
|
return [n for n in names if n.startswith(MPRIS_PREFIX)]
|
|
|
|
async def _get_player_interface(self, bus_name: str) -> Any:
|
|
bus = await self._ensure_bus()
|
|
intro = await bus.introspect(bus_name, MPRIS_PATH)
|
|
obj = bus.get_proxy_object(bus_name, MPRIS_PATH, intro)
|
|
return obj.get_interface(PLAYER_IFACE)
|
|
|
|
async def _pause_player(self, bus_name: str) -> str | None:
|
|
"""Pausiert einen Player, wenn er im Status 'Playing' ist.
|
|
Gibt den Bus-Namen zurück, wenn pausiert wurde, sonst None.
|
|
"""
|
|
try:
|
|
player = await self._get_player_interface(bus_name)
|
|
status = await player.get_playback_status()
|
|
if status != "Playing":
|
|
return None
|
|
await player.call_pause()
|
|
return bus_name
|
|
except Exception as e:
|
|
logger.warning("Konnte Player %s nicht pausieren: %s", bus_name, e)
|
|
return None
|
|
|
|
async def _resume_player(self, bus_name: str) -> None:
|
|
try:
|
|
player = await self._get_player_interface(bus_name)
|
|
await player.call_play()
|
|
except Exception as e:
|
|
logger.warning("Konnte Player %s nicht fortsetzen: %s", bus_name, e)
|
|
|
|
async def pause(self) -> None:
|
|
try:
|
|
names = await self._list_player_names()
|
|
except Exception as e:
|
|
if not self._bus_broken:
|
|
logger.warning(
|
|
"D-Bus nicht erreichbar, Media-Pause dauerhaft deaktiviert: %s", e
|
|
)
|
|
self._bus_broken = True
|
|
self._paused = []
|
|
return
|
|
results = await asyncio.gather(
|
|
*(self._pause_player(n) for n in names),
|
|
return_exceptions=True,
|
|
)
|
|
self._paused = [name for name in results if isinstance(name, str)]
|
|
|
|
async def resume(self) -> None:
|
|
if not self._paused:
|
|
return
|
|
to_resume = self._paused
|
|
self._paused = []
|
|
await asyncio.gather(
|
|
*(self._resume_player(n) for n in to_resume),
|
|
return_exceptions=True,
|
|
)
|