128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
"""Bridges the Spotify client's MPRIS state to the HA media_player entity via playerctl.
|
|
|
|
Identical in shape and reasoning to
|
|
hosts/thin-client/agent/thinclient_agent/mpris_bridge.py, with PLAYER_PRIORITY pointed
|
|
at the Spotify GUI client's own MPRIS bus name instead of mpv/spotifyd — see that
|
|
file's docstring for why this polls via subprocess rather than holding a dbus
|
|
connection open.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import subprocess
|
|
import threading
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
PLAYER_PRIORITY = "spotify,%any"
|
|
|
|
_STATUS_TO_HA = {
|
|
"Playing": "playing",
|
|
"Paused": "paused",
|
|
"Stopped": "idle",
|
|
}
|
|
|
|
_METADATA_FORMAT = "{{title}}\x1f{{artist}}\x1f{{album}}\x1f{{mpris:length}}\x1f{{mpris:artUrl}}"
|
|
|
|
|
|
class MprisBridge:
|
|
def __init__(self, publish_state, env_provider, poll_interval: float = 2.0):
|
|
self._publish_state = publish_state
|
|
# playerctl needs DBUS_SESSION_BUS_ADDRESS, which this system service does not
|
|
# inherit; SwayControl.session_env() derives it from the kiosk user's runtime dir.
|
|
self._env_provider = env_provider
|
|
self._poll_interval = poll_interval
|
|
self._last_state: dict | None = None
|
|
|
|
def _playerctl(self, *args: str) -> str | None:
|
|
try:
|
|
result = subprocess.run(
|
|
["playerctl", "-p", PLAYER_PRIORITY, *args],
|
|
env=self._env_provider(),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
log.debug("playerctl %s failed: %s", " ".join(args), exc)
|
|
return None
|
|
if result.returncode != 0:
|
|
return None
|
|
return result.stdout.strip()
|
|
|
|
def read_state(self) -> dict:
|
|
status = self._playerctl("status")
|
|
if status is None:
|
|
return {"state": "off"}
|
|
|
|
state = {"state": _STATUS_TO_HA.get(status, "idle")}
|
|
|
|
metadata = self._playerctl("metadata", "--format", _METADATA_FORMAT)
|
|
if metadata:
|
|
title, artist, album, length, art_url = (metadata.split("\x1f") + [""] * 5)[:5]
|
|
state["title"] = title
|
|
state["artist"] = artist
|
|
state["album"] = album
|
|
state["art_url"] = art_url
|
|
if length.isdigit():
|
|
state["duration"] = int(length) // 1_000_000
|
|
|
|
position = self._playerctl("position")
|
|
if position:
|
|
try:
|
|
state["position"] = int(float(position))
|
|
except ValueError:
|
|
pass
|
|
|
|
volume = self._playerctl("volume")
|
|
if volume:
|
|
try:
|
|
state["volume"] = round(float(volume), 3)
|
|
except ValueError:
|
|
pass
|
|
|
|
return state
|
|
|
|
def poll_once(self) -> None:
|
|
state = self.read_state()
|
|
if state != self._last_state:
|
|
self._last_state = state
|
|
self._publish_state(state)
|
|
|
|
def run_forever(self, stop_event: threading.Event) -> None:
|
|
while not stop_event.is_set():
|
|
try:
|
|
self.poll_once()
|
|
except Exception:
|
|
log.exception("MPRIS poll failed")
|
|
stop_event.wait(self._poll_interval)
|
|
|
|
# --- command side -------------------------------------------------------
|
|
def handle_command(self, command: str) -> None:
|
|
command = command.strip().upper()
|
|
action = {
|
|
"PLAY": ("play",),
|
|
"PAUSE": ("pause",),
|
|
"PLAY_PAUSE": ("play-pause",),
|
|
"TOGGLE": ("play-pause",),
|
|
"STOP": ("stop",),
|
|
"NEXT": ("next",),
|
|
"PREVIOUS": ("previous",),
|
|
"PREV": ("previous",),
|
|
}.get(command)
|
|
|
|
if action is None:
|
|
log.warning("ignoring unknown media command %r", command)
|
|
return
|
|
|
|
log.info("media command %s", command)
|
|
self._playerctl(*action)
|
|
self.poll_once()
|
|
|
|
def set_volume(self, level: float) -> None:
|
|
level = max(0.0, min(1.0, level))
|
|
self._playerctl("volume", f"{level:.3f}")
|
|
self.poll_once()
|