diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index 3164cdd..360352e 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -354,12 +354,16 @@ def main(): stop_sound_guard_ms = saved_settings.get("stop_sound_guard_ms", 200) voice_commands_enabled = saved_settings.get("voice_commands_enabled") # None = auto audio_device_index = audio_settings.get("device_index", None) + audio_device_name = audio_settings.get("device_name", None) advanced_settings = config_manager.get_settings().get("advanced", {}) logger.info(f"Final settings: engine={engine}, language={language}, model={model_size}") if audio_device_index is not None: - logger.info(f"Using audio device index={audio_device_index} (from saved config)") + logger.info( + f"Using audio device index={audio_device_index} " + f"(name={audio_device_name}, from saved config)" + ) # Initialize main components logger.info("Initializing Vocalinux...") @@ -375,6 +379,7 @@ def main(): stop_sound_guard_ms=stop_sound_guard_ms, voice_commands_enabled=voice_commands_enabled, audio_device_index=audio_device_index, + audio_device_name=audio_device_name, whispercpp_no_timestamps=advanced_settings.get("whispercpp_no_timestamps", True), whispercpp_no_context=advanced_settings.get("whispercpp_no_context", True), whispercpp_initial_prompt=advanced_settings.get("whispercpp_initial_prompt", ""), diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index a6ecaa3..3abe9c0 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -180,9 +180,36 @@ def _preload_pywhispercpp_shared_libraries() -> None: ) +def _is_virtual_device(device_name: str) -> bool: + """Check if a device name corresponds to a virtual audio device. + + Virtual devices (e.g. speech-dispatcher-dummy, PulseAudio monitors, + PipeWire null sinks) cannot be used for recording and may cause crashes + if PortAudio tries to open them. + + Args: + device_name: The device name to check. + + Returns: + True if the device appears to be virtual, False otherwise. + """ + if not device_name: + return False + name_lower = device_name.lower() + virtual_patterns = [ + "speech-dispatcher", + "speechdispatcher", + "dummy", + "null sink", + "monitor of", + "alsa_output", # ALSA output devices exposed as monitors + ] + return any(pattern in name_lower for pattern in virtual_patterns) + + def get_audio_input_devices() -> list: """ - Get a list of available audio input devices. + Get a list of available audio input devices, excluding virtual devices. Returns: List of tuples: (device_index, device_name, is_default) @@ -206,6 +233,10 @@ def get_audio_input_devices() -> list: # Only include devices that have input channels if info.get("maxInputChannels", 0) > 0: name = info.get("name", f"Device {i}") + # Skip virtual devices that can cause crashes + if _is_virtual_device(name): + logger.debug(f"Filtering virtual device [{i}]: {name}") + continue is_default = i == default_input_device devices.append((i, name, is_default)) except (IOError, OSError): @@ -220,6 +251,37 @@ def get_audio_input_devices() -> list: return devices +def _resolve_device_by_name( + audio, device_name: Optional[str], fallback_index: Optional[int] = None +) -> Optional[int]: + """Resolve a device index by name, falling back to index, then system default. + + Device indices shift when USB devices are replugged or virtual devices are + added/removed. Name matching is more stable. + """ + if not device_name: + return _resolve_valid_input_device(audio, fallback_index) + + try: + device_count = int(audio.get_device_count()) + except (IOError, OSError, TypeError, ValueError, AttributeError): + return _resolve_valid_input_device(audio, fallback_index) + + # ponytail: just find the index by name, delegate validation to _resolve_valid_input_device + for i in range(device_count): + try: + info = audio.get_device_info_by_index(i) + except (IOError, OSError, TypeError, ValueError, AttributeError): + continue + if isinstance(info, dict) and info.get("name", "") == device_name: + return _resolve_valid_input_device(audio, i) + + logger.warning( + f"Audio device '{device_name}' not found, falling back to index {fallback_index}" + ) + return _resolve_valid_input_device(audio, fallback_index) + + def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) -> Optional[int]: """Resolve a valid audio input device, skipping output-only devices (e.g. HDMI). @@ -765,6 +827,7 @@ class SpeechRecognitionManager: # Audio device selection (None means use system default) self.audio_device_index = kwargs.get("audio_device_index", None) + self.audio_device_name = kwargs.get("audio_device_name", None) # whisper.cpp advanced parameters self.whispercpp_no_timestamps = kwargs.get("whispercpp_no_timestamps", True) @@ -2026,21 +2089,30 @@ class SpeechRecognitionManager: except ValueError: pass - def set_audio_device(self, device_index: Optional[int]): + def set_audio_device(self, device_index: Optional[int], device_name: Optional[str] = None): """ Set the audio input device to use. Args: device_index: The device index to use, or None for system default + device_name: The device name (for stable re-resolution on next recording) """ - if device_index != self.audio_device_index: - logger.info(f"Audio device changed from {self.audio_device_index} to {device_index}") - self.audio_device_index = device_index + if device_index != self.audio_device_index or device_name != self.audio_device_name: + logger.info( + f"Audio device changed from {self.audio_device_index} " + f"to {device_index} (name: {device_name})" + ) + self.audio_device_index = device_index + self.audio_device_name = device_name def get_audio_device(self) -> Optional[int]: """Get the currently configured audio device index.""" return self.audio_device_index + def get_audio_device_name(self) -> Optional[str]: + """Get the currently configured audio device name.""" + return self.audio_device_name + def get_last_audio_level(self) -> float: """Get the last recorded audio level (0-100).""" return self._last_audio_level @@ -2199,8 +2271,14 @@ class SpeechRecognitionManager: self._pyaudio_instance = pyaudio.PyAudio() audio = self._pyaudio_instance - # Resolve a valid input device — skip output-only devices (e.g. HDMI) - resolved_device_index = _resolve_valid_input_device(audio, self.audio_device_index) + # Resolve the input device by name first (indices can shift between + # sessions due to USB replugging or virtual devices being added). + # Fall back to the stored index, then to the system default. + resolved_device_index = _resolve_device_by_name( + audio, self.audio_device_name, self.audio_device_index + ) + if resolved_device_index is None: + resolved_device_index = _resolve_valid_input_device(audio, None) if resolved_device_index is None: logger.error("No audio input devices found with input channels.") logger.error( @@ -2211,15 +2289,16 @@ class SpeechRecognitionManager: self._update_state(RecognitionState.ERROR) return - # Log available devices for debugging + # Log available devices for debugging (skip virtual devices) logger.debug("Available audio input devices:") for i in range(audio.get_device_count()): try: info = audio.get_device_info_by_index(i) if info.get("maxInputChannels", 0) > 0: - logger.debug( - f" [{i}] {info.get('name')} (inputs: {info.get('maxInputChannels')})" - ) + name = info.get("name", "") + if _is_virtual_device(name): + continue + logger.debug(f" [{i}] {name} (inputs: {info.get('maxInputChannels')})") except (IOError, OSError): continue @@ -2658,6 +2737,7 @@ class SpeechRecognitionManager: vad_sensitivity: Optional[int] = None, silence_timeout: Optional[float] = None, audio_device_index: Optional[int] = None, + audio_device_name: Optional[str] = None, force_download: bool = True, **kwargs, # Allow for future expansion ): @@ -2671,10 +2751,13 @@ class SpeechRecognitionManager: vad_sensitivity: New VAD sensitivity (for VOSK). silence_timeout: New silence timeout (for VOSK). audio_device_index: Audio input device index (None for default, -1 to clear). + audio_device_name: Audio device name for stable re-resolution. force_download: If True, download missing models (default: True for UI-triggered reconfigures). """ logger.info( - f"Reconfiguring speech engine. New settings: engine={engine}, model_size={model_size}, language={language}, vad={vad_sensitivity}, silence={silence_timeout}, audio_device={audio_device_index}" + f"Reconfiguring speech engine. New settings: engine={engine}, model_size={model_size}, " + f"language={language}, vad={vad_sensitivity}, silence={silence_timeout}, " + f"audio_device={audio_device_index}, audio_device_name={audio_device_name}" ) restart_needed = False @@ -2704,8 +2787,11 @@ class SpeechRecognitionManager: if audio_device_index is not None: if audio_device_index == -1: self.audio_device_index = None + self.audio_device_name = None else: self.audio_device_index = audio_device_index + if audio_device_name is not None: + self.audio_device_name = audio_device_name if "voice_commands_enabled" in kwargs: self._voice_commands_preference = kwargs.get("voice_commands_enabled") @@ -2828,10 +2914,12 @@ class SpeechRecognitionManager: except Exception as e: logger.debug(f"Error closing old audio stream: {e}") - # Resolve a valid input device — skip output-only devices (e.g. HDMI) - resolved_device_index = _resolve_valid_input_device( - audio_instance, self.audio_device_index + # Resolve a valid input device — by name first, then by index + resolved_device_index = _resolve_device_by_name( + audio_instance, self.audio_device_name, self.audio_device_index ) + if resolved_device_index is None: + resolved_device_index = _resolve_valid_input_device(audio_instance, None) if resolved_device_index is None: logger.error("Reconnection failed: no input devices available.") return False diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index 9ec00cd..305277a 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -3251,13 +3251,27 @@ For now, the engine has been reverted to VOSK.""" self.audio_device_combo.append(str(device_index), label) saved_device = self.config_manager.get_optional_int("audio", "device_index", None) + saved_device_name = self.config_manager.get("audio", "device_name", None) if saved_device is None: self.audio_device_combo.set_active_id("-1") else: - if not self.audio_device_combo.set_active_id(str(saved_device)): - logger.warning(f"Saved audio device {saved_device} no longer available") - self.audio_device_combo.set_active_id("-1") + # Try to match by saved device name first (more stable across reboots) + matched = False + if saved_device_name: + for idx, name, _ in devices: + if name == saved_device_name: + self.audio_device_combo.set_active_id(str(idx)) + matched = True + break + if not matched: + # Fall back to stored index + if not self.audio_device_combo.set_active_id(str(saved_device)): + logger.warning( + f"Saved audio device {saved_device} " + f"(name: {saved_device_name}) no longer available" + ) + self.audio_device_combo.set_active_id("-1") logger.info(f"Found {len(devices)} audio input devices") @@ -3288,9 +3302,9 @@ For now, the engine has been reverted to VOSK.""" self.config_manager.save_settings() if device_index == -1: - self.speech_engine.set_audio_device(None) + self.speech_engine.set_audio_device(None, None) else: - self.speech_engine.set_audio_device(device_index) + self.speech_engine.set_audio_device(device_index, device_name) logger.info(f"Audio device changed to: [{device_index}] {device_name}") self.audio_test_status.set_markup(f"Selected: {device_name}") diff --git a/tests/test_main.py b/tests/test_main.py index 53e9d95..c1e3a35 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -231,6 +231,7 @@ class TestMainModule(unittest.TestCase): stop_sound_guard_ms=200, voice_commands_enabled=None, audio_device_index=None, + audio_device_name=None, whispercpp_no_timestamps=True, whispercpp_no_context=True, whispercpp_initial_prompt="", diff --git a/tests/test_recognition_manager.py b/tests/test_recognition_manager.py index b392fb6..92d0ef6 100644 --- a/tests/test_recognition_manager.py +++ b/tests/test_recognition_manager.py @@ -518,6 +518,39 @@ class TestSpeechRecognition(unittest.TestCase): self.assertEqual(manager.silence_timeout, 1.5) self.assertEqual(manager.audio_device_index, 2) + def test_init_with_device_name(self): + """Test initialization with audio_device_name kwarg.""" + manager = SpeechRecognitionManager( + engine="vosk", audio_device_index=3, audio_device_name="USB Mic" + ) + + self.assertEqual(manager.audio_device_index, 3) + self.assertEqual(manager.get_audio_device_name(), "USB Mic") + + def test_set_audio_device_with_name(self): + """Test set_audio_device stores both index and name.""" + manager = SpeechRecognitionManager(engine="vosk") + + manager.set_audio_device(5, "BRIO Webcam") + self.assertEqual(manager.get_audio_device(), 5) + self.assertEqual(manager.get_audio_device_name(), "BRIO Webcam") + + def test_set_audio_device_clears_name(self): + """Test set_audio_device(None, None) clears both.""" + manager = SpeechRecognitionManager(engine="vosk", audio_device_name="Mic") + + manager.set_audio_device(None, None) + self.assertIsNone(manager.get_audio_device()) + self.assertIsNone(manager.get_audio_device_name()) + + def test_reconfigure_with_device_name(self): + """Test reconfigure accepts audio_device_name.""" + manager = SpeechRecognitionManager(engine="vosk") + + manager.reconfigure(audio_device_index=2, audio_device_name="Headset") + self.assertEqual(manager.audio_device_index, 2) + self.assertEqual(manager.get_audio_device_name(), "Headset") + class TestModuleLevelFunctions(unittest.TestCase): """Test module-level functions in recognition_manager.""" @@ -572,6 +605,238 @@ class TestModuleLevelFunctions(unittest.TestCase): with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}): devices = recognition_manager.get_audio_input_devices() + def test_get_audio_input_devices_skips_unreadable_device(self): + """Test unreadable devices are skipped during enumeration.""" + from vocalinux.speech_recognition import recognition_manager + + mock_pyaudio_instance = MagicMock() + mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 1} + mock_pyaudio_instance.get_device_count.return_value = 2 + mock_pyaudio_instance.get_device_info_by_index.side_effect = [ + IOError("device disappeared"), + {"name": "USB Mic", "maxInputChannels": 1}, + ] + + mock_pyaudio = MagicMock() + mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance + + with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}): + devices = recognition_manager.get_audio_input_devices() + + self.assertEqual(devices, [(1, "USB Mic", True)]) + + def test_get_audio_input_devices_skips_output_only_device(self): + """Test output-only devices are skipped during enumeration.""" + from vocalinux.speech_recognition import recognition_manager + + mock_pyaudio_instance = MagicMock() + mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 1} + mock_pyaudio_instance.get_device_count.return_value = 2 + mock_pyaudio_instance.get_device_info_by_index.side_effect = [ + {"name": "HDMI Output", "maxInputChannels": 0}, + {"name": "USB Mic", "maxInputChannels": 1}, + ] + + mock_pyaudio = MagicMock() + mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance + + with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}): + devices = recognition_manager.get_audio_input_devices() + + self.assertEqual(devices, [(1, "USB Mic", True)]) + + def test_get_audio_input_devices_import_error(self): + """Test missing PyAudio returns an empty device list.""" + from vocalinux.speech_recognition import recognition_manager + + with patch.dict(sys.modules, {"pyaudio": None}): + devices = recognition_manager.get_audio_input_devices() + + self.assertEqual(devices, []) + + def test_get_audio_input_devices_pyaudio_error(self): + """Test PyAudio enumeration errors return an empty device list.""" + from vocalinux.speech_recognition import recognition_manager + + mock_pyaudio = MagicMock() + mock_pyaudio.PyAudio.side_effect = OSError("boom") + + with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}): + devices = recognition_manager.get_audio_input_devices() + + self.assertEqual(devices, []) + + def test_is_virtual_device(self): + """Test _is_virtual_device detects known virtual device patterns.""" + from vocalinux.speech_recognition.recognition_manager import _is_virtual_device + + # Virtual devices + self.assertTrue(_is_virtual_device("speech-dispatcher-dummy")) + self.assertTrue(_is_virtual_device("speech-dispatcher-generic")) + self.assertTrue(_is_virtual_device("Dummy Output")) + self.assertTrue(_is_virtual_device("Null Sink")) + self.assertTrue(_is_virtual_device("Monitor of Built-in Audio")) + self.assertTrue(_is_virtual_device("alsa_output.pci-0000")) + + # Real devices + self.assertFalse(_is_virtual_device("BRIO Ultra HD Webcam")) + self.assertFalse(_is_virtual_device("Built-in Microphone")) + self.assertFalse(_is_virtual_device("USB Headset")) + + # Edge cases + self.assertFalse(_is_virtual_device("")) + self.assertFalse(_is_virtual_device(None)) + + def test_is_virtual_device_case_insensitive(self): + """Test _is_virtual_device is case-insensitive.""" + from vocalinux.speech_recognition.recognition_manager import _is_virtual_device + + self.assertTrue(_is_virtual_device("SPEECH-DISPATCHER-DUMMY")) + self.assertTrue(_is_virtual_device("NULL SINK")) + self.assertTrue(_is_virtual_device("Monitor Of Something")) + + def test_get_audio_input_devices_filters_virtual(self): + """Test that get_audio_input_devices excludes virtual devices.""" + from vocalinux.speech_recognition import recognition_manager + + mock_pyaudio_instance = MagicMock() + mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 0} + mock_pyaudio_instance.get_device_count.return_value = 3 + mock_pyaudio_instance.get_device_info_by_index.side_effect = [ + {"name": "Real Mic", "maxInputChannels": 2}, + {"name": "speech-dispatcher-dummy", "maxInputChannels": 2}, + {"name": "USB Headset", "maxInputChannels": 1}, + ] + + mock_pyaudio = MagicMock() + mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance + + with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}): + devices = recognition_manager.get_audio_input_devices() + + # Should only have 2 devices (virtual one filtered out) + device_names = [d[1] for d in devices] + self.assertIn("Real Mic", device_names) + self.assertIn("USB Headset", device_names) + self.assertNotIn("speech-dispatcher-dummy", device_names) + + @patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device") + def test_resolve_device_by_name_found(self, mock_valid): + """Test _resolve_device_by_name finds a matching device and delegates validation.""" + from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name + + mock_valid.return_value = 1 + mock_audio = MagicMock() + mock_audio.get_device_count.return_value = 3 + mock_audio.get_device_info_by_index.side_effect = [ + {"name": "Built-in Mic", "maxInputChannels": 1}, + {"name": "BRIO Webcam", "maxInputChannels": 2}, + {"name": "speech-dispatcher-dummy", "maxInputChannels": 2}, + ] + + result = _resolve_device_by_name(mock_audio, "BRIO Webcam") + self.assertEqual(result, 1) + mock_valid.assert_called_once_with(mock_audio, 1) + + @patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device") + def test_resolve_device_by_name_not_found_with_fallback(self, mock_valid): + """Test _resolve_device_by_name falls back to index when name not found.""" + from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name + + mock_valid.return_value = 0 + mock_audio = MagicMock() + mock_audio.get_device_count.return_value = 2 + mock_audio.get_device_info_by_index.return_value = { + "name": "Mic", + "maxInputChannels": 1, + } + + result = _resolve_device_by_name(mock_audio, "Missing Device", fallback_index=0) + self.assertEqual(result, 0) + mock_valid.assert_called_once_with(mock_audio, 0) + + @patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device") + def test_resolve_device_by_name_skips_unreadable_device(self, mock_valid): + """Test _resolve_device_by_name skips devices that cannot be read.""" + from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name + + mock_valid.return_value = 1 + mock_audio = MagicMock() + mock_audio.get_device_count.return_value = 2 + mock_audio.get_device_info_by_index.side_effect = [ + OSError("device disappeared"), + {"name": "USB Mic", "maxInputChannels": 1}, + ] + + result = _resolve_device_by_name(mock_audio, "USB Mic") + self.assertEqual(result, 1) + mock_valid.assert_called_once_with(mock_audio, 1) + + @patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device") + def test_resolve_device_by_name_ignores_non_dict_device_info(self, mock_valid): + """Test _resolve_device_by_name ignores malformed device info.""" + from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name + + mock_valid.return_value = 0 + mock_audio = MagicMock() + mock_audio.get_device_count.return_value = 1 + mock_audio.get_device_info_by_index.return_value = None + + result = _resolve_device_by_name(mock_audio, "USB Mic", fallback_index=0) + self.assertEqual(result, 0) + mock_valid.assert_called_once_with(mock_audio, 0) + + @patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device") + def test_resolve_device_by_name_not_found_no_fallback(self, mock_valid): + """Test _resolve_device_by_name returns None when name not found and no fallback.""" + from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name + + mock_valid.return_value = None + mock_audio = MagicMock() + mock_audio.get_device_count.return_value = 1 + mock_audio.get_device_info_by_index.return_value = { + "name": "Mic", + "maxInputChannels": 1, + } + + result = _resolve_device_by_name(mock_audio, "Missing Device") + self.assertIsNone(result) + + @patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device") + def test_resolve_device_by_name_empty_name_with_fallback(self, mock_valid): + """Test _resolve_device_by_name with empty name uses fallback index.""" + from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name + + mock_valid.return_value = 0 + mock_audio = MagicMock() + + result = _resolve_device_by_name(mock_audio, None, fallback_index=0) + self.assertEqual(result, 0) + mock_valid.assert_called_once_with(mock_audio, 0) + + @patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device") + def test_resolve_device_by_name_empty_name_no_fallback(self, mock_valid): + """Test _resolve_device_by_name with empty name and no fallback returns None.""" + from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name + + mock_valid.return_value = None + mock_audio = MagicMock() + result = _resolve_device_by_name(mock_audio, None) + self.assertIsNone(result) + + @patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device") + def test_resolve_device_by_name_get_count_error(self, mock_valid): + """Test _resolve_device_by_name handles get_device_count errors.""" + from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name + + mock_valid.return_value = 0 + mock_audio = MagicMock() + mock_audio.get_device_count.side_effect = IOError("boom") + + result = _resolve_device_by_name(mock_audio, "Some Device", fallback_index=0) + self.assertEqual(result, 0) + mock_valid.assert_called_once_with(mock_audio, 0) + def test_show_notification(self): """Test _show_notification helper function.""" from vocalinux.speech_recognition.recognition_manager import _show_notification diff --git a/tests/test_recognition_manager_downloads.py b/tests/test_recognition_manager_downloads.py index 92da61a..6027bb8 100644 --- a/tests/test_recognition_manager_downloads.py +++ b/tests/test_recognition_manager_downloads.py @@ -238,6 +238,69 @@ 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.""" + 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.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 + + with ( + patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}), + patch("time.sleep"), + patch( + "vocalinux.speech_recognition.recognition_manager._resolve_device_by_name", + return_value=None, + ) as mock_resolve_name, + patch( + "vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device", + return_value=1, + ) as mock_resolve_default, + patch( + "vocalinux.speech_recognition.recognition_manager._get_supported_channels", + return_value=1, + ), + patch( + "vocalinux.speech_recognition.recognition_manager._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): + """Test reconnection fails when no resolver finds an input device.""" + manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic") + + mock_pyaudio_mod = MagicMock() + mock_audio_instance = MagicMock() + + with ( + patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}), + patch("time.sleep"), + patch( + "vocalinux.speech_recognition.recognition_manager._resolve_device_by_name", + return_value=None, + ), + patch( + "vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device", + return_value=None, + ), + ): + result = manager._attempt_audio_reconnection(mock_audio_instance) + + assert result is False + mock_audio_instance.open.assert_not_called() + def test_attempt_audio_reconnection_max_attempts(self): """Test reconnection stops after max attempts.""" manager = _make_manager(engine="whisper_cpp")