diff --git a/src/vocalinux/audio/__init__.py b/src/vocalinux/audio/__init__.py index 16a882b..0a54f05 100644 --- a/src/vocalinux/audio/__init__.py +++ b/src/vocalinux/audio/__init__.py @@ -1 +1 @@ -"""Playback helpers that are not the dictation microphone.""" +"""Audio helpers — playback ducking and PipeWire system-audio capture.""" diff --git a/src/vocalinux/audio/capture.py b/src/vocalinux/audio/capture.py index ed41f22..3274678 100644 --- a/src/vocalinux/audio/capture.py +++ b/src/vocalinux/audio/capture.py @@ -13,9 +13,17 @@ without inheriting dictation behavior. import ctypes import logging +import subprocess import time from typing import TYPE_CHECKING, Any, Optional +from .pipewire import ( + PIPEWIRE_INDEX_BASE, + _test_pipewire_input, + get_system_audio_sources, + is_pipewire_device_index, +) + if TYPE_CHECKING: import numpy as np @@ -209,6 +217,22 @@ def get_audio_input_devices() -> list: except OSError as e: logger.error(f"Error enumerating audio devices: {e}") + # PipeWire sinks exposed as system-audio sources. Indices count down from + # PIPEWIRE_INDEX_BASE so they can never collide with a PortAudio index. + # Kept outside the PortAudio block so a missing PyAudio still lists them. + try: + for position, source in enumerate(get_system_audio_sources()): + devices.append((PIPEWIRE_INDEX_BASE - position, source.display_name, source.is_default)) + except ( + OSError, + subprocess.SubprocessError, + ValueError, + TypeError, + KeyError, + AttributeError, + ): + logger.warning("PipeWire source enumeration failed", exc_info=True) + return devices @@ -620,6 +644,9 @@ def test_audio_input(device_index: Optional[int] = None, duration: float = 1.0) "error": None, } + if is_pipewire_device_index(device_index): + return _test_pipewire_input(device_index, duration) + try: import numpy as np import pyaudio @@ -739,6 +766,9 @@ class PortAudioCaptureSource: self.channels = 1 self.downmix_channel: Optional[int] = None + #: Reads through a PyAudio instance — the caller must supply one. + requires_pyaudio: bool = True + def open(self, audio: Any = None) -> None: """Resolve the input device and open the negotiated capture stream. @@ -884,6 +914,7 @@ class PortAudioCaptureSource: FORMAT = pyaudio.paInt16 self.audio = audio_instance + new_stream: Any = None try: # Close existing stream if it exists if self.stream: @@ -952,9 +983,11 @@ class PortAudioCaptureSource: except (IOError, OSError) as e: logger.error(f"Audio reconnection failed: {e}") + _safe_close_stream(new_stream) return False except Exception as e: logger.error(f"Unexpected error during audio reconnection: {e}") + _safe_close_stream(new_stream) return False def close(self) -> None: diff --git a/src/vocalinux/audio/pipewire.py b/src/vocalinux/audio/pipewire.py new file mode 100644 index 0000000..37947c3 --- /dev/null +++ b/src/vocalinux/audio/pipewire.py @@ -0,0 +1,531 @@ +"""PipeWire capture: monitor/system-audio sources through ``pw-record``. + +PortAudio enumerates ALSA and PulseAudio devices, not PipeWire nodes, so a +sink's monitor (everything playing through it) is unreachable through the +microphone path. ``pw-record`` is the native PipeWire recorder: pointed at an +``Audio/Sink`` node it captures that sink's monitor, and PipeWire converts to +the requested format in the graph, so no PortAudio-style format negotiation is +needed. With ``--raw`` and ``-`` it streams raw s16 PCM on stdout, which this +module wraps in the same ``read``/``stop_stream``/``close`` shape the capture +loop uses for PortAudio streams. + +Sources are enumerated with ``pw-dump`` (one JSON document describing every +object in the daemon). Each ``Audio/Sink`` — physical or virtual — becomes a +selectable "System audio" device. Per-application stream capture is a +different problem and deliberately out of scope here. +""" + +from __future__ import annotations + +import json +import logging +import os +import select +import shutil +import subprocess +import time +from dataclasses import dataclass +from typing import Any, Callable, List, Optional, Tuple + +from ..utils.host_process import host_env + +logger = logging.getLogger(__name__) + +#: Display-name prefix marking a PipeWire system-audio entry. Kept free of +#: the substrings ``_is_virtual_device`` filters on ("pipewire", "monitor"), +#: which only apply to PortAudio pseudo-devices. +SYSTEM_AUDIO_PREFIX = "System audio: " + +#: Device index assigned to the first PipeWire source. PortAudio indices are +#: non-negative and -1 already means "system default", so PipeWire sources +#: count down from -2. +PIPEWIRE_INDEX_BASE = -2 + +#: pw-dump media classes whose monitor a user can meaningfully record. +_MONITOR_CLASSES = frozenset({"Audio/Sink", "Audio/Sink/Virtual"}) + +#: Metadata keys carrying the effective and configured default sink. +_DEFAULT_SINK_KEYS = ("default.audio.sink", "default.configured.audio.sink") + +_COMMAND_TIMEOUT_SECONDS = 3.0 + +CommandResult = Tuple[int, str, str] +CommandRunner = Callable[[List[str]], CommandResult] +Which = Callable[[str], Optional[str]] +Popen = Callable[..., "subprocess.Popen[bytes]"] + + +@dataclass(frozen=True) +class PipeWireSource: + """An ``Audio/Sink`` node whose monitor can be recorded. + + Attributes: + node_name: ``node.name`` — the stable identifier ``pw-record + --target`` accepts (e.g. ``alsa_output.pci-0000_00_1f.3.analog-stereo``). + description: Human-readable ``node.description``. + serial: ``object.serial``, used for deterministic ordering. + is_default: True when the daemon's default sink metadata names this node. + label: Display label with collisions already disambiguated. + """ + + node_name: str + description: str + serial: int + is_default: bool = False + label: str = "" + + @property + def display_name(self) -> str: + """Device-list name: prefix plus the (disambiguated) label.""" + return f"{SYSTEM_AUDIO_PREFIX}{self.label or self.description or self.node_name}" + + +def _run_host_command(args: List[str]) -> CommandResult: + """Run a host PipeWire tool. The caller handles a non-zero status.""" + completed = subprocess.run( + args, + capture_output=True, + text=True, + timeout=_COMMAND_TIMEOUT_SECONDS, + check=False, + env=host_env(), + ) + return completed.returncode, completed.stdout or "", completed.stderr or "" + + +def _parse_pw_dump(objects: object) -> List[PipeWireSource]: + """Extract monitor-able sink nodes and default-sink names from ``pw-dump``.""" + if not isinstance(objects, list): + return [] + sinks: List[PipeWireSource] = [] + default_names = set() + for obj in objects: + if not isinstance(obj, dict): + continue + info = obj.get("info") or {} + if not isinstance(info, dict): + continue + if obj.get("type") == "PipeWire:Interface:Node": + props = info.get("props") or {} + if not isinstance(props, dict): + continue + if props.get("media.class") not in _MONITOR_CLASSES: + continue + node_name = props.get("node.name") + if not node_name: + continue + try: + serial = int(props.get("object.serial") or obj.get("id") or 0) + except (TypeError, ValueError): + serial = 0 + sinks.append( + PipeWireSource( + node_name=str(node_name), + description=str(props.get("node.description") or node_name), + serial=serial, + ) + ) + elif obj.get("type") == "PipeWire:Interface:Metadata": + metadata = info.get("metadata") or [] + if not isinstance(metadata, list): + continue + for entry in metadata: + if not isinstance(entry, dict) or entry.get("key") not in _DEFAULT_SINK_KEYS: + continue + value = entry.get("value") + if isinstance(value, dict) and value.get("name"): + default_names.add(str(value["name"])) + + sinks.sort(key=lambda source: source.serial) + + # Two sinks can share a description; include the node name so entries and + # stored device names stay resolvable. + description_counts: dict = {} + for source in sinks: + description_counts[source.description] = description_counts.get(source.description, 0) + 1 + labelled: List[PipeWireSource] = [] + for source in sinks: + label = source.description + if description_counts.get(source.description, 0) > 1: + label = f"{source.description} ({source.node_name})" + labelled.append( + PipeWireSource( + node_name=source.node_name, + description=source.description, + serial=source.serial, + is_default=source.node_name in default_names, + label=label, + ) + ) + return labelled + + +def get_system_audio_sources( + runner: Optional[CommandRunner] = None, + which: Optional[Which] = None, +) -> List[PipeWireSource]: + """List PipeWire sinks whose monitor can be captured, in stable order. + + Returns an empty list when ``pw-dump`` is missing, the daemon is not + running, or the output cannot be parsed — PipeWire absent must never break + microphone enumeration. + """ + which = which or shutil.which + # Both tools are required: pw-dump enumerates, pw-record captures. A + # system missing pw-record would otherwise list sources that can never + # be opened. + if which("pw-dump") is None or which("pw-record") is None: + return [] + runner = runner or _run_host_command + try: + code, stdout, _stderr = runner(["pw-dump"]) + except (OSError, subprocess.SubprocessError) as exc: + logger.debug("pw-dump failed: %s", exc) + return [] + if code != 0 or not stdout.strip(): + return [] + try: + objects = json.loads(stdout) + except ValueError: + logger.debug("Could not parse pw-dump output as JSON") + return [] + return _parse_pw_dump(objects) + + +def pipewire_available(which: Optional[Which] = None) -> bool: + """Whether the host has the PipeWire command-line tools needed to capture.""" + which = which or shutil.which + return which("pw-dump") is not None and which("pw-record") is not None + + +def is_pipewire_device_index(device_index: Optional[int]) -> bool: + """Whether a device index falls in the PipeWire (system-audio) range.""" + return ( + isinstance(device_index, int) + and not isinstance(device_index, bool) + and device_index <= PIPEWIRE_INDEX_BASE + ) + + +def is_pipewire_device(device_index: Optional[int], device_name: Optional[str]) -> bool: + """Whether a stored device selection refers to a PipeWire source.""" + if is_pipewire_device_index(device_index): + return True + return bool(device_name) and str(device_name).startswith(SYSTEM_AUDIO_PREFIX) + + +def _source_at_position( + device_index: Optional[int], sources: List[PipeWireSource] +) -> Optional[PipeWireSource]: + """Return the source at a stored negative index position, if in range.""" + if device_index is None or not is_pipewire_device_index(device_index): + return None + position = PIPEWIRE_INDEX_BASE - device_index + if 0 <= position < len(sources): + return sources[position] + return None + + +def _resolve_by_stored_name( + device_name: str, + device_index: Optional[int], + sources: List[PipeWireSource], +) -> Optional[PipeWireSource]: + """Match a stored display name to a live source by stable identifiers. + + Match order, strongest evidence first: + + 1. Exact ``display_name`` equality. + 2. ``node.description`` equality on the stored remainder. When several + sources share the description (dedup labels changed since the name was + stored), the stored index position disambiguates: a sink added later + gets a higher ``object.serial`` and sorts after the original. + 3. A trailing ``(node.name)`` — the dedup format written when two sinks + shared a description. ``node.name`` is stable across description + changes, so a renamed sink still resolves. Description matching runs + first so a description that itself ends in parentheses is not + mistaken for the suffix. + """ + for source in sources: + if source.display_name == device_name: + return source + remainder = device_name + if remainder.startswith(SYSTEM_AUDIO_PREFIX): + remainder = remainder[len(SYSTEM_AUDIO_PREFIX) :] + matches = [s for s in sources if s.description == remainder] + if len(matches) == 1: + return matches[0] + if len(matches) > 1: + positioned = _source_at_position(device_index, sources) + if positioned in matches: + return positioned + return None # Genuinely ambiguous — refuse rather than guess. + if remainder.endswith(")") and " (" in remainder: + node_part = remainder.rpartition(" (")[2][:-1] + node_matches = [s for s in sources if s.node_name == node_part] + if len(node_matches) == 1: + return node_matches[0] + return None + + +def resolve_pipewire_source( + device_index: Optional[int], + device_name: Optional[str], + sources: Optional[List[PipeWireSource]] = None, +) -> Optional[PipeWireSource]: + """Resolve a stored selection to a live PipeWire source. + + A stored name that no longer resolves returns ``None`` rather than + falling back to a positional index: index positions shift as sinks come + and go, and guessing wrong would capture a different device. The index + fallback only applies when no name was stored (legacy selections). + """ + if sources is None: + sources = get_system_audio_sources() + if device_name: + return _resolve_by_stored_name(device_name, device_index, sources) + return _source_at_position(device_index, sources) + + +class PipeWireCaptureSource: + """``CaptureSource`` over a PipeWire sink monitor via ``pw-record --raw``. + + ``pw-record --raw`` emits raw s16 PCM on stdout at the rate/channels the + dictation loop already expects (mono 16 kHz by default), so no downmixing + or resampling is needed. ``read_chunk`` raises ``IOError`` when the + process dies — the same class of event as a PortAudio device loss, so the + existing reconnection path applies. + + Attributes: + device_index: Stored negative device index (see PIPEWIRE_INDEX_BASE). + device_name: Stored display name, re-resolved first on open. + audio: Unused — kept None; present for parity with PortAudioCaptureSource. + stream: The capture process while open, else None. + sample_rate: Capture rate in Hz. + channels: Channel count (pw-record is opened mono by default). + downmix_channel: Always None — the stream is already mono. + """ + + #: Spawns ``pw-record`` itself — the PortAudio instance can be absent. + requires_pyaudio: bool = False + + _CHUNK_FRAMES = 1024 + + def __init__( + self, + device_index: Optional[int] = None, + device_name: Optional[str] = None, + *, + sample_rate: int = 16000, + channels: int = 1, + popen: Optional[Popen] = None, + which: Optional[Which] = None, + ) -> None: + self.device_index = device_index + self.device_name = device_name + self.audio: Any = None + self.stream: Optional["subprocess.Popen[bytes]"] = None + self.sample_rate = sample_rate + self.channels = channels + self.downmix_channel: Optional[int] = None + self.source: Optional[PipeWireSource] = None + self._popen: Popen = popen or subprocess.Popen + self._which = which or shutil.which + + def open(self, audio: Any = None) -> None: + """Resolve the PipeWire source and spawn ``pw-record``. + + Args: + audio: Ignored — present for ``CaptureSource`` parity. + + Raises: + FileNotFoundError: when ``pw-record`` is not installed. + IOError: when the stored selection no longer resolves to a source. + """ + if self.stream is not None: + return + if self._which("pw-record") is None: + raise FileNotFoundError("pw-record is not installed") + source = resolve_pipewire_source(self.device_index, self.device_name) + if source is None: + # Deliberately no PortAudio fallback: silently switching to a + # microphone would capture the wrong audio. + raise IOError("PipeWire source is no longer available") + self.source = source + self.stream = self._popen( + [ + "pw-record", + "--raw", + "--format", + "s16", + "--rate", + str(self.sample_rate), + "--channels", + str(self.channels), + "--target", + source.node_name, + "-", + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + env=host_env(), + ) + + def reopen(self, audio_instance: Any) -> bool: + """Close the current process and spawn a fresh one. True on success.""" + self.close() + try: + self.open() + except (IOError, OSError) as exc: + logger.error("PipeWire reconnection failed: %s", exc) + return False + return True + + #: Seconds a chunk read may wait for samples before the stream is + #: declared stalled. pw-record streams continuously — even a silent sink + #: emits zeros — so a long gap means the process is wedged, not quiet. + _STALL_TIMEOUT_SECONDS = 5.0 + + def read_chunk(self) -> bytes: + """Return one 1024-frame chunk of mono int16 PCM. + + Raises IOError when the process dies or when no samples arrive within + ``_STALL_TIMEOUT_SECONDS`` — a blocking ``read`` would keep the + capture thread (and ``_buffer_lock``) held forever on a wedged + pw-record, preventing a clean stop. + """ + process = self.stream + if process is None or process.stdout is None: + raise IOError("PipeWire capture stream is not open") + need = self._CHUNK_FRAMES * self.channels * 2 + try: + fd = process.stdout.fileno() + except (OSError, ValueError, AttributeError): + fd = None + if not isinstance(fd, int) or fd < 0: + fd = None + deadline = time.monotonic() + self._STALL_TIMEOUT_SECONDS + parts: List[bytes] = [] + remaining = need + while remaining > 0: + left = deadline - time.monotonic() + if left <= 0: + code = process.poll() + raise IOError( + "pw-record stalled: no audio for " + f"{self._STALL_TIMEOUT_SECONDS:.0f}s (exit status {code})" + ) + if fd is not None: + try: + ready, _, _ = select.select([fd], [], [], left) + except (OSError, ValueError) as exc: + raise IOError(f"PipeWire capture read failed: {exc}") from exc + if not ready: + continue + try: + chunk = os.read(fd, remaining) + except OSError as exc: + raise IOError(f"PipeWire capture read failed: {exc}") from exc + else: + # File-like substitutes without a descriptor (tests) cannot + # block, so they read directly. + try: + chunk = process.stdout.read(remaining) + except (OSError, ValueError) as exc: + raise IOError(f"PipeWire capture read failed: {exc}") from exc + if not chunk: + code = process.poll() + raise IOError(f"pw-record stopped streaming (exit status {code})") + parts.append(chunk) + remaining -= len(chunk) + return b"".join(parts) + + def close(self) -> None: + """Terminate the process, reap it (killing if needed), release the pipe.""" + process = self.stream + if process is None: + return + try: + if process.poll() is None: + process.terminate() + except OSError: + pass + try: + process.wait(timeout=1.0) + except subprocess.TimeoutExpired: + try: + process.kill() + except OSError as exc: + logger.warning("Could not kill stalled pw-record: %s", exc) + try: + process.wait(timeout=1.0) + except subprocess.TimeoutExpired: + logger.warning("pw-record did not exit after kill") + except OSError as exc: + logger.warning("Error waiting for pw-record after kill: %s", exc) + except OSError as exc: + logger.warning("Error waiting for pw-record to exit: %s", exc) + if process.stdout is not None: + try: + process.stdout.close() + except OSError as exc: + logger.debug("Error closing pw-record stdout: %s", exc) + self.stream = None + + +def _test_pipewire_input(device_index: int, duration: float = 1.0) -> dict: + """ + Test capture from a PipeWire system-audio source. + + Args: + device_index: Negative PipeWire source index (see PIPEWIRE_INDEX_BASE) + duration: How long to record in seconds + + Returns: + Dictionary in the same shape as :func:`test_audio_input`. + """ + result = { + "success": False, + "device_name": "Unknown", + "device_index": device_index, + "sample_count": 0, + "max_amplitude": 0.0, + "mean_amplitude": 0.0, + "has_signal": False, + "error": None, + } + + if resolve_pipewire_source(device_index, None) is None: + result["error"] = "PipeWire system-audio source is not available" + return result + + source = PipeWireCaptureSource(device_index=device_index) + try: + import numpy as np + + source.open() + result["device_name"] = source.source.display_name if source.source else "Unknown" + result["sample_rate"] = source.sample_rate + all_amplitudes = [] + chunks_to_read = max(1, int(source.sample_rate * duration / source._CHUNK_FRAMES)) + for _ in range(chunks_to_read): + data = source.read_chunk() + audio_data = np.frombuffer(data, dtype=np.int16) + all_amplitudes.extend(np.abs(audio_data)) + + if all_amplitudes: + all_amplitudes = np.array(all_amplitudes) + max_amplitude = float(np.max(all_amplitudes)) + result["success"] = True + result["sample_count"] = len(all_amplitudes) + result["max_amplitude"] = max_amplitude + result["mean_amplitude"] = float(np.mean(all_amplitudes)) + result["has_signal"] = max_amplitude > 200 + except (IOError, OSError) as e: + result["error"] = f"PipeWire capture failed: {e}" + except ImportError as e: + result["error"] = f"Missing dependency: {e}" + finally: + source.close() + + return result diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 853d96b..f31c329 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -18,7 +18,7 @@ import sys import threading import time from pathlib import Path -from typing import TYPE_CHECKING, Callable, Optional +from typing import TYPE_CHECKING, Callable, Optional, Union if TYPE_CHECKING: import numpy as np @@ -42,6 +42,7 @@ from ..audio.capture import ( # noqa: F401 (re-exported for compatibility) get_audio_input_devices, test_audio_input, ) +from ..audio.pipewire import PipeWireCaptureSource, is_pipewire_device from ..audio.playback_ducker import default_dictation_duck_session, duck_delay_seconds from ..common_types import RecognitionState from ..ui.audio_feedback import play_error_sound, play_start_sound, play_stop_sound @@ -517,7 +518,7 @@ class SpeechRecognitionManager: self._capture_channels = 1 # Default, updated when device is opened self._capture_downmix_channel = None # Speech-gated sticky N>=3 channel for open stream # The active capture source while a recording session is open. - self._capture_source: Optional[PortAudioCaptureSource] = None + self._capture_source: Optional[Union[PortAudioCaptureSource, PipeWireCaptureSource]] = None # Recover a sink left quiet by a crash before doing anything slow. # Tests inject a session so this never touches a real audio server. @@ -2978,6 +2979,10 @@ class SpeechRecognitionManager: try: if not self._playback_duck.enabled(): return + if is_pipewire_device(self.audio_device_index, self.audio_device_name): + # Ducking lowers the default sink — the very audio a + # system-audio source is capturing. + return # The capture thread may already have failed and released the duck; # arming now would lower playback in the error state with nothing # left to put it back. @@ -3289,7 +3294,6 @@ class SpeechRecognitionManager: try: import numpy as np - import pyaudio except ImportError as e: logger.error(f"Failed to import required audio libraries: {e}") logger.error("Please install required dependencies: pip install pyaudio numpy") @@ -3305,17 +3309,29 @@ class SpeechRecognitionManager: # PyAudio configuration CHUNK = 1024 - # Initialize PyAudio with reconnection support - self._pyaudio_instance = pyaudio.PyAudio() - audio = self._pyaudio_instance - # The capture source owns the device: resolution, format - # negotiation, downmixing, and resampling. This loop only consumes - # mono 16 kHz chunks and applies dictation segmentation policy. - source = PortAudioCaptureSource( - device_index=self.audio_device_index, - device_name=self.audio_device_name, - ) + # negotiation, downmixing, and resampling. PipeWire sources spawn + # pw-record themselves and need no PyAudio instance; PortAudio + # sources get one passed to open()/reopen(). This loop only + # consumes mono 16 kHz chunks and applies dictation segmentation + # policy. + source = self._new_capture_source() + audio = None + if getattr(source, "requires_pyaudio", True): + try: + import pyaudio + except ImportError as e: + logger.error(f"Failed to import required audio libraries: {e}") + logger.error("Please install required dependencies: pip install pyaudio numpy") + self.should_record = False + self.release_playback_duck() + play_error_sound() + self._buffered_capture_failed = True + self._update_state(RecognitionState.ERROR) + self._signal_buffered_capture_done() + return + self._pyaudio_instance = pyaudio.PyAudio() + audio = self._pyaudio_instance self._capture_source = source # The attempt count belongs to this session — a previous thread # may still be finishing and must not leave its retries here. @@ -3338,7 +3354,8 @@ class SpeechRecognitionManager: self.should_record = False self.release_playback_duck() play_error_sound() - audio.terminate() + if audio is not None: + audio.terminate() self._buffered_capture_failed = True self._update_state(RecognitionState.ERROR) return @@ -3504,6 +3521,10 @@ class SpeechRecognitionManager: if self._attempt_audio_reconnection(audio): logger.info("Audio reconnection successful, continuing recording") + # The session source may have been rebuilt (e.g. + # the configured device switched between a mic and + # a PipeWire sink) — keep reading the new source. + source = self._capture_source continue # Continue recording with the reopened source else: logger.error("Audio reconnection failed, stopping recording") @@ -3961,18 +3982,36 @@ class SpeechRecognitionManager: # If only VOSK params changed, just log it logger.info("Applied VAD/silence timeout changes.") - def _capture_source_for_session(self) -> PortAudioCaptureSource: + def _new_capture_source(self) -> Union[PortAudioCaptureSource, PipeWireCaptureSource]: + """Build the capture source matching the configured device.""" + if is_pipewire_device(self.audio_device_index, self.audio_device_name): + return PipeWireCaptureSource( + device_index=self.audio_device_index, + device_name=self.audio_device_name, + ) + return PortAudioCaptureSource( + device_index=self.audio_device_index, + device_name=self.audio_device_name, + ) + + def _capture_source_for_session(self) -> Union[PortAudioCaptureSource, PipeWireCaptureSource]: """Return the session's capture source, creating one when absent. Device selection is refreshed from the manager's configured device so - reconnections track the current setting. + reconnections track the current setting. The source is rebuilt when + the configured device switched families (PortAudio <-> PipeWire). """ + want_pipewire = is_pipewire_device( + getattr(self, "audio_device_index", None), + getattr(self, "audio_device_name", None), + ) source = getattr(self, "_capture_source", None) - if source is None: - source = PortAudioCaptureSource() + if source is None or isinstance(source, PipeWireCaptureSource) != want_pipewire: + source = self._new_capture_source() self._capture_source = source - source.device_index = getattr(self, "audio_device_index", None) - source.device_name = getattr(self, "audio_device_name", None) + else: + source.device_index = getattr(self, "audio_device_index", None) + source.device_name = getattr(self, "audio_device_name", None) return source def _sync_capture_state(self) -> None: @@ -4018,7 +4057,21 @@ class SpeechRecognitionManager: # The source may not know about the stream it is asked to replace # (e.g. when tests drive this method directly). source.stream = getattr(self, "_audio_stream", None) - ok = source.reopen(audio_instance) + audio = audio_instance + if getattr(source, "requires_pyaudio", True) and audio is None: + # A session that started on a PipeWire sink never built a PyAudio + # instance; a PortAudio source still needs one to reopen. + audio = getattr(self, "_pyaudio_instance", None) + if audio is None: + try: + import pyaudio + + audio = pyaudio.PyAudio() + self._pyaudio_instance = audio + except (ImportError, OSError, AttributeError) as e: + logger.error(f"Failed to initialize PyAudio for reconnection: {e}") + return False + ok = source.reopen(audio) self._sync_capture_state() return ok diff --git a/tests/conftest.py b/tests/conftest.py index 0d81a20..1bc61d5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,6 +10,7 @@ import subprocess import sys import tempfile import threading +from collections.abc import Generator from unittest.mock import MagicMock, patch import pytest @@ -151,6 +152,23 @@ def _no_remote_desktop_portal(monkeypatch): monkeypatch.setattr("vocalinux.text_injection.remote_desktop_portal.PORTAL_AVAILABLE", False) +@pytest.fixture(autouse=True) +def _no_real_pipewire_enumeration() -> Generator[None, None, None]: + """Keep host PipeWire enumeration out of unit tests. + + get_audio_input_devices appends the machine's real PipeWire sinks when + pw-dump is present and a daemon answers; tests exercising PipeWire + sources patch the seam with the sources they want. + """ + try: + from vocalinux.audio import capture as _capture + except ImportError: + yield + return + with patch.object(_capture, "get_system_audio_sources", return_value=[]): + yield + + @pytest.fixture(autouse=True) def _reset_shared_config_manager(): """Drop the process-wide ConfigManager between tests. diff --git a/tests/test_pipewire_capture.py b/tests/test_pipewire_capture.py new file mode 100644 index 0000000..6eada40 --- /dev/null +++ b/tests/test_pipewire_capture.py @@ -0,0 +1,686 @@ +""" +Tests for PipeWire system-audio capture (audio/pipewire.py) and its +integration into recognition_manager (issue #751). +""" + +import io +import json +import os +import sys +import unittest +from typing import Any, Callable, Dict, List, Optional, Tuple +from unittest.mock import MagicMock, patch + +from vocalinux.audio.pipewire import ( + PIPEWIRE_INDEX_BASE, + SYSTEM_AUDIO_PREFIX, + PipeWireCaptureSource, + PipeWireSource, + _parse_pw_dump, + _test_pipewire_input, + get_system_audio_sources, + is_pipewire_device, + is_pipewire_device_index, + pipewire_available, + resolve_pipewire_source, +) +from vocalinux.common_types import RecognitionState +from vocalinux.speech_recognition.recognition_manager import ( + SpeechRecognitionManager, + get_audio_input_devices, +) +from vocalinux.speech_recognition.recognition_manager import ( + test_audio_input as _run_test_audio_input, +) + + +def _node( + object_id: int, serial: int, media_class: str, node_name: str, description: str +) -> Dict[str, Any]: + return { + "id": object_id, + "type": "PipeWire:Interface:Node", + "info": { + "props": { + "media.class": media_class, + "node.name": node_name, + "node.description": description, + "object.serial": serial, + } + }, + } + + +PW_DUMP = [ + _node(30, 30, "Audio/Source", "alsa_input.mic", "USB Microphone"), + _node(41, 41, "Audio/Sink", "alsa_output.dac", "Built-in Audio Analog Stereo"), + _node(55, 55, "Audio/Sink/Virtual", "obs_sink", "OBS Output"), + _node(60, 60, "Stream/Output/Audio", "Firefox", "Firefox"), + { + "id": 10, + "type": "PipeWire:Interface:Metadata", + "info": { + "metadata": [ + { + "subject": 0, + "key": "default.audio.sink", + "value": {"name": "alsa_output.dac"}, + }, + { + "subject": 0, + "key": "default.audio.source", + "value": {"name": "alsa_input.mic"}, + }, + ] + }, + }, +] + + +def _source( + node_name: str = "alsa_output.dac", + description: str = "Built-in Audio Analog Stereo", + **kw: Any, +) -> PipeWireSource: + return PipeWireSource( + node_name=node_name, + description=description, + serial=kw.pop("serial", 41), + **kw, + ) + + +class _FakeProcess: + """Minimal Popen stand-in for PipeWireCaptureSource tests.""" + + def __init__(self, payload: bytes = b"", exit_code: Optional[int] = None) -> None: + self.stdout = io.BytesIO(payload) + self._exit_code = exit_code + self.terminated = False + self.killed = False + self.waited = False + + def poll(self) -> Optional[int]: + return self._exit_code + + def terminate(self) -> None: + self.terminated = True + self._exit_code = 0 + + def kill(self) -> None: + self.killed = True + self._exit_code = -9 + + def wait(self, timeout: Optional[float] = None) -> int: + self.waited = True + return self._exit_code if self._exit_code is not None else 0 + + +def _popen_factory( + process: _FakeProcess, calls: List[Tuple[List[str], Dict[str, Any]]] +) -> Callable[..., _FakeProcess]: + def _fake_popen(args: List[str], **kwargs: Any) -> _FakeProcess: + calls.append((args, kwargs)) + return process + + return _fake_popen + + +def _make_manager(**kw: Any) -> SpeechRecognitionManager: + """Create a manager with engine init patched out.""" + with patch.object(SpeechRecognitionManager, "_init_vosk"): + with patch.object(SpeechRecognitionManager, "_init_whisper"): + with patch.object(SpeechRecognitionManager, "_init_whispercpp"): + return SpeechRecognitionManager( + engine="whisper_cpp", + model_size="small", + language="en-us", + defer_download=True, + **kw, + ) + + +class TestParsePwDump(unittest.TestCase): + def test_sinks_only_sorted_by_serial(self) -> None: + sources = _parse_pw_dump(PW_DUMP) + names = [s.node_name for s in sources] + # Mic sources and app streams are excluded; order follows serials. + assert names == ["alsa_output.dac", "obs_sink"] + + def test_default_sink_flagged(self) -> None: + sources = _parse_pw_dump(PW_DUMP) + by_name = {s.node_name: s for s in sources} + assert by_name["alsa_output.dac"].is_default is True + assert by_name["obs_sink"].is_default is False + + def test_display_name_uses_prefix(self) -> None: + sources = _parse_pw_dump(PW_DUMP) + assert sources[0].display_name == f"{SYSTEM_AUDIO_PREFIX}Built-in Audio Analog Stereo" + + def test_duplicate_descriptions_get_disambiguated(self) -> None: + objects = [ + _node(1, 1, "Audio/Sink", "sink_a", "Speakers"), + _node(2, 2, "Audio/Sink", "sink_b", "Speakers"), + ] + sources = _parse_pw_dump(objects) + labels = sorted(s.label for s in sources) + assert labels == ["Speakers (sink_a)", "Speakers (sink_b)"] + + def test_missing_description_falls_back_to_node_name(self) -> None: + objects = [ + { + "id": 1, + "type": "PipeWire:Interface:Node", + "info": { + "props": { + "media.class": "Audio/Sink", + "node.name": "sink_x", + "object.serial": 1, + } + }, + } + ] + sources = _parse_pw_dump(objects) + assert sources[0].display_name == f"{SYSTEM_AUDIO_PREFIX}sink_x" + + def test_non_list_and_malformed_entries(self) -> None: + assert _parse_pw_dump("not a list") == [] + assert _parse_pw_dump([None, "x", {"type": "PipeWire:Interface:Node"}]) == [] + + +class TestGetSystemAudioSources(unittest.TestCase): + def test_missing_pw_dump_returns_empty(self) -> None: + assert get_system_audio_sources(which=lambda name: None) == [] + + def test_missing_pw_record_returns_empty(self) -> None: + # pw-dump alone cannot capture — without pw-record the picker must + # not offer sources that always fail to open. + def _which(name: str) -> Optional[str]: + return "/usr/bin/pw-dump" if name == "pw-dump" else None + + assert get_system_audio_sources(which=_which) == [] + + def test_nonzero_exit_returns_empty(self) -> None: + sources = get_system_audio_sources( + runner=lambda args: (1, "", "boom"), which=lambda name: "/usr/bin/pw-dump" + ) + assert sources == [] + + def test_invalid_json_returns_empty(self) -> None: + sources = get_system_audio_sources( + runner=lambda args: (0, "{nope", ""), which=lambda name: "/usr/bin/pw-dump" + ) + assert sources == [] + + def test_runner_exception_returns_empty(self) -> None: + def _boom(args: List[str]) -> Tuple[int, str, str]: + raise OSError("spawn failed") + + sources = get_system_audio_sources(runner=_boom, which=lambda name: "/usr/bin/pw-dump") + assert sources == [] + + def test_happy_path_parses_dump(self) -> None: + sources = get_system_audio_sources( + runner=lambda args: (0, json.dumps(PW_DUMP), ""), + which=lambda name: "/usr/bin/pw-dump", + ) + assert [s.node_name for s in sources] == ["alsa_output.dac", "obs_sink"] + + +class TestDevicePredicates(unittest.TestCase): + def test_is_pipewire_device_index(self) -> None: + assert is_pipewire_device_index(PIPEWIRE_INDEX_BASE) + assert is_pipewire_device_index(-3) + assert not is_pipewire_device_index(-1) # -1 is PortAudio "system default" + assert not is_pipewire_device_index(0) + assert not is_pipewire_device_index(None) + assert not is_pipewire_device_index(True) + + def test_is_pipewire_device(self) -> None: + assert is_pipewire_device(-2, None) + assert is_pipewire_device(None, f"{SYSTEM_AUDIO_PREFIX}Speakers") + assert is_pipewire_device(3, f"{SYSTEM_AUDIO_PREFIX}Speakers") + assert not is_pipewire_device(3, "USB Microphone") + assert not is_pipewire_device(None, None) + + +class TestResolvePipewireSource(unittest.TestCase): + def setUp(self) -> None: + self.sources = [ + _source("alsa_output.dac", "DAC", serial=41), + _source("obs_sink", "OBS Output", serial=55), + ] + + def test_resolves_by_display_name(self) -> None: + name = self.sources[1].display_name + assert resolve_pipewire_source(None, name, self.sources) is self.sources[1] + + def test_resolves_by_index_position(self) -> None: + assert resolve_pipewire_source(PIPEWIRE_INDEX_BASE, None, self.sources) is self.sources[0] + assert ( + resolve_pipewire_source(PIPEWIRE_INDEX_BASE - 1, None, self.sources) is self.sources[1] + ) + + def test_name_wins_over_index(self) -> None: + name = self.sources[1].display_name + # Stale index pointing at position 0, name pointing at OBS. + assert resolve_pipewire_source(PIPEWIRE_INDEX_BASE, name, self.sources) is self.sources[1] + + def test_out_of_range_and_missing(self) -> None: + assert resolve_pipewire_source(PIPEWIRE_INDEX_BASE - 9, None, self.sources) is None + assert resolve_pipewire_source(PIPEWIRE_INDEX_BASE, "System audio: Gone", []) is None + assert resolve_pipewire_source(None, None, self.sources) is None + + def test_unresolved_name_never_falls_back_to_index(self) -> None: + # The named sink is gone; another sink now sits at the stored index + # position. Index fallback would silently capture the wrong device. + assert ( + resolve_pipewire_source(PIPEWIRE_INDEX_BASE, "System audio: Gone", self.sources) is None + ) + + def test_duplicate_description_disambiguated_by_position(self) -> None: + # A second sink with the same description was added after the name + # was stored; its higher serial sorts it last, so the stored index + # position still points at the originally selected sink. + dup_a = _source("sink_a", "Speakers", serial=10, label="Speakers (sink_a)") + dup_b = _source("sink_b", "Speakers", serial=20, label="Speakers (sink_b)") + sources = [dup_a, dup_b] + assert ( + resolve_pipewire_source(PIPEWIRE_INDEX_BASE, "System audio: Speakers", sources) is dup_a + ) + assert ( + resolve_pipewire_source(PIPEWIRE_INDEX_BASE - 1, "System audio: Speakers", sources) + is dup_b + ) + + def test_ambiguous_description_without_position_returns_none(self) -> None: + # Labels as _parse_pw_dump emits them once descriptions collide. + dup_a = _source("sink_a", "Speakers", serial=10, label="Speakers (sink_a)") + dup_b = _source("sink_b", "Speakers", serial=20, label="Speakers (sink_b)") + sources = [dup_a, dup_b] + assert resolve_pipewire_source(None, "System audio: Speakers", sources) is None + # Index pointing at a non-candidate must not pick one anyway. + assert ( + resolve_pipewire_source(PIPEWIRE_INDEX_BASE - 5, "System audio: Speakers", sources) + is None + ) + + def test_node_name_suffix_survives_description_change(self) -> None: + renamed = _source("alsa_output.dac", "New DAC", serial=41, label="New DAC") + assert ( + resolve_pipewire_source(None, "System audio: DAC (alsa_output.dac)", [renamed]) + is renamed + ) + + def test_description_parens_not_parsed_as_node_name(self) -> None: + intended = _source("alsa_output.a", "Speakers (HDMI)", serial=10) + other = _source("HDMI", "Different sink", serial=20) + sources = [intended, other] + assert resolve_pipewire_source(None, "System audio: Speakers (HDMI)", sources) is intended + + +class TestPipewireAvailable(unittest.TestCase): + def test_requires_both_tools(self) -> None: + tools = {"pw-dump": "/usr/bin/pw-dump"} + assert not pipewire_available(which=tools.get) + tools["pw-record"] = "/usr/bin/pw-record" + assert pipewire_available(which=tools.get) + + +class TestPipeWireCaptureSource(unittest.TestCase): + def test_open_builds_pw_record_command(self) -> None: + process = _FakeProcess() + calls = [] + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(process, calls), + which=lambda n: "/usr/bin/pw-record", + ) + capture.open() + args, kwargs = calls[0] + assert args[:2] == ["pw-record", "--raw"] + assert "--target" in args + assert args[args.index("--target") + 1] == "alsa_output.dac" + assert args[-1] == "-" # raw PCM on stdout + assert kwargs["stdout"] is not None + assert "env" in kwargs # host binaries get the de-bundled environment + assert capture.stream is process + + def test_open_requires_pw_record(self) -> None: + capture = PipeWireCaptureSource(device_index=PIPEWIRE_INDEX_BASE, which=lambda name: None) + with self.assertRaises(FileNotFoundError): + capture.open() + + def test_open_raises_when_source_gone(self) -> None: + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=None): + capture = PipeWireCaptureSource(device_index=PIPEWIRE_INDEX_BASE, which=lambda n: "x") + with self.assertRaises(IOError): + capture.open() + + def test_read_chunk_returns_exact_frame_bytes(self) -> None: + payload = b"\x01\x02" * 4096 + process = _FakeProcess(payload) + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(process, []), + which=lambda n: "x", + ) + capture.open() + data = capture.read_chunk() + assert data == payload[:2048] + assert capture.read_chunk() == payload[2048:4096] + + def test_read_chunk_raises_ioerror_at_eof(self) -> None: + process = _FakeProcess(b"\x01\x02" * 512, exit_code=0) + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(process, []), + which=lambda n: "x", + ) + capture.open() + with self.assertRaises(IOError): + capture.read_chunk() + + def test_read_chunk_raises_when_not_open(self) -> None: + capture = PipeWireCaptureSource(device_index=PIPEWIRE_INDEX_BASE, which=lambda n: "x") + with self.assertRaises(IOError): + capture.read_chunk() + + def test_read_chunk_raises_when_process_died(self) -> None: + process = _FakeProcess(exit_code=1) + process.stdout = io.BytesIO(b"") # dead process yields EOF + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(process, []), + which=lambda n: "x", + ) + capture.open() + with self.assertRaises(IOError) as ctx: + capture.read_chunk() + assert "exit status" in str(ctx.exception) + + def test_read_chunk_raises_when_stream_stalls(self) -> None: + # A wedged pw-record that stays alive but emits nothing must not + # block the read forever — stop_recognition needs the lock the + # capture thread is holding. + read_fd, write_fd = os.pipe() + process = _FakeProcess() + process.stdout = os.fdopen(read_fd, "rb") + try: + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(process, []), + which=lambda n: "x", + ) + capture._STALL_TIMEOUT_SECONDS = 0.05 + capture.open() + with self.assertRaises(IOError) as ctx: + capture.read_chunk() + assert "stalled" in str(ctx.exception) + finally: + os.close(write_fd) + process.stdout.close() + + def test_close_terminates_process(self) -> None: + process = _FakeProcess() + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(process, []), + which=lambda n: "x", + ) + capture.open() + capture.close() + assert process.terminated + assert process.waited + assert capture.stream is None + # close is idempotent + capture.close() + + def test_reopen_respawns_process(self) -> None: + calls = [] + first, second = _FakeProcess(), _FakeProcess() + processes = iter([first, second]) + + def _factory(args: List[str], **kwargs: Any) -> _FakeProcess: + calls.append(args) + return next(processes) + + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, popen=_factory, which=lambda n: "x" + ) + capture.open() + assert capture.reopen(MagicMock()) is True + assert len(calls) == 2 + assert first.terminated + assert capture.stream is second + + def test_reopen_returns_false_when_source_gone(self) -> None: + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(_FakeProcess(), []), + which=lambda n: "x", + ) + with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=None): + assert capture.reopen(MagicMock()) is False + + +class TestManagerPipeWireIntegration(unittest.TestCase): + def test_get_audio_input_devices_appends_system_audio(self) -> None: + sources = [ + _source("alsa_output.dac", "DAC", serial=41, is_default=True), + _source("obs_sink", "OBS Output", serial=55), + ] + mock_audio = MagicMock() + mock_audio.get_device_count.return_value = 1 + mock_audio.get_default_input_device_info.return_value = {"index": 0} + mock_audio.get_device_info_by_index.return_value = { + "index": 0, + "name": "USB Microphone", + "maxInputChannels": 1, + } + mock_pyaudio = MagicMock(PyAudio=MagicMock(return_value=mock_audio)) + + with ( + patch.dict("sys.modules", {"pyaudio": mock_pyaudio}), + patch( + "vocalinux.audio.capture.get_system_audio_sources", + return_value=sources, + ), + ): + devices = get_audio_input_devices() + + assert devices[0] == (0, "USB Microphone", True) + assert devices[1] == (PIPEWIRE_INDEX_BASE, "System audio: DAC", True) + assert devices[2] == (PIPEWIRE_INDEX_BASE - 1, "System audio: OBS Output", False) + + def test_record_audio_uses_pipewire_source_for_system_audio(self) -> None: + manager = _make_manager( + audio_device_index=PIPEWIRE_INDEX_BASE, + audio_device_name="System audio: DAC", + ) + manager.should_record = True + manager.state = RecognitionState.LISTENING + manager.silence_timeout = 0.05 + manager._silero_vad = None + + fake_source = MagicMock() + fake_source.channels = 1 + fake_source.sample_rate = 16000 + fake_source.downmix_channel = None + fake_source.stream = MagicMock() + fake_source.audio = None + + def _read_once() -> bytes: + manager.should_record = False + return b"\x00" * 2048 + + fake_source.read_chunk.side_effect = _read_once + mock_audio = MagicMock() + mock_pyaudio = MagicMock(paInt16=8) + mock_pyaudio.PyAudio.return_value = mock_audio + + with ( + patch.dict("sys.modules", {"pyaudio": mock_pyaudio}), + patch( + "vocalinux.speech_recognition.recognition_manager.PipeWireCaptureSource", + return_value=fake_source, + ) as mock_source_cls, + patch( + "vocalinux.speech_recognition.recognition_manager.PortAudioCaptureSource" + ) as mock_portaudio_cls, + patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"), + ): + manager._record_audio() + + mock_source_cls.assert_called_once_with( + device_index=PIPEWIRE_INDEX_BASE, device_name="System audio: DAC" + ) + fake_source.open.assert_called_once_with(mock_audio) + fake_source.read_chunk.assert_called() + fake_source.close.assert_called_once_with() + mock_portaudio_cls.assert_not_called() + mock_audio.open.assert_not_called() + + def test_reconnect_reopens_pipewire_source(self) -> None: + manager = _make_manager( + audio_device_index=PIPEWIRE_INDEX_BASE, + audio_device_name="System audio: DAC", + ) + manager._reconnection_delay = 0 + manager._max_reconnection_attempts = 5 + + fake_source = MagicMock() + fake_source.sample_rate = 16000 + fake_source.channels = 1 + fake_source.downmix_channel = None + fake_source.stream = MagicMock() + fake_source.audio = None + fake_source.reopen.return_value = True + + with ( + patch( + "vocalinux.speech_recognition.recognition_manager.PipeWireCaptureSource", + return_value=fake_source, + ) as mock_source_cls, + patch( + "vocalinux.speech_recognition.recognition_manager.PortAudioCaptureSource" + ) as mock_portaudio_cls, + ): + assert manager._attempt_audio_reconnection(MagicMock()) is True + + mock_source_cls.assert_called_once_with( + device_index=PIPEWIRE_INDEX_BASE, device_name="System audio: DAC" + ) + fake_source.reopen.assert_called_once() + mock_portaudio_cls.assert_not_called() + assert manager._audio_stream is fake_source.stream + assert manager._capture_sample_rate == 16000 + assert manager._capture_channels == 1 + + def test_reconnect_fails_when_pipewire_source_gone(self) -> None: + manager = _make_manager( + audio_device_index=PIPEWIRE_INDEX_BASE, + audio_device_name="System audio: DAC", + ) + manager._reconnection_delay = 0 + manager._max_reconnection_attempts = 5 + + fake_source = MagicMock() + fake_source.reopen.return_value = False + fake_source.sample_rate = 16000 + fake_source.channels = 1 + fake_source.downmix_channel = None + fake_source.stream = None + fake_source.audio = None + + with ( + patch( + "vocalinux.speech_recognition.recognition_manager.PipeWireCaptureSource", + return_value=fake_source, + ), + patch( + "vocalinux.speech_recognition.recognition_manager.PortAudioCaptureSource" + ) as mock_portaudio_cls, + ): + # A vanished sink must not silently reopen a microphone. + assert manager._attempt_audio_reconnection(MagicMock()) is False + mock_portaudio_cls.assert_not_called() + + def test_audio_input_routes_negative_index_to_pipewire(self) -> None: + source = _source() + fake_capture = MagicMock() + fake_capture.sample_rate = 16000 + fake_capture.channels = 1 + fake_capture._CHUNK_FRAMES = 1024 + fake_capture.source = source + fake_capture.read_chunk.return_value = b"\xff\x7f" * 1024 # int16 max amplitude + + with ( + patch( + "vocalinux.audio.pipewire.resolve_pipewire_source", + return_value=source, + ), + patch( + "vocalinux.audio.pipewire.PipeWireCaptureSource", + return_value=fake_capture, + ), + ): + if isinstance(sys.modules.get("numpy"), MagicMock): + # Restore the stashed real module — re-importing it would + # leave sys.modules["numpy.*"] bound to an orphan. + real_numpy = getattr(sys, "_vocalinux_real_numpy", None) + if real_numpy is not None: + sys.modules["numpy"] = real_numpy + else: + del sys.modules["numpy"] + result = _run_test_audio_input(device_index=PIPEWIRE_INDEX_BASE, duration=0.1) + + assert result["success"] is True + assert result["device_name"] == "System audio: Built-in Audio Analog Stereo" + assert result["has_signal"] is True + assert result["error"] is None + + def test_pipewire_input_reports_missing_source(self) -> None: + with patch( + "vocalinux.audio.pipewire.resolve_pipewire_source", + return_value=None, + ): + result = _test_pipewire_input(PIPEWIRE_INDEX_BASE, duration=0.1) + assert result["success"] is False + assert result["error"] + + def test_playback_duck_skipped_for_system_audio(self) -> None: + manager = _make_manager( + audio_device_index=PIPEWIRE_INDEX_BASE, + audio_device_name="System audio: DAC", + ) + manager.should_record = True + manager.state = RecognitionState.LISTENING + manager._playback_duck = MagicMock() + manager._playback_duck.enabled.return_value = True + + manager._arm_playback_duck() + + manager._playback_duck.start.assert_not_called() + + def test_playback_duck_still_arms_for_microphone(self) -> None: + manager = _make_manager(audio_device_index=0, audio_device_name="USB Microphone") + manager.should_record = True + manager.state = RecognitionState.LISTENING + manager._playback_duck = MagicMock() + manager._playback_duck.enabled.return_value = True + + manager._arm_playback_duck() + + manager._playback_duck.start.assert_called_once() + + +if __name__ == "__main__": + unittest.main()