tueit_Transkriptor/pipewire.py

"""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)