whisper-local/whisper_local/media/_mpris.py

"""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

    async def _ensure_bus(self) -> Any:
        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:
            logger.warning("D-Bus nicht erreichbar, überspringe Media-Pause: %s", e)
            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,
        )