diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py
index 9defcd7..f32a184 100644
--- a/src/vocalinux/speech_recognition/recognition_manager.py
+++ b/src/vocalinux/speech_recognition/recognition_manager.py
@@ -217,16 +217,39 @@ def _is_virtual_device(device_name: str) -> bool:
"""
if not device_name:
return False
- name_lower = device_name.lower()
+ name_lower = device_name.strip().lower()
+ virtual_exact_names = {
+ "default",
+ "monitor",
+ "paplay",
+ "pipewire",
+ }
+ virtual_prefixes = ("default:", "pipewire:", "pipewire ")
virtual_patterns = [
+ ".monitor",
+ "_monitor",
+ "-monitor",
+ "alsa_output", # ALSA output devices exposed as monitors
+ "deepfilternet",
+ "dummy",
+ "filter-chain",
+ "filter_chain",
+ "monitor of",
+ "null sink",
+ "null source",
+ "null-sink",
+ "null-source",
+ "null_sink",
+ "null_source",
+ "paplay",
"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)
+ return (
+ name_lower in virtual_exact_names
+ or name_lower.startswith(virtual_prefixes)
+ or any(pattern in name_lower for pattern in virtual_patterns)
+ )
def _is_bluetooth_device(device_name: Optional[str]) -> bool:
@@ -356,7 +379,7 @@ def _resolve_device_by_name(
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).
+ """Resolve a valid audio input device, skipping unsafe or output-only devices.
Checks that the device has maxInputChannels > 0. Falls back from
preferred_index → system default → first available input device.
@@ -399,6 +422,11 @@ def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) ->
input_device_indices.append(i)
continue
+ device_name = info.get("name", "")
+ if _is_virtual_device(device_name):
+ logger.debug(f"Filtering virtual input device [{i}]: {device_name}")
+ continue
+
channels = info.get("maxInputChannels", 0)
if isinstance(channels, (int, float)) and channels > 0:
input_device_indices.append(i)
@@ -415,7 +443,7 @@ def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) ->
except (IOError, OSError):
device_name = "unknown"
logger.warning(
- "Configured audio device [%s] (%s) has no input channels. "
+ "Configured audio device [%s] (%s) is not a safe input device. "
"Falling back to a valid input device.",
preferred_index,
device_name,
@@ -640,15 +668,17 @@ def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict:
audio = pyaudio.PyAudio()
- # Get device info
+ # Get device info for display. Keep open_device_index as None for
+ # System Default so PortAudio opens the host default instead of an
+ # explicit pseudo-device index like "default" / DeepFilterNet (#624).
+ open_device_index = device_index
try:
if device_index is not None:
info = audio.get_device_info_by_index(device_index)
else:
info = audio.get_default_input_device_info()
- device_index = info.get("index")
result["device_name"] = info.get("name", "Unknown")
- result["device_index"] = device_index
+ result["device_index"] = info.get("index", device_index)
except (IOError, OSError) as e:
result["error"] = f"Cannot get device info: {e}"
audio.terminate()
@@ -657,7 +687,7 @@ def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict:
# Negotiate the format and open the capture stream in ONE PortAudio
# open — Bluetooth SCO devices abort with heap corruption when the
# stream is opened, closed, and quickly reopened (issue #567).
- CHANNELS, RATE, stream = _open_capture_stream(audio, device_index)
+ CHANNELS, RATE, stream = _open_capture_stream(audio, open_device_index)
logger.info(f"Using {CHANNELS} channel(s) for audio test")
result["sample_rate"] = RATE
@@ -672,8 +702,8 @@ def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict:
"input": True,
"frames_per_buffer": CHUNK,
}
- if device_index is not None:
- stream_kwargs["input_device_index"] = device_index
+ if open_device_index is not None:
+ stream_kwargs["input_device_index"] = open_device_index
stream = audio.open(**stream_kwargs)
except (IOError, OSError) as e:
@@ -2528,20 +2558,24 @@ class SpeechRecognitionManager:
# 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(
- "Please connect a microphone and ensure it is recognized by the system."
+ use_system_default = self.audio_device_index is None and self.audio_device_name is None
+ if use_system_default:
+ resolved_device_index = None
+ else:
+ resolved_device_index = _resolve_device_by_name(
+ audio, self.audio_device_name, self.audio_device_index
)
- play_error_sound()
- audio.terminate()
- self._update_state(RecognitionState.ERROR)
- return
+ 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 = None
+ use_system_default = True
# Log available devices for debugging (skip virtual devices)
logger.debug("Available audio input devices:")
@@ -2565,10 +2599,14 @@ class SpeechRecognitionManager:
logger.info(f"Using sample rate: {RATE}Hz")
try:
- device_info = audio.get_device_info_by_index(resolved_device_index)
- logger.info(
- f"Using audio device [{resolved_device_index}]: {device_info.get('name')}"
- )
+ if resolved_device_index is None:
+ device_info = audio.get_default_input_device_info()
+ logger.info(f"Using system default audio device: {device_info.get('name')}")
+ else:
+ device_info = audio.get_device_info_by_index(resolved_device_index)
+ logger.info(
+ f"Using audio device [{resolved_device_index}]: {device_info.get('name')}"
+ )
except (IOError, OSError):
logger.warning(f"Could not get info for device index {resolved_device_index}")
@@ -2591,7 +2629,7 @@ class SpeechRecognitionManager:
default_idx = audio.get_default_input_device_info().get("index")
except (IOError, OSError):
default_idx = None
- if resolved_device_index != default_idx:
+ if resolved_device_index is not None and resolved_device_index != default_idx:
stream_kwargs["input_device_index"] = resolved_device_index
try:
@@ -2795,12 +2833,7 @@ class SpeechRecognitionManager:
break
# Clean up
- if stream and hasattr(stream, "is_active") and stream.is_active():
- try:
- stream.stop_stream()
- stream.close()
- except Exception as e:
- logger.warning(f"Error closing audio stream: {e}")
+ _safe_close_stream(stream)
if audio and hasattr(audio, "terminate"):
try:
@@ -3171,21 +3204,25 @@ class SpeechRecognitionManager:
try:
# Close existing stream if it exists
if self._audio_stream:
- try:
- self._audio_stream.stop_stream()
- self._audio_stream.close()
- except Exception as e:
- logger.debug(f"Error closing old audio stream: {e}")
+ _safe_close_stream(self._audio_stream)
+ self._audio_stream = None
# 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
+ use_system_default = self.audio_device_index is None and self.audio_device_name is None
+ if use_system_default:
+ resolved_device_index = None
+ else:
+ 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 and not use_system_default:
+ logger.warning(
+ "Reconnection: no safe input devices enumerated; "
+ "falling back to system default capture."
+ )
+ resolved_device_index = None
# Stream configuration
CHUNK = 1024
@@ -3214,7 +3251,7 @@ class SpeechRecognitionManager:
default_idx = audio_instance.get_default_input_device_info().get("index")
except (IOError, OSError):
default_idx = None
- if resolved_device_index != default_idx:
+ if resolved_device_index is not None and resolved_device_index != default_idx:
stream_kwargs["input_device_index"] = resolved_device_index
new_stream = audio_instance.open(**stream_kwargs)
diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py
index 9f4cae2..9a37a32 100644
--- a/src/vocalinux/ui/settings_dialog.py
+++ b/src/vocalinux/ui/settings_dialog.py
@@ -65,6 +65,36 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
+
+def _raw_audio_device_name(device_name: Optional[str]) -> Optional[str]:
+ """Return the persisted device name without UI-only suffixes."""
+ if device_name is None:
+ return None
+ return device_name.removesuffix(" (default)")
+
+
+def _resolve_audio_device_selection(
+ devices: list,
+ saved_index: Optional[int],
+ saved_name: Optional[str],
+) -> Optional[int]:
+ """Resolve a saved audio device to a currently listed index.
+
+ Returns the matched device index, or None when the saved device is gone
+ (caller should fall back to System Default).
+ """
+ saved_raw_name = _raw_audio_device_name(saved_name)
+ if saved_raw_name:
+ for idx, name, _is_default in devices:
+ if name == saved_raw_name:
+ return idx
+ if saved_index is not None:
+ for idx, _name, _is_default in devices:
+ if idx == saved_index:
+ return saved_index
+ return None
+
+
# Define available models for each engine
ENGINE_MODELS = {
"vosk": [
@@ -4325,22 +4355,29 @@ For now, the engine has been reverted to VOSK."""
if saved_device is None:
self.audio_device_combo.set_active_id("-1")
else:
- # 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")
+ matched_index = _resolve_audio_device_selection(
+ devices, saved_device, saved_device_name
+ )
+ if matched_index is not None:
+ self.audio_device_combo.set_active_id(str(matched_index))
+ # Migrate legacy configs that stored the UI "(default)" suffix.
+ saved_raw_name = _raw_audio_device_name(saved_device_name)
+ if saved_device_name and saved_raw_name and saved_device_name != saved_raw_name:
+ self.config_manager.set("audio", "device_name", saved_raw_name)
+ self.config_manager.set("audio", "device_index", matched_index)
+ self.config_manager.save_settings()
+ else:
+ logger.warning(
+ f"Saved audio device {saved_device} "
+ f"(name: {saved_device_name}) no longer available"
+ )
+ self.audio_device_combo.set_active_id("-1")
+ # Keep combo, config, and engine aligned on System Default.
+ self.config_manager.set("audio", "device_index", None)
+ self.config_manager.set("audio", "device_name", None)
+ self.config_manager.save_settings()
+ if self.speech_engine is not None:
+ self.speech_engine.set_audio_device(None, None)
logger.info(f"Found {len(devices)} audio input devices")
@@ -4359,7 +4396,8 @@ For now, the engine has been reverted to VOSK."""
return
device_index = int(device_id)
- device_name = self.audio_device_combo.get_active_text()
+ device_label = self.audio_device_combo.get_active_text()
+ device_name = _raw_audio_device_name(device_label)
if device_index == -1:
self.config_manager.set("audio", "device_index", None)
@@ -4376,7 +4414,7 @@ For now, the engine has been reverted to VOSK."""
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}")
+ self.audio_test_status.set_markup(f"Selected: {device_label}")
def _on_test_audio_clicked(self, widget):
"""Handle test audio button click."""
diff --git a/tests/test_recognition_manager_device_config.py b/tests/test_recognition_manager_device_config.py
index a29b50a..8e4e87b 100644
--- a/tests/test_recognition_manager_device_config.py
+++ b/tests/test_recognition_manager_device_config.py
@@ -32,10 +32,14 @@ from vocalinux.speech_recognition.recognition_manager import (
_get_supported_channels,
_get_supported_sample_rate,
_is_bluetooth_device,
+ _is_virtual_device,
_open_capture_stream,
+ _resolve_valid_input_device,
_safe_close_stream,
get_audio_input_devices,
- test_audio_input,
+)
+from vocalinux.speech_recognition.recognition_manager import (
+ test_audio_input as _run_test_audio_input,
)
@@ -52,6 +56,209 @@ def _make_manager(engine="whisper_cpp", **kw):
class TestAudioDeviceDetection(unittest.TestCase):
"""Test audio device enumeration functions."""
+ def test_is_virtual_device_filters_pipewire_pseudo_sources(self):
+ """PipeWire pseudo sources from issue #624 should not be opened by index."""
+ unsafe_names = [
+ "default",
+ "DeepFilterNet Source",
+ "paplay",
+ "pipewire",
+ "PipeWire filter-chain source",
+ "Monitor of Built-in Audio Analog Stereo",
+ "alsa_output.pci-0000_00_1f.3.analog-stereo.monitor",
+ "Null Sink",
+ "dummy",
+ "speech-dispatcher-dummy",
+ ]
+
+ for name in unsafe_names:
+ assert _is_virtual_device(name), name
+
+ assert not _is_virtual_device("USB Microphone")
+ assert not _is_virtual_device("Built-in Audio Analog Stereo")
+ assert not _is_virtual_device("Built-in Mic (PipeWire)")
+ assert not _is_virtual_device("")
+
+ def test_get_audio_input_devices_filters_unsafe_virtual_sources(self):
+ """Device enumeration should hide unsafe virtual sources but keep real mics."""
+ mock_audio = MagicMock()
+ mock_audio.get_device_count.return_value = 4
+ mock_audio.get_default_input_device_info.return_value = {"index": 2}
+ devices = [
+ {"index": 0, "name": "DeepFilterNet Source", "maxInputChannels": 2},
+ {"index": 1, "name": "Monitor of Built-in Audio Analog Stereo", "maxInputChannels": 2},
+ {"index": 2, "name": "USB Microphone", "maxInputChannels": 1},
+ {"index": 3, "name": "default", "maxInputChannels": 32},
+ ]
+ mock_audio.get_device_info_by_index.side_effect = lambda i: devices[i]
+ mock_pyaudio = MagicMock(PyAudio=MagicMock(return_value=mock_audio))
+
+ with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
+ assert get_audio_input_devices() == [(2, "USB Microphone", True)]
+
+ def test_resolve_valid_input_device_skips_stale_virtual_index(self):
+ """A saved virtual source index should fall back to a physical microphone."""
+ mock_audio = MagicMock()
+ mock_audio.get_device_count.return_value = 3
+ mock_audio.get_default_input_device_info.return_value = {"index": 1}
+ devices = [
+ {"index": 0, "name": "DeepFilterNet Source", "maxInputChannels": 2},
+ {"index": 1, "name": "default", "maxInputChannels": 32},
+ {"index": 2, "name": "USB Microphone", "maxInputChannels": 1},
+ ]
+ mock_audio.get_device_info_by_index.side_effect = lambda i: devices[i]
+
+ assert _resolve_valid_input_device(mock_audio, preferred_index=0) == 2
+
+ def test_system_default_selection_uses_no_explicit_device_index(self):
+ """System Default should still open through PortAudio's default device path."""
+ manager = _make_manager(
+ engine="whisper_cpp", audio_device_index=None, audio_device_name=None
+ )
+ manager.should_record = True
+ manager.state = RecognitionState.LISTENING
+ 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.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 = {
+ "index": 0,
+ "name": "default",
+ "maxInputChannels": 32,
+ }
+ mock_audio.get_default_input_device_info.return_value = {"index": 0, "name": "default"}
+
+ 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._open_capture_stream",
+ return_value=(1, 16000, mock_stream),
+ ) as mock_open,
+ patch(
+ "vocalinux.speech_recognition.recognition_manager.play_error_sound",
+ ),
+ ):
+ if isinstance(sys.modules.get("numpy"), MagicMock):
+ del sys.modules["numpy"]
+ manager._record_audio()
+
+ 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."""
+ manager = _make_manager(
+ engine="whisper_cpp",
+ audio_device_index=0,
+ audio_device_name="DeepFilterNet Source",
+ )
+ manager.should_record = True
+ manager.state = RecognitionState.LISTENING
+ 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.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 = {
+ "index": 0,
+ "name": "DeepFilterNet Source",
+ "maxInputChannels": 2,
+ }
+ mock_audio.get_default_input_device_info.return_value = {
+ "index": 0,
+ "name": "default",
+ }
+ 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._resolve_device_by_name",
+ return_value=None,
+ ),
+ patch(
+ "vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device",
+ return_value=None,
+ ),
+ patch(
+ "vocalinux.speech_recognition.recognition_manager._open_capture_stream",
+ return_value=(1, 16000, mock_stream),
+ ) as mock_open,
+ patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"),
+ ):
+ if isinstance(sys.modules.get("numpy"), MagicMock):
+ 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()
+
+ 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."""
+ mock_stream = MagicMock()
+ mock_stream.read.return_value = b"\x00\x01" * 1024
+ mock_audio = MagicMock()
+ mock_audio.open.return_value = mock_stream
+ mock_audio.get_default_input_device_info.return_value = {
+ "index": 7,
+ "name": "default",
+ "defaultSampleRate": 48000,
+ "maxInputChannels": 1,
+ }
+ mock_pyaudio = MagicMock(paInt16=8)
+ mock_pyaudio.PyAudio.return_value = mock_audio
+
+ with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
+ if isinstance(sys.modules.get("numpy"), MagicMock):
+ del sys.modules["numpy"]
+ with patch(
+ "vocalinux.speech_recognition.recognition_manager._open_capture_stream",
+ return_value=(1, 48000, None),
+ ):
+ result = _run_test_audio_input(device_index=None, duration=0.1)
+
+ assert result["success"] is True
+ kwargs = mock_audio.open.call_args.kwargs
+ assert "input_device_index" not in kwargs
+
+ def test_settings_persist_raw_audio_device_name(self):
+ """Settings should persist raw PortAudio names, not UI-only suffixes."""
+ from vocalinux.ui.settings_dialog import (
+ _raw_audio_device_name,
+ _resolve_audio_device_selection,
+ )
+
+ assert _raw_audio_device_name("USB Microphone (default)") == "USB Microphone"
+ assert _raw_audio_device_name("USB Microphone") == "USB Microphone"
+ assert _raw_audio_device_name(None) is None
+
+ devices = [(3, "USB Microphone", True), (5, "Webcam Mic", False)]
+ # Legacy configs may still store the UI-only "(default)" suffix.
+ assert _resolve_audio_device_selection(devices, 3, "USB Microphone (default)") == 3
+ assert _resolve_audio_device_selection(devices, 99, "Webcam Mic") == 5
+ # Filtered-out / missing devices resolve to System Default.
+ assert _resolve_audio_device_selection(devices, 1, "DeepFilterNet") is None
+ assert _resolve_audio_device_selection(devices, 1, None) is None
+
def test_is_bluetooth_device_detection(self):
"""Bluetooth headset/mic names should be recognized (Issue #567)."""
assert _is_bluetooth_device("Bluetooth internal capture stream for HUAWEI FreeBuds Pro 3")
@@ -192,6 +399,38 @@ class TestAudioDeviceDetection(unittest.TestCase):
mock_stream.stop_stream.assert_called_once()
mock_stream.close.assert_called_once()
+ def test_test_audio_input_system_default_keeps_null_device_index(self):
+ """System Default must open PortAudio without an explicit device index."""
+ from vocalinux.speech_recognition.recognition_manager import test_audio_input
+
+ mock_stream = MagicMock()
+ mock_stream.read.return_value = b"\x00\x01" * 1024
+ mock_audio = MagicMock()
+ mock_audio.open.return_value = mock_stream
+ mock_audio.get_default_input_device_info.return_value = {
+ "index": 7,
+ "name": "default",
+ "defaultSampleRate": 48000,
+ "maxInputChannels": 1,
+ }
+ mock_pyaudio = MagicMock(paInt16=8)
+ mock_pyaudio.PyAudio.return_value = mock_audio
+
+ with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
+ if isinstance(sys.modules.get("numpy"), MagicMock):
+ del sys.modules["numpy"]
+ with patch(
+ "vocalinux.speech_recognition.recognition_manager._open_capture_stream",
+ return_value=(1, 48000, mock_stream),
+ ) as mock_open:
+ result = test_audio_input(device_index=None, duration=0.1)
+
+ assert result["success"] is True
+ mock_open.assert_called_once_with(mock_audio, None)
+ # Display metadata can still report the host default index/name.
+ assert result["device_index"] == 7
+ assert result["device_name"] == "default"
+
def test_test_audio_input_negotiation_fallback_open(self):
"""When negotiation returns no stream, mic test falls back to a plain open."""
mock_stream = MagicMock()
@@ -214,7 +453,7 @@ class TestAudioDeviceDetection(unittest.TestCase):
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
return_value=(1, 16000, None),
):
- result = test_audio_input(device_index=14, duration=0.1)
+ result = _run_test_audio_input(device_index=14, duration=0.1)
assert result["success"] is True
mock_audio.open.assert_called_once()
@@ -237,7 +476,7 @@ class TestAudioDeviceDetection(unittest.TestCase):
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
return_value=(1, 16000, None),
):
- result = test_audio_input(device_index=14, duration=0.1)
+ result = _run_test_audio_input(device_index=14, duration=0.1)
assert result["success"] is False
assert "Cannot open audio stream" in result["error"]
diff --git a/tests/test_recognition_manager_downloads.py b/tests/test_recognition_manager_downloads.py
index 4a3e5bb..a55bf47 100644
--- a/tests/test_recognition_manager_downloads.py
+++ b/tests/test_recognition_manager_downloads.py
@@ -401,11 +401,13 @@ class TestAudioReconnection:
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."""
+ """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()
with (
patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
@@ -418,11 +420,15 @@ class TestAudioReconnection:
"vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device",
return_value=None,
),
+ patch(
+ "vocalinux.speech_recognition.recognition_manager._open_capture_stream",
+ return_value=(1, 16000, mock_stream),
+ ) as mock_open,
):
result = manager._attempt_audio_reconnection(mock_audio_instance)
- assert result is False
- mock_audio_instance.open.assert_not_called()
+ assert result is True
+ mock_open.assert_called_once_with(mock_audio_instance, None)
def test_attempt_audio_reconnection_max_attempts(self):
"""Test reconnection stops after max attempts."""