From 1f8b3ee2cf84f0230a818e8dd045766760f61c02 Mon Sep 17 00:00:00 2001 From: bhack Date: Tue, 13 Oct 2026 01:02:00 +0200 Subject: [PATCH 1/2] Optimize idle analyzer and DSP work for 0.8.9 --- CHANGELOG.md | 8 ++ data/io.github.bhack.mini-eq.metainfo.xml | 13 ++- docs/performance.md | 17 ++++ pyproject.toml | 2 +- src/mini_eq/analyzer.py | 23 ++++- src/mini_eq/filter_chain.py | 27 ++++-- src/mini_eq/routing.py | 10 ++ src/mini_eq/window.py | 7 ++ src/mini_eq/window_analyzer.py | 112 ++++++++++++++++++++-- tests/test_mini_eq_analyzer.py | 72 ++++++++++++++ tests/test_mini_eq_filter_chain.py | 40 ++++++++ tests/test_mini_eq_routing.py | 23 +++++ tests/test_mini_eq_window_analyzer.py | 79 +++++++++++++++ tools/check_headless_pipewire_runtime.py | 44 ++++++++- 14 files changed, 455 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 97896ea..2c2c3bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## 0.8.9 - 2026-10-13 + +- Avoid processing unused Off bands while keeping all 32 band slots available. +- Stop analyzer animation and spectrum calculations while frozen, and stop + animation while the window is hidden. +- Use audio-driven analyzer wakeups and coalesced display updates while + preserving loudness measurement and background Shell meters. + ## 0.8.8 - 2026-09-17 - Harden single-instance lock handling against unsafe files and concurrent diff --git a/data/io.github.bhack.mini-eq.metainfo.xml b/data/io.github.bhack.mini-eq.metainfo.xml index 91ada6d..af3093d 100644 --- a/data/io.github.bhack.mini-eq.metainfo.xml +++ b/data/io.github.bhack.mini-eq.metainfo.xml @@ -33,11 +33,11 @@ - https://raw.githubusercontent.com/bhack/mini-eq/v0.8.8/docs/screenshots/mini-eq.png + https://raw.githubusercontent.com/bhack/mini-eq/v0.8.9/docs/screenshots/mini-eq.png Adjust sound output with equalizer controls - https://raw.githubusercontent.com/bhack/mini-eq/v0.8.8/docs/screenshots/mini-eq-dark.png + https://raw.githubusercontent.com/bhack/mini-eq/v0.8.9/docs/screenshots/mini-eq-dark.png Use the equalizer with dark style @@ -45,6 +45,15 @@ https://github.com/bhack/mini-eq/issues https://github.com/bhack/mini-eq + + +
    +
  • Avoid processing unused Off bands while keeping all 32 band slots available.
    • +
  • Stop analyzer animation and spectrum calculations while frozen, and stop animation while the window is hidden.
    • +
  • Use audio-driven analyzer wakeups and coalesced display updates while preserving loudness measurement and background Shell meters.
    • +
+ +
    diff --git a/docs/performance.md b/docs/performance.md index 52cc058..4ef1b83 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -83,3 +83,20 @@ https://docs.gtk.org/gtk4/running.html Those environment variables are debugging knobs, not stable application settings, so do not depend on them from Mini EQ code. + +Compare runtime CPU separately from profiler captures. Use a private +PipeWire/WirePlumber graph, synthetic playback, and temporary application +settings so the measurements do not change desktop audio or presets. Compare +Monitor off, live, frozen, and hidden-window states at both 48 and 192 kHz. +Warm up first, repeat the baseline after the changed build, and record the +runtime versions and CPU frequency/power policy. Express process CPU as a +percentage of one logical core; include native audio threads in that total. + +Use Sysprof's GTK trace to distinguish frame updates, layout, snapshots, and +rendering, and its native samples to distinguish PipeWire DSP from Python +callbacks. Profile startup separately from steady playback: shader compilation +and font loading can dominate the first frame. `cProfile` is useful for Python +call attribution, but its elapsed waiting time is not CPU consumption and its +instrumentation overhead must not be used as a benchmark result. The official +references are GTK's [drawing model](https://docs.gtk.org/gtk4/drawing-model.html) +and Python's [profiler guidance](https://docs.python.org/3/library/profile.html). diff --git a/pyproject.toml b/pyproject.toml index d0ad586..a286603 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "mini-eq" -version = "0.8.8" +version = "0.8.9" description = "Compact PipeWire system-wide parametric equalizer for Linux desktops." readme = "README.md" requires-python = ">=3.11" diff --git a/src/mini_eq/analyzer.py b/src/mini_eq/analyzer.py index d2760a1..6688c55 100644 --- a/src/mini_eq/analyzer.py +++ b/src/mini_eq/analyzer.py @@ -51,7 +51,6 @@ ANALYZER_FFT_MIN_SIZE = 8192 ANALYZER_FFT_MAX_SIZE = 32768 ANALYZER_SAMPLE_WIDTH_BYTES = 4 -ANALYZER_QUEUE_WAIT_SECONDS = 0.005 ANALYZER_READER_JOIN_TIMEOUT_SECONDS = 0.2 ANALYZER_DISPLAY_GAIN_MIN = -12.0 ANALYZER_DISPLAY_GAIN_MAX = 32.0 @@ -432,6 +431,8 @@ def __init__( self.stop_event = threading.Event() self.stop_event.set() self.audio_blocks = deque(maxlen=ANALYZER_CAPTURE_QUEUE_BLOCKS) + self.audio_blocks_ready = threading.Event() + self.spectrum_enabled = True self.sample_rate = SAMPLE_RATE self.response_speed = ANALYZER_RESPONSE_DEFAULT @@ -442,6 +443,10 @@ def client(self): def set_levels_callback(self, callback: Callable[[list[float]], None] | None) -> None: self.levels_callback = callback + def set_spectrum_enabled(self, enabled: bool) -> None: + # Display freeze does not interrupt integrated loudness measurement. + self.spectrum_enabled = bool(enabled) + def set_loudness_callback(self, callback: Callable[[AnalyzerLoudnessSnapshot | None], None] | None) -> None: self.loudness_callback = callback @@ -476,6 +481,7 @@ def start(self) -> bool: return True self.audio_blocks.clear() + self.audio_blocks_ready.clear() self.stop_event.clear() try: @@ -511,6 +517,7 @@ def stop(self, *, close_stream: bool = False, close_client: bool | None = None) self.reader_thread = None self.stop_event.set() + self.audio_blocks_ready.set() self.audio_blocks.clear() self.emit_loudness_snapshot(None) @@ -640,6 +647,7 @@ def process_audio_block(self, _stream, block) -> None: self.sample_rate = float(audio_format.get_rate() or self.sample_rate or SAMPLE_RATE) data = block.get_data().get_data() self.audio_blocks.append(interleaved_f32le_bytes_to_channel_payloads(data, audio_format.get_channels())) + self.audio_blocks_ready.set() def read_audio_levels(self) -> None: pending_samples = array("f") @@ -655,7 +663,12 @@ def read_audio_levels(self) -> None: try: left_payload, right_payload = self.audio_blocks.popleft() except IndexError: - self.stop_event.wait(ANALYZER_QUEUE_WAIT_SECONDS) + # Clear before rechecking the queue so a producer cannot + # enqueue between the empty check and wait unnoticed. + self.audio_blocks_ready.clear() + if self.audio_blocks or self.stop_event.is_set(): + continue + self.audio_blocks_ready.wait() continue if self.loudness_callback is None: @@ -683,6 +696,12 @@ def read_audio_levels(self) -> None: loudness_meter = None loudness_stopped = True + if not self.spectrum_enabled: + pending_samples = array("f") + fft_samples = array("f") + smoothed_powers = () + continue + pending_samples.extend(stereo_f32le_bytes_to_mono_samples(left_payload, right_payload)) frame_count = analyzer_frame_count(self.sample_rate) diff --git a/src/mini_eq/filter_chain.py b/src/mini_eq/filter_chain.py index a763088..87540b4 100644 --- a/src/mini_eq/filter_chain.py +++ b/src/mini_eq/filter_chain.py @@ -43,6 +43,8 @@ def native_biquad_band_control_values( sample_rate: float = SAMPLE_RATE, solo_active: bool = False, ) -> dict[str, float]: + if band.filter_type == FILTER_TYPES["Off"] or band.filter_type not in NATIVE_BIQUAD_LABELS: + return {} # The native filter computes its coefficients at the DSP clock rate. wet = float(eq_enabled and band_is_effective(band, solo_active) and band.filter_type in NATIVE_BIQUAD_LABELS) controls: dict[str, float] = {} @@ -67,8 +69,9 @@ def native_biquad_control_values( sample_rate: float = SAMPLE_RATE, ) -> dict[str, float]: controls = builtin_biquad_preamp_control_values(preamp_db, eq_enabled) + solo_active = bands_have_solo(bands) for index, band in enumerate(bands[:MAX_BANDS]): - controls.update(native_biquad_band_control_values(index, band, eq_enabled, sample_rate, bands_have_solo(bands))) + controls.update(native_biquad_band_control_values(index, band, eq_enabled, sample_rate, solo_active)) return controls @@ -77,6 +80,8 @@ def build_native_biquad_nodes(bands: list[EqBand], preamp_db: float, eq_enabled: for side in ("l", "r"): nodes.append(build_biquad_node(preamp_node_name(side), preamp_coefficients_by_rate(preamp_db, eq_enabled))) for index, band in enumerate(bands): + if band.filter_type == FILTER_TYPES["Off"] or band.filter_type not in NATIVE_BIQUAD_LABELS: + continue name = biquad_node_name(side, index) label = NATIVE_BIQUAD_LABELS.get(band.filter_type, "bq_peaking") controls = native_biquad_band_control_values(index, band, eq_enabled, solo_active=bands_have_solo(bands)) @@ -90,11 +95,11 @@ def build_native_biquad_nodes(bands: list[EqBand], preamp_db: float, eq_enabled: return "\n".join(nodes) -def build_native_biquad_links(band_count: int) -> str: +def build_native_biquad_links(band_indexes: list[int]) -> str: links: list[str] = [] for side in ("l", "r"): previous = preamp_node_name(side) - for index in range(band_count): + for index in band_indexes: name = biquad_node_name(side, index) links.extend( [ @@ -295,10 +300,20 @@ def build_builtin_biquad_filter_chain_module_args( nodes = (build_native_biquad_nodes if native_biquads else build_builtin_biquad_nodes)( graph_bands, preamp_db, eq_enabled ) - links = (build_native_biquad_links if native_biquads else build_builtin_biquad_links)(band_count) + if native_biquads: + band_indexes = [ + index + for index, band in enumerate(graph_bands) + if band.filter_type != FILTER_TYPES["Off"] and band.filter_type in NATIVE_BIQUAD_LABELS + ] + links = build_native_biquad_links(band_indexes) + output_index = band_indexes[-1] if band_indexes else None + else: + links = build_builtin_biquad_links(band_count) + output_index = band_count - 1 if band_count else None rate_property = "" if native_biquads else f" audio.rate = {int(SAMPLE_RATE)}\n" - output_l = biquad_node_name("l", band_count - 1) if band_count else preamp_node_name("l") - output_r = biquad_node_name("r", band_count - 1) if band_count else preamp_node_name("r") + output_l = biquad_node_name("l", output_index) if output_index is not None else preamp_node_name("l") + output_r = biquad_node_name("r", output_index) if output_index is not None else preamp_node_name("r") return f"""{{ node.description = {pipewire_quote(VIRTUAL_SINK_DESCRIPTION)} diff --git a/src/mini_eq/routing.py b/src/mini_eq/routing.py index 6b71f43..365b909 100644 --- a/src/mini_eq/routing.py +++ b/src/mini_eq/routing.py @@ -147,6 +147,7 @@ def __init__(self, output_sink: str | None) -> None: self.stream_router: PipeWireStreamRouter | None = None self.output_analyzer: OutputSpectrumAnalyzer | None = None self.analyzer_response_speed = ANALYZER_RESPONSE_DEFAULT + self.analyzer_spectrum_enabled = True if not self.is_valid_output_sink(self.output_sink): raise AudioBackendError("output sink cannot be a Mini EQ virtual sink") @@ -188,6 +189,11 @@ def set_analyzer_loudness_callback( if self.output_analyzer is not None: self.output_analyzer.set_loudness_callback(callback) + def set_analyzer_spectrum_enabled(self, enabled: bool) -> None: + self.analyzer_spectrum_enabled = bool(enabled) + if self.output_analyzer is not None: + self.output_analyzer.set_spectrum_enabled(enabled) + def list_sinks(self) -> list[PipeWireNode]: return self.output_backend.list_audio_sinks() @@ -328,6 +334,7 @@ def ensure_output_analyzer(self) -> OutputSpectrumAnalyzer: self.output_analyzer.set_levels_callback(self.analyzer_levels_callback) self.output_analyzer.set_loudness_callback(self.analyzer_loudness_callback) self.output_analyzer.set_response_speed(self.analyzer_response_speed) + self.output_analyzer.set_spectrum_enabled(getattr(self, "analyzer_spectrum_enabled", True)) return self.output_analyzer def prepare_output_analyzer(self) -> bool: @@ -912,6 +919,9 @@ def set_filter_controls(self, controls: dict[str, float]) -> None: self.restart_engine() return + if not controls: + return + try: self.output_backend.set_node_params(self.filter_node_id, controls) except Exception as exc: diff --git a/src/mini_eq/window.py b/src/mini_eq/window.py index 9a23eb3..18a7353 100644 --- a/src/mini_eq/window.py +++ b/src/mini_eq/window.py @@ -1,5 +1,6 @@ from __future__ import annotations +import threading import time import gi @@ -180,9 +181,13 @@ def __init__( self.analyzer_frozen = False self.analyzer_smoothing = 0.40 self.analyzer_preview_source_id = 0 + self.analyzer_background_decay_source_id = 0 self.analyzer_preview_uses_tick_callback = False self.analyzer_preview_last_tick_time = 0.0 self.analyzer_levels = [0.0] * ANALYZER_BIN_COUNT + self.analyzer_levels_lock = threading.Lock() + self.analyzer_pending_levels: tuple[float, ...] | None = None + self.analyzer_levels_idle_pending = False self.analyzer_db_floor = ANALYZER_DB_FLOOR self.analyzer_display_gain_db = ANALYZER_DISPLAY_GAIN_DEFAULT self.analyzer_loudness_snapshot = None @@ -264,6 +269,8 @@ def __init__( self.install_css() self.sync_appearance_css_class() self.build_window_content(auto_route) + self.connect("map", self.on_analyzer_visibility_changed) + self.connect("unmap", self.on_analyzer_visibility_changed) self.style_dark_notify_handler_id = self.style_manager.connect( "notify::dark", self.on_style_manager_dark_changed ) diff --git a/src/mini_eq/window_analyzer.py b/src/mini_eq/window_analyzer.py index 4813b3a..bf432d1 100644 --- a/src/mini_eq/window_analyzer.py +++ b/src/mini_eq/window_analyzer.py @@ -26,6 +26,7 @@ ANALYZER_ATTACK_SMOOTHING_MAX = 0.25 ANALYZER_PIXEL_REDRAW_THRESHOLD = 1.0 CONTROL_ANALYZER_EMIT_INTERVAL_SECONDS = 0.10 +ANALYZER_BACKGROUND_DECAY_INTERVAL_MS = 100 LOUDNESS_METER_MIN_LUFS = -60.0 LOUDNESS_METER_MAX_LUFS = 0.0 @@ -188,7 +189,8 @@ def start_analyzer_preview(self) -> None: self.refresh_after_monitor_state_changed(monitor_visibility_changed=True) return - self.start_analyzer_preview_clock() + self.sync_analyzer_spectrum_enabled() + self.sync_analyzer_preview_clock() def stop_analyzer_preview(self, *, stop_backend: bool = True) -> None: if stop_backend: @@ -203,8 +205,14 @@ def stop_analyzer_preview(self, *, stop_backend: bool = True) -> None: destroy_glib_source(self.analyzer_preview_source_id) self.analyzer_preview_source_id = 0 self.analyzer_preview_uses_tick_callback = False + decay_source_id = getattr(self, "analyzer_background_decay_source_id", 0) + if decay_source_id > 0: + destroy_glib_source(decay_source_id) + self.analyzer_background_decay_source_id = 0 def start_analyzer_preview_clock(self) -> None: + if self.analyzer_preview_source_id > 0: + return add_tick_callback = getattr(self.analyzer_area, "add_tick_callback", None) if callable(add_tick_callback): self.analyzer_preview_source_id = add_tick_callback(self.on_analyzer_preview_frame) @@ -214,6 +222,67 @@ def start_analyzer_preview_clock(self) -> None: self.analyzer_preview_source_id = GLib.timeout_add(ANALYZER_PREVIEW_INTERVAL_MS, self.on_analyzer_preview_tick) self.analyzer_preview_uses_tick_callback = False + def sync_analyzer_spectrum_enabled(self) -> None: + callback = getattr(self.controller, "set_analyzer_spectrum_enabled", None) + if callable(callback): + callback(not self.analyzer_frozen) + + def sync_analyzer_preview_clock(self) -> None: + if ( + not getattr(self, "ui_shutting_down", False) + and self.analyzer_enabled + and not self.analyzer_frozen + and self.analyzer_area_is_drawable() + ): + decay_source_id = getattr(self, "analyzer_background_decay_source_id", 0) + if decay_source_id > 0: + destroy_glib_source(decay_source_id) + self.analyzer_background_decay_source_id = 0 + self.start_analyzer_preview_clock() + else: + self.stop_analyzer_preview(stop_backend=False) + + def on_analyzer_visibility_changed(self, *_args: object) -> None: + self.sync_analyzer_preview_clock() + self.schedule_background_analyzer_decay() + self.queue_analyzer_draw(force=True) + + def schedule_background_analyzer_decay(self) -> None: + if ( + getattr(self, "ui_shutting_down", False) + or not self.analyzer_enabled + or self.analyzer_frozen + or not hasattr(self, "analyzer_area") + or self.analyzer_area_is_drawable() + or not any(level > 0.01 for level in self.analyzer_levels) + or getattr(self, "analyzer_background_decay_source_id", 0) > 0 + ): + return + # A hidden window has no animation clock. Keep the Shell meter's + # stale-data decay at its 10 Hz publication cadence instead. + self.analyzer_background_decay_source_id = GLib.timeout_add( + ANALYZER_BACKGROUND_DECAY_INTERVAL_MS, self.on_background_analyzer_decay + ) + + def on_background_analyzer_decay(self) -> bool: + if ( + self.ui_shutting_down + or not self.analyzer_enabled + or self.analyzer_frozen + or self.analyzer_area_is_drawable() + ): + self.analyzer_background_decay_source_id = 0 + return False + now = GLib.get_monotonic_time() / 1_000_000.0 + if now - self.analyzer_last_frame_time < 0.18: + return True + keep_running = self.decay_analyzer_levels( + now, interval_scale=ANALYZER_BACKGROUND_DECAY_INTERVAL_MS / 1000.0 / ANALYZER_PREVIEW_INTERVAL_S + ) + if not keep_running: + self.analyzer_background_decay_source_id = 0 + return keep_running + def on_analyzer_preview_frame(self, _widget: Gtk.Widget, frame_clock: object) -> bool: get_frame_time = getattr(frame_clock, "get_frame_time", None) now = ( @@ -312,10 +381,24 @@ def analyzer_pixels_changed(self, current: tuple[float, ...]) -> bool: return False def on_analyzer_levels(self, levels: list[float]) -> None: - if self.ui_shutting_down: + if self.ui_shutting_down or self.analyzer_frozen: return - GLib.idle_add(self.on_analyzer_levels_idle, tuple(levels)) + with self.analyzer_levels_lock: + self.analyzer_pending_levels = tuple(levels) + if self.analyzer_levels_idle_pending: + return + self.analyzer_levels_idle_pending = True + GLib.idle_add(self.dispatch_pending_analyzer_levels) + + def dispatch_pending_analyzer_levels(self) -> bool: + with self.analyzer_levels_lock: + levels = self.analyzer_pending_levels + self.analyzer_pending_levels = None + self.analyzer_levels_idle_pending = False + if levels is not None: + self.on_analyzer_levels_idle(levels) + return False def on_analyzer_levels_idle(self, levels: tuple[float, ...]) -> bool: if self.ui_shutting_down: @@ -340,6 +423,7 @@ def on_analyzer_levels_idle(self, levels: tuple[float, ...]) -> bool: now = GLib.get_monotonic_time() / 1_000_000.0 self.analyzer_last_frame_time = now + self.schedule_background_analyzer_decay() self.queue_analyzer_draw() self.maybe_emit_control_analyzer_levels_changed(now) return False @@ -377,6 +461,11 @@ def on_analyzer_preview_tick(self, now: float | None = None) -> bool: self.analyzer_preview_uses_tick_callback = False return False + if self.analyzer_frozen or not self.analyzer_area_is_drawable(): + self.analyzer_preview_source_id = 0 + self.analyzer_preview_uses_tick_callback = False + return False + if not self.analyzer_enabled and not any(level > 0.01 for level in self.analyzer_levels): self.analyzer_preview_source_id = 0 self.analyzer_preview_uses_tick_callback = False @@ -389,18 +478,21 @@ def on_analyzer_preview_tick(self, now: float | None = None) -> bool: if self.analyzer_enabled and not self.analyzer_frozen and age < 0.18: return True + self.decay_analyzer_levels(now) + return True + + def decay_analyzer_levels(self, now: float, *, interval_scale: float = 1.0) -> bool: still_visible = False for index, current in enumerate(self.analyzer_levels): - decayed = current * (0.78 if self.analyzer_enabled else 0.68) + decayed = current * (0.78 if self.analyzer_enabled else 0.68) ** interval_scale if decayed > 0.01: still_visible = True - self.analyzer_levels[index] = decayed + self.analyzer_levels[index] = decayed if decayed > 0.01 else 0.0 - if still_visible: - self.queue_analyzer_draw() - self.maybe_emit_control_analyzer_levels_changed(now) + self.queue_analyzer_draw() + self.maybe_emit_control_analyzer_levels_changed(now) - return True + return still_visible def _sync_monitor_controls_unlocked(self) -> None: set_switch_confirmed_state(self.analyzer_switch, self.analyzer_enabled) @@ -496,6 +588,8 @@ def on_analyzer_freeze_changed(self, switch: Gtk.Switch, state: object | None) - return False self.analyzer_frozen = frozen + self.sync_analyzer_spectrum_enabled() + self.sync_analyzer_preview_clock() self.refresh_after_monitor_state_changed() return True diff --git a/tests/test_mini_eq_analyzer.py b/tests/test_mini_eq_analyzer.py index b0c28ba..63ee1a0 100644 --- a/tests/test_mini_eq_analyzer.py +++ b/tests/test_mini_eq_analyzer.py @@ -1,6 +1,7 @@ from __future__ import annotations import math +import threading from array import array import pytest @@ -583,3 +584,74 @@ def levels_callback(_levels: list[float]) -> None: assert snapshots == [analyzer.AnalyzerLoudnessSnapshot(-20.0, -18.0, -17.0)] assert len(created_meters) == 1 assert created_meters[0].closed is True + + +def test_analyzer_empty_reader_waits_for_data_and_stop_wakes_it(monkeypatch) -> None: + spectrum = analyzer.OutputSpectrumAnalyzer("test_sink", None, lambda _message: None) + entered = threading.Event() + original_wait = spectrum.audio_blocks_ready.wait + + def wait_for_audio(): + entered.set() + return original_wait() + + monkeypatch.setattr(spectrum.audio_blocks_ready, "wait", wait_for_audio) + spectrum.stop_event.clear() + reader = threading.Thread(target=spectrum.read_audio_levels, daemon=True) + spectrum.reader_thread = reader + reader.start() + try: + assert entered.wait(1) + finally: + spectrum.stop() + assert not reader.is_alive() + + +def test_frozen_analyzer_preserves_loudness_without_fft(monkeypatch) -> None: + snapshots = [] + fft_calls = [] + spectrum = analyzer.OutputSpectrumAnalyzer("test_sink", None, lambda _message: None) + spectrum.stop_event.clear() + spectrum.set_spectrum_enabled(False) + spectrum.audio_blocks.append((array("f", [0.2] * 2048).tobytes(), array("f", [0.1] * 2048).tobytes())) + + class Meter: + def close(self): + pass + + snapshot = analyzer.AnalyzerLoudnessSnapshot(-20, -18, -17) + + def on_loudness(value): + snapshots.append(value) + spectrum.stop_event.set() + + spectrum.set_loudness_callback(on_loudness) + monkeypatch.setattr(spectrum, "create_loudness_meter", Meter) + monkeypatch.setattr(spectrum, "feed_loudness_meter", lambda *_args: True) + monkeypatch.setattr(spectrum, "read_loudness_snapshot", lambda _meter: snapshot) + monkeypatch.setattr(analyzer, "samples_to_log_band_powers", lambda *_args, **_kwargs: fft_calls.append(True)) + spectrum.read_audio_levels() + assert snapshots == [snapshot] + assert fft_calls == [] + + +def test_analyzer_rechecks_queue_when_data_arrives_during_empty_clear(monkeypatch) -> None: + spectrum = analyzer.OutputSpectrumAnalyzer("test_sink", None, lambda _message: None) + spectrum.stop_event.clear() + original_clear = spectrum.audio_blocks_ready.clear + frames = [] + + def enqueue_during_clear(): + original_clear() + payload = array("f", [0.1] * 2048).tobytes() + spectrum.audio_blocks.append((payload, payload)) + spectrum.audio_blocks_ready.set() + + def on_levels(levels): + frames.append(levels) + spectrum.stop_event.set() + + monkeypatch.setattr(spectrum.audio_blocks_ready, "clear", enqueue_during_clear) + spectrum.set_levels_callback(on_levels) + spectrum.read_audio_levels() + assert frames diff --git a/tests/test_mini_eq_filter_chain.py b/tests/test_mini_eq_filter_chain.py index de43e1c..a1e3b6e 100644 --- a/tests/test_mini_eq_filter_chain.py +++ b/tests/test_mini_eq_filter_chain.py @@ -45,6 +45,46 @@ def test_native_biquad_solo_and_mute_use_dry_path() -> None: assert filter_chain.native_biquad_control_values(bands, 0, True)["band_r_1:Gain 1"] == 0 +def test_native_graph_skips_off_slots_without_renumbering() -> None: + bands = core.inactive_eq_bands() + bands[2] = core.EqBand(core.FILTER_TYPES["Lo-pass"], 1000, 0, 1) + bands[7] = core.EqBand(core.FILTER_TYPES["Bell"], 2000, 3, 1, mute=True) + args = filter_chain.build_native_biquad_filter_chain_module_args( + bands=bands, + preamp_db=-3, + eq_enabled=True, + virtual_sink_name="mini_eq_sink", + filter_output_name="mini_eq_sink_output", + output_sink="test", + ) + for side in ("l", "r"): + assert f'output = "preamp_{side}:Out" input = "band_{side}_2_filter:In"' in args + assert f'output = "band_{side}_2:Out" input = "band_{side}_7_filter:In"' in args + assert f'"band_{side}_7:Out"' in args + assert f"name = band_{side}_7_filter" in args + assert "band_l_0" not in args + assert "band_l_31" not in args + controls = filter_chain.native_biquad_control_values(bands, -3, True) + assert controls["band_l_2:Gain 1"] == 1 + assert controls["band_l_7:Gain 1"] == 0 + assert not any("band_l_0" in key or "band_l_31" in key for key in controls) + assert filter_chain.native_biquad_band_control_values(31, bands[31], True) == {} + assert len(bands) == core.MAX_BANDS + + +def test_native_all_off_graph_connects_preamp_to_outputs() -> None: + args = filter_chain.build_native_biquad_filter_chain_module_args( + bands=core.inactive_eq_bands(), + preamp_db=-6, + eq_enabled=True, + virtual_sink_name="mini_eq_sink", + filter_output_name="mini_eq_sink_output", + output_sink="test", + ) + assert "band_l_" not in args + assert 'outputs = [ "preamp_l:Out" "preamp_r:Out" ]' in args + + def test_builtin_biquad_filter_chain_uses_pipewire_raw_biquads() -> None: bands = [ core.EqBand(core.FILTER_TYPES["Bell"], 1000.0, 6.0, 1.4), diff --git a/tests/test_mini_eq_routing.py b/tests/test_mini_eq_routing.py index 691e46c..6d58eda 100644 --- a/tests/test_mini_eq_routing.py +++ b/tests/test_mini_eq_routing.py @@ -1055,8 +1055,12 @@ def set_loudness_callback(self, callback) -> None: def set_response_speed(self, speed: float) -> None: self.response_speed = speed + def set_spectrum_enabled(self, enabled: bool) -> None: + self.spectrum_enabled = enabled + monkeypatch.setattr(routing, "OutputSpectrumAnalyzer", FakeAnalyzer) controller.emit_status = lambda _message: None + controller.analyzer_spectrum_enabled = False analyzer_instance = routing.SystemWideEqController.ensure_output_analyzer(controller) @@ -1064,6 +1068,7 @@ def set_response_speed(self, speed: float) -> None: assert analyzer_instance.output_sink_name == "speakers" assert analyzer_instance.levels_callback is controller.analyzer_levels_callback assert analyzer_instance.loudness_callback is controller.analyzer_loudness_callback + assert analyzer_instance.spectrum_enabled is False def test_enabling_prepared_analyzer_does_not_restart_running_engine() -> None: @@ -1818,6 +1823,24 @@ def test_filter_control_param_change_ignores_recent_local_control_update( assert controller.applying_filter_control_verification is False +@pytest.mark.parametrize("type_changed", [False, True]) +def test_empty_off_band_controls_only_rebuild_when_topology_changed(type_changed: bool) -> None: + controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController) + controller.running = True + controller.filter_node_id = 42 + controller.bands = core.inactive_eq_bands() + built_types = [band.filter_type for band in controller.bands] + if type_changed: + built_types[31] = core.FILTER_TYPES["Bell"] + controller._engine_band_types = tuple(built_types) + calls = [] + controller.restart_engine = lambda: calls.append("restart") + # An empty control set must not reach the backend, including on a rebuild. + controller.output_backend = object() + controller.set_filter_controls({}) + assert calls == (["restart"] if type_changed else []) + + def test_filter_control_verification_echo_does_not_schedule_another_verification( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/test_mini_eq_window_analyzer.py b/tests/test_mini_eq_window_analyzer.py index b7f7146..52f5803 100644 --- a/tests/test_mini_eq_window_analyzer.py +++ b/tests/test_mini_eq_window_analyzer.py @@ -1,5 +1,7 @@ from __future__ import annotations +import threading + import pytest from tests._mini_eq_imports import import_mini_eq_module @@ -29,6 +31,9 @@ def __init__(self) -> None: self.analyzer_enabled = True self.analyzer_frozen = False self.analyzer_levels = [0.0, 0.0] + self.analyzer_levels_lock = threading.Lock() + self.analyzer_pending_levels = None + self.analyzer_levels_idle_pending = False self.analyzer_smoothing = 0.5 self.analyzer_last_frame_time = 0.0 self.control_analyzer_last_emit_time = 0.0 @@ -76,16 +81,24 @@ class FakeAnalyzerController: def __init__(self, *, start_result: bool = True) -> None: self.enabled_values: list[bool] = [] self.start_result = start_result + self.spectrum_enabled = True def set_analyzer_enabled(self, enabled: bool) -> bool: self.enabled_values.append(enabled) return self.start_result + def set_analyzer_spectrum_enabled(self, enabled: bool) -> None: + self.spectrum_enabled = enabled + class TickAnalyzerArea: def __init__(self) -> None: self.callback = None self.removed_ids: list[int] = [] + self.mapped = True + + def get_mapped(self) -> bool: + return self.mapped def add_tick_callback(self, callback): self.callback = callback @@ -381,6 +394,20 @@ def test_analyzer_level_frames_emit_compact_control_signal() -> None: assert window.application.state_count == 0 +def test_analyzer_dispatch_coalesces_frames_to_latest_levels(monkeypatch) -> None: + queued = [] + monkeypatch.setattr(window_analyzer.GLib, "idle_add", lambda callback: queued.append(callback)) + window = AnalyzerSignalWindow() + window.on_analyzer_levels([1.0, 0.2]) + window.on_analyzer_levels([0.6, 0.8]) + assert len(queued) == 1 + assert queued.pop()() is False + assert window.analyzer_levels == pytest.approx([0.45, 0.6]) + assert window.application.analyzer_count == 1 + window.on_analyzer_levels([0.3, 0.4]) + assert len(queued) == 1 + + def test_analyzer_redraw_skips_unmapped_area() -> None: window = AnalyzerDrawWindow() @@ -406,6 +433,58 @@ def test_analyzer_preview_uses_frame_clock_tick_callback() -> None: assert window.analyzer_preview_uses_tick_callback is False +def test_analyzer_hidden_clock_stops_but_capture_continues() -> None: + window = AnalyzerPreviewWindow() + window.start_analyzer_preview() + window.analyzer_area.mapped = False + window.on_analyzer_visibility_changed() + assert window.analyzer_area.removed_ids == [42] + assert window.analyzer_preview_source_id == 0 + assert window.controller.enabled_values == [True] + window.analyzer_area.mapped = True + window.on_analyzer_visibility_changed() + assert window.analyzer_preview_source_id == 42 + assert window.controller.enabled_values == [True] + + +def test_analyzer_freeze_stops_clock_and_fft_then_resumes() -> None: + window = AnalyzerPreviewWindow() + window.start_analyzer_preview() + window.on_analyzer_freeze_changed(FakeSwitch(True), None) + assert window.analyzer_preview_source_id == 0 + assert window.controller.spectrum_enabled is False + assert window.controller.enabled_values == [True] + window.on_analyzer_freeze_changed(FakeSwitch(False), None) + assert window.analyzer_preview_source_id == 42 + assert window.controller.spectrum_enabled is True + assert window.controller.enabled_values == [True] + + +def test_hidden_analyzer_stale_levels_decay_to_zero_for_shell(monkeypatch) -> None: + window = AnalyzerPreviewWindow() + window.ui_shutting_down = False + window.analyzer_last_frame_time = 0.0 + window.analyzer_area.mapped = False + window.analyzer_levels = [0.8, 0.4] + queued = [] + monkeypatch.setattr( + window_analyzer.GLib, "timeout_add", lambda interval, callback: queued.append((interval, callback)) or 9 + ) + window.schedule_background_analyzer_decay() + assert queued == [(100, window.on_background_analyzer_decay)] + window.schedule_background_analyzer_decay() + assert len(queued) == 1 + now = iter(range(1_000_000, 5_000_000, 100_000)) + monkeypatch.setattr(window_analyzer.GLib, "get_monotonic_time", lambda: next(now)) + for _ in range(40): + if not window.on_background_analyzer_decay(): + break + assert window.analyzer_levels == [0.0, 0.0] + assert window.analyzer_background_decay_source_id == 0 + assert window.application.analyzer_count > 0 + assert window.analyzer_preview_source_id == 0 + + def test_analyzer_preview_failure_refreshes_monitor_controls() -> None: window = AnalyzerPreviewWindow(start_result=False) diff --git a/tools/check_headless_pipewire_runtime.py b/tools/check_headless_pipewire_runtime.py index 4dd409b..c18aecc 100755 --- a/tools/check_headless_pipewire_runtime.py +++ b/tools/check_headless_pipewire_runtime.py @@ -875,6 +875,48 @@ def run_controller_flow( tmp_dir / "mini-eq-headless-baseline.raw", timeout_seconds, ) + from mini_eq.core import FILTER_TYPES + + print("## sparse native graph: enable and remove slot 31", flush=True) + controller.set_band_type(31, FILTER_TYPES["Bell"]) + controller.set_band_frequency(31, 440.0) + controller.set_band_q(31, 4.0) + controller.set_band_gain(31, -6.0) + dispatch_until("sparse high-index topology ready", lambda: controller.running, timeout_seconds) + wait_for_stream_routed_and_processing( + smoke_id, virtual_sink_name, filter_output_name, timeout_seconds, "after adding slot 31" + ) + sparse_rms = capture_sink_monitor_rms(controller.output_sink, tmp_dir / "sparse-slot.raw", timeout_seconds) + sparse_db = 20.0 * math.log10(max(sparse_rms, 1e-12) / max(baseline_rms, 1e-12)) + if not -7.0 <= sparse_db <= -5.0: + raise RuntimeError(f"Sparse slot 31 response was {sparse_db:.2f} dB; expected -6 dB") + controller.set_band_type(31, FILTER_TYPES["Off"]) + dispatch_until("sparse high-index topology removed", lambda: controller.running, timeout_seconds) + wait_for_stream_routed_and_processing( + smoke_id, virtual_sink_name, filter_output_name, timeout_seconds, "after removing slot 31" + ) + + print("## all-Off native graph retains preamp", flush=True) + payload = controller.build_preset_payload() + payload["preamp_db"] = -6.0 + for band in payload["bands"]: + band["filter_type"] = FILTER_TYPES["Off"] + controller.apply_preset_payload(payload) + dispatch_until("all-Off topology ready", lambda: controller.running, timeout_seconds) + wait_for_stream_routed_and_processing( + smoke_id, virtual_sink_name, filter_output_name, timeout_seconds, "with all bands Off" + ) + all_off_rms = capture_sink_monitor_rms(controller.output_sink, tmp_dir / "all-off.raw", timeout_seconds) + all_off_db = 20.0 * math.log10(max(all_off_rms, 1e-12) / max(baseline_rms, 1e-12)) + if not -7.0 <= all_off_db <= -5.0: + raise RuntimeError(f"All-Off preamp response was {all_off_db:.2f} dB; expected -6 dB") + controller.reset_state() + dispatch_until("default topology restored", lambda: controller.running, timeout_seconds) + virtual_serial = wait_for_stream_routed_and_processing( + smoke_id, virtual_sink_name, filter_output_name, timeout_seconds, "after restoring default bands" + ) + print(f"Sparse slot 31: {sparse_db:.2f} dB; all-Off preamp: {all_off_db:.2f} dB", flush=True) + # Exercise a frequency-selective live update, not only preamp gain: # coefficients computed at the wrong rate still pass a preamp test. controller.set_band_frequency(0, 440.0) @@ -889,8 +931,6 @@ def run_controller_flow( if not -13.0 <= bell_db <= -11.0: raise RuntimeError(f"440 Hz bell response was {bell_db:.2f} dB; expected -12 dB") print(f"Live 440 Hz bell response: {bell_db:.2f} dB", flush=True) - from mini_eq.core import FILTER_TYPES - controller.set_band_type(0, FILTER_TYPES["Lo-pass"]) controller.set_band_frequency(0, 100.0) controller.set_band_q(0, 0.707) From ac050018c08d4684b9efa416b63cf3f214db863e Mon Sep 17 00:00:00 2001 From: bhack Date: Tue, 13 Oct 2026 01:55:49 +0200 Subject: [PATCH 2/2] Keep the EQ input device stable across DSP rebuilds --- CHANGELOG.md | 2 + data/io.github.bhack.mini-eq.metainfo.xml | 1 + docs/development.md | 2 +- docs/release.md | 13 + io.github.bhack.mini-eq.yaml | 2 + src/mini_eq/deps.py | 8 +- src/mini_eq/filter_chain.py | 28 +- src/mini_eq/pipewire_backend.py | 76 +++++- src/mini_eq/pipewire_stream_router.py | 11 +- src/mini_eq/routing.py | 103 ++++++-- tests/test_mini_eq_filter_chain.py | 28 ++ tests/test_mini_eq_pipewire_backend.py | 54 ++++ tests/test_mini_eq_pipewire_stream_router.py | 5 +- tests/test_mini_eq_routing.py | 104 ++++++-- tools/check_mpris_preset_runtime.py | 257 +++++++++++++++++++ 15 files changed, 636 insertions(+), 58 deletions(-) create mode 100644 tools/check_mpris_preset_runtime.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c2c3bf..19b76bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## 0.8.9 - 2026-10-13 +- Keep the audio device stable when changing presets or filter types, avoiding + media-player pauses caused by device removal. - Avoid processing unused Off bands while keeping all 32 band slots available. - Stop analyzer animation and spectrum calculations while frozen, and stop animation while the window is hidden. diff --git a/data/io.github.bhack.mini-eq.metainfo.xml b/data/io.github.bhack.mini-eq.metainfo.xml index af3093d..fe40e61 100644 --- a/data/io.github.bhack.mini-eq.metainfo.xml +++ b/data/io.github.bhack.mini-eq.metainfo.xml @@ -48,6 +48,7 @@
      +
    • Keep the audio device stable when changing presets or filter types, avoiding media-player pauses caused by device removal.
    • Avoid processing unused Off bands while keeping all 32 band slots available.
    • Stop analyzer animation and spectrum calculations while frozen, and stop animation while the window is hidden.
    • Use audio-driven analyzer wakeups and coalesced display updates while preserving loudness measurement and background Shell meters.
      • diff --git a/docs/development.md b/docs/development.md index f1923ea..a85dd5d 100644 --- a/docs/development.md +++ b/docs/development.md @@ -172,7 +172,7 @@ runtime. ## Local Flatpak Build The Flatpak manifest uses the GNOME runtime. It does not ship a full PipeWire -daemon or session manager; it builds only the local PipeWire filter-chain module +daemon or session manager; it builds the local PipeWire filter-chain and loopback modules and SPA builtin filter-graph support that Mini EQ loads inside the app process. It also builds pipewire-gobject for Mini EQ's app-facing PipeWire access and bundles NumPy and libebur128 for analyzer and live LUFS support. diff --git a/docs/release.md b/docs/release.md index 206e83c..bc5290a 100644 --- a/docs/release.md +++ b/docs/release.md @@ -222,6 +222,19 @@ a focused local loop: python3 tools/check_headless_pipewire_runtime.py --timeout 35 --cycles 2 ``` +For preset topology or virtual-device lifecycle changes, also run the isolated +PulseAudio/MPRIS smoke on a recent WirePlumber with `linking.pause-playback` +support. It checks stable device identity, no player pause on preset/type edits +or normal shutdown, and a positive device-removal control: + +```bash +python3 tools/check_mpris_preset_runtime.py +``` + +This requires `pipewire-pulse` and `paplay` in addition to the native headless +smoke dependencies. It uses a private session bus and audio graph and does not +change the host's playback or WirePlumber settings. + For background portal changes, run one clean-permission Flatpak smoke in a real GNOME session. Enable **Keep Running in Background**, approve the portal prompt, enable **Start at Login**, close the window, and confirm the app stays diff --git a/io.github.bhack.mini-eq.yaml b/io.github.bhack.mini-eq.yaml index 156e33b..dce95e5 100644 --- a/io.github.bhack.mini-eq.yaml +++ b/io.github.bhack.mini-eq.yaml @@ -105,11 +105,13 @@ modules: - -Dflatpak=enabled make-args: - src/modules/libpipewire-module-filter-chain.so + - src/modules/libpipewire-module-loopback.so - spa/plugins/filter-graph/libspa-filter-graph.so - spa/plugins/filter-graph/libspa-filter-graph-plugin-builtin.so no-make-install: true build-commands: - install -Dm755 src/modules/libpipewire-module-filter-chain.so ${FLATPAK_DEST}/lib/pipewire-0.3/libpipewire-module-filter-chain.so + - install -Dm755 src/modules/libpipewire-module-loopback.so ${FLATPAK_DEST}/lib/pipewire-0.3/libpipewire-module-loopback.so - install -Dm755 spa/plugins/filter-graph/libspa-filter-graph.so ${FLATPAK_DEST}/lib/spa-0.2/filter-graph/libspa-filter-graph.so - install -Dm755 spa/plugins/filter-graph/libspa-filter-graph-plugin-builtin.so ${FLATPAK_DEST}/lib/spa-0.2/filter-graph/libspa-filter-graph-plugin-builtin.so sources: diff --git a/src/mini_eq/deps.py b/src/mini_eq/deps.py index 767097d..4864e37 100644 --- a/src/mini_eq/deps.py +++ b/src/mini_eq/deps.py @@ -54,7 +54,7 @@ ) PIPEWIRE_FILTER_CHAIN_HINT = ( "Ubuntu/Debian: pipewire; Fedora: pipewire; Arch: pipewire. " - "Flatpak builds bundle only the filter-chain module and SPA builtin filter support." + "Flatpak builds bundle the filter-chain and loopback modules and SPA builtin filter support." ) NUMPY_HINT = ( "Install the package with Python dependencies: python -m pip install mini-eq, or python -m pip install numpy." @@ -399,6 +399,12 @@ def collect_dependency_checks() -> list[DependencyCheck]: True, PIPEWIRE_FILTER_CHAIN_HINT, ), + check_pipewire_module( + "libpipewire-module-loopback.so", + "PipeWire loopback module", + True, + PIPEWIRE_FILTER_CHAIN_HINT, + ), check_spa_plugin( "filter-graph/libspa-filter-graph-plugin-builtin.so", "PipeWire SPA builtin filter graph plugin", diff --git a/src/mini_eq/filter_chain.py b/src/mini_eq/filter_chain.py index 87540b4..9e9dae6 100644 --- a/src/mini_eq/filter_chain.py +++ b/src/mini_eq/filter_chain.py @@ -120,6 +120,30 @@ def pipewire_quote(value: str) -> str: return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' +def build_eq_input_module_args(virtual_sink_name: str, dsp_input_name: str) -> str: + """Keep the player-facing device alive independently of DSP topology.""" + return f"""{{ + audio.channels = 2 + audio.position = [ FL FR ] + capture.props = {{ + node.name = {pipewire_quote(virtual_sink_name)} + node.description = {pipewire_quote(VIRTUAL_SINK_DESCRIPTION)} + media.class = Audio/Sink + state.restore-props = false + state.restore-target = false + }} + playback.props = {{ + node.name = {pipewire_quote(virtual_sink_name + ".input")} + target.object = {pipewire_quote(dsp_input_name)} + node.passive = true + node.dont-fallback = true + node.linger = true + state.restore-props = false + state.restore-target = false + }} +}}""" + + def spa_float(value: float) -> str: return f"{float(value):.8g}" @@ -294,6 +318,7 @@ def build_builtin_biquad_filter_chain_module_args( filter_output_name: str, output_sink: str, native_biquads: bool = False, + capture_as_stream: bool = False, ) -> str: graph_bands = bands[:MAX_BANDS] band_count = len(graph_bands) @@ -334,7 +359,8 @@ def build_builtin_biquad_filter_chain_module_args( capture.props = {{ node.name = {pipewire_quote(virtual_sink_name)} node.description = {pipewire_quote(VIRTUAL_SINK_DESCRIPTION)} - media.class = Audio/Sink + media.class = {"Stream/Input/Audio" if capture_as_stream else "Audio/Sink"} + {"node.autoconnect = false" if capture_as_stream else ""} state.restore-props = false state.restore-target = false audio.channels = 2 diff --git a/src/mini_eq/pipewire_backend.py b/src/mini_eq/pipewire_backend.py index 5397e42..8f93e93 100644 --- a/src/mini_eq/pipewire_backend.py +++ b/src/mini_eq/pipewire_backend.py @@ -485,6 +485,75 @@ def watch_for_audio_sink( timeout_ms=timeout_ms, ) + def wait_for_input_unlinked(self, node_id: int, stream_ids: set[int], timeout_ms: int = 1500) -> bool: + """Drain device links on shutdown, not just the metadata write. + + The session manager moves streams asynchronously. Dispatch its graph + events before destroying an app-owned device that can trigger MPRIS + pause policy. This bounded nested loop is for shutdown only. + """ + self._ensure_connected() + loop = self._GLib.MainLoop() + pending_id = 0 + timeout_id = 0 + done = False + detached = False + link_handlers: dict[int, int] = {} + + def check() -> bool: + nonlocal pending_id, done, detached + pending_id = 0 + if done: + return False + links = self.list_links() + detached = not any(link.input_node_id == node_id for link in links) + existing = {node.bound_id for node in self.list_output_streams()} + for stream_id in stream_ids & existing: + restored = [ + link for link in links if link.output_node_id == stream_id and link.input_node_id != node_id + ] + if not restored: + detached = False + for link in restored: + if link.bound_id not in link_handlers: + link_handlers[link.bound_id] = self.connect_link_state_changed(link.bound_id, changed) + proxy = self._link_proxy_by_bound_id(link.bound_id) + if proxy is None or proxy.get_state() not in ("active", "paused"): + detached = False + if detached: + done = True + loop.quit() + return False + + def changed(*_args) -> None: + nonlocal pending_id + if not done and pending_id == 0: + pending_id = self._GLib.idle_add(check) + + def timeout() -> bool: + nonlocal timeout_id, done + timeout_id = 0 + done = True + loop.quit() + return False + + added = self.connect_object_added(changed) + removed = self.connect_object_removed(changed) + try: + timeout_id = self._GLib.timeout_add(max(1, timeout_ms), timeout) + changed() + loop.run() + finally: + done = True + self.disconnect_node_manager_handler(added) + self.disconnect_node_manager_handler(removed) + for handler_id in link_handlers.values(): + self.disconnect_link_handler(handler_id) + for source_id in (pending_id, timeout_id): + if source_id: + self._GLib.source_remove(source_id) + return detached + def connect_object_added(self, callback) -> int: self._ensure_connected() handler_id = self._GObject.Object.connect(self._registry, "global-added", callback) @@ -986,11 +1055,14 @@ def _node_param_id_by_name(self, node, name: str) -> int | None: return None def load_filter_chain_module(self, arguments: str): + return self.load_module(FILTER_CHAIN_MODULE_NAME, arguments) + + def load_module(self, name: str, arguments: str): self._ensure_connected() - module = self._core.load_module(FILTER_CHAIN_MODULE_NAME, arguments) + module = self._core.load_module(name, arguments) if module is None: - raise PipeWireBackendError(f"failed to load PipeWire module: {FILTER_CHAIN_MODULE_NAME}") + raise PipeWireBackendError(f"failed to load PipeWire module: {name}") self._loaded_modules.append(module) return module diff --git a/src/mini_eq/pipewire_stream_router.py b/src/mini_eq/pipewire_stream_router.py index cb1fd3b..0fcc7f1 100644 --- a/src/mini_eq/pipewire_stream_router.py +++ b/src/mini_eq/pipewire_stream_router.py @@ -47,6 +47,7 @@ def __init__( self.accept_stream_events = False self.event_source_id = 0 self.pending_route_applied_callback = False + self.pending_processing_links: dict[int, PipeWireLink] = {} self.object_added_handler_id = 0 self.object_removed_handler_id = 0 self.link_state_handler_ids: dict[int, int] = {} @@ -294,6 +295,12 @@ def on_stream_event_idle(self) -> bool: route_applied = self.pending_route_applied_callback self.pending_route_applied_callback = False + # Registry signals can arrive on the PipeWire loop thread. Binding and + # synchronizing link proxies must happen on the GLib thread. + pending_links = list(self.pending_processing_links.values()) + self.pending_processing_links.clear() + for link in pending_links: + self.track_processing_link_state(link) self.refresh() if route_applied and self.route_applied_callback is not None: self.route_applied_callback() @@ -319,7 +326,7 @@ def handle_object_added(self, _manager, node) -> None: return if self._link_touches_processing_path(link): - self.track_processing_link_state(link) + self.pending_processing_links[link.bound_id] = link self.schedule_refresh(route_applied=True) elif self._link_touches_routed_stream(link): self.schedule_refresh() @@ -331,6 +338,7 @@ def handle_object_removed(self, _manager, node) -> None: return should_refresh = self._link_touches_processing_path(link) or self._link_touches_routed_stream(link) + self.pending_processing_links.pop(link.bound_id, None) self.untrack_processing_link_state(link) if should_refresh: self.schedule_refresh() @@ -355,6 +363,7 @@ def start_monitoring(self, *, require_initial_route: bool = False) -> None: def stop_monitoring(self) -> None: self.accept_stream_events = False + self.pending_processing_links.clear() if self.event_source_id > 0: destroy_glib_source(self.event_source_id) diff --git a/src/mini_eq/routing.py b/src/mini_eq/routing.py index 365b909..09c649f 100644 --- a/src/mini_eq/routing.py +++ b/src/mini_eq/routing.py @@ -35,10 +35,11 @@ sanitize_preset_name, ) from .filter_chain import ( - build_native_biquad_filter_chain_module_args as build_builtin_biquad_filter_chain_module_args, + build_eq_input_module_args, + builtin_biquad_preamp_control_values, ) from .filter_chain import ( - builtin_biquad_preamp_control_values, + build_native_biquad_filter_chain_module_args as build_builtin_biquad_filter_chain_module_args, ) from .filter_chain import native_biquad_band_control_values as builtin_biquad_band_control_values from .filter_chain import native_biquad_control_values as builtin_biquad_control_values @@ -112,6 +113,7 @@ def __init__(self, output_sink: str | None) -> None: self._observed_output_preset_target_snapshot: OutputPresetTargetSnapshot | None = None self.filter_output_name = f"{self.virtual_sink_name}{FILTER_OUTPUT_SUFFIX}" self.engine_module = None + self.input_module = None self.engine_start_watch = None self.engine_start_pending = False self.filter_node_id: int | None = None @@ -727,11 +729,23 @@ def build_filter_chain_module_args(self) -> str: bands=self.bands, preamp_db=self.preamp_db, eq_enabled=self.eq_enabled, - virtual_sink_name=self.virtual_sink_name, + virtual_sink_name=self.dsp_input_name, filter_output_name=self.filter_output_name, output_sink=self.output_sink, + capture_as_stream=True, ) + @property + def dsp_input_name(self) -> str: + return f"{self.virtual_sink_name}.dsp" + + def ensure_input_module(self) -> None: + if getattr(self, "input_module", None) is None: + self.input_module = self.output_backend.load_module( + "libpipewire-module-loopback", + build_eq_input_module_args(self.virtual_sink_name, self.dsp_input_name), + ) + def cancel_pending_engine_start(self) -> None: self._engine_start_token = None watch = getattr(self, "engine_start_watch", None) @@ -754,6 +768,7 @@ def start_engine( if getattr(self, "engine_start_pending", False): return + self.ensure_input_module() self.engine_module = self.output_backend.load_filter_chain_module(self.build_filter_chain_module_args()) self.engine_start_pending = True start_token = object() @@ -765,8 +780,14 @@ def fail(exc: Exception) -> None: self._engine_start_token = None self.engine_start_watch = None self.engine_start_pending = False + module = self.engine_module self.engine_module = None self.filter_node_id = None + if module is not None: + try: + self.output_backend.unload_filter_chain_module(module) + except Exception: + pass try: self.output_backend.sync() except Exception: @@ -786,7 +807,7 @@ def on_sink_ready(sink: PipeWireNode | None) -> None: return if sink is None: - fail(RuntimeError(f"filter-chain did not create {self.virtual_sink_name}")) + fail(RuntimeError(f"filter-chain did not create {self.dsp_input_name}")) return # A preset/type edit during creation makes this topology obsolete. @@ -820,9 +841,32 @@ def on_sink_ready(sink: PipeWireNode | None) -> None: on_ready() try: - self.engine_start_watch = self.output_backend.watch_for_audio_sink( - self.virtual_sink_name, - on_sink_ready, + + def on_dsp_ready(node: PipeWireNode | None) -> None: + if self._engine_start_token is not start_token: + return + if node is None: + on_sink_ready(None) + return + + def on_input_ready(sink: PipeWireNode | None) -> None: + if self._engine_start_token is not start_token: + return + if sink is None: + fail(RuntimeError(f"loopback did not create {self.virtual_sink_name}")) + else: + on_sink_ready(node) + + try: + self.engine_start_watch = self.output_backend.watch_for_audio_sink( + self.virtual_sink_name, on_input_ready, timeout_ms=3000 + ) + except Exception as exc: + fail(exc) + + self.engine_start_watch = self.output_backend.watch_for_node( + lambda node: node.node_name == self.dsp_input_name, + on_dsp_ready, timeout_ms=3000, ) except Exception as exc: @@ -886,27 +930,23 @@ def restart_engine(self) -> None: if not was_running: return - stream_router = self.stream_router if self.routed else None - if stream_router is not None: - stream_router.stop_monitoring() - try: - stream_router.restore_output_streams() - except Exception as exc: - stream_router.emit_warning(exc) - + # Only stream nodes disappear here. The player-facing Audio/Sink stays + # alive, so WirePlumber's device-removal MPRIS policy cannot pause players. self.stop_engine(announce=False) - def on_ready() -> None: - if stream_router is not None: - stream_router.start_monitoring(require_initial_route=True) - def on_error(exc: Exception) -> None: - if stream_router is not None: - stream_router.emit_warning(exc) - else: - self.emit_status(str(exc)) + # Keep the stable sink alive, but don't strand music in a broken DSP. + if self.routed: + try: + self.route_system_audio(False, refresh_output=False) + except Exception as restore_exc: + self.emit_status(f"filter-chain routing restore failed: {restore_exc}") + self.emit_status(str(exc)) - self.start_engine(on_ready=on_ready, on_error=on_error) + try: + self.start_engine(on_error=on_error) + except Exception as exc: + on_error(exc) def set_filter_controls(self, controls: dict[str, float]) -> None: if self.filter_node_id is None or not self.running: @@ -1137,6 +1177,7 @@ def on_engine_error(exc: Exception) -> None: raise def shutdown(self) -> None: + routed_stream_ids = set(getattr(self.stream_router, "routed_stream_ids", ())) self.shutting_down = True self.status_callback = None self.outputs_changed_callback = None @@ -1156,6 +1197,13 @@ def shutdown(self) -> None: pass finally: try: + # Remove internal streams first. Keep the device until the + # session manager has linked restored playback to its output. + self.stop_engine(announce=False) + if getattr(self, "input_module", None) is not None: + sink = self.output_backend.audio_sink_by_name(self.virtual_sink_name) + if sink is not None: + self.output_backend.wait_for_input_unlinked(sink.bound_id, routed_stream_ids) if self.stream_router is not None: self.stream_router.close() if self.output_analyzer is not None: @@ -1165,6 +1213,13 @@ def shutdown(self) -> None: self.stop_engine(announce=False) except Exception: pass + input_module = getattr(self, "input_module", None) + self.input_module = None + if input_module is not None: + try: + self.output_backend.unload_filter_chain_module(input_module) + except Exception: + pass try: self.output_backend.close() except Exception: diff --git a/tests/test_mini_eq_filter_chain.py b/tests/test_mini_eq_filter_chain.py index a1e3b6e..254437e 100644 --- a/tests/test_mini_eq_filter_chain.py +++ b/tests/test_mini_eq_filter_chain.py @@ -9,6 +9,34 @@ def test_pipewire_quote_escapes_module_argument_strings() -> None: assert filter_chain.pipewire_quote('a"b\\c') == '"a\\"b\\\\c"' +def test_stable_input_has_no_hardware_fallback_or_fixed_sample_rate() -> None: + args = filter_chain.build_eq_input_module_args("mini_eq_sink", "mini_eq_sink.dsp") + assert 'node.name = "mini_eq_sink"' in args + assert 'target.object = "mini_eq_sink.dsp"' in args + assert "media.class = Audio/Sink" in args + assert "node.dont-fallback = true" in args + assert "node.linger = true" in args + assert "node.passive = true" in args + assert "audio.rate" not in args + + +def test_dsp_input_is_a_stream_not_a_removable_device() -> None: + args = filter_chain.build_native_biquad_filter_chain_module_args( + bands=core.default_eq_bands(), + preamp_db=0, + eq_enabled=True, + virtual_sink_name="mini_eq_sink.dsp", + filter_output_name="mini_eq_sink_output", + output_sink="speakers", + capture_as_stream=True, + ) + assert "media.class = Stream/Input/Audio" in args + assert "media.class = Audio/Sink" not in args + assert "node.autoconnect = false" in args + assert 'target.object = "speakers"' in args + assert "audio.rate" not in args + + def test_native_biquads_follow_graph_rate_and_have_true_bypass() -> None: bands = [core.EqBand(core.FILTER_TYPES["Lo-pass"], 1000.0, 0.0, 1.4)] args = filter_chain.build_native_biquad_filter_chain_module_args( diff --git a/tests/test_mini_eq_pipewire_backend.py b/tests/test_mini_eq_pipewire_backend.py index f367189..15237c8 100644 --- a/tests/test_mini_eq_pipewire_backend.py +++ b/tests/test_mini_eq_pipewire_backend.py @@ -8,6 +8,60 @@ from tests._mini_eq_imports import pipewire_routes as pw_routes +@pytest.mark.parametrize("initial", ["old-link", "gap"]) +@pytest.mark.parametrize("restored_state", ["active", "paused", "error"]) +def test_shutdown_wait_requires_a_restored_link_and_cleans_up(initial, restored_state) -> None: + backend = pw_backend.PipeWireBackend.__new__(pw_backend.PipeWireBackend) + links = [pw_backend.PipeWireLink(1, 7, 42, False, False)] if initial == "old-link" else [] + events = {} + disconnected = [] + + class Loop: + stopped = False + + def quit(self): + self.stopped = True + + def run(self): + glib.idle() + # An unlinked gap is not a completed restore. + assert not self.stopped + links[:] = [pw_backend.PipeWireLink(2, 7, 43, False, False)] + events[10]() + glib.idle() + if not self.stopped: + glib.timeout() + + class GLib: + MainLoop = Loop + + def idle_add(self, callback): + self.idle = callback + return 88 + + def timeout_add(self, milliseconds, callback): + assert milliseconds == 1500 + self.timeout = callback + return 77 + + def source_remove(self, source_id): + pass + + glib = GLib() + backend._GLib = glib + backend._ensure_connected = lambda: None + backend.list_links = lambda: list(links) + backend.list_output_streams = lambda: [SimpleNamespace(bound_id=7)] + backend.connect_object_added = lambda callback: events.setdefault(10, callback) and 10 + backend.connect_object_removed = lambda callback: events.setdefault(11, callback) and 11 + backend.disconnect_node_manager_handler = disconnected.append + backend.connect_link_state_changed = lambda _link_id, _callback: 12 + backend.disconnect_link_handler = disconnected.append + backend._link_proxy_by_bound_id = lambda _link_id: SimpleNamespace(get_state=lambda: restored_state) + assert backend.wait_for_input_unlinked(42, {7}) is (restored_state != "error") + assert set(disconnected) == {10, 11, 12} + + class FakeCore: calls: int = 0 diff --git a/tests/test_mini_eq_pipewire_stream_router.py b/tests/test_mini_eq_pipewire_stream_router.py index af6b375..5c9443e 100644 --- a/tests/test_mini_eq_pipewire_stream_router.py +++ b/tests/test_mini_eq_pipewire_stream_router.py @@ -488,11 +488,12 @@ def test_pipewire_router_reapplies_controls_when_virtual_sink_link_appears(monke assert router.event_source_id == 321 assert len(scheduled_callbacks) == 1 - assert 5092 in backend.link_state_handlers + assert backend.link_state_handlers == {} keep_source = scheduled_callbacks[0]() assert keep_source is False + assert 5092 in backend.link_state_handlers assert backend.moves == [(1, "mini_eq_sink")] assert applied == ["apply"] @@ -608,6 +609,8 @@ def test_pipewire_router_tracks_internal_output_links(monkeypatch: pytest.Monkey router.handle_object_added(None, make_link(93, output_node_id=internal_output.bound_id, input_node_id=22)) assert router.event_source_id == 321 + assert backend.link_state_handlers == {} + scheduled_callbacks[0]() assert 5093 in backend.link_state_handlers diff --git a/tests/test_mini_eq_routing.py b/tests/test_mini_eq_routing.py index 6d58eda..2f07ec6 100644 --- a/tests/test_mini_eq_routing.py +++ b/tests/test_mini_eq_routing.py @@ -1364,9 +1364,9 @@ def close(self) -> None: "stop-monitor", ("target", "speakers"), ("disable", False), - "close-router", "unload-engine", "sync-engine", + "close-router", "close-backend", ] assert controller.routed is False @@ -1570,15 +1570,25 @@ class FakeBackend: def __init__(self) -> None: self.callback = None + def load_module(self, name, arguments): + assert name == "libpipewire-module-loopback" + return object() + def load_filter_chain_module(self, arguments: str): calls.append(f"load:{arguments}") return module - def watch_for_audio_sink(self, sink_name: str, callback, *, timeout_ms: int): - calls.append(f"wait:{sink_name}:{timeout_ms}") + def watch_for_node(self, predicate, callback, *, timeout_ms: int): + assert predicate(make_node(42, "mini_eq_sink.dsp")) + assert not predicate(make_node(43, "mini_eq_sink")) + calls.append(f"wait:mini_eq_sink.dsp:{timeout_ms}") self.callback = callback return FakeWatch() + def watch_for_audio_sink(self, name, callback, **kwargs): + callback(make_node(43, name)) + return FakeWatch() + backend = FakeBackend() controller.engine_module = None controller.engine_start_pending = False @@ -1598,7 +1608,7 @@ def watch_for_audio_sink(self, sink_name: str, callback, *, timeout_ms: int): assert calls == [ "load:module args", - "wait:mini_eq_sink:3000", + "wait:mini_eq_sink.dsp:3000", "status:filter-chain PipeWire EQ ready: mini_eq_sink -> speakers", "apply", ] @@ -1607,6 +1617,49 @@ def watch_for_audio_sink(self, sink_name: str, callback, *, timeout_ms: int): assert controller.running is True +def test_input_module_is_loaded_once_across_dsp_replacements() -> None: + controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController) + calls = [] + module = object() + + class Backend: + def load_module(self, name, arguments): + calls.append((name, arguments)) + return module + + controller.virtual_sink_name = "mini_eq_sink" + controller.output_backend = Backend() + controller.ensure_input_module() + controller.ensure_input_module() + assert len(calls) == 1 + assert calls[0][0] == "libpipewire-module-loopback" + assert 'target.object = "mini_eq_sink.dsp"' in calls[0][1] + assert controller.input_module is module + + +def test_stop_engine_preserves_the_input_device() -> None: + controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController) + dsp = object() + device = object() + unloaded = [] + + class Backend: + def unload_filter_chain_module(self, module): + unloaded.append(module) + + def sync(self): + pass + + controller.engine_module = dsp + controller.input_module = device + controller.output_backend = Backend() + controller.running = True + controller.stop_engine(announce=False) + assert unloaded == [dsp] + assert controller.input_module is device + assert not controller.running + + @pytest.mark.parametrize("replacement_fails", [False, True]) def test_start_engine_preserves_completion_across_pending_type_changes(replacement_fails: bool) -> None: controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController) @@ -1621,6 +1674,9 @@ def cancel(self): pass class FakeBackend: + def load_module(self, name, arguments): + return object() + def load_filter_chain_module(self, arguments): return object() @@ -1630,10 +1686,14 @@ def unload_filter_chain_module(self, module): def sync(self): pass - def watch_for_audio_sink(self, name, callback, **kwargs): + def watch_for_node(self, predicate, callback, **kwargs): callbacks.append(callback) return FakeWatch() + def watch_for_audio_sink(self, name, callback, **kwargs): + callback(make_node(41, name)) + return FakeWatch() + def set_node_params(self, node_id, controls): pass @@ -1674,7 +1734,7 @@ def set_node_params(self, node_id, controls): callbacks[2](None if replacement_fails else make_node(44, "mini_eq_sink")) assert ready == ([] if replacement_fails else [44]) assert monitored == ([] if replacement_fails else [44]) - assert errors == (["filter-chain did not create mini_eq_sink"] if replacement_fails else []) + assert errors == (["filter-chain did not create mini_eq_sink.dsp"] if replacement_fails else []) assert controller.running is not replacement_fails @@ -1890,12 +1950,15 @@ class FakeBackend: def __init__(self) -> None: self.callback = None + def load_module(self, name, arguments): + return object() + def load_filter_chain_module(self, arguments: str): calls.append(f"load:{arguments}") return module - def watch_for_audio_sink(self, sink_name: str, callback, *, timeout_ms: int): - calls.append(f"wait:{sink_name}:{timeout_ms}") + def watch_for_node(self, predicate, callback, *, timeout_ms: int): + calls.append(f"wait:mini_eq_sink.dsp:{timeout_ms}") self.callback = callback return FakeWatch() @@ -1918,8 +1981,8 @@ def sync(self) -> None: assert backend.callback is not None backend.callback(None) - assert calls == ["load:module args", "wait:mini_eq_sink:3000", "sync"] - assert errors == ["filter-chain did not create mini_eq_sink"] + assert calls == ["load:module args", "wait:mini_eq_sink.dsp:3000", "sync"] + assert errors == ["filter-chain did not create mini_eq_sink.dsp"] assert controller.engine_module is None assert controller.filter_node_id is None assert controller.running is False @@ -1952,7 +2015,7 @@ def sync(self) -> None: assert controller.running is False -def test_restart_engine_pauses_stream_router_monitoring_during_restart() -> None: +def test_restart_engine_preserves_player_routing_during_restart() -> None: controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController) controller.running = True controller.routed = True @@ -1980,16 +2043,10 @@ def start_monitoring(self, *, require_initial_route: bool = False) -> None: routing.SystemWideEqController.restart_engine(controller) - assert calls == [ - "stop-monitoring", - "restore", - "stop-engine:False", - "start-engine", - "start-monitoring:True", - ] + assert calls == ["stop-engine:False", "start-engine"] -def test_restart_engine_continues_when_routed_stream_restore_fails() -> None: +def test_restart_engine_does_not_restore_routed_streams() -> None: controller = routing.SystemWideEqController.__new__(routing.SystemWideEqController) controller.running = True controller.routed = True @@ -2018,14 +2075,7 @@ def start_monitoring(self, *, require_initial_route: bool = False) -> None: routing.SystemWideEqController.restart_engine(controller) - assert calls == [ - "stop-monitoring", - "restore", - "warning:restore failed", - "stop-engine:False", - "start-engine", - "start-monitoring:True", - ] + assert calls == ["stop-engine:False", "start-engine"] def test_emit_status_is_silent_during_shutdown(capsys: pytest.CaptureFixture[str]) -> None: diff --git a/tools/check_mpris_preset_runtime.py b/tools/check_mpris_preset_runtime.py new file mode 100644 index 0000000..01f307b --- /dev/null +++ b/tools/check_mpris_preset_runtime.py @@ -0,0 +1,257 @@ +#!/usr/bin/env python3 +"""Check preset rebuilds against WirePlumber's real MPRIS pause policy.""" + +from __future__ import annotations + +import argparse +import os +import signal +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +try: + from tools import check_headless_pipewire_runtime as head + from tools import check_live_ui_runtime as live +except ModuleNotFoundError: # direct script execution + import check_headless_pipewire_runtime as head + import check_live_ui_runtime as live + +MPRIS_XML = """ + + + + + + + + + + + + + + + + +""" + + +class MprisPlayer: + def __init__(self) -> None: + from gi.repository import Gio + + self.pause_calls = 0 + self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None) + self.registrations = [ + self.bus.register_object("/org/mpris/MediaPlayer2", interface, self.method, self.property, None) + for interface in Gio.DBusNodeInfo.new_for_xml(MPRIS_XML).interfaces + ] + self.owner = Gio.bus_own_name_on_connection( + self.bus, "org.mpris.MediaPlayer2.minieqtest", Gio.BusNameOwnerFlags.NONE, None, None + ) + + def method(self, _connection, _sender, _path, _interface, name, _parameters, invocation): + from gi.repository import GLib + + if name == "Pause": + self.pause_calls += 1 + invocation.return_value(GLib.Variant("()", ())) + + def property(self, _connection, _sender, _path, _interface, name): + from gi.repository import GLib + + if name in ("Identity", "DesktopEntry"): + return GLib.Variant("s", "mini-eq-mpris-test") + if name == "PlaybackStatus": + return GLib.Variant("s", "Playing") + if name == "PauseCount": + return GLib.Variant("u", self.pause_calls) + if name in ("SupportedUriSchemes", "SupportedMimeTypes"): + return GLib.Variant("as", []) + return GLib.Variant("b", name in ("CanPause", "CanControl")) + + def close(self) -> None: + from gi.repository import Gio + + Gio.bus_unown_name(self.owner) + for registration in self.registrations: + self.bus.unregister_object(registration) + + +class PlayerProbe: + """Query a player in a separate process, never Mini EQ's own PID tree.""" + + def __init__(self): + from gi.repository import Gio + + self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None) + + @property + def pause_calls(self) -> int: + from gi.repository import Gio, GLib + + return self.bus.call_sync( + "org.mpris.MediaPlayer2.minieqtest", + "/org/mpris/MediaPlayer2", + "org.freedesktop.DBus.Properties", + "Get", + GLib.Variant("(ss)", ("org.mpris.MediaPlayer2.Player", "PauseCount")), + GLib.VariantType.new("(v)"), + Gio.DBusCallFlags.NONE, + 1000, + None, + ).unpack()[0] + + +def run_player_helper(audio: str) -> int: + from gi.repository import GLib + + player = MprisPlayer() + playback = subprocess.Popen(["paplay", "--client-name=mini-eq-mpris-test", audio]) + loop = GLib.MainLoop() + GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGTERM, lambda: loop.quit() or False) + try: + loop.run() + finally: + live.terminate_process(playback, "Pulse playback") + player.close() + return 0 + + +def observe(seconds: float) -> None: + from gi.repository import GLib + + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + head.drain_main_context(GLib.MainContext.default()) + time.sleep(0.02) + + +def run_helper() -> int: + for tool in ("pipewire", "pipewire-pulse", "wireplumber", "paplay", "pw-dump", "wpctl", "pw-metadata"): + live.require_tool(tool) + head.ensure_source_path() + from mini_eq.core import FILTER_TYPES, inactive_eq_bands + from mini_eq.routing import SystemWideEqController + + pw = wp = pulse = playback = controller = player = None + with tempfile.TemporaryDirectory(prefix="mini-eq-mpris-runtime-") as directory: + root = Path(directory) + for name in ("runtime", "config", "data", "cache"): + (root / name).mkdir(mode=0o700) + for key, name in ( + ("XDG_RUNTIME_DIR", "runtime"), + ("XDG_CONFIG_HOME", "config"), + ("XDG_DATA_HOME", "data"), + ("XDG_CACHE_HOME", "cache"), + ): + os.environ[key] = str(root / name) + os.environ["PULSE_SERVER"] = "unix:" + str(root / "runtime/pulse/native") + live.write_pipewire_config(root / "config") + # The private graph must not enumerate or compete with host devices. + wp_config = root / "config/wireplumber/wireplumber.conf.d" + wp_config.mkdir(parents=True) + (wp_config / "10-private-audio.conf").write_text( + "wireplumber.profiles = { main = { " + "monitor.alsa = disabled monitor.bluez = disabled " + "monitor.v4l2 = disabled monitor.libcamera = disabled } }\n", + encoding="utf-8", + ) + with (root / "pulse.log").open("w") as pulse_log, (root / "player.log").open("w") as player_log: + try: + pw, wp = head.start_pipewire_session(root, 15) + pulse = subprocess.Popen(["pipewire-pulse"], stdout=pulse_log, stderr=subprocess.STDOUT) + live.wait_for("private Pulse socket", lambda: (root / "runtime/pulse/native").exists(), 15) + subprocess.run(["wpctl", "settings", "linking.pause-playback", "true"], check=True) + audio = live.create_sine_wav(root / "signal.wav", 90) + playback = subprocess.Popen( + [sys.executable, str(Path(__file__).resolve()), "--player-helper", str(audio)], + stdout=player_log, + stderr=subprocess.STDOUT, + ) + player = PlayerProbe() + head.dispatch_until("MPRIS player ready", lambda: player.pause_calls == 0, 15) + node = live.wait_for( + "Pulse player node", + lambda: next( + ( + n + for n in live.node_items() + if n.get("info", {}).get("props", {}).get("application.name") == "mini-eq-mpris-test" + ), + None, + ), + 15, + ) + stream_id = live.node_id(node) + controller = SystemWideEqController(live.PRIMARY_SINK_NAME) + controller.bands = inactive_eq_bands() + head.wait_for_controller_ready(controller, 15) + head.wait_for_smoke_routed(stream_id, controller.virtual_sink_name, 15) + observe(2) + serial = live.object_serial(live.node_by_name(controller.virtual_sink_name)) + for name in ("Bell", "Lo-pass", "Off"): + controller.set_band_type(0, FILTER_TYPES[name]) + head.dispatch_until("DSP replacement ready", lambda: controller.running, 15) + head.wait_for_processing_path_active( + controller.virtual_sink_name, controller.filter_output_name, 15 + ) + observe(2) + if live.object_serial(live.node_by_name(controller.virtual_sink_name)) != serial: + raise RuntimeError("Preset change replaced the player-facing device") + if player.pause_calls: + raise RuntimeError(f"WirePlumber paused the player after {name}") + if playback.poll() is not None: + raise RuntimeError("Pulse playback disconnected") + print("Preset/type replacements: stable sink, active audio path, zero MPRIS Pause calls", flush=True) + controller.shutdown() + controller = None + observe(2) + if player.pause_calls: + raise RuntimeError("WirePlumber paused the player on normal shutdown") + print("Normal shutdown: zero MPRIS Pause calls", flush=True) + # A positive control prevents a broken/unmatched MPRIS fixture + # from making the zero-pause assertions pass vacuously. + controller = SystemWideEqController(live.PRIMARY_SINK_NAME) + head.wait_for_controller_ready(controller, 15) + head.wait_for_smoke_routed(stream_id, controller.virtual_sink_name, 15) + observe(2) + controller.stream_router.stop_monitoring() + controller.output_backend.unload_filter_chain_module(controller.input_module) + controller.input_module = None + head.dispatch_until("device-removal MPRIS positive control", lambda: player.pause_calls > 0, 15) + print("Positive control: device removal invokes MPRIS Pause", flush=True) + finally: + try: + if controller is not None: + controller.shutdown() + finally: + for process, label in ( + (playback, "Pulse player"), + (pulse, "Pulse server"), + (wp, "WirePlumber"), + (pw, "PipeWire"), + ): + live.terminate_process(process, label) + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--helper", action="store_true", help=argparse.SUPPRESS) + parser.add_argument("--player-helper", help=argparse.SUPPRESS) + args = parser.parse_args() + if args.player_helper: + return run_player_helper(args.player_helper) + if args.helper: + return run_helper() + return subprocess.run( + ["dbus-run-session", "--", sys.executable, str(Path(__file__).resolve()), "--helper"] + ).returncode + + +if __name__ == "__main__": + raise SystemExit(main())