120 lines
4.5 KiB
Python
120 lines
4.5 KiB
Python
"""Live peak/amplitude metering, independent of the volume *setting* — the
|
|
whole point is to make "the app is playing but nothing is reaching the mix"
|
|
visually obvious even when volume sliders read 100%, instead of only ever
|
|
showing what the volume knob is set to.
|
|
|
|
PipeWire/pipewire-pulse has no `pactl` subcommand for this (pactl only reports
|
|
configured volume, never signal level), so this shells out to `parec`
|
|
(pipewire-pulse's PulseAudio-protocol recorder) in raw-PCM mode and computes
|
|
peak amplitude client-side from the byte stream — the same technique
|
|
pavucontrol's meters use under the hood. `parec --monitor-stream=<id>` taps a
|
|
single application's own stream (Applications tab); `parec --device=<name>`
|
|
taps a physical device directly (`<sink>.monitor` for outputs, the source
|
|
name itself for inputs).
|
|
|
|
One `parec` process per distinct target, shared across every card watching
|
|
it, and only running while that card is on the currently-visible tab of a
|
|
currently-visible panel (see ui/card_base.py's start_peak/stop_peak, driven by
|
|
ui/scroll_row.py's activate/deactivate) — a mixer panel comfortably shows a
|
|
dozen live meters, but there's no reason to keep any of them decoding audio
|
|
while the panel is hidden.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import array
|
|
from typing import Callable
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import Gio, GLib # noqa: E402
|
|
|
|
CHUNK_BYTES = 512 # ~256 s16le mono samples per read, small enough to feel live
|
|
RATE_HZ = 8000 # plenty of resolution for a peak/VU bar, cheap to decode
|
|
LATENCY_MSEC = 60
|
|
|
|
|
|
def _peak_from_pcm16(data: bytes) -> float:
|
|
usable = len(data) - (len(data) % 2)
|
|
if usable <= 0:
|
|
return 0.0
|
|
samples = array.array("h")
|
|
samples.frombytes(data[:usable])
|
|
peak = max((abs(s) for s in samples), default=0)
|
|
return min(1.0, peak / 32768.0)
|
|
|
|
|
|
class PeakMonitorManager:
|
|
def __init__(self) -> None:
|
|
self._monitors: dict[str, dict] = {}
|
|
|
|
def watch(self, key: str, target_args: list[str],
|
|
on_level: Callable[[float], None]) -> Callable[[], None]:
|
|
"""target_args is the parec argv tail identifying what to tap, e.g.
|
|
["--device=alsa_output.foo.monitor"] or ["--monitor-stream=42"].
|
|
Returns an unwatch() callback; call it to stop receiving levels."""
|
|
entry = self._monitors.get(key)
|
|
if entry is None:
|
|
entry = self._start(key, target_args)
|
|
entry["callbacks"].add(on_level)
|
|
|
|
def unwatch() -> None:
|
|
live = self._monitors.get(key)
|
|
if live is None:
|
|
return
|
|
live["callbacks"].discard(on_level)
|
|
if not live["callbacks"]:
|
|
self._stop(key)
|
|
|
|
return unwatch
|
|
|
|
def _start(self, key: str, target_args: list[str]) -> dict:
|
|
argv = ["parec", "--format=s16le", f"--rate={RATE_HZ}", "--channels=1",
|
|
f"--latency-msec={LATENCY_MSEC}", *target_args]
|
|
entry = {"proc": None, "stream": None, "callbacks": set()}
|
|
self._monitors[key] = entry
|
|
try:
|
|
proc = Gio.Subprocess.new(
|
|
argv, Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE)
|
|
except GLib.Error:
|
|
return entry
|
|
entry["proc"] = proc
|
|
entry["stream"] = proc.get_stdout_pipe()
|
|
self._read_next(key)
|
|
return entry
|
|
|
|
def _read_next(self, key: str) -> None:
|
|
entry = self._monitors.get(key)
|
|
if entry is None or entry["stream"] is None:
|
|
return
|
|
entry["stream"].read_bytes_async(CHUNK_BYTES, GLib.PRIORITY_DEFAULT, None,
|
|
self._on_bytes, key)
|
|
|
|
def _on_bytes(self, stream, result, key: str) -> None:
|
|
entry = self._monitors.get(key)
|
|
if entry is None:
|
|
return
|
|
try:
|
|
data = stream.read_bytes_finish(result).get_data() or b""
|
|
except GLib.Error:
|
|
data = b""
|
|
if not data:
|
|
# target went away (app closed, device unplugged) — the owning
|
|
# card's next refresh will drop it entirely; just stop quietly.
|
|
self._stop(key)
|
|
return
|
|
level = _peak_from_pcm16(data)
|
|
for cb in list(entry["callbacks"]):
|
|
cb(level)
|
|
self._read_next(key)
|
|
|
|
def _stop(self, key: str) -> None:
|
|
entry = self._monitors.pop(key, None)
|
|
if entry is not None and entry.get("proc") is not None:
|
|
entry["proc"].force_exit()
|
|
|
|
def stop_all(self) -> None:
|
|
for key in list(self._monitors.keys()):
|
|
self._stop(key)
|