From eba8e66879acfdd6eae297da9033e88996966131 Mon Sep 17 00:00:00 2001 From: Sean Doran Date: Fri, 27 Feb 2026 18:28:56 -0500 Subject: [PATCH] Add MPRIS2 volume routing and PA event watcher; fix Spotify song transition reset - Add MPRISController class: wraps playerctl with 3s player-list cache, fuzzy app-name matching, and thread-safe get/set_volume via MPRIS2 D-Bus - Add PulseController event watcher thread: separate pulsectl connection listens for sink-input events and queues them so reapply fires within ~0.1s for PA-only apps instead of waiting up to 1s - Route set_app_volume and get_app_volume_norm through MPRIS first, falling back to PA stream for non-MPRIS apps (e.g. Brave/Chromium) - Update reapply_app_volumes to prefer MPRIS path, preventing turnupd from fighting Spotify on the PA layer during song transitions - Wire MPRISController and start_watching() into main(); drain_events() triggers immediate reapply on any PA sink-input event --- src/turnup/turnupd.py | 258 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 253 insertions(+), 5 deletions(-) diff --git a/src/turnup/turnupd.py b/src/turnup/turnupd.py index f165c2e..bcf5809 100755 --- a/src/turnup/turnupd.py +++ b/src/turnup/turnupd.py @@ -14,9 +14,11 @@ even when no knob is being moved. import logging import os +import queue import signal import subprocess import sys +import threading import time import pulsectl @@ -131,17 +133,156 @@ def all_led_colors( ] +# ── MPRIS2 controller (playerctl-backed) ────────────────────────────────────── + +class MPRISController: + """Uses *playerctl* to read/write per-app volume via the MPRIS2 D-Bus interface. + + Caches the player list for ``_CACHE_TTL`` seconds to avoid spawning a new + subprocess on every single call. + """ + + _CACHE_TTL: float = 3.0 # seconds between ``playerctl --list-all`` calls + + def __init__(self) -> None: + self._players: list[str] = [] + self._players_ts: float = 0.0 + self._lock = threading.Lock() + + # ── internal helpers ────────────────────────────────────────────────────── + + def _run(self, *args: str, timeout: float = 2.0) -> tuple[bool, str]: + """Run ``playerctl `` and return ``(success, stdout.strip())``.""" + try: + result = subprocess.run( + ["playerctl", *args], + capture_output=True, + text=True, + timeout=timeout, + ) + return result.returncode == 0, result.stdout.strip() + except (FileNotFoundError, subprocess.TimeoutExpired) as exc: + log.debug("playerctl call failed: %s", exc) + return False, "" + + def _refresh_players(self, *, force: bool = False) -> None: + """Refresh the cached player list if it has expired (or *force* is set).""" + now = time.monotonic() + if not force and (now - self._players_ts) < self._CACHE_TTL: + return + ok, out = self._run("--list-all") + with self._lock: + self._players = [p.strip() for p in out.splitlines() if p.strip()] if ok else [] + self._players_ts = now + + # ── public API ──────────────────────────────────────────────────────────── + + def find_player(self, app_name: str) -> str | None: + """Return the first cached player whose name contains *app_name* (case-insensitive).""" + self._refresh_players() + needle = app_name.lower() + with self._lock: + for player in self._players: + if needle in player.lower(): + return player + return None + + def get_volume(self, app_name: str) -> float | None: + """Return the MPRIS volume (0.0–1.0) for *app_name*, or ``None`` if unavailable.""" + player = self.find_player(app_name) + if player is None: + return None + ok, out = self._run("--player", player, "volume") + if not ok or not out: + return None + try: + return max(0.0, min(1.0, float(out))) + except ValueError: + return None + + def set_volume(self, app_name: str, volume: float) -> bool: + """Set the MPRIS volume for *app_name*. Returns ``True`` on success.""" + player = self.find_player(app_name) + if player is None: + return False + volume = max(0.0, min(1.0, volume)) + ok, _ = self._run("--player", player, "volume", f"{volume:.4f}") + if ok: + log.debug("MPRIS: %r volume → %.4f", player, volume) + return ok + + # ── PulseAudio / PipeWire controller ────────────────────────────────────────── class PulseController: - """Thin wrapper around :class:`pulsectl.Pulse` for volume and mute control.""" + """Thin wrapper around :class:`pulsectl.Pulse` for volume and mute control. - def __init__(self) -> None: + When *mpris* is supplied, ``set_app_volume`` and ``get_app_volume_norm`` + will prefer the MPRIS2 path for any app that has a live playerctl player, + falling back to the PulseAudio stream only when MPRIS is unavailable. + + A background watcher thread (started by :meth:`start_watching`) listens + for PulseAudio sink-input events and pushes indices onto ``_event_q`` so + the main loop can trigger an immediate reapply for PA-only apps instead of + waiting for the 1-second timer. + """ + + def __init__(self, mpris: MPRISController | None = None) -> None: self._pulse = pulsectl.Pulse("turnupd") + self._mpris = mpris + self._event_q: queue.Queue[int] = queue.Queue() + self._watcher_thread: threading.Thread | None = None + self._stop_event = threading.Event() def close(self) -> None: + self._stop_event.set() self._pulse.close() + # ── PA event watcher ────────────────────────────────────────────────────── + + def start_watching(self) -> None: + """Spawn the background PA event listener thread (idempotent).""" + if self._watcher_thread and self._watcher_thread.is_alive(): + return + self._stop_event.clear() + t = threading.Thread(target=self._event_loop, daemon=True, name="pa-watcher") + t.start() + self._watcher_thread = t + log.debug("PA watcher thread started") + + def _event_loop(self) -> None: + """Background thread: open a *separate* Pulse connection and listen for events.""" + try: + with pulsectl.Pulse("turnupd-watcher") as watch_pulse: + def _cb(ev: pulsectl.PulseEventInfo) -> None: # type: ignore[name-defined] + if ev.facility == "sink_input": + self._event_q.put(int(ev.index)) + raise pulsectl.PulseLoopStop + + watch_pulse.event_mask_set("sink_input") + watch_pulse.event_callback_set(_cb) + while not self._stop_event.is_set(): + try: + watch_pulse.event_listen(timeout=1.0) + except pulsectl.PulseLoopStop: + pass + except Exception as exc: + log.debug("PA event loop error: %s", exc) + time.sleep(0.5) + except Exception as exc: + log.warning("PA watcher thread exiting: %s", exc) + + def drain_events(self) -> bool: + """Drain all pending PA events. Returns ``True`` if any events were present.""" + had_events = False + while True: + try: + self._event_q.get_nowait() + had_events = True + except queue.Empty: + break + return had_events + def set_sink_volume(self, sink_name: str, volume: float) -> None: volume = max(0.0, min(VOLUME_MAX, volume)) try: @@ -192,15 +333,25 @@ class PulseController: def set_app_volume(self, app_name: str, volume: float) -> None: volume = max(0.0, min(VOLUME_MAX, volume)) + + # Prefer the MPRIS2 path — it writes to the app's internal slider so the + # volume survives song transitions (e.g. Spotify resetting on new tracks). + if self._mpris and self._mpris.set_volume(app_name, volume): + log.debug("MPRIS set_volume: %r = %.4f", app_name, volume) + return + + # Fall back to PulseAudio stream volume. needle = app_name.lower() + found = False try: for inp in self._pulse.sink_input_list(): name = inp.proplist.get("application.name", "") binary = inp.proplist.get("application.process.binary", "") if needle in name.lower() or needle in binary.lower(): self._pulse.volume_set_all_chans(inp, volume) - return - log.debug("App %r not found in sink inputs", app_name) + found = True + if not found: + log.debug("App %r not found in sink inputs", app_name) except Exception as exc: log.warning("set_app_volume(%r) failed: %s", app_name, exc) @@ -230,6 +381,13 @@ class PulseController: def get_app_volume_norm(self, app_name: str) -> float | None: """Return the current app volume normalised to 0.0–1.0, or None if not found.""" + # Prefer MPRIS — more accurate for apps like Spotify. + if self._mpris: + vol = self._mpris.get_volume(app_name) + if vol is not None: + return vol + + # Fall back to PulseAudio stream. needle = app_name.lower() try: for inp in self._pulse.sink_input_list(): @@ -278,6 +436,83 @@ def init_knob_norms(config: dict, pulse: PulseController) -> list[float]: return norms +def build_app_volume_map(config: dict, knob_norms: list[float]) -> dict[str, float]: + """Return ``{app_name_lower: volume}`` for every app/group knob in *config*. + + Used by :func:`reapply_app_volumes` to know what volume each configured + application should currently be at, based on the last knob positions. + """ + app_volumes: dict[str, float] = {} + for knob_id_str, knob_cfg in config.get("knobs", {}).items(): + try: + knob_id = int(knob_id_str) + except ValueError: + continue + if knob_id >= NUM_KNOBS: + continue + action = knob_cfg.get("action", "") + vol = round(knob_norms[knob_id] * VOLUME_MAX, 4) + if action == "app_volume": + t = knob_cfg.get("target", "") + if t: + app_volumes[t.lower()] = vol + elif action == "group_volume": + for t in knob_cfg.get("targets", []): + if t: + app_volumes[t.lower()] = vol + return app_volumes + + +def reapply_app_volumes(config: dict, pulse: PulseController, knob_norms: list[float]) -> None: + """Re-apply stored knob volumes to every matching active sink input. + + Called on a 1-second timer and whenever a PA sink-input event fires so + that new streams (e.g. Spotify starting a new song) are brought back to + the last knob position rather than being left at the 100 % default that + ``module-stream-restore`` restores them to. + + For MPRIS-capable apps the volume is written via playerctl (which updates + the app's own internal slider). For PA-only apps (e.g. Brave) the volume + is corrected on the PulseAudio stream level. + """ + app_volumes = build_app_volume_map(config, knob_norms) + if not app_volumes: + return + + # Split targets into MPRIS-handled vs PA-only. + mpris = pulse._mpris + pa_only: dict[str, float] = {} + + for app_name, vol in app_volumes.items(): + if mpris and mpris.set_volume(app_name, vol): + log.debug("reapply MPRIS: %r → %.4f", app_name, vol) + else: + pa_only[app_name] = vol + + if not pa_only: + return + + # PA stream correction for non-MPRIS apps. + try: + for inp in pulse._pulse.sink_input_list(): + name = inp.proplist.get("application.name", "").lower() + binary = inp.proplist.get("application.process.binary", "").lower() + for needle, vol in pa_only.items(): + if needle in name or needle in binary: + current = inp.volume.value_flat + if abs(current - vol) > 0.01: + pulse._pulse.volume_set_all_chans(inp, vol) + log.debug( + "reapply PA: %r volume %.2f → %.2f", + inp.proplist.get("application.name", needle), + current, + vol, + ) + break # one needle is enough per stream + except Exception as exc: + log.debug("reapply_app_volumes (PA) failed: %s", exc) + + # ── Event handlers ───────────────────────────────────────────────────────────── def handle_knob( @@ -356,7 +591,9 @@ def main() -> None: log.info("Turn Up daemon starting — %s @ %d baud", port, baud) - pulse = PulseController() + mpris = MPRISController() + pulse = PulseController(mpris) + pulse.start_watching() knob_norms = init_knob_norms(config, pulse) buf = bytearray() @@ -366,6 +603,9 @@ def main() -> None: except OSError: config_mtime = None last_config_check = time.monotonic() + # Re-apply app volumes periodically so new streams (e.g. Spotify new song) + # are brought to the last knob position rather than resetting to 100 %. + last_reapply = time.monotonic() def _shutdown(sig: int, _frame: object) -> None: log.info("Received signal %d — shutting down", sig) @@ -413,6 +653,14 @@ def main() -> None: except OSError: pass + # Re-apply configured app volumes every 1 s to catch new + # streams (e.g. Spotify starting a new song resets to 100 %). + # Also trigger immediately on any PA sink-input event so + # PA-only apps (e.g. Brave) are corrected within ~0.1 s. + if pulse.drain_events() or now - last_reapply >= 1.0: + last_reapply = now + reapply_app_volumes(config, pulse, knob_norms) + except serial.SerialException as exc: log.warning("Serial error: %s — retrying in 3 s", exc) time.sleep(3)