From b0d8afad921709837111726dd21ec6db43f0f136 Mon Sep 17 00:00:00 2001 From: Xavier Karma Date: Fri, 6 Nov 2026 18:35:19 +0000 Subject: [PATCH] fix(audio): never open the microphone for system audio, never swap a configured mic (#48) System-audio capture sets stream.capture.sink on the spawned pw-record, so PipeWire links only the sink monitor. An explicit microphone that is missing, renamed, virtual or output-only now fails with an Audio Input Unavailable notification instead of recording from the default input. Fixes #41 Fixes #42 --- docs/USER_GUIDE.md | 5 +- src/vocalinux/audio/capture.py | 116 ++++++++++---- src/vocalinux/audio/errors.py | 9 ++ src/vocalinux/audio/pipewire.py | 37 ++++- .../speech_recognition/recognition_manager.py | 22 ++- tests/test_capture_recovery.py | 76 ++++++++++ tests/test_pipewire_capture.py | 46 ++++++ tests/test_pulse_sources.py | 19 ++- .../test_recognition_manager_device_config.py | 141 +++++++++++++++--- tests/test_recognition_manager_downloads.py | 98 ++++++------ 10 files changed, 453 insertions(+), 116 deletions(-) diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index 827d50a..414e6f9 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -44,7 +44,10 @@ The clipboard copy behind these pastes is temporary: your previous clipboard com Open **Settings → Audio → Input Device** to select a specific microphone or choose **System Default**. A valid explicit microphone selection uses that device and -bypasses the System Default preference below. +bypasses the System Default preference below. If the selected microphone is +unplugged, renamed, or is a virtual device, dictation does not switch to another +input: Voculinux shows an "Audio Input Unavailable" notification and records +nothing. Choose a microphone again, or System Default, in Settings. With **System Default**, Voculinux leaves valid default or virtual inputs under the audio system's control rather than pinning a hardware device. diff --git a/src/vocalinux/audio/capture.py b/src/vocalinux/audio/capture.py index db7c69c..34bf392 100644 --- a/src/vocalinux/audio/capture.py +++ b/src/vocalinux/audio/capture.py @@ -17,7 +17,7 @@ import subprocess import time from typing import TYPE_CHECKING, Any, Callable, Optional, cast -from .errors import CaptureStalledError +from .errors import AudioDeviceUnavailableError, CaptureStalledError from .pipewire import ( PIPEWIRE_INDEX_BASE, _test_pipewire_input, @@ -427,7 +427,7 @@ def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = Non device_name = "unknown" logger.warning( "Configured audio device [%s] (%s) is not a safe input device. " - "Falling back to a valid input device.", + "Choosing another input device.", preferred_index, device_name, ) @@ -438,6 +438,86 @@ def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = Non return input_device_indices[0] +def _selects_system_default(device_index: Optional[int], device_name: Optional[str]) -> bool: + """Whether a stored selection means System Default. + + Settings stores System Default as -1 and clears both fields; an unnamed + index of -1 (PortAudio's own default sentinel) is the same choice. + """ + return not device_name and (device_index is None or device_index == -1) + + +def _configured_label(device_name: Optional[str], device_index: Optional[int]) -> str: + """Name a configured selection in user-facing messages.""" + label = device_name or f"index {device_index}" + return f"Configured audio device '{label}'" + + +def _require_usable_configured_input( + audio: Any, device_name: Optional[str], device_index: Optional[int] +) -> None: + """Refuse an explicit microphone that cannot be used exactly as configured. + + The device must be found by name (or by index when no name was stored) and + must be a real input. Matching follows :func:`_resolve_device_by_name`, so + a device accepted here is the one it returns. Returns quietly when device + enumeration itself fails, leaving the stored index to PortAudio as before. + + Raises: + AudioDeviceUnavailableError: the device is missing, renamed, virtual or + output-only. Nothing is opened, so no other input is recorded. + """ + label = _configured_label(device_name, device_index) + try: + device_count = int(audio.get_device_count()) + except (IOError, OSError, TypeError, ValueError, AttributeError): + return + + match: Optional[tuple[int, Any]] = None + for index in range(max(device_count, 0)): + try: + info = audio.get_device_info_by_index(index) + except (IOError, OSError, TypeError, ValueError, AttributeError): + continue + if device_name: + if isinstance(info, dict) and info.get("name", "") == device_name: + match = (index, info) + break + elif index == device_index: + match = (index, info) + break + + if match is None: + raise AudioDeviceUnavailableError( + f"{label} is not connected or was renamed; nothing was recorded from another input." + ) + _, info = match + if not isinstance(info, dict): + return # Cannot classify it; keep the same leniency as _resolve_valid_input_device. + name = str(info.get("name", "")) + channels = info.get("maxInputChannels", 0) + has_input = isinstance(channels, (int, float)) and channels > 0 + if _is_virtual_device(name) or not has_input: + raise AudioDeviceUnavailableError( + f"{label} is not a usable microphone (virtual or output-only); " + "nothing was recorded from another input." + ) + + +def _resolve_configured_input( + audio: Any, device_name: Optional[str], device_index: Optional[int] +) -> int: + """Resolve an explicit microphone selection, or raise. Never substitutes a device.""" + _require_usable_configured_input(audio, device_name, device_index) + resolved = _resolve_device_by_name(audio, device_name, device_index) + if resolved is None: + raise AudioDeviceUnavailableError( + f"{_configured_label(device_name, device_index)} " + "could not be opened; nothing was recorded from another input." + ) + return resolved + + def _resolve_system_default(audio: Any) -> Optional[int]: """Prefer ALSA's sound-server PCM when PortAudio substituted hardware. @@ -971,25 +1051,16 @@ class PortAudioCaptureSource: # PulseAudio source opens the default device with PULSE_SOURCE set. pulse_selected, pulse_source = _resolve_pulse_selection(self.device_index, self.device_name) pulse_name = pulse_source.name if pulse_source is not None else None - use_system_default = pulse_selected or ( - self.device_index is None and self.device_name is None + use_system_default = pulse_selected or _selects_system_default( + self.device_index, self.device_name ) if use_system_default: resolved_device_index = _resolve_system_default(audio) else: - resolved_device_index = _resolve_device_by_name( + # An explicit selection is used exactly or refused (issue #42). + resolved_device_index = _resolve_configured_input( audio, self.device_name, self.device_index ) - if resolved_device_index is None: - resolved_device_index = _resolve_valid_input_device(audio, None) - if resolved_device_index is None and not use_system_default: - # No safe enumerated mic left (e.g. only PipeWire pseudo devices). - # Fall back to PortAudio system default instead of aborting. - logger.warning( - "No safe audio input devices enumerated; " "falling back to system default capture." - ) - resolved_device_index = _resolve_system_default(audio) - use_system_default = True # Log available devices for debugging (skip virtual devices) logger.debug("Available audio input devices:") @@ -1117,23 +1188,16 @@ class PortAudioCaptureSource: self.device_index, self.device_name ) pulse_name = pulse_source.name if pulse_source is not None else None - use_system_default = pulse_selected or ( - self.device_index is None and self.device_name is None + use_system_default = pulse_selected or _selects_system_default( + self.device_index, self.device_name ) if use_system_default: resolved_device_index = _resolve_system_default(audio_instance) else: - resolved_device_index = _resolve_device_by_name( + # An explicit selection is used exactly or refused (issue #42). + resolved_device_index = _resolve_configured_input( audio_instance, self.device_name, self.device_index ) - if resolved_device_index is None: - resolved_device_index = _resolve_valid_input_device(audio_instance, None) - if resolved_device_index is None and not use_system_default: - logger.warning( - "Reconnection: no safe input devices enumerated; " - "falling back to system default capture." - ) - resolved_device_index = _resolve_system_default(audio_instance) # Negotiate the format and open the capture stream in ONE # PortAudio open — Bluetooth SCO devices abort with heap diff --git a/src/vocalinux/audio/errors.py b/src/vocalinux/audio/errors.py index 72fe98f..ec3bded 100644 --- a/src/vocalinux/audio/errors.py +++ b/src/vocalinux/audio/errors.py @@ -3,3 +3,12 @@ class CaptureStalledError(IOError): """The capture device stopped delivering audio frames.""" + + +class AudioDeviceUnavailableError(IOError): + """The configured capture device cannot be used as configured. + + Raised instead of recording from another input: a configured device that + is missing, renamed, virtual or output-only must fail visibly (issue #42). + The message is user-facing and names the configured device. + """ diff --git a/src/vocalinux/audio/pipewire.py b/src/vocalinux/audio/pipewire.py index 50d9d47..0d6ca9a 100644 --- a/src/vocalinux/audio/pipewire.py +++ b/src/vocalinux/audio/pipewire.py @@ -25,10 +25,10 @@ import shutil import subprocess import time from dataclasses import dataclass -from typing import Any, Callable, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple from ..utils.host_process import host_env -from .errors import CaptureStalledError +from .errors import AudioDeviceUnavailableError, CaptureStalledError logger = logging.getLogger(__name__) @@ -52,6 +52,28 @@ _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") +#: Stream property that makes ``--target `` record the sink's monitor. +#: Without it PipeWire also links the default microphone to the stream (issue +#: #41), so system-audio capture must always set it. +_CAPTURE_SINK_PROPERTY = "stream.capture.sink = true" + + +def _capture_sink_env(base: Mapping[str, str]) -> Dict[str, str]: + """Return ``base`` with ``PIPEWIRE_PROPS`` pinning capture to the sink monitor. + + This is the form verified on the real machine (PipeWire 1.6.10): the stream + linked the sink's monitor and no microphone. A user's existing + ``PIPEWIRE_PROPS`` is kept, and only this child process sees the change. + """ + env = dict(base) + existing = env.get("PIPEWIRE_PROPS", "").strip() + if existing.startswith("{") and existing.endswith("}"): + existing = existing[1:-1].strip() + parts = [part for part in (existing, _CAPTURE_SINK_PROPERTY) if part] + env["PIPEWIRE_PROPS"] = "{ " + " ".join(parts) + " }" + return env + + _COMMAND_TIMEOUT_SECONDS = 3.0 CommandResult = Tuple[int, str, str] @@ -346,7 +368,8 @@ class PipeWireCaptureSource: Raises: FileNotFoundError: when ``pw-record`` is not installed. - IOError: when the stored selection no longer resolves to a source. + AudioDeviceUnavailableError: when the stored selection no longer + resolves to a source (an IOError subclass). """ if self.stream is not None: return @@ -356,7 +379,11 @@ class PipeWireCaptureSource: 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") + label = self.device_name or f"index {self.device_index}" + raise AudioDeviceUnavailableError( + f"System audio source '{label}' is no longer available; " + "nothing was recorded from another input." + ) self.source = source self.stream = self._popen( [ @@ -374,7 +401,7 @@ class PipeWireCaptureSource: ], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, - env=host_env(), + env=host_env(_capture_sink_env(os.environ)), ) def reopen( diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 7b45704..fb0fcb5 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -30,6 +30,7 @@ if TYPE_CHECKING: from ..app_identity import APP_NAME, RESOURCE_DIR_NAME, user_agent from ..audio.capture import ( # noqa: F401 (re-exported for compatibility) _STICKY_LOCK_MIN_MEAN_SQUARE, + AudioDeviceUnavailableError, CaptureStalledError, PortAudioCaptureSource, _alsa_handler, @@ -112,6 +113,9 @@ class _CaptureSession: self.stopping = False self.stall_recovery_attempted = False self.stall_error: Optional[OSError] = None + # A configured device that cannot be used (issue #42), kept for the + # failure notice when the read error that ended the session is not it. + self.unavailable_error: Optional[AudioDeviceUnavailableError] = None self.device_name = "System Default" self.segments: queue.Queue = queue.Queue(maxsize=32) # A dispatched callback can outlive the bounded recognition join. @@ -3955,6 +3959,10 @@ class SpeechRecognitionManager: def recover(error: OSError) -> bool: if not recording(): return False + if isinstance(error, AudioDeviceUnavailableError): + # A configured device that cannot be used does not recover by retrying. + session.unavailable_error = session.unavailable_error or error + return False if session.stall_error is None: session.device_name = self._capture_device_name(source) if isinstance(error, CaptureStalledError): @@ -3965,6 +3973,8 @@ class SpeechRecognitionManager: session.stall_recovery_attempted = True recovered = self._attempt_audio_reconnection(audio, should_continue=recording) validation_error = getattr(self._capture_source, "last_error", None) + if isinstance(validation_error, AudioDeviceUnavailableError): + session.unavailable_error = session.unavailable_error or validation_error if isinstance(validation_error, CaptureStalledError): session.stall_error = session.stall_error or validation_error session.stall_recovery_attempted = True @@ -4324,7 +4334,17 @@ class SpeechRecognitionManager: play_error_sound() if isinstance(error, CaptureStalledError): session.stall_error = session.stall_error or error - if session.stall_error is not None: + unavailable = session.unavailable_error + if unavailable is None and isinstance(error, AudioDeviceUnavailableError): + unavailable = error + if unavailable is not None: + logger.error("Audio input unavailable: %s", unavailable) + _show_notification( + "Audio Input Unavailable", + f"{unavailable} Choose a microphone in Settings, then try dictation again.", + "dialog-warning", + ) + elif session.stall_error is not None: logger.error("Microphone '%s' stalled: %s", session.device_name, session.stall_error) _show_notification( "Microphone Stalled", diff --git a/tests/test_capture_recovery.py b/tests/test_capture_recovery.py index 3e884a6..38a3311 100644 --- a/tests/test_capture_recovery.py +++ b/tests/test_capture_recovery.py @@ -385,3 +385,79 @@ def test_successful_reopen_updates_identity_for_a_later_stall( for stream in streams: stream.close.assert_called_once() audio.terminate.assert_called_once() + + +def test_unusable_configured_device_is_reported_without_retrying( + manager: rm.SpeechRecognitionManager, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A configured virtual mic fails visibly; it is not reopened or replaced (#42).""" + source = capture.PortAudioCaptureSource(device_index=0, device_name="pipewire") + devices = [ + {"index": 0, "name": "pipewire", "maxInputChannels": 32}, + {"index": 1, "name": "Built-in Mic", "maxInputChannels": 2}, + ] + audio = Mock() + audio.get_device_count.return_value = 2 + audio.get_device_info_by_index.side_effect = lambda i: devices[i] + audio.get_default_input_device_info.return_value = {"index": 1, "name": "Built-in Mic"} + monkeypatch.setitem(sys.modules, "pyaudio", Mock(PyAudio=lambda: audio, paInt16=8)) + monkeypatch.setattr(manager, "_new_capture_source", lambda: source) + opener = Mock(return_value=(1, 16000, Mock())) + monkeypatch.setattr(capture, "_open_capture_stream", opener) + reconnect = Mock(return_value=False) + monkeypatch.setattr(manager, "_attempt_audio_reconnection", reconnect) + manager.should_record = True + manager.state = RecognitionState.LISTENING + + manager._record_audio() + + opener.assert_not_called() + audio.open.assert_not_called() + reconnect.assert_not_called() + title, message = rm._show_notification.call_args.args[:2] + assert title == "Audio Input Unavailable" + assert "'pipewire'" in message + assert "Choose a microphone in Settings" in message + rm.play_error_sound.assert_called() + assert manager.state == RecognitionState.IDLE + + +def test_configured_device_lost_mid_dictation_is_reported( + manager: rm.SpeechRecognitionManager, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A read error followed by a refused reconnect still reports the cause (#42).""" + source = capture.PortAudioCaptureSource(device_index=0, device_name="USB Microphone") + devices = { + 0: {"index": 0, "name": "USB Microphone", "maxInputChannels": 1}, + 1: {"index": 1, "name": "Built-in Mic", "maxInputChannels": 2}, + } + audio = Mock() + audio.get_device_count.return_value = 2 + audio.get_device_info_by_index.side_effect = lambda i: devices[i] + audio.get_default_input_device_info.return_value = {"index": 1, "name": "Built-in Mic"} + monkeypatch.setitem(sys.modules, "pyaudio", Mock(PyAudio=lambda: audio, paInt16=8)) + monkeypatch.setattr(manager, "_new_capture_source", lambda: source) + opener = Mock(return_value=(1, 16000, Mock())) + monkeypatch.setattr(capture, "_open_capture_stream", opener) + + def read(**_kwargs: object) -> bytes: + # The microphone is renamed while dictating: the reconnect must refuse it. + devices[0] = {"index": 0, "name": "Renamed USB", "maxInputChannels": 1} + raise OSError("device lost") + + monkeypatch.setattr(source, "read_chunk", read) + # Reconnection rebuilds the source from the manager's configured device. + manager.audio_device_index = 0 + manager.audio_device_name = "USB Microphone" + manager.should_record = True + manager.state = RecognitionState.LISTENING + + manager._record_audio() + + opener.assert_called_once() + title, message = rm._show_notification.call_args.args[:2] + assert title == "Audio Input Unavailable" + assert "'USB Microphone'" in message + assert manager.state == RecognitionState.IDLE diff --git a/tests/test_pipewire_capture.py b/tests/test_pipewire_capture.py index c2b1d01..e2fda6e 100644 --- a/tests/test_pipewire_capture.py +++ b/tests/test_pipewire_capture.py @@ -11,6 +11,7 @@ import unittest from typing import Any, Callable, Dict, List, Optional, Tuple from unittest.mock import MagicMock, patch +from vocalinux.audio.errors import AudioDeviceUnavailableError from vocalinux.audio.pipewire import ( PIPEWIRE_INDEX_BASE, SYSTEM_AUDIO_PREFIX, @@ -347,6 +348,43 @@ class TestPipeWireCaptureSource(unittest.TestCase): assert "env" in kwargs # host binaries get the de-bundled environment assert capture.stream is process + def test_open_pins_capture_to_the_sink_monitor(self) -> None: + # Without stream.capture.sink, pw-record --target also links the + # default microphone (issue #41). Set it through PIPEWIRE_PROPS, the + # form verified on the real machine. + calls: List[Tuple[List[str], Dict[str, Any]]] = [] + with ( + patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()), + patch.dict("os.environ", {"PATH": "/usr/bin"}, clear=True), + ): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(_FakeProcess(), calls), + which=lambda n: "/usr/bin/pw-record", + ) + capture.open() + args, kwargs = calls[0] + assert args[args.index("--target") + 1] == "alsa_output.dac" + assert args[-1] == "-" + assert "stream.capture.sink = true" in kwargs["env"]["PIPEWIRE_PROPS"] + assert kwargs["env"]["PATH"] == "/usr/bin" + + def test_open_keeps_user_pipewire_props_when_pinning_the_monitor(self) -> None: + calls: List[Tuple[List[str], Dict[str, Any]]] = [] + user_props = "{ node.latency = 1024/48000 }" + with ( + patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()), + patch.dict("os.environ", {"PIPEWIRE_PROPS": user_props}, clear=True), + ): + capture = PipeWireCaptureSource( + device_index=PIPEWIRE_INDEX_BASE, + popen=_popen_factory(_FakeProcess(), calls), + which=lambda n: "/usr/bin/pw-record", + ) + capture.open() + props = calls[0][1]["env"]["PIPEWIRE_PROPS"] + assert props == "{ node.latency = 1024/48000 stream.capture.sink = true }" + def test_open_requires_pw_record(self) -> None: capture = PipeWireCaptureSource(device_index=PIPEWIRE_INDEX_BASE, which=lambda name: None) with self.assertRaises(FileNotFoundError): @@ -358,6 +396,14 @@ class TestPipeWireCaptureSource(unittest.TestCase): with self.assertRaises(IOError): capture.open() + def test_open_reports_a_gone_source_as_unavailable(self) -> None: + # No PortAudio fallback: the typed error lets the manager tell the user. + 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(AudioDeviceUnavailableError) as ctx: + capture.open() + assert "no longer available" in str(ctx.exception) + def test_read_chunk_returns_exact_frame_bytes(self) -> None: payload = b"\x01\x02" * 4096 process = _FakeProcess(payload) diff --git a/tests/test_pulse_sources.py b/tests/test_pulse_sources.py index 35a91aa..8b8ef6f 100644 --- a/tests/test_pulse_sources.py +++ b/tests/test_pulse_sources.py @@ -389,13 +389,13 @@ class TestPulseSourceEnv: assert leaked is None -def _pyaudio_with_one_mic() -> Tuple[MagicMock, MagicMock]: +def _pyaudio_with_one_mic(name: str = "default") -> Tuple[MagicMock, MagicMock]: mock_audio = MagicMock() mock_audio.get_device_count.return_value = 1 - mock_audio.get_default_input_device_info.return_value = {"index": 0, "name": "default"} + mock_audio.get_default_input_device_info.return_value = {"index": 0, "name": name} mock_audio.get_device_info_by_index.return_value = { "index": 0, - "name": "default", + "name": name, "maxInputChannels": 32, } return mock_audio, MagicMock(PyAudio=MagicMock(return_value=mock_audio)) @@ -434,9 +434,13 @@ class TestCaptureIntegration: assert devices == [(PULSE_INDEX_BASE, "Yeti", False)] def _open_recording_env( - self, device_index: int, device_name: str, sources: List[PulseSource] + self, + device_index: int, + device_name: str, + sources: List[PulseSource], + mic_name: str = "default", ) -> Tuple[Dict[str, object], Optional[str]]: - mock_audio, mock_pyaudio = _pyaudio_with_one_mic() + mock_audio, mock_pyaudio = _pyaudio_with_one_mic(mic_name) seen: Dict[str, object] = {} def fake_open(audio: Any, index: Optional[int]) -> Tuple[int, int, MagicMock]: @@ -467,7 +471,10 @@ class TestCaptureIntegration: assert leaked is None def test_portaudio_selection_is_untouched(self) -> None: - seen, _leaked = self._open_recording_env(0, "default", self.sources) + # A real configured microphone; an explicit virtual "default" is now refused (#42). + seen, _leaked = self._open_recording_env( + 0, "USB Microphone", self.sources, mic_name="USB Microphone" + ) assert seen["env"] is None def test_audio_test_uses_the_chosen_source(self) -> None: diff --git a/tests/test_recognition_manager_device_config.py b/tests/test_recognition_manager_device_config.py index f570697..73750dd 100644 --- a/tests/test_recognition_manager_device_config.py +++ b/tests/test_recognition_manager_device_config.py @@ -160,8 +160,8 @@ class TestAudioDeviceDetection(unittest.TestCase): mock_open.assert_called_once_with(mock_audio, None) - def test_record_audio_falls_back_to_system_default_when_no_safe_device(self): - """Stale/unsafe saved devices should reopen via PortAudio system default.""" + def test_record_audio_refuses_system_default_for_unusable_configured_device(self): + """An unusable configured mic must not reopen the system default (issue #42).""" manager = _make_manager( engine="whisper_cpp", audio_device_index=0, @@ -172,14 +172,6 @@ class TestAudioDeviceDetection(unittest.TestCase): manager.silence_timeout = 0.05 manager._silero_vad = None - mock_stream = MagicMock() - - def _read_once(*_args, **_kwargs): - manager.should_record = False - return b"\x00" * 2048 - - mock_stream.get_read_available.return_value = 1024 - mock_stream.read.side_effect = _read_once mock_audio = MagicMock() mock_audio.get_device_count.return_value = 1 mock_audio.get_device_info_by_index.return_value = { @@ -196,17 +188,9 @@ class TestAudioDeviceDetection(unittest.TestCase): with ( patch.dict("sys.modules", {"pyaudio": mock_pyaudio}), - patch( - "vocalinux.audio.capture._resolve_device_by_name", - return_value=None, - ), - patch( - "vocalinux.audio.capture._resolve_valid_input_device", - return_value=None, - ), patch( "vocalinux.audio.capture._open_capture_stream", - return_value=(1, 16000, mock_stream), + return_value=(1, 16000, MagicMock()), ) as mock_open, patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"), ): @@ -214,8 +198,9 @@ class TestAudioDeviceDetection(unittest.TestCase): del sys.modules["numpy"] manager._record_audio() - mock_open.assert_called_once_with(mock_audio, None) - mock_audio.get_default_input_device_info.assert_called() + mock_open.assert_not_called() + mock_audio.open.assert_not_called() + assert manager._buffered_capture_failed def test_test_audio_input_system_default_fallback_open_omits_device_index(self): """Fallback mic-test open for System Default must omit input_device_index.""" @@ -946,6 +931,120 @@ class TestDownmixToMono(unittest.TestCase): assert sticky3 == 2 +class TestExplicitMicrophoneSelection(unittest.TestCase): + """A configured microphone is recorded exactly, or not at all (issue #42).""" + + def _audio(self, devices): + audio = MagicMock() + audio.get_device_count.return_value = len(devices) + audio.get_device_info_by_index.side_effect = lambda i: devices[i] + audio.get_default_input_device_info.return_value = {"index": 2, "name": "Built-in Mic"} + return audio + + def _pyaudio_module(self): + module = MagicMock(paInt16=8) + module.PyAudio.return_value = MagicMock() + return module + + def _devices(self): + return [ + {"index": 0, "name": "pipewire", "maxInputChannels": 32}, + {"index": 1, "name": "USB Microphone", "maxInputChannels": 1}, + {"index": 2, "name": "Built-in Mic", "maxInputChannels": 2}, + ] + + def test_virtual_configured_device_is_refused_not_replaced(self): + """A configured virtual device must not open the hardware default instead.""" + from vocalinux.audio.capture import PortAudioCaptureSource + from vocalinux.audio.errors import AudioDeviceUnavailableError + + audio = self._audio(self._devices()) + source = PortAudioCaptureSource(device_index=0, device_name="pipewire") + with ( + patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}), + patch( + "vocalinux.audio.capture._open_capture_stream", + return_value=(1, 16000, MagicMock()), + ) as opener, + ): + with self.assertRaises(AudioDeviceUnavailableError) as ctx: + source.open(audio) + opener.assert_not_called() + audio.open.assert_not_called() + assert "'pipewire'" in str(ctx.exception) + + def test_configured_device_that_disappeared_is_not_switched(self): + """A missing configured mic must not reuse the stored index for another input.""" + from vocalinux.audio.capture import PortAudioCaptureSource + from vocalinux.audio.errors import AudioDeviceUnavailableError + + devices = self._devices() + devices[1] = {"index": 1, "name": "Built-in Mic", "maxInputChannels": 2} + devices[2] = {"index": 2, "name": "Other Mic", "maxInputChannels": 1} + audio = self._audio(devices) + source = PortAudioCaptureSource(device_index=1, device_name="USB Microphone") + with ( + patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}), + patch( + "vocalinux.audio.capture._open_capture_stream", + return_value=(1, 16000, MagicMock()), + ) as opener, + ): + with self.assertRaises(AudioDeviceUnavailableError): + source.open(audio) + opener.assert_not_called() + + def test_configured_usable_device_opens_by_name(self): + """A configured microphone that is present still opens at its index.""" + from vocalinux.audio.capture import PortAudioCaptureSource + + audio = self._audio(self._devices()) + source = PortAudioCaptureSource(device_index=1, device_name="USB Microphone") + with ( + patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}), + patch( + "vocalinux.audio.capture._open_capture_stream", + return_value=(1, 16000, MagicMock()), + ) as opener, + ): + source.open(audio) + opener.assert_called_once_with(audio, 1) + + def test_system_default_sentinel_still_opens_the_default_device(self): + """A stored -1 is PortAudio's System Default, not a device to look up.""" + from vocalinux.audio.capture import PortAudioCaptureSource + + audio = self._audio(self._devices()) + source = PortAudioCaptureSource(device_index=-1, device_name=None) + with ( + patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}), + patch( + "vocalinux.audio.capture._open_capture_stream", + return_value=(1, 16000, MagicMock()), + ) as opener, + ): + source.open(audio) + opener.assert_called_once_with(audio, None) + + def test_reopen_refuses_a_virtual_configured_device(self): + """Reconnection reports the refusal through last_error and opens nothing.""" + from vocalinux.audio.capture import PortAudioCaptureSource + from vocalinux.audio.errors import AudioDeviceUnavailableError + + audio = self._audio(self._devices()) + source = PortAudioCaptureSource(device_index=0, device_name="pipewire") + with ( + patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}), + patch( + "vocalinux.audio.capture._open_capture_stream", + return_value=(1, 16000, MagicMock()), + ) as opener, + ): + assert source.reopen(audio) is False + opener.assert_not_called() + assert isinstance(source.last_error, AudioDeviceUnavailableError) + + class TestRecordAudioNegotiationFallback(unittest.TestCase): """Cover _record_audio fallback when stream negotiation returns None.""" diff --git a/tests/test_recognition_manager_downloads.py b/tests/test_recognition_manager_downloads.py index 9200abf..f6546d3 100644 --- a/tests/test_recognition_manager_downloads.py +++ b/tests/test_recognition_manager_downloads.py @@ -1295,76 +1295,62 @@ class TestAudioReconnection: assert result is True assert manager._audio_stream == mock_stream - def test_attempt_audio_reconnection_falls_back_to_default_resolver(self): - """Test reconnection falls back when saved device name/index cannot resolve.""" + def test_attempt_audio_reconnection_refuses_missing_configured_device(self): + """A configured mic that is gone must fail visibly, not reopen the default (#42).""" manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic") - mock_pyaudio_mod = MagicMock() mock_pyaudio_mod.paInt16 = 8 - mock_stream = MagicMock() - mock_stream.get_read_available.return_value = 1024 - mock_stream.read.return_value = b"\x00" * 1024 mock_audio_instance = MagicMock() - mock_audio_instance.get_default_input_device_info.return_value = {"index": 0} - mock_audio_instance.open.return_value = mock_stream + mock_audio_instance.get_device_count.return_value = 1 + mock_audio_instance.get_device_info_by_index.return_value = { + "index": 0, + "name": "Built-in Mic", + "maxInputChannels": 2, + } + mock_audio_instance.get_default_input_device_info.return_value = { + "index": 0, + "name": "Built-in Mic", + } with ( patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}), patch("time.sleep"), - patch( - "vocalinux.audio.capture._resolve_device_by_name", - return_value=None, - ) as mock_resolve_name, - patch( - "vocalinux.audio.capture._resolve_valid_input_device", - return_value=1, - ) as mock_resolve_default, - patch( - "vocalinux.audio.capture._get_supported_channels", - return_value=1, - ), - patch( - "vocalinux.audio.capture._get_supported_sample_rate", - return_value=16000, - ), - ): - result = manager._attempt_audio_reconnection(mock_audio_instance) - - assert result is True - assert manager._audio_stream == mock_stream - mock_resolve_name.assert_called_once_with(mock_audio_instance, "Missing Mic", None) - mock_resolve_default.assert_called_once_with(mock_audio_instance, None) - - def test_attempt_audio_reconnection_no_resolved_device(self): - """When no safe device is enumerated, reconnect via system default.""" - manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic") - - mock_pyaudio_mod = MagicMock() - mock_pyaudio_mod.paInt16 = 8 - mock_audio_instance = MagicMock() - mock_stream = MagicMock() - mock_stream.get_read_available.return_value = 1024 - - with ( - patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}), - patch("time.sleep"), - patch( - "vocalinux.audio.capture._resolve_device_by_name", - return_value=None, - ), - patch( - "vocalinux.audio.capture._resolve_valid_input_device", - return_value=None, - ), patch( "vocalinux.audio.capture._open_capture_stream", - return_value=(1, 16000, mock_stream), + return_value=(1, 16000, MagicMock()), ) as mock_open, ): result = manager._attempt_audio_reconnection(mock_audio_instance) - assert result is True - mock_open.assert_called_once_with(mock_audio_instance, None) + assert result is False + mock_open.assert_not_called() + mock_audio_instance.open.assert_not_called() + + def test_attempt_audio_reconnection_refuses_when_no_safe_input_exists(self): + """With no usable input enumerated, reconnection refuses rather than using the default.""" + manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic") + mock_pyaudio_mod = MagicMock() + mock_pyaudio_mod.paInt16 = 8 + mock_audio_instance = MagicMock() + mock_audio_instance.get_device_count.return_value = 1 + mock_audio_instance.get_device_info_by_index.return_value = { + "index": 0, + "name": "pipewire", + "maxInputChannels": 32, + } + + with ( + patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}), + patch("time.sleep"), + patch( + "vocalinux.audio.capture._open_capture_stream", + return_value=(1, 16000, MagicMock()), + ) as mock_open, + ): + result = manager._attempt_audio_reconnection(mock_audio_instance) + + assert result is False + mock_open.assert_not_called() def test_attempt_audio_reconnection_max_attempts(self): """Test reconnection stops after max attempts."""