feat: move to virtual sink and loopback audio device as default methodHEADmain

- creates a virtual audio sink on startup and a loopback
- enables listening on loopback
- provide finegrain control over what you want the AI to hear

1 files changed, 385 insertions, 0 deletions

diff --git a/linux_audio.py b/linux_audio.py
new file mode 100644
index 0000000..ea162bb
--- /dev/null
+++ b/linux_audio.py
@@ -0,0 +1,385 @@
+import os
+import select
+import shutil
+import subprocess
+import sys
+import threading
+import time
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+import numpy as np
+import sounddevice as sd
+
+LINUX_VIRTUAL_SINK_MODULE: str = "module-null-sink"
+LINUX_VIRTUAL_SOURCE_MODULE: str = "module-remap-source"
+LINUX_VIRTUAL_LOOPBACK_MODULE: str = "module-loopback"
+LINUX_VIRTUAL_SINK_NAME: str = "auto_live_tl_sink"
+LINUX_VIRTUAL_SOURCE_NAME: str = "auto_live_tl_input"
+LINUX_VIRTUAL_SINK_DESCRIPTION: str = "auto-live-tl-virtual-sink"
+LINUX_VIRTUAL_SOURCE_DESCRIPTION: str = "auto-live-tl-virtual-input"
+LINUX_VIRTUAL_DEVICE_WAIT_SECONDS: float = 2.5
+LINUX_VIRTUAL_DIRECT_DEVICE_NAME: str = "auto-live-tl-virtual-input (PipeWire direct)"
+LINUX_VIRTUAL_DIRECT_CAPTURE_RATE: int = 48000
+LINUX_DIRECT_CAPTURE_IDLE_TIMEOUT_SECONDS: float = 3.0
+
+_virtual_sink_module_id: Optional[int] = None
+_virtual_source_module_id: Optional[int] = None
+_virtual_loopback_module_id: Optional[int] = None
+_virtual_sink_created_by_app: bool = False
+_virtual_source_created_by_app: bool = False
+_virtual_loopback_created_by_app: bool = False
+
+
+def is_linux() -> bool:
+    return sys.platform.startswith("linux")
+
+
+def is_virtual_input_name(device_name: str) -> bool:
+    lowered = device_name.lower()
+    return (
+        LINUX_VIRTUAL_SOURCE_NAME.lower() in lowered
+        or LINUX_VIRTUAL_SOURCE_DESCRIPTION.lower() in lowered
+        or LINUX_VIRTUAL_SINK_NAME.lower() in lowered
+        or LINUX_VIRTUAL_SINK_DESCRIPTION.lower() in lowered
+    )
+
+
+def _run_pactl(args: List[str]) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["pactl", *args],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+
+
+def _find_existing_module_id(module_name: str, arg_match: str) -> Optional[int]:
+    result = _run_pactl(["list", "short", "modules"])
+    if result.returncode != 0:
+        return None
+
+    for line in result.stdout.splitlines():
+        parts = line.split("\t")
+        if len(parts) < 3:
+            continue
+        module_id_raw, listed_module_name, module_args = parts[0], parts[1], parts[2]
+        if listed_module_name != module_name:
+            continue
+        if arg_match not in module_args:
+            continue
+        try:
+            return int(module_id_raw)
+        except ValueError:
+            continue
+    return None
+
+
+def _wait_for_virtual_input_device() -> bool:
+    deadline = time.monotonic() + LINUX_VIRTUAL_DEVICE_WAIT_SECONDS
+    while time.monotonic() < deadline:
+        try:
+            devices = sd.query_devices()
+        except Exception:
+            time.sleep(0.1)
+            continue
+        for dev in devices:
+            if dev.get("max_input_channels", 0) > 0 and is_virtual_input_name(str(dev.get("name", ""))):
+                return True
+        time.sleep(0.1)
+    return False
+
+
+def _get_default_sink_name() -> str:
+    result = _run_pactl(["get-default-sink"])
+    if result.returncode == 0:
+        sink_name = (result.stdout or "").strip()
+        if sink_name:
+            return sink_name
+    return "@DEFAULT_SINK@"
+
+
+def ensure_virtual_input_sink() -> None:
+    global _virtual_sink_module_id, _virtual_source_module_id, _virtual_loopback_module_id
+    global _virtual_sink_created_by_app, _virtual_source_created_by_app, _virtual_loopback_created_by_app
+
+    if not is_linux():
+        return
+
+    if shutil.which("pactl") is None:
+        print("[audio] 'pactl' not found, skipping PipeWire virtual device creation.")
+        return
+
+    existing_sink_id = _find_existing_module_id(
+        LINUX_VIRTUAL_SINK_MODULE,
+        f"sink_name={LINUX_VIRTUAL_SINK_NAME}",
+    )
+    if existing_sink_id is not None:
+        _virtual_sink_module_id = existing_sink_id
+        _virtual_sink_created_by_app = False
+        print(f"[audio] Reusing existing virtual sink module id={existing_sink_id}.")
+    else:
+        sink_result = _run_pactl(
+            [
+                "load-module",
+                LINUX_VIRTUAL_SINK_MODULE,
+                f"sink_name={LINUX_VIRTUAL_SINK_NAME}",
+                f"sink_properties=device.description={LINUX_VIRTUAL_SINK_DESCRIPTION}",
+            ]
+        )
+        if sink_result.returncode != 0:
+            stderr = (sink_result.stderr or "").strip()
+            print(f"[audio] Failed to create PipeWire virtual sink: {stderr or 'unknown error'}")
+            return
+
+        sink_id_text = (sink_result.stdout or "").strip()
+        try:
+            _virtual_sink_module_id = int(sink_id_text)
+            _virtual_sink_created_by_app = True
+            print(
+                "[audio] Created PipeWire virtual sink "
+                f"'{LINUX_VIRTUAL_SINK_DESCRIPTION}' (module id={_virtual_sink_module_id})."
+            )
+        except ValueError:
+            _virtual_sink_module_id = None
+            _virtual_sink_created_by_app = True
+            print("[audio] Created PipeWire virtual sink, but could not parse module id.")
+
+    existing_source_id = _find_existing_module_id(
+        LINUX_VIRTUAL_SOURCE_MODULE,
+        f"source_name={LINUX_VIRTUAL_SOURCE_NAME}",
+    )
+    if existing_source_id is not None:
+        _virtual_source_module_id = existing_source_id
+        _virtual_source_created_by_app = False
+        print(f"[audio] Reusing existing virtual input module id={existing_source_id}.")
+    else:
+        source_result = _run_pactl(
+            [
+                "load-module",
+                LINUX_VIRTUAL_SOURCE_MODULE,
+                f"master={LINUX_VIRTUAL_SINK_NAME}.monitor",
+                f"source_name={LINUX_VIRTUAL_SOURCE_NAME}",
+                f"source_properties=device.description={LINUX_VIRTUAL_SOURCE_DESCRIPTION}",
+            ]
+        )
+        if source_result.returncode != 0:
+            stderr = (source_result.stderr or "").strip()
+            print(f"[audio] Failed to create virtual input source: {stderr or 'unknown error'}")
+            return
+
+        source_id_text = (source_result.stdout or "").strip()
+        try:
+            _virtual_source_module_id = int(source_id_text)
+            _virtual_source_created_by_app = True
+            print(
+                "[audio] Created virtual input source "
+                f"'{LINUX_VIRTUAL_SOURCE_DESCRIPTION}' (module id={_virtual_source_module_id})."
+            )
+        except ValueError:
+            _virtual_source_module_id = None
+            _virtual_source_created_by_app = True
+            print("[audio] Created virtual input source, but could not parse module id.")
+
+    existing_loopback_id = _find_existing_module_id(
+        LINUX_VIRTUAL_LOOPBACK_MODULE,
+        f"source={LINUX_VIRTUAL_SINK_NAME}.monitor",
+    )
+    if existing_loopback_id is not None:
+        _virtual_loopback_module_id = existing_loopback_id
+        _virtual_loopback_created_by_app = False
+        print(f"[audio] Reusing existing headphone loopback module id={existing_loopback_id}.")
+    else:
+        default_sink = _get_default_sink_name()
+        loopback_result = _run_pactl(
+            [
+                "load-module",
+                LINUX_VIRTUAL_LOOPBACK_MODULE,
+                f"source={LINUX_VIRTUAL_SINK_NAME}.monitor",
+                f"sink={default_sink}",
+                "latency_msec=60",
+            ]
+        )
+        if loopback_result.returncode != 0:
+            stderr = (loopback_result.stderr or "").strip()
+            print(f"[audio] Failed to create headphone loopback: {stderr or 'unknown error'}")
+        else:
+            loopback_id_text = (loopback_result.stdout or "").strip()
+            try:
+                _virtual_loopback_module_id = int(loopback_id_text)
+                _virtual_loopback_created_by_app = True
+                print(
+                    "[audio] Mirroring virtual sink to default output "
+                    f"(module id={_virtual_loopback_module_id}, sink={default_sink})."
+                )
+            except ValueError:
+                _virtual_loopback_module_id = None
+                _virtual_loopback_created_by_app = True
+                print("[audio] Headphone loopback created, but could not parse module id.")
+
+    if not _wait_for_virtual_input_device():
+        print("[audio] Virtual input device was not detected by PortAudio yet.")
+
+
+def teardown_virtual_input_sink() -> None:
+    global _virtual_sink_module_id, _virtual_source_module_id, _virtual_loopback_module_id
+    global _virtual_sink_created_by_app, _virtual_source_created_by_app, _virtual_loopback_created_by_app
+
+    if not is_linux() or shutil.which("pactl") is None:
+        return
+
+    if _virtual_loopback_created_by_app and _virtual_loopback_module_id is not None:
+        result = _run_pactl(["unload-module", str(_virtual_loopback_module_id)])
+        if result.returncode != 0:
+            stderr = (result.stderr or "").strip()
+            print(f"[audio] Failed to unload headphone loopback module {_virtual_loopback_module_id}: {stderr or 'unknown error'}")
+        else:
+            print(f"[audio] Unloaded headphone loopback module {_virtual_loopback_module_id}.")
+
+    if _virtual_source_created_by_app and _virtual_source_module_id is not None:
+        result = _run_pactl(["unload-module", str(_virtual_source_module_id)])
+        if result.returncode != 0:
+            stderr = (result.stderr or "").strip()
+            print(f"[audio] Failed to unload virtual input module {_virtual_source_module_id}: {stderr or 'unknown error'}")
+        else:
+            print(f"[audio] Unloaded virtual input module {_virtual_source_module_id}.")
+
+    if _virtual_sink_created_by_app and _virtual_sink_module_id is not None:
+        result = _run_pactl(["unload-module", str(_virtual_sink_module_id)])
+        if result.returncode != 0:
+            stderr = (result.stderr or "").strip()
+            print(f"[audio] Failed to unload virtual sink module {_virtual_sink_module_id}: {stderr or 'unknown error'}")
+        else:
+            print(f"[audio] Unloaded virtual sink module {_virtual_sink_module_id}.")
+
+    _virtual_loopback_module_id = None
+    _virtual_source_module_id = None
+    _virtual_sink_module_id = None
+    _virtual_loopback_created_by_app = False
+    _virtual_source_created_by_app = False
+    _virtual_sink_created_by_app = False
+
+
+def append_pipewire_direct_device_option(
+    input_devices: List[Tuple[int, Dict[str, Any]]],
+    has_virtual_input: bool,
+) -> None:
+    if not has_virtual_input:
+        print(
+            "[audio] Virtual input device not visible in sounddevice device list. "
+            "Adding PipeWire direct capture option instead."
+        )
+
+    input_devices.append(
+        (
+            -1,
+            {
+                "name": LINUX_VIRTUAL_DIRECT_DEVICE_NAME,
+                "max_input_channels": 1,
+                "default_samplerate": float(LINUX_VIRTUAL_DIRECT_CAPTURE_RATE),
+            },
+        )
+    )
+
+
+def _build_pipewire_direct_commands(sample_rate: int) -> List[List[str]]:
+    commands: List[List[str]] = []
+
+    if shutil.which("ffmpeg") is not None:
+        for source_name in [LINUX_VIRTUAL_SOURCE_NAME, f"{LINUX_VIRTUAL_SINK_NAME}.monitor"]:
+            commands.append(
+                [
+                    "ffmpeg",
+                    "-nostdin",
+                    "-hide_banner",
+                    "-loglevel",
+                    "error",
+                    "-f",
+                    "pulse",
+                    "-i",
+                    source_name,
+                    "-ac",
+                    "1",
+                    "-ar",
+                    str(sample_rate),
+                    "-f",
+                    "f32le",
+                    "-",
+                ]
+            )
+
+    return commands
+
+
+def has_pipewire_direct_backend(sample_rate: int) -> bool:
+    return bool(_build_pipewire_direct_commands(sample_rate))
+
+
+def pipewire_direct_capture_worker(
+    stop_event: threading.Event,
+    sample_rate: int,
+    on_chunk: Callable[[np.ndarray], None],
+    add_runtime_log: Callable[[str, str], None],
+) -> None:
+    try:
+        _run_pipewire_direct_capture(stop_event, sample_rate, on_chunk)
+    except Exception as exc:
+        message = f"PipeWire direct capture failed: {exc}"
+        print(f"[audio] {message}")
+        add_runtime_log("AUDIO", message)
+
+
+def _run_pipewire_direct_capture(
+    stop_event: threading.Event,
+    sample_rate: int,
+    on_chunk: Callable[[np.ndarray], None],
+) -> None:
+    commands = _build_pipewire_direct_commands(sample_rate)
+    if not commands:
+        raise RuntimeError("PipeWire direct capture needs 'ffmpeg' installed.")
+
+    bytes_per_chunk = int(sample_rate * 0.5) * 4
+
+    for command in commands:
+        process: Optional[subprocess.Popen[bytes]] = None
+        try:
+            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            time.sleep(0.2)
+            if process.poll() is not None:
+                continue
+
+            assert process.stdout is not None
+            print(f"[audio] PipeWire direct capture backend: {' '.join(command[:2])}")
+
+            silent_windows = 0
+            while not stop_event.is_set():
+                ready, _, _ = select.select([process.stdout], [], [], 1.0)
+                if not ready:
+                    silent_windows += 1
+                    if silent_windows * 1.0 >= LINUX_DIRECT_CAPTURE_IDLE_TIMEOUT_SECONDS:
+                        raise RuntimeError("backend produced no audio frames")
+                    continue
+
+                raw = os.read(process.stdout.fileno(), bytes_per_chunk)
+                if not raw:
+                    raise RuntimeError("backend stream ended")
+
+                silent_windows = 0
+                chunk = np.frombuffer(raw, dtype=np.float32).copy()
+                if chunk.size == 0:
+                    continue
+                on_chunk(chunk)
+
+            return
+        except Exception as exc:
+            backend_name = " ".join(command[:2])
+            print(f"[audio] Direct capture backend failed ({backend_name}): {exc}")
+            continue
+        finally:
+            if process is not None and process.poll() is None:
+                process.terminate()
+                try:
+                    process.wait(timeout=1.5)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+
+    raise RuntimeError("All PipeWire direct capture backends failed.")