139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
"""Bridges mpv's and Spotify'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 — see that file's docstring
|
|
for why this polls via subprocess rather than holding a dbus connection open.
|
|
|
|
PLAYER_PRIORITY lists mpv first, then the Spotify GUI client. That order is the answer
|
|
to "both are alive, which one does the remote's play button reach?", and mpv wins
|
|
because it is the one holding something somebody deliberately opened.
|
|
|
|
What is NOT in that list is any game. Steam exposes no MPRIS bus, and a game's audio is
|
|
not something you "pause" from a phone — the volume controls in audio_control.py are
|
|
the right surface for a game making noise, not these transport buttons.
|
|
|
|
This bridge reports "off" for most of a gaming session, and that is correct rather than
|
|
broken: on this image the media players genuinely do not exist until somebody leaves
|
|
Big Picture (see configs/session/media-session). The entity to watch for "what is this
|
|
box doing" is the session-mode sensor in session_mode.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import subprocess
|
|
import threading
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
PLAYER_PRIORITY = "mpv,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()
|