70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
"""Pause whatever is playing when focus starts — and only that.
|
|
|
|
Every modern player (Spotify, VLC, mpv, browsers, Elisa) speaks MPRIS, so this
|
|
needs no per-app support: list the `org.mpris.MediaPlayer2.*` names on the
|
|
session bus, pause the ones that are actually playing, and remember exactly
|
|
which ones so the break can hand the music back.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from PyQt6.QtDBus import QDBusConnection, QDBusInterface
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_PREFIX = "org.mpris.MediaPlayer2."
|
|
_PATH = "/org/mpris/MediaPlayer2"
|
|
_PLAYER_IFACE = "org.mpris.MediaPlayer2.Player"
|
|
_PROPS_IFACE = "org.freedesktop.DBus.Properties"
|
|
|
|
|
|
class MediaController:
|
|
def __init__(self) -> None:
|
|
self._bus = QDBusConnection.sessionBus()
|
|
self._paused_by_us: list[str] = []
|
|
|
|
def _players(self) -> list[str]:
|
|
interface = QDBusInterface(
|
|
"org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", self._bus
|
|
)
|
|
reply = interface.call("ListNames")
|
|
arguments = reply.arguments()
|
|
names = list(arguments[0]) if arguments else []
|
|
return [name for name in names if name.startswith(_PREFIX)]
|
|
|
|
def _status(self, service: str) -> str:
|
|
properties = QDBusInterface(service, _PATH, _PROPS_IFACE, self._bus)
|
|
reply = properties.call("Get", _PLAYER_IFACE, "PlaybackStatus")
|
|
arguments = reply.arguments()
|
|
return str(arguments[0]) if arguments else ""
|
|
|
|
def pause_all(self) -> list[str]:
|
|
"""Pause every playing player; returns the ones that were paused."""
|
|
paused: list[str] = []
|
|
for service in self._players():
|
|
try:
|
|
if self._status(service) != "Playing":
|
|
continue
|
|
QDBusInterface(service, _PATH, _PLAYER_IFACE, self._bus).call("Pause")
|
|
paused.append(service)
|
|
except Exception as exc: # pragma: no cover - a rude player is not our problem
|
|
log.debug("could not pause %s: %s", service, exc)
|
|
self._paused_by_us = paused
|
|
if paused:
|
|
log.info("paused %s media player(s)", len(paused))
|
|
return paused
|
|
|
|
def resume_paused(self) -> None:
|
|
"""Resume only what we paused, and only if it is still around."""
|
|
alive = set(self._players())
|
|
for service in self._paused_by_us:
|
|
if service not in alive:
|
|
continue
|
|
try:
|
|
QDBusInterface(service, _PATH, _PLAYER_IFACE, self._bus).call("Play")
|
|
except Exception as exc: # pragma: no cover
|
|
log.debug("could not resume %s: %s", service, exc)
|
|
self._paused_by_us = []
|