From 52e55e112f066b5ba80f7992f47e736168fb552e Mon Sep 17 00:00:00 2001 From: "thomas.kopp" Date: Sun, 2 Aug 2026 01:25:15 +0200 Subject: [PATCH] fix: build PipeWire capture chain on demand instead of permanently MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The null sink and its two loopbacks were loaded at startup and stayed loaded forever. While loaded they link the microphone and the speaker device into a single PipeWire driver group: the microphone drives the graph and the speaker device runs as a clock follower, so every playback stream pays for continuous cross-device resampling. Nothing needs the chain while idle — recording is triggered manually. Build it in toggle_recording() and drop it as soon as the audio is captured, before transcription runs for minutes. Module ids are now derived from pactl instead of tracked in a state file, so a crashed process cannot leave stale ids behind. The state file only holds the device selection. Startup tears down any leftover chain. Loopbacks get latency_msec=200; transcription is offline, and a small buffer would put these nodes on the graph's realtime deadline. --- api/pipeline.py | 9 ++- api/router.py | 33 +++------ main.py | 44 ++---------- pipewire.py | 155 +++++++++++++++++++++++++++++++++++++++++ tests/test_pipewire.py | 145 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 322 insertions(+), 64 deletions(-) create mode 100644 pipewire.py create mode 100644 tests/test_pipewire.py diff --git a/api/pipeline.py b/api/pipeline.py index f44c5f3..bba3bcc 100644 --- a/api/pipeline.py +++ b/api/pipeline.py @@ -5,6 +5,7 @@ import tempfile import traceback from datetime import datetime +import pipewire from api.state import state, Status from api.router import broadcast from config import load as load_config @@ -26,7 +27,13 @@ async def run_pipeline(): diar_cfg = cfg.get("diarization", {}) use_diarization = diar_cfg.get("enabled") and diar_cfg.get("hf_token") - recorder.stop() + # Drop the capture chain as soon as the audio is captured — transcription + # runs for minutes and must not keep the PipeWire clock domains coupled. + try: + recorder.stop() + finally: + pipewire.teardown() + await state.set_status(Status.PROCESSING) await broadcast({"event": "processing"}) diff --git a/api/router.py b/api/router.py index 64a091e..053a9df 100644 --- a/api/router.py +++ b/api/router.py @@ -112,8 +112,11 @@ async def toggle_recording(user: dict = Depends(current_user)): return {"action": "reset"} if state.status == Status.IDLE: from audio import AudioRecorder + import pipewire cfg = load_config() audio_device = cfg.get("audio", {}).get("device") or None + # Build the capture chain only for the duration of the recording. + pipewire.setup() state._recorder = AudioRecorder(device=audio_device) state._recorder.start() state.recording_user = user["username"] @@ -303,7 +306,8 @@ async def list_audio_devices(user: dict = Depends(current_user)): @router.post("/audio/combined") async def create_combined_source(body: dict, user: dict = Depends(current_user)): - import subprocess, json, pathlib + import subprocess + import pipewire if not user.get("is_admin"): raise HTTPException(status_code=403, detail="Nur Administratoren") mic_sd = body.get("mic", "") @@ -320,28 +324,11 @@ async def create_combined_source(body: dict, user: dict = Depends(current_user)) known = {line.split("\t")[1] for line in out.strip().splitlines() if "\t" in line} if mic not in known or monitor not in known: raise HTTPException(status_code=400, detail="Unbekanntes Audio-Device") - # Use description without spaces so sounddevice name == sink_name - sink_id = subprocess.check_output([ - "pactl", "load-module", "module-null-sink", - "sink_name=transkriptor-combined", - "sink_properties=device.description=transkriptor-combined", - ], timeout=5).decode().strip() - mic_id = subprocess.check_output([ - "pactl", "load-module", "module-loopback", - f"source={mic}", "sink=transkriptor-combined", - ], timeout=5).decode().strip() - mon_id = subprocess.check_output([ - "pactl", "load-module", "module-loopback", - f"source={monitor}", "sink=transkriptor-combined", - ], timeout=5).decode().strip() - state_path = pathlib.Path( - os.path.expanduser("~/.config/tueit-transcriber/pipewire-modules.json") - ) - state_path.parent.mkdir(parents=True, exist_ok=True) - ids = [int(sink_id), int(mic_id), int(mon_id)] - # Store pactl names for restore, sounddevice name as device - state_path.write_text(json.dumps({"ids": ids, "mic": mic, "monitor": monitor})) - return {"device": "transkriptor-combined", "module_ids": ids} + # Only remember the selection. The chain itself is loaded when a recording + # starts and unloaded when it ends, so it does not couple the microphone + # and the speaker device into one PipeWire clock domain while idle. + pipewire.save_devices(mic, monitor) + return {"device": pipewire.SINK_NAME} @router.get("/settings") diff --git a/main.py b/main.py index 819e85b..c431df4 100644 --- a/main.py +++ b/main.py @@ -12,6 +12,7 @@ from fastapi.staticfiles import StaticFiles import pystray from PIL import Image, ImageDraw +import pipewire from api.router import router from api.state import state, Status from config import load as load_config @@ -49,45 +50,6 @@ async def settingsjs(): return FileResponse(str(FRONTEND_DIR / "settings.js")) -# ── PipeWire combined source restore ────────────────────────────────────────── - -def _restore_pipewire_combined(): - """Recreate transkriptor-combined.monitor on startup if it was previously configured.""" - import json, subprocess, logging - state_path = Path(os.path.expanduser("~/.config/tueit-transcriber/pipewire-modules.json")) - if not state_path.exists(): - return - try: - data = json.loads(state_path.read_text()) - mic = data.get("mic") - monitor = data.get("monitor") - if not mic or not monitor: - return - sources = subprocess.check_output( - ["pactl", "list", "sources", "short"], stderr=subprocess.DEVNULL, timeout=5 - ).decode() - if "transkriptor-combined.monitor" in sources: - return # already loaded - sink_id = subprocess.check_output([ - "pactl", "load-module", "module-null-sink", - "sink_name=transkriptor-combined", - "sink_properties=device.description=transkriptor-combined", - ], timeout=5).decode().strip() - mic_id = subprocess.check_output([ - "pactl", "load-module", "module-loopback", - f"source={mic}", "sink=transkriptor-combined", - ], timeout=5).decode().strip() - mon_id = subprocess.check_output([ - "pactl", "load-module", "module-loopback", - f"source={monitor}", "sink=transkriptor-combined", - ], timeout=5).decode().strip() - ids = [int(sink_id), int(mic_id), int(mon_id)] - state_path.write_text(json.dumps({"ids": ids, "mic": mic, "monitor": monitor})) - logging.getLogger(__name__).info("Restored PipeWire combined source (ids: %s)", ids) - except Exception as e: - logging.getLogger(__name__).warning("Could not restore PipeWire combined source: %s", e) - - # ── PID file ─────────────────────────────────────────────────────────────────── def write_pid(pid_path: str): @@ -184,7 +146,9 @@ if __name__ == "__main__": pid_path = cfg.get("pid_file", os.path.expanduser("~/.local/run/tueit-transcriber.pid")) write_pid(pid_path) - _restore_pipewire_combined() + # Clear a chain left behind by a previous crash. It is built on demand when + # a recording starts, never while idle. + pipewire.teardown() signal.signal(signal.SIGUSR1, _sigusr1_handler) uvicorn_cfg = uvicorn.Config(app, host=host, port=port, log_level="debug") diff --git a/pipewire.py b/pipewire.py new file mode 100644 index 0000000..ea38d81 --- /dev/null +++ b/pipewire.py @@ -0,0 +1,155 @@ +"""On-demand PipeWire capture chain. + +The chain is a null sink fed by two loopbacks: the microphone and the monitor +of the speaker device. Recording reads the null sink's monitor, so both sources +end up in one WAV. + +While the chain is loaded it links the microphone and the speaker device into a +single PipeWire driver group. The microphone then drives the graph and the +speaker device runs as a clock follower, so every playback stream pays for +continuous cross-device resampling. The chain is therefore loaded only while a +recording actually runs, and torn down again afterwards. +""" + +import json +import logging +import os +import pathlib +import subprocess +import time + +logger = logging.getLogger(__name__) + +SINK_NAME = "transkriptor-combined" + +# Transcription is offline, so loopback latency is irrelevant for the result. +# A small buffer would put these nodes on the graph's realtime deadline instead. +LOOPBACK_LATENCY_MSEC = 200 + +STATE_PATH = pathlib.Path( + os.path.expanduser("~/.config/tueit-transcriber/pipewire-modules.json") +) + +_TIMEOUT = 5 + + +def load_devices() -> dict | None: + """Return the configured {"mic", "monitor"} pactl source names, or None.""" + try: + data = json.loads(STATE_PATH.read_text()) + except (OSError, ValueError): + return None + mic, monitor = data.get("mic"), data.get("monitor") + if not mic or not monitor: + return None + return {"mic": mic, "monitor": monitor} + + +def save_devices(mic: str, monitor: str) -> None: + """Persist the device selection. Loads nothing — setup() does that.""" + STATE_PATH.parent.mkdir(parents=True, exist_ok=True) + STATE_PATH.write_text(json.dumps({"mic": mic, "monitor": monitor})) + + +def is_loaded() -> bool: + """True if the combined sink currently exists in PipeWire.""" + try: + out = subprocess.check_output( + ["pactl", "list", "short", "sources"], + stderr=subprocess.DEVNULL, timeout=_TIMEOUT, + ).decode() + except (subprocess.SubprocessError, OSError): + return False + return f"{SINK_NAME}.monitor" in out + + +def setup(timeout: float = 3.0) -> list[int]: + """Load the capture chain. Returns the loaded module ids. + + Returns an empty list if no devices are configured, the chain is already + loaded, or pactl fails — recording then falls back to the raw device. + """ + devices = load_devices() + if devices is None: + logger.info("No combined source configured, skipping capture chain") + return [] + if is_loaded(): + return [] + + try: + ids = [ + _load_module( + "module-null-sink", + f"sink_name={SINK_NAME}", + f"sink_properties=device.description={SINK_NAME}", + ), + _load_module( + "module-loopback", + f"source={devices['mic']}", f"sink={SINK_NAME}", + f"latency_msec={LOOPBACK_LATENCY_MSEC}", + ), + _load_module( + "module-loopback", + f"source={devices['monitor']}", f"sink={SINK_NAME}", + f"latency_msec={LOOPBACK_LATENCY_MSEC}", + ), + ] + except (subprocess.SubprocessError, OSError, ValueError) as e: + logger.warning("Could not load capture chain: %s", e) + teardown() + return [] + + _wait_until_visible(timeout) + logger.info("Capture chain loaded (module ids: %s)", ids) + return ids + + +def teardown() -> None: + """Unload the capture chain. Safe to call when nothing is loaded.""" + for module_id in sorted(_chain_module_ids(), reverse=True): + subprocess.run( + ["pactl", "unload-module", str(module_id)], + stderr=subprocess.DEVNULL, timeout=_TIMEOUT, check=False, + ) + + +def _chain_module_ids() -> list[int]: + """Ids of every loaded module belonging to the chain. + + Derived from pactl rather than tracked in a file, so a crashed or killed + process cannot leave stale ids behind. + """ + try: + out = subprocess.check_output( + ["pactl", "list", "short", "modules"], + stderr=subprocess.DEVNULL, timeout=_TIMEOUT, + ).decode() + except (subprocess.SubprocessError, OSError): + return [] + ids = [] + for line in out.splitlines(): + if SINK_NAME not in line: + continue + try: + ids.append(int(line.split("\t")[0])) + except (IndexError, ValueError): + continue + return ids + + +def _load_module(name: str, *args: str) -> int: + out = subprocess.check_output( + ["pactl", "load-module", name, *args], + stderr=subprocess.DEVNULL, timeout=_TIMEOUT, + ) + return int(out.decode().strip()) + + +def _wait_until_visible(timeout: float) -> None: + """Block until the sink shows up, so sounddevice can find it.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if is_loaded(): + return + time.sleep(0.1) + logger.warning("Capture chain did not appear within %.1fs", timeout) diff --git a/tests/test_pipewire.py b/tests/test_pipewire.py new file mode 100644 index 0000000..df11874 --- /dev/null +++ b/tests/test_pipewire.py @@ -0,0 +1,145 @@ +import json +from unittest.mock import patch, MagicMock + +import pytest + + +@pytest.fixture +def state_path(tmp_path, monkeypatch): + import pipewire + p = tmp_path / "pipewire-modules.json" + monkeypatch.setattr(pipewire, "STATE_PATH", p) + return p + + +def test_save_devices_persists_selection(state_path): + import pipewire + pipewire.save_devices("mic-src", "monitor-src") + data = json.loads(state_path.read_text()) + assert data["mic"] == "mic-src" + assert data["monitor"] == "monitor-src" + + +def test_save_devices_does_not_load_modules(state_path): + import pipewire + with patch("subprocess.check_output") as run: + pipewire.save_devices("mic-src", "monitor-src") + run.assert_not_called() + + +def test_load_devices_returns_none_without_state(state_path): + import pipewire + assert pipewire.load_devices() is None + + +def test_load_devices_returns_none_on_incomplete_state(state_path): + import pipewire + state_path.write_text(json.dumps({"mic": "mic-src"})) + assert pipewire.load_devices() is None + + +def test_load_devices_roundtrip(state_path): + import pipewire + pipewire.save_devices("mic-src", "monitor-src") + assert pipewire.load_devices() == {"mic": "mic-src", "monitor": "monitor-src"} + + +def test_setup_without_configured_devices_does_nothing(state_path): + import pipewire + with patch("subprocess.check_output") as run: + assert pipewire.setup() == [] + run.assert_not_called() + + +def test_setup_loads_sink_and_two_loopbacks(state_path): + import pipewire + pipewire.save_devices("mic-src", "monitor-src") + with patch.object(pipewire, "is_loaded", side_effect=[False, True]), \ + patch("subprocess.check_output", side_effect=[b"11\n", b"12\n", b"13\n"]) as run: + ids = pipewire.setup() + assert ids == [11, 12, 13] + cmds = [c.args[0] for c in run.call_args_list] + assert cmds[0][:3] == ["pactl", "load-module", "module-null-sink"] + assert cmds[1][:3] == ["pactl", "load-module", "module-loopback"] + assert cmds[2][:3] == ["pactl", "load-module", "module-loopback"] + + +def test_setup_uses_configured_sources(state_path): + import pipewire + pipewire.save_devices("mic-src", "monitor-src") + with patch.object(pipewire, "is_loaded", side_effect=[False, True]), \ + patch("subprocess.check_output", side_effect=[b"11\n", b"12\n", b"13\n"]) as run: + pipewire.setup() + cmds = [c.args[0] for c in run.call_args_list] + assert "source=mic-src" in cmds[1] + assert "source=monitor-src" in cmds[2] + + +def test_setup_applies_large_loopback_latency(state_path): + """A small buffer would put the loopbacks on the graph's realtime deadline.""" + import pipewire + pipewire.save_devices("mic-src", "monitor-src") + with patch.object(pipewire, "is_loaded", side_effect=[False, True]), \ + patch("subprocess.check_output", side_effect=[b"11\n", b"12\n", b"13\n"]) as run: + pipewire.setup() + expected = f"latency_msec={pipewire.LOOPBACK_LATENCY_MSEC}" + for cmd in [c.args[0] for c in run.call_args_list][1:]: + assert expected in cmd + assert pipewire.LOOPBACK_LATENCY_MSEC >= 100 + + +def test_setup_is_noop_when_already_loaded(state_path): + import pipewire + pipewire.save_devices("mic-src", "monitor-src") + with patch.object(pipewire, "is_loaded", return_value=True), \ + patch("subprocess.check_output") as run: + assert pipewire.setup() == [] + run.assert_not_called() + + +def test_teardown_unloads_chain_modules_highest_id_first(): + import pipewire + listing = ( + b"7\tmodule-something\tunrelated\n" + b"11\tmodule-null-sink\tsink_name=transkriptor-combined\n" + b"12\tmodule-loopback\tsource=mic-src sink=transkriptor-combined\n" + b"13\tmodule-loopback\tsource=monitor-src sink=transkriptor-combined\n" + ) + with patch("subprocess.check_output", return_value=listing), \ + patch("subprocess.run") as run: + pipewire.teardown() + unloaded = [c.args[0][-1] for c in run.call_args_list] + assert unloaded == ["13", "12", "11"] + + +def test_teardown_ignores_unrelated_modules(): + import pipewire + listing = b"7\tmodule-something\tunrelated\n" + with patch("subprocess.check_output", return_value=listing), \ + patch("subprocess.run") as run: + pipewire.teardown() + run.assert_not_called() + + +def test_teardown_survives_pactl_failure(): + import pipewire + import subprocess as sp + with patch("subprocess.check_output", side_effect=sp.SubprocessError("boom")): + pipewire.teardown() # must not raise + + +def test_setup_survives_pactl_failure(state_path): + import pipewire + import subprocess as sp + pipewire.save_devices("mic-src", "monitor-src") + with patch.object(pipewire, "is_loaded", return_value=False), \ + patch("subprocess.check_output", side_effect=sp.SubprocessError("boom")): + assert pipewire.setup() == [] + + +def test_is_loaded_detects_sink(): + import pipewire + with patch("subprocess.check_output", return_value=b"42\ttranskriptor-combined.monitor\n"): + assert pipewire.is_loaded() is True + with patch("subprocess.check_output", return_value=b"42\tsomething-else.monitor\n"): + assert pipewire.is_loaded() is False