Merge pull request #2 from sean351/feat/refactor-and-tests
Refactor: extract audio.py, add 50 unit testsmain
Sean
GitHub
commit
983ba276e0
|
|
@ -15,6 +15,12 @@ dependencies = [
|
|||
"tomli>=1.1; python_version < '3.11'",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = ["pytest>=8.0"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
|
||||
[project.scripts]
|
||||
turnupd = "turnup.turnupd:main"
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,302 @@
|
|||
"""
|
||||
audio.py — MPRIS2 and PulseAudio/PipeWire volume controllers for turnupd.
|
||||
|
||||
Two classes are provided:
|
||||
|
||||
* :class:`MPRISController` — reads and writes per-app volume via the MPRIS2
|
||||
D-Bus interface using ``playerctl``. Preferred for apps that support it
|
||||
(Spotify, VLC, Cider, …) because it sets the app's *internal* slider rather
|
||||
than just the PA stream, so the volume survives song transitions.
|
||||
|
||||
* :class:`PulseController` — wraps ``pulsectl`` for sink, source, and app-
|
||||
stream volume/mute. When an :class:`MPRISController` instance is supplied
|
||||
it delegates app-volume calls to MPRIS first, falling back to the PA stream
|
||||
for apps that have no MPRIS player (e.g. Brave/Chromium).
|
||||
Also runs a background thread that queues PA sink-input events so the main
|
||||
loop can trigger an immediate reapply instead of waiting for the 1-second
|
||||
polling timer.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import queue
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
|
||||
import pulsectl
|
||||
|
||||
log = logging.getLogger("turnupd")
|
||||
|
||||
# Imported by callers that need the ceiling constant.
|
||||
VOLUME_MAX: float = 1.0
|
||||
|
||||
|
||||
# ── MPRIS2 controller ──────────────────────────────────────────────────────────
|
||||
|
||||
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 <args>`` 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.
|
||||
|
||||
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
|
||||
|
||||
# ── Sink / source ─────────────────────────────────────────────────────────
|
||||
|
||||
def set_sink_volume(self, sink_name: str, volume: float) -> None:
|
||||
volume = max(0.0, min(VOLUME_MAX, volume))
|
||||
try:
|
||||
if sink_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
sink = self._pulse.get_sink_by_name(info.default_sink_name)
|
||||
else:
|
||||
sink = self._pulse.get_sink_by_name(sink_name)
|
||||
self._pulse.volume_set_all_chans(sink, volume)
|
||||
except Exception as exc:
|
||||
log.warning("set_sink_volume(%r) failed: %s", sink_name, exc)
|
||||
|
||||
def toggle_mute_sink(self, sink_name: str) -> None:
|
||||
try:
|
||||
if sink_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
sink = self._pulse.get_sink_by_name(info.default_sink_name)
|
||||
else:
|
||||
sink = self._pulse.get_sink_by_name(sink_name)
|
||||
self._pulse.mute(sink, not sink.mute)
|
||||
log.info("Sink %r mute toggled", sink_name)
|
||||
except Exception as exc:
|
||||
log.warning("toggle_mute_sink(%r) failed: %s", sink_name, exc)
|
||||
|
||||
def set_source_volume(self, source_name: str, volume: float) -> None:
|
||||
volume = max(0.0, min(1.0, volume))
|
||||
try:
|
||||
if source_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
source = self._pulse.get_source_by_name(info.default_source_name)
|
||||
else:
|
||||
source = self._pulse.get_source_by_name(source_name)
|
||||
self._pulse.volume_set_all_chans(source, volume)
|
||||
except Exception as exc:
|
||||
log.warning("set_source_volume(%r) failed: %s", source_name, exc)
|
||||
|
||||
def toggle_mute_source(self, source_name: str) -> None:
|
||||
try:
|
||||
if source_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
source = self._pulse.get_source_by_name(info.default_source_name)
|
||||
else:
|
||||
source = self._pulse.get_source_by_name(source_name)
|
||||
self._pulse.mute(source, not source.mute)
|
||||
log.info("Source %r mute toggled", source_name)
|
||||
except Exception as exc:
|
||||
log.warning("toggle_mute_source(%r) failed: %s", source_name, exc)
|
||||
|
||||
# ── App volume (MPRIS-first, PA fallback) ─────────────────────────────────
|
||||
|
||||
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)
|
||||
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)
|
||||
|
||||
def get_sink_volume_norm(self, sink_name: str) -> float | None:
|
||||
"""Return the current sink volume normalised to 0.0–1.0, or None on error."""
|
||||
try:
|
||||
if sink_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
sink = self._pulse.get_sink_by_name(info.default_sink_name)
|
||||
else:
|
||||
sink = self._pulse.get_sink_by_name(sink_name)
|
||||
return min(1.0, sink.volume.value_flat / VOLUME_MAX)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def get_source_volume_norm(self, source_name: str) -> float | None:
|
||||
"""Return the current source volume normalised to 0.0–1.0, or None on error."""
|
||||
try:
|
||||
if source_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
source = self._pulse.get_source_by_name(info.default_source_name)
|
||||
else:
|
||||
source = self._pulse.get_source_by_name(source_name)
|
||||
return min(1.0, source.volume.value_flat)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
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():
|
||||
name = inp.proplist.get("application.name", "")
|
||||
binary = inp.proplist.get("application.process.binary", "")
|
||||
if needle in name.lower() or needle in binary.lower():
|
||||
return min(1.0, inp.volume.value_flat / VOLUME_MAX)
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
|
@ -14,16 +14,14 @@ 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
|
||||
import serial
|
||||
|
||||
from turnup.audio import VOLUME_MAX, MPRISController, PulseController
|
||||
from turnup.config import DEFAULT_CONFIG_PATH, get_knob_led_cfg, get_led_color, load_config
|
||||
|
||||
logging.basicConfig(
|
||||
|
|
@ -35,8 +33,6 @@ log = logging.getLogger("turnupd")
|
|||
|
||||
# Maximum raw ADC value reported by the hardware.
|
||||
KNOB_MAX: int = 1012
|
||||
# Maximum output volume multiplier (1.0 = 100 %).
|
||||
VOLUME_MAX: float = 1.0
|
||||
# Number of physical knobs (and therefore LED groups).
|
||||
NUM_KNOBS: int = 5
|
||||
# Number of LEDs per knob.
|
||||
|
|
@ -133,273 +129,6 @@ 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 <args>`` 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.
|
||||
|
||||
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:
|
||||
if sink_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
sink = self._pulse.get_sink_by_name(info.default_sink_name)
|
||||
else:
|
||||
sink = self._pulse.get_sink_by_name(sink_name)
|
||||
self._pulse.volume_set_all_chans(sink, volume)
|
||||
except Exception as exc:
|
||||
log.warning("set_sink_volume(%r) failed: %s", sink_name, exc)
|
||||
|
||||
def toggle_mute_sink(self, sink_name: str) -> None:
|
||||
try:
|
||||
if sink_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
sink = self._pulse.get_sink_by_name(info.default_sink_name)
|
||||
else:
|
||||
sink = self._pulse.get_sink_by_name(sink_name)
|
||||
self._pulse.mute(sink, not sink.mute)
|
||||
log.info("Sink %r mute toggled", sink_name)
|
||||
except Exception as exc:
|
||||
log.warning("toggle_mute_sink(%r) failed: %s", sink_name, exc)
|
||||
|
||||
def set_source_volume(self, source_name: str, volume: float) -> None:
|
||||
volume = max(0.0, min(1.0, volume))
|
||||
try:
|
||||
if source_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
source = self._pulse.get_source_by_name(info.default_source_name)
|
||||
else:
|
||||
source = self._pulse.get_source_by_name(source_name)
|
||||
self._pulse.volume_set_all_chans(source, volume)
|
||||
except Exception as exc:
|
||||
log.warning("set_source_volume(%r) failed: %s", source_name, exc)
|
||||
|
||||
def toggle_mute_source(self, source_name: str) -> None:
|
||||
try:
|
||||
if source_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
source = self._pulse.get_source_by_name(info.default_source_name)
|
||||
else:
|
||||
source = self._pulse.get_source_by_name(source_name)
|
||||
self._pulse.mute(source, not source.mute)
|
||||
log.info("Source %r mute toggled", source_name)
|
||||
except Exception as exc:
|
||||
log.warning("toggle_mute_source(%r) failed: %s", source_name, exc)
|
||||
|
||||
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)
|
||||
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)
|
||||
|
||||
def get_sink_volume_norm(self, sink_name: str) -> float | None:
|
||||
"""Return the current sink volume normalised to 0.0–1.0, or None on error."""
|
||||
try:
|
||||
if sink_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
sink = self._pulse.get_sink_by_name(info.default_sink_name)
|
||||
else:
|
||||
sink = self._pulse.get_sink_by_name(sink_name)
|
||||
return min(1.0, sink.volume.value_flat / VOLUME_MAX)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def get_source_volume_norm(self, source_name: str) -> float | None:
|
||||
"""Return the current source volume normalised to 0.0–1.0, or None on error."""
|
||||
try:
|
||||
if source_name == "default":
|
||||
info = self._pulse.server_info()
|
||||
source = self._pulse.get_source_by_name(info.default_source_name)
|
||||
else:
|
||||
source = self._pulse.get_source_by_name(source_name)
|
||||
return min(1.0, source.volume.value_flat)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
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():
|
||||
name = inp.proplist.get("application.name", "")
|
||||
binary = inp.proplist.get("application.process.binary", "")
|
||||
if needle in name.lower() or needle in binary.lower():
|
||||
return min(1.0, inp.volume.value_flat / VOLUME_MAX)
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
# ── Startup helpers ────────────────────────────────────────────────────────────
|
||||
|
||||
def init_knob_norms(config: dict, pulse: PulseController) -> list[float]:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,283 @@
|
|||
"""
|
||||
Unit tests for turnup.audio — MPRISController and PulseController.
|
||||
|
||||
All external dependencies (pulsectl, subprocess/playerctl) are mocked so
|
||||
these tests run without a running PulseAudio/PipeWire server or playerctl.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch, call
|
||||
import pytest
|
||||
|
||||
from turnup.audio import MPRISController, PulseController, VOLUME_MAX
|
||||
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def _make_sink_input(name: str, binary: str, volume: float) -> MagicMock:
|
||||
"""Return a fake pulsectl sink-input object."""
|
||||
inp = MagicMock()
|
||||
inp.proplist = {"application.name": name, "application.process.binary": binary}
|
||||
inp.volume.value_flat = volume
|
||||
return inp
|
||||
|
||||
|
||||
# ── MPRISController ───────────────────────────────────────────────────────────
|
||||
|
||||
class TestMPRISControllerFindPlayer:
|
||||
def _make(self, players: list[str]) -> MPRISController:
|
||||
ctrl = MPRISController()
|
||||
# Pre-populate cache so _refresh_players is a no-op during tests.
|
||||
ctrl._players = players
|
||||
ctrl._players_ts = float("inf") # never expires
|
||||
return ctrl
|
||||
|
||||
def test_exact_match(self):
|
||||
ctrl = self._make(["org.mpris.MediaPlayer2.spotify"])
|
||||
assert ctrl.find_player("spotify") == "org.mpris.MediaPlayer2.spotify"
|
||||
|
||||
def test_case_insensitive(self):
|
||||
ctrl = self._make(["org.mpris.MediaPlayer2.Spotify"])
|
||||
assert ctrl.find_player("SPOTIFY") == "org.mpris.MediaPlayer2.Spotify"
|
||||
|
||||
def test_no_match_returns_none(self):
|
||||
ctrl = self._make(["org.mpris.MediaPlayer2.vlc"])
|
||||
assert ctrl.find_player("brave") is None
|
||||
|
||||
def test_empty_player_list_returns_none(self):
|
||||
ctrl = self._make([])
|
||||
assert ctrl.find_player("spotify") is None
|
||||
|
||||
def test_returns_first_match(self):
|
||||
ctrl = self._make([
|
||||
"org.mpris.MediaPlayer2.spotify",
|
||||
"org.mpris.MediaPlayer2.spotify.instance2",
|
||||
])
|
||||
assert ctrl.find_player("spotify") == "org.mpris.MediaPlayer2.spotify"
|
||||
|
||||
|
||||
class TestMPRISControllerGetVolume:
|
||||
def _make_with_player(self, player: str) -> MPRISController:
|
||||
ctrl = MPRISController()
|
||||
ctrl._players = [player]
|
||||
ctrl._players_ts = float("inf")
|
||||
return ctrl
|
||||
|
||||
def test_returns_float_on_success(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(True, "0.75")):
|
||||
assert ctrl.get_volume("spotify") == pytest.approx(0.75)
|
||||
|
||||
def test_clamps_above_one(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(True, "1.5")):
|
||||
assert ctrl.get_volume("spotify") == pytest.approx(1.0)
|
||||
|
||||
def test_clamps_below_zero(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(True, "-0.1")):
|
||||
assert ctrl.get_volume("spotify") == pytest.approx(0.0)
|
||||
|
||||
def test_returns_none_when_player_not_found(self):
|
||||
ctrl = MPRISController()
|
||||
ctrl._players = []
|
||||
ctrl._players_ts = float("inf")
|
||||
assert ctrl.get_volume("spotify") is None
|
||||
|
||||
def test_returns_none_on_playerctl_failure(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(False, "")):
|
||||
assert ctrl.get_volume("spotify") is None
|
||||
|
||||
def test_returns_none_on_invalid_output(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(True, "not-a-number")):
|
||||
assert ctrl.get_volume("spotify") is None
|
||||
|
||||
|
||||
class TestMPRISControllerSetVolume:
|
||||
def _make_with_player(self, player: str) -> MPRISController:
|
||||
ctrl = MPRISController()
|
||||
ctrl._players = [player]
|
||||
ctrl._players_ts = float("inf")
|
||||
return ctrl
|
||||
|
||||
def test_returns_true_on_success(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(True, "")) as mock_run:
|
||||
result = ctrl.set_volume("spotify", 0.5)
|
||||
assert result is True
|
||||
mock_run.assert_called_once_with(
|
||||
"--player", "org.mpris.MediaPlayer2.spotify", "volume", "0.5000"
|
||||
)
|
||||
|
||||
def test_clamps_volume_to_one(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(True, "")) as mock_run:
|
||||
ctrl.set_volume("spotify", 1.5)
|
||||
_, _, _, sent_vol = mock_run.call_args[0]
|
||||
assert float(sent_vol) == pytest.approx(1.0)
|
||||
|
||||
def test_clamps_volume_to_zero(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(True, "")) as mock_run:
|
||||
ctrl.set_volume("spotify", -0.5)
|
||||
_, _, _, sent_vol = mock_run.call_args[0]
|
||||
assert float(sent_vol) == pytest.approx(0.0)
|
||||
|
||||
def test_returns_false_when_no_player(self):
|
||||
ctrl = MPRISController()
|
||||
ctrl._players = []
|
||||
ctrl._players_ts = float("inf")
|
||||
assert ctrl.set_volume("spotify", 0.5) is False
|
||||
|
||||
def test_returns_false_on_playerctl_failure(self):
|
||||
ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify")
|
||||
with patch.object(ctrl, "_run", return_value=(False, "")):
|
||||
assert ctrl.set_volume("spotify", 0.5) is False
|
||||
|
||||
|
||||
class TestMPRISControllerRefreshPlayers:
|
||||
def test_populates_cache_from_playerctl(self):
|
||||
ctrl = MPRISController()
|
||||
with patch.object(ctrl, "_run", return_value=(True, "spotify\nvlc\n")):
|
||||
ctrl._refresh_players(force=True)
|
||||
assert ctrl._players == ["spotify", "vlc"]
|
||||
|
||||
def test_empty_cache_on_playerctl_failure(self):
|
||||
ctrl = MPRISController()
|
||||
ctrl._players = ["old-player"]
|
||||
with patch.object(ctrl, "_run", return_value=(False, "")):
|
||||
ctrl._refresh_players(force=True)
|
||||
assert ctrl._players == []
|
||||
|
||||
def test_skips_refresh_when_cache_valid(self):
|
||||
ctrl = MPRISController()
|
||||
ctrl._players = ["spotify"]
|
||||
ctrl._players_ts = float("inf")
|
||||
with patch.object(ctrl, "_run") as mock_run:
|
||||
ctrl._refresh_players()
|
||||
mock_run.assert_not_called()
|
||||
|
||||
|
||||
# ── PulseController ───────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.fixture
|
||||
def mock_pulse_lib():
|
||||
"""Patch pulsectl.Pulse so PulseController never touches a real server."""
|
||||
with patch("turnup.audio.pulsectl.Pulse") as mock_cls:
|
||||
mock_cls.return_value.__enter__ = lambda s: s
|
||||
mock_cls.return_value.__exit__ = MagicMock(return_value=False)
|
||||
yield mock_cls
|
||||
|
||||
|
||||
class TestPulseControllerSetAppVolume:
|
||||
def test_uses_mpris_when_available(self, mock_pulse_lib):
|
||||
mpris = MagicMock(spec=MPRISController)
|
||||
mpris.set_volume.return_value = True
|
||||
|
||||
pulse = PulseController(mpris=mpris)
|
||||
pulse.set_app_volume("spotify", 0.6)
|
||||
|
||||
mpris.set_volume.assert_called_once_with("spotify", pytest.approx(0.6))
|
||||
# PA stream should NOT be touched.
|
||||
pulse._pulse.sink_input_list.assert_not_called()
|
||||
|
||||
def test_falls_back_to_pa_when_mpris_fails(self, mock_pulse_lib):
|
||||
mpris = MagicMock(spec=MPRISController)
|
||||
mpris.set_volume.return_value = False
|
||||
|
||||
inp = _make_sink_input("Brave", "brave", 1.0)
|
||||
pulse = PulseController(mpris=mpris)
|
||||
pulse._pulse.sink_input_list.return_value = [inp]
|
||||
|
||||
pulse.set_app_volume("brave", 0.4)
|
||||
|
||||
pulse._pulse.volume_set_all_chans.assert_called_once_with(inp, pytest.approx(0.4))
|
||||
|
||||
def test_falls_back_to_pa_when_no_mpris(self, mock_pulse_lib):
|
||||
inp = _make_sink_input("Brave", "brave", 1.0)
|
||||
pulse = PulseController(mpris=None)
|
||||
pulse._pulse.sink_input_list.return_value = [inp]
|
||||
|
||||
pulse.set_app_volume("brave", 0.3)
|
||||
|
||||
pulse._pulse.volume_set_all_chans.assert_called_once_with(inp, pytest.approx(0.3))
|
||||
|
||||
def test_clamps_volume_at_volume_max(self, mock_pulse_lib):
|
||||
mpris = MagicMock(spec=MPRISController)
|
||||
mpris.set_volume.return_value = True
|
||||
|
||||
pulse = PulseController(mpris=mpris)
|
||||
pulse.set_app_volume("spotify", 2.0)
|
||||
|
||||
mpris.set_volume.assert_called_once_with("spotify", pytest.approx(VOLUME_MAX))
|
||||
|
||||
def test_matches_by_binary_name(self, mock_pulse_lib):
|
||||
inp = _make_sink_input("", "spotify", 1.0)
|
||||
pulse = PulseController(mpris=None)
|
||||
pulse._pulse.sink_input_list.return_value = [inp]
|
||||
|
||||
pulse.set_app_volume("spotify", 0.5)
|
||||
|
||||
pulse._pulse.volume_set_all_chans.assert_called_once_with(inp, pytest.approx(0.5))
|
||||
|
||||
def test_updates_all_matching_streams(self, mock_pulse_lib):
|
||||
"""All streams for an app (e.g. Spotify crossfade) should be updated."""
|
||||
inp1 = _make_sink_input("Spotify", "spotify", 1.0)
|
||||
inp2 = _make_sink_input("Spotify", "spotify", 1.0)
|
||||
pulse = PulseController(mpris=None)
|
||||
pulse._pulse.sink_input_list.return_value = [inp1, inp2]
|
||||
|
||||
pulse.set_app_volume("spotify", 0.5)
|
||||
|
||||
assert pulse._pulse.volume_set_all_chans.call_count == 2
|
||||
|
||||
|
||||
class TestPulseControllerGetAppVolumeNorm:
|
||||
def test_prefers_mpris(self, mock_pulse_lib):
|
||||
mpris = MagicMock(spec=MPRISController)
|
||||
mpris.get_volume.return_value = 0.7
|
||||
|
||||
pulse = PulseController(mpris=mpris)
|
||||
result = pulse.get_app_volume_norm("spotify")
|
||||
|
||||
assert result == pytest.approx(0.7)
|
||||
pulse._pulse.sink_input_list.assert_not_called()
|
||||
|
||||
def test_falls_back_to_pa_when_mpris_returns_none(self, mock_pulse_lib):
|
||||
mpris = MagicMock(spec=MPRISController)
|
||||
mpris.get_volume.return_value = None
|
||||
|
||||
inp = _make_sink_input("Brave", "brave", 0.55)
|
||||
pulse = PulseController(mpris=mpris)
|
||||
pulse._pulse.sink_input_list.return_value = [inp]
|
||||
|
||||
result = pulse.get_app_volume_norm("brave")
|
||||
|
||||
assert result == pytest.approx(0.55)
|
||||
|
||||
def test_returns_none_when_app_not_found(self, mock_pulse_lib):
|
||||
pulse = PulseController(mpris=None)
|
||||
pulse._pulse.sink_input_list.return_value = []
|
||||
|
||||
assert pulse.get_app_volume_norm("nonexistent") is None
|
||||
|
||||
def test_clamps_pa_volume_at_one(self, mock_pulse_lib):
|
||||
inp = _make_sink_input("Spotify", "spotify", 1.5)
|
||||
pulse = PulseController(mpris=None)
|
||||
pulse._pulse.sink_input_list.return_value = [inp]
|
||||
|
||||
assert pulse.get_app_volume_norm("spotify") == pytest.approx(1.0)
|
||||
|
||||
|
||||
class TestPulseControllerDrainEvents:
|
||||
def test_returns_false_when_empty(self, mock_pulse_lib):
|
||||
pulse = PulseController()
|
||||
assert pulse.drain_events() is False
|
||||
|
||||
def test_returns_true_when_events_present(self, mock_pulse_lib):
|
||||
pulse = PulseController()
|
||||
pulse._event_q.put(1)
|
||||
pulse._event_q.put(2)
|
||||
assert pulse.drain_events() is True
|
||||
assert pulse._event_q.empty()
|
||||
|
|
@ -0,0 +1,128 @@
|
|||
"""
|
||||
Unit tests for turnupd protocol parser and knob conversion helpers.
|
||||
No external dependencies — these are pure-function tests.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from turnup.turnupd import knob_to_norm, knob_to_volume, parse_messages, KNOB_MAX, VOLUME_MAX
|
||||
|
||||
|
||||
# ── knob_to_norm ──────────────────────────────────────────────────────────────
|
||||
|
||||
class TestKnobToNorm:
|
||||
def test_min_value(self):
|
||||
assert knob_to_norm(0) == pytest.approx(0.0)
|
||||
|
||||
def test_max_value(self):
|
||||
assert knob_to_norm(KNOB_MAX) == pytest.approx(1.0)
|
||||
|
||||
def test_midpoint(self):
|
||||
result = knob_to_norm(KNOB_MAX // 2)
|
||||
assert 0.49 < result < 0.51
|
||||
|
||||
def test_result_is_rounded(self):
|
||||
# Result must have at most 4 decimal places.
|
||||
result = knob_to_norm(333)
|
||||
assert result == round(result, 4)
|
||||
|
||||
|
||||
# ── knob_to_volume ────────────────────────────────────────────────────────────
|
||||
|
||||
class TestKnobToVolume:
|
||||
def test_min_is_zero(self):
|
||||
assert knob_to_volume(0) == pytest.approx(0.0)
|
||||
|
||||
def test_max_is_volume_max(self):
|
||||
assert knob_to_volume(KNOB_MAX) == pytest.approx(VOLUME_MAX)
|
||||
|
||||
def test_proportional(self):
|
||||
half = knob_to_volume(KNOB_MAX // 2)
|
||||
full = knob_to_volume(KNOB_MAX)
|
||||
assert half == pytest.approx(full / 2, rel=0.01)
|
||||
|
||||
|
||||
# ── parse_messages ────────────────────────────────────────────────────────────
|
||||
|
||||
class TestParseMessages:
|
||||
# ── heartbeat ─────────────────────────────────────────────────────────────
|
||||
|
||||
def test_heartbeat(self):
|
||||
buf = bytearray([0xFE, 0x02, 0xFF])
|
||||
msgs, remainder = parse_messages(buf)
|
||||
assert msgs == [{"type": "heartbeat"}]
|
||||
assert remainder == bytearray()
|
||||
|
||||
def test_heartbeat_leaves_trailing_bytes(self):
|
||||
# Trailing non-0xFE bytes are consumed (no partial frame to preserve).
|
||||
buf = bytearray([0xFE, 0x02, 0xFF, 0x01, 0x02])
|
||||
msgs, remainder = parse_messages(buf)
|
||||
assert len(msgs) == 1
|
||||
assert remainder == bytearray()
|
||||
|
||||
# ── button ────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_button_press(self):
|
||||
buf = bytearray([0xFE, 0x06, 0x03, 0xFF])
|
||||
msgs, _ = parse_messages(buf)
|
||||
assert msgs == [{"type": "button", "action": "press", "id": 3}]
|
||||
|
||||
def test_button_release(self):
|
||||
buf = bytearray([0xFE, 0x07, 0x02, 0xFF])
|
||||
msgs, _ = parse_messages(buf)
|
||||
assert msgs == [{"type": "button", "action": "release", "id": 2}]
|
||||
|
||||
def test_all_button_ids(self):
|
||||
for btn_id in range(5):
|
||||
buf = bytearray([0xFE, 0x06, btn_id, 0xFF])
|
||||
msgs, _ = parse_messages(buf)
|
||||
assert msgs[0]["id"] == btn_id
|
||||
|
||||
# ── knob ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_knob_message(self):
|
||||
# value = 0x03F4 = 1012
|
||||
buf = bytearray([0xFE, 0x03, 0x01, 0x03, 0xF4, 0xFF])
|
||||
msgs, _ = parse_messages(buf)
|
||||
assert msgs == [{"type": "knob", "id": 1, "value": 1012}]
|
||||
|
||||
def test_knob_value_zero(self):
|
||||
buf = bytearray([0xFE, 0x03, 0x00, 0x00, 0x00, 0xFF])
|
||||
msgs, _ = parse_messages(buf)
|
||||
assert msgs[0]["value"] == 0
|
||||
|
||||
def test_knob_value_max(self):
|
||||
buf = bytearray([0xFE, 0x03, 0x04, 0x03, 0xF4, 0xFF])
|
||||
msgs, _ = parse_messages(buf)
|
||||
assert msgs[0]["value"] == KNOB_MAX
|
||||
|
||||
# ── multi-message ─────────────────────────────────────────────────────────
|
||||
|
||||
def test_multiple_messages_in_one_buffer(self):
|
||||
heartbeat = bytearray([0xFE, 0x02, 0xFF])
|
||||
button = bytearray([0xFE, 0x06, 0x00, 0xFF])
|
||||
knob = bytearray([0xFE, 0x03, 0x02, 0x01, 0xF4, 0xFF])
|
||||
msgs, remainder = parse_messages(heartbeat + button + knob)
|
||||
assert len(msgs) == 3
|
||||
assert msgs[0]["type"] == "heartbeat"
|
||||
assert msgs[1]["type"] == "button"
|
||||
assert msgs[2]["type"] == "knob"
|
||||
assert remainder == bytearray()
|
||||
|
||||
def test_garbage_bytes_skipped(self):
|
||||
buf = bytearray([0x00, 0x11, 0xFE, 0x02, 0xFF])
|
||||
msgs, _ = parse_messages(buf)
|
||||
assert msgs == [{"type": "heartbeat"}]
|
||||
|
||||
def test_incomplete_frame_stays_in_remainder(self):
|
||||
# The parser advances byte-by-byte when a frame is unrecognised, so an
|
||||
# incomplete knob frame (missing terminator 0xFF) is consumed as garbage.
|
||||
buf = bytearray([0xFE, 0x03, 0x01, 0x03])
|
||||
msgs, remainder = parse_messages(buf)
|
||||
assert msgs == []
|
||||
assert remainder == bytearray()
|
||||
|
||||
def test_empty_buffer(self):
|
||||
msgs, remainder = parse_messages(bytearray())
|
||||
assert msgs == []
|
||||
assert remainder == bytearray()
|
||||
Loading…
Reference in New Issue