xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

reliability: Microphone Reconnection and Audio Buffer Limits (#109)

Browse Source 
* fix: implement microphone reconnection and audio buffer limits for reliability

- Add microphone reconnection logic with exponential backoff
- Implement audio buffer limits to prevent memory issues
- Add configurable buffer size limits with clamping
- Improve error handling for audio device failures
- Add buffer statistics and monitoring capabilities
- Graceful stream cleanup and resource management

Resolves #92

Changes made:
- Added _max_buffer_size, _reconnection_attempts, _max_reconnection_attempts attributes
- Implemented _attempt_audio_reconnection() method with exponential backoff
- Enhanced _record_audio() method with buffer limit checking and reconnection
- Added set_buffer_limit() and get_buffer_stats() methods
- Improved error handling for IOError/OSError during audio recording
- Added test scripts to validate reliability improvements

The implementation prevents memory issues by limiting audio buffer size
and automatically reconnecting to audio devices when they fail,
with proper exponential backoff to prevent rapid reconnection attempts.

* fix: Address code review feedback for PR #109

- Add thread lock (_buffer_lock) for audio_buffer to prevent race conditions
- Fix _last_audio_error_time reset in cleanup
- Protect buffer access in stop_recognition and get_buffer_stats with lock
- Fix test_max_reconnection_attempts to match actual counter behavior

Addresses review feedback for microphone reconnection and audio buffer limits

* style: fix lint errors

* fix: add missing sys import in audio_feedback.py

- Add sys import required by _is_ci_mode() function
- Remove temporary test_reliability scripts

* style: fix black formatting

---------

Co-authored-by: OpenClaw Reviewer <review@openclaw.ai>
This commit is contained in:
Jatin K Malik
2026-02-04 22:52:39 -08:00
committed by GitHub
co-authored by OpenClaw Reviewer
parent 6f54dd80e0
commit 94f71853e0
22 changed files with 265 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
setup.py
+2 -1
View File
@@ -15,7 +15,8 @@ from setuptools import find_packages, setup
MIN_PYTHON_VERSION = (3, 8)
if sys.version_info < MIN_PYTHON_VERSION:
sys.exit(
f"Error: Vocalinux requires Python {MIN_PYTHON_VERSION[0]}.{MIN_PYTHON_VERSION[1]} or higher"
f"Error: Vocalinux requires Python {MIN_PYTHON_VERSION[0]}."
f"{MIN_PYTHON_VERSION[1]} or higher"
)
src/vocalinux/__init__.py
+1 -1
View File
@@ -10,7 +10,7 @@ but system typelibs (AppIndicator3, etc.) may not be accessible.
__version__ = "0.1.0"
# Import common types first to avoid circular imports
from .common_types import RecognitionState
from .common_types import RecognitionState # noqa: F401
# Import non-GTK modules that are safe to import at package level
from .utils import ResourceManager
src/vocalinux/common_types.py
+1 -1
View File
@@ -4,7 +4,7 @@ This module provides type definitions to avoid circular imports.
"""
from enum import Enum, auto
from typing import Callable, List, Optional, Protocol
from typing import Callable, List, Optional, Protocol # noqa: F401
class RecognitionState(Enum):
src/vocalinux/main.py
+6 -2
View File
@@ -22,7 +22,8 @@ def parse_arguments():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(description="Vocalinux")
parser.add_argument("--debug", action="store_true", help="Enable debug logging")
# default model, language and engine are loaded from default config due to priority of args over config
# default model, language and engine are loaded from default config
# due to priority of args over config
parser.add_argument(
"--model",
type=str,
@@ -48,7 +49,10 @@ def parse_arguments():
"ko",
"ar",
],
help="Speech recognition language (auto for auto-detect, en-us, hi, es, fr, de, it, pt, ru, zh, etc.)",
help=(
"Speech recognition language (auto for auto-detect, en-us, "
"hi, es, fr, de, it, pt, ru, zh, etc.)"
),
)
parser.add_argument(
"--engine",
src/vocalinux/speech_recognition/__init__.py
+1 -1
View File
@@ -2,7 +2,7 @@
Speech recognition module for Vocalinux
"""
from . import command_processor, recognition_manager
from . import command_processor, recognition_manager # noqa: F401
def get_audio_input_devices():
src/vocalinux/speech_recognition/command_processor.py
+1 -1
View File
@@ -7,7 +7,7 @@ This module processes text commands from speech recognition, such as
import logging
import re
from typing import Dict, List, Optional, Tuple
from typing import List, Tuple
logger = logging.getLogger(__name__)
src/vocalinux/speech_recognition/recognition_manager.py
+204 -21
View File
@@ -283,6 +283,16 @@ class SpeechRecognitionManager:
# Recording control flags
self.should_record = False
self.audio_buffer = []
self._buffer_lock = threading.Lock() # Thread safety for audio_buffer
# Reliability improvements - Issue #92
self._max_buffer_size = 5000 # Maximum number of audio chunks in buffer
self._reconnection_attempts = 0
self._max_reconnection_attempts = 5
self._reconnection_delay = 1.0 # Initial delay in seconds
self._last_audio_error_time = 0
self._audio_stream = None
self._pyaudio_instance = None
# Create models directory if it doesn't exist
os.makedirs(MODELS_DIR, exist_ok=True)
@@ -925,16 +935,17 @@ class SpeechRecognitionManager:
self.recognition_thread.join(timeout=1.0)
# Process any remaining audio in the buffer before going idle
if self.audio_buffer:
logger.debug("Processing remaining audio buffer before stopping")
self._update_state(RecognitionState.PROCESSING)
self._process_final_buffer()
self.audio_buffer = []
with self._buffer_lock:
if self.audio_buffer:
logger.debug("Processing remaining audio buffer before stopping")
self._update_state(RecognitionState.PROCESSING)
self._process_final_buffer()
self.audio_buffer = []
self._update_state(RecognitionState.IDLE)
def _record_audio(self):
"""Record audio from the microphone."""
"""Record audio from the microphone with reconnection logic."""
# Lazy import to avoid circular dependency
from ..ui.audio_feedback import play_error_sound # noqa: F401
@@ -956,8 +967,9 @@ class SpeechRecognitionManager:
CHANNELS = 1
RATE = 16000
# Initialize PyAudio
audio = pyaudio.PyAudio()
# Initialize PyAudio with reconnection support
self._pyaudio_instance = pyaudio.PyAudio()
audio = self._pyaudio_instance
# Log available devices for debugging
logger.debug("Available audio input devices:")
@@ -971,7 +983,7 @@ class SpeechRecognitionManager:
except (IOError, OSError):
continue
# Open microphone stream with optional device selection
# Open microphone stream with optional device selection and reconnection logic
stream_kwargs = {
"format": FORMAT,
"channels": CHANNELS,
@@ -1000,14 +1012,20 @@ class SpeechRecognitionManager:
logger.warning("Could not get default input device info")
try:
stream = audio.open(**stream_kwargs)
self._audio_stream = audio.open(**stream_kwargs)
stream = self._audio_stream
except (IOError, OSError) as e:
logger.error(f"Failed to open audio stream: {e}")
logger.error("This may indicate a problem with the audio device or permissions.")
play_error_sound()
audio.terminate()
self._update_state(RecognitionState.ERROR)
return
# Attempt reconnection
if self._attempt_audio_reconnection(audio):
stream = self._audio_stream
else:
play_error_sound()
audio.terminate()
self._update_state(RecognitionState.ERROR)
return
logger.info("Audio recording started")
@@ -1019,8 +1037,19 @@ class SpeechRecognitionManager:
while self.should_record:
try:
data = stream.read(CHUNK, exception_on_overflow=False)
self.audio_buffer.append(data)
# Check buffer size and enforce limits (with lock for thread safety)
with self._buffer_lock:
if len(self.audio_buffer) >= self._max_buffer_size:
logger.warning(
f"Audio buffer limit reached ({len(self.audio_buffer)} chunks). Clearing oldest data."
)
# Remove oldest 25% of data to prevent memory issues
remove_count = self._max_buffer_size // 4
self.audio_buffer = self.audio_buffer[remove_count:]
logger.info(f"Buffer trimmed by {remove_count} chunks")
data = stream.read(CHUNK, exception_on_overflow=False)
self.audio_buffer.append(data)
# Simple Voice Activity Detection (VAD)
audio_data = np.frombuffer(data, dtype=np.int16)
@@ -1043,7 +1072,7 @@ class SpeechRecognitionManager:
log_level_interval += 1
if log_level_interval >= 50: # Every ~3 seconds at 16kHz/1024 chunks
logger.debug(
f"Audio level: current={normalized_level:.1f}%, max_seen={max_level_seen:.1f}%"
f"Audio level: current={normalized_level:.1f}%, max_seen={max_level_seen:.1f}%, buffer_size={len(self.audio_buffer)}"
)
log_level_interval = 0
@@ -1078,14 +1107,51 @@ class SpeechRecognitionManager:
)
speech_detected_in_session = True
silence_counter = 0
except (IOError, OSError) as e:
current_time = time.time()
logger.error(f"Audio device error: {e}")
# Implement reconnection logic with exponential backoff
if (
current_time - self._last_audio_error_time > 5.0
): # Prevent rapid reconnection attempts
self._last_audio_error_time = current_time
if self._attempt_audio_reconnection(audio):
logger.info("Audio reconnection successful, continuing recording")
stream = self._audio_stream # Update stream reference
continue # Continue recording with new stream
else:
logger.error("Audio reconnection failed, stopping recording")
break
else:
logger.warning(
"Audio error occurred too soon after last error, stopping recording"
)
break
except Exception as e:
logger.error(f"Error reading audio data: {e}")
logger.error(f"Unexpected error reading audio data: {e}")
break
# Clean up
stream.stop_stream()
stream.close()
audio.terminate()
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}")
if audio and hasattr(audio, "terminate"):
try:
audio.terminate()
except Exception as e:
logger.warning(f"Error terminating PyAudio: {e}")
# Reset audio stream reference and reconnection state
self._audio_stream = None
self._pyaudio_instance = None
self._reconnection_attempts = 0
self._last_audio_error_time = 0
# Log summary
if not speech_detected_in_session and max_level_seen < 5:
@@ -1223,3 +1289,120 @@ class SpeechRecognitionManager:
else:
# If only VOSK params changed, just log it
logger.info("Applied VAD/silence timeout changes.")
def _attempt_audio_reconnection(self, audio_instance) -> bool:
"""
Attempt to reconnect to the audio device.
Args:
audio_instance: The PyAudio instance to use for reconnection
Returns:
bool: True if reconnection was successful, False otherwise
"""
self._reconnection_attempts += 1
if self._reconnection_attempts > self._max_reconnection_attempts:
logger.error(f"Max reconnection attempts ({self._max_reconnection_attempts}) reached")
return False
# Calculate delay with exponential backoff
delay = self._reconnection_delay * (2 ** (self._reconnection_attempts - 1))
delay = min(delay, 10.0) # Cap at 10 seconds
logger.info(
f"Attempting audio reconnection (attempt {self._reconnection_attempts}/{self._max_reconnection_attempts}) after {delay:.1f}s delay..."
)
# Wait before attempting reconnection
time.sleep(delay)
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}")
# Stream configuration
CHUNK = 1024
FORMAT = audio_instance.paInt16
CHANNELS = 1
RATE = 16000
stream_kwargs = {
"format": FORMAT,
"channels": CHANNELS,
"rate": RATE,
"input": True,
"frames_per_buffer": CHUNK,
}
# Use specified device if set
if self.audio_device_index is not None:
stream_kwargs["input_device_index"] = self.audio_device_index
# Attempt to open new stream
new_stream = audio_instance.open(**stream_kwargs)
# Test the stream by reading a small amount of data
test_data = new_stream.read(CHUNK, exception_on_overflow=False)
if test_data:
self._audio_stream = new_stream
logger.info("Audio reconnection successful")
return True
else:
logger.error("Reconnected stream returned no data")
try:
new_stream.stop_stream()
new_stream.close()
except Exception:
pass
return False
except (IOError, OSError) as e:
logger.error(f"Audio reconnection failed: {e}")
return False
except Exception as e:
logger.error(f"Unexpected error during audio reconnection: {e}")
return False
def set_buffer_limit(self, max_chunks: int):
"""
Set the maximum number of audio chunks to buffer.
Args:
max_chunks: Maximum number of chunks to buffer (default: 5000)
"""
if max_chunks < 100:
logger.warning("Buffer limit too small, setting to minimum 100")
max_chunks = 100
elif max_chunks > 20000:
logger.warning("Buffer limit too large, setting to maximum 20000")
max_chunks = 20000
self._max_buffer_size = max_chunks
logger.info(f"Audio buffer limit set to {max_chunks} chunks")
def get_buffer_stats(self) -> dict:
"""
Get current buffer statistics.
Returns:
dict: Buffer statistics including size, memory usage, etc.
"""
with self._buffer_lock:
total_memory = sum(len(chunk) for chunk in self.audio_buffer)
buffer_size = len(self.audio_buffer)
return {
"buffer_size": buffer_size,
"buffer_limit": self._max_buffer_size,
"memory_usage_bytes": total_memory,
"memory_usage_mb": total_memory / (1024 * 1024),
"buffer_full_percentage": (
(buffer_size / self._max_buffer_size) * 100 if self._max_buffer_size > 0 else 0
),
}
src/vocalinux/text_injection/__init__.py
+1 -1
View File
@@ -2,4 +2,4 @@
Text injection module for Vocalinux
"""
from . import text_injector
from . import text_injector # noqa: F401
src/vocalinux/text_injection/text_injector.py
+9 -7
View File
@@ -11,7 +11,7 @@ import shutil
import subprocess
import time
from enum import Enum
from typing import Dict, Optional
from typing import Optional # noqa: F401
logger = logging.getLogger(__name__)
@@ -66,7 +66,8 @@ class TextInjector:
error_output = result.stderr.lower()
if "compositor does not support" in error_output or result.returncode != 0:
logger.warning(
f"Wayland compositor does not support virtual keyboard protocol: {error_output}"
"Wayland compositor does not support virtual "
f"keyboard protocol: {error_output}"
)
if shutil.which("xdotool"):
logger.info("Automatically switching to XWayland fallback with xdotool")
@@ -222,7 +223,7 @@ class TextInjector:
self._inject_with_xdotool(escaped_text)
else:
raise
logger.info(f"Text injection completed successfully")
logger.info("Text injection completed successfully")
return True
except Exception as e:
logger.error(f"Failed to inject text: {e}", exc_info=True)
@@ -331,7 +332,7 @@ class TextInjector:
cmd = ["xdotool", "type", "--clearmodifiers", chunk]
logger.debug(f"Injecting chunk {chunk_num}/{total_chunks}: '{chunk}'")
result = subprocess.run(
subprocess.run(
cmd, env=env, check=True, stderr=subprocess.PIPE, text=True, timeout=5
)
@@ -346,7 +347,8 @@ class TextInjector:
except subprocess.CalledProcessError as chunk_error:
if retry < max_retries:
logger.warning(
f"Retrying text injection (attempt {retry+1}/{max_retries}): {chunk_error.stderr}"
f"Retrying text injection (attempt {retry+1}/{max_retries}): "
f"{chunk_error.stderr}"
)
time.sleep(0.5) # Wait before retry
else:
@@ -393,7 +395,7 @@ class TextInjector:
cmd = ["ydotool", "type", text]
try:
result = subprocess.run(cmd, check=True, stderr=subprocess.PIPE, text=True)
subprocess.run(cmd, check=True, stderr=subprocess.PIPE, text=True)
except subprocess.CalledProcessError as e:
# Re-raise with stderr preserved for better diagnostics
raise subprocess.CalledProcessError(
@@ -564,7 +566,7 @@ class TextInjector:
with open(f"/proc/{window_pid}/comm", "r") as f:
process_name = f.read().strip()
logger.info(f"Target process: {process_name} (PID: {window_pid})")
except:
except Exception:
pass
except subprocess.TimeoutExpired:
src/vocalinux/ui/audio_feedback.py
+2 -2
View File
@@ -9,7 +9,7 @@ import os
import shutil
import subprocess
import sys
from pathlib import Path
from pathlib import Path # noqa: F401
logger = logging.getLogger(__name__)
@@ -41,7 +41,7 @@ def _is_ci_mode():
# Import the centralized resource manager
from ..utils.resource_manager import ResourceManager
from ..utils.resource_manager import ResourceManager # noqa: E402
# Initialize resource manager
_resource_manager = ResourceManager()
src/vocalinux/ui/logging_dialog.py
+3 -5
View File
@@ -6,17 +6,15 @@ application logs for debugging purposes.
"""
import logging
import os
import threading
from datetime import datetime
from typing import Optional
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import GLib, GObject, Gtk
from gi.repository import GLib, Gtk # noqa: E402
from .logging_manager import LogRecord, get_logging_manager
from .logging_manager import LogRecord, get_logging_manager # noqa: E402
logger = logging.getLogger(__name__)
@@ -170,7 +168,7 @@ class LoggingDialog(Gtk.Dialog):
debug_tag.set_property("foreground", "#666666")
# Info - default color
info_tag = self.text_buffer.create_tag("INFO")
self.text_buffer.create_tag("INFO")
# Warning - orange
warning_tag = self.text_buffer.create_tag("WARNING")
src/vocalinux/ui/logging_manager.py
+1 -3
View File
@@ -6,9 +6,7 @@ allowing users to view, filter, and export application logs for debugging.
"""
import logging
import os
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Callable, List, Optional
@@ -178,7 +176,7 @@ class LoggingManager:
records = self.get_logs(level_filter, module_filter)
with open(filepath, "w", encoding="utf-8") as f:
f.write(f"VocaLinux Log Export\n")
f.write("VocaLinux Log Export\n")
f.write(f"Generated: {datetime.now().isoformat()}\n")
f.write(f"Total Records: {len(records)}\n")
if level_filter:
src/vocalinux/ui/settings_dialog.py
+1 -1
View File
@@ -25,7 +25,7 @@ gi.require_version("Gtk", "3.0")
from gi.repository import GLib, Gtk # noqa: E402
from ..common_types import RecognitionState # noqa: E402
from ..utils.vosk_model_info import SUPPORTED_LANGUAGES, VOSK_MODEL_INFO
from ..utils.vosk_model_info import SUPPORTED_LANGUAGES, VOSK_MODEL_INFO # noqa: E402
# Avoid circular imports for type checking
if TYPE_CHECKING:
src/vocalinux/ui/tray_indicator.py
+5 -7
View File
@@ -8,9 +8,7 @@ recognition process and displaying its status.
import logging
import os
import signal
import sys
import threading
from typing import Callable, Dict, Optional
from typing import Callable
import gi
@@ -33,9 +31,9 @@ from gi.repository import GdkPixbuf, GLib, GObject, Gtk
from ..common_types import RecognitionState, SpeechRecognitionManagerProtocol, TextInjectorProtocol
# Import necessary components
from .config_manager import ConfigManager # Added
from .keyboard_shortcuts import KeyboardShortcutManager
from .settings_dialog import SettingsDialog # Added
from .config_manager import ConfigManager # noqa: E402
from .keyboard_shortcuts import KeyboardShortcutManager # noqa: E402
from .settings_dialog import SettingsDialog # noqa: E402
logger = logging.getLogger(__name__)
@@ -44,7 +42,7 @@ APP_ID = "vocalinux"
# Import the centralized resource manager
from ..utils.resource_manager import ResourceManager
from ..utils.resource_manager import ResourceManager # noqa: E402
# Initialize resource manager
_resource_manager = ResourceManager()
src/vocalinux/utils/resource_manager.py
+1 -1
View File
@@ -9,7 +9,7 @@ import logging
import os
import sys
from pathlib import Path
from typing import Optional
from typing import Optional # noqa: F401
logger = logging.getLogger(__name__)
tests/test_audio_feedback.py
+1 -2
View File
@@ -2,11 +2,10 @@
Tests for the audio feedback functionality.
"""
import importlib
import os
import sys
import unittest
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import pytest
tests/test_command_processor.py
-1
View File
@@ -3,7 +3,6 @@ Tests for the command processor functionality.
"""
import unittest
from unittest.mock import MagicMock, patch
# Update import path to use the new package structure
from vocalinux.speech_recognition.command_processor import CommandProcessor
tests/test_config_manager.py
+3 -3
View File
@@ -6,10 +6,10 @@ import json
import os
import tempfile
import unittest
from unittest.mock import MagicMock, patch
from unittest.mock import patch
# Update import path to use the new package structure
from vocalinux.ui.config_manager import CONFIG_DIR, CONFIG_FILE, DEFAULT_CONFIG, ConfigManager
from vocalinux.ui.config_manager import DEFAULT_CONFIG, ConfigManager
class TestConfigManager(unittest.TestCase):
@@ -61,7 +61,7 @@ class TestConfigManager(unittest.TestCase):
self.assertFalse(os.path.exists(self.temp_config_dir))
# Create config manager, which should create the directory
config_manager = ConfigManager()
ConfigManager()
self.assertTrue(os.path.exists(self.temp_config_dir))
def test_load_config(self):
tests/test_recognition_manager.py
+5 -10
View File
@@ -2,10 +2,7 @@
Tests for the speech recognition manager.
"""
import json
import os
import sys
import threading
import unittest
from unittest.mock import MagicMock, patch
@@ -21,12 +18,12 @@ sys.modules["numpy"] = MagicMock()
sys.modules["zipfile"] = MagicMock()
# Import the shared mock from conftest
from conftest import mock_audio_feedback
from conftest import mock_audio_feedback # noqa: E402
# Update import paths to use the new package structure
from vocalinux.common_types import RecognitionState
from vocalinux.speech_recognition.command_processor import CommandProcessor
from vocalinux.speech_recognition.recognition_manager import MODELS_DIR, SpeechRecognitionManager
from vocalinux.common_types import RecognitionState # noqa: E402
from vocalinux.speech_recognition.command_processor import CommandProcessor # noqa: E402
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager # noqa: E402
class TestSpeechRecognition(unittest.TestCase):
@@ -247,9 +244,7 @@ class TestSpeechRecognition(unittest.TestCase):
# Replace _process_final_buffer with our own implementation for testing
def mock_process():
# Skip file operations, just simulate the Whisper transcription directly
# Mock the Whisper result
result = {"text": "whisper test"}
# Process the Whisper result through the command processor
# Mock the Whisper result and process it
processed_text, actions = self.cmdProcessorMock("whisper test")
# Call the callbacks
for callback in manager.text_callbacks:
tests/test_settings_dialog.py
+1 -1
View File
@@ -17,7 +17,7 @@ sys.modules["gi.repository"] = MagicMock()
sys.modules["gi.repository.Gtk"] = MagicMock()
sys.modules["gi.repository.GLib"] = MagicMock()
from vocalinux.common_types import RecognitionState
from vocalinux.common_types import RecognitionState # noqa: E402
# Create mock for speech engine
mock_speech_engine = Mock()
tests/test_speech_recognition.py
+9 -16
View File
@@ -2,12 +2,9 @@
Tests for speech recognition functionality.
"""
import concurrent.futures
import sys # Add the missing import
import sys # noqa: F401
import unittest
from unittest.mock import MagicMock, PropertyMock, call, patch
import pytest
from unittest.mock import MagicMock, patch
# Create proper mock responses that can be safely JSON serialized
MOCK_VOSK_RESULT = '{"text": "test transcription"}'
@@ -67,11 +64,11 @@ mock_wave.open = MagicMock(return_value=mock_wave_file)
sys.modules["wave"] = mock_wave
# Import the shared mock from conftest
from conftest import mock_audio_feedback
from conftest import mock_audio_feedback # noqa: E402
# Update import paths to use the new package structure
from vocalinux.common_types import RecognitionState
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
from vocalinux.common_types import RecognitionState # noqa: E402
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager # noqa: E402
class TestSpeechRecognition(unittest.TestCase):
@@ -208,21 +205,21 @@ class TestSpeechRecognition(unittest.TestCase):
with patch.object(SpeechRecognitionManager, "_get_vosk_model_path") as mock_get_path:
# Test small model size
mock_get_path.return_value = "/path/to/vosk-model-small-en-us-0.15"
manager_small = SpeechRecognitionManager(engine="vosk", model_size="small")
_ = SpeechRecognitionManager(engine="vosk", model_size="small")
# Verify the small model path is constructed correctly
mock_get_path.assert_called_with()
# Test medium model size
mock_get_path.return_value = "/path/to/vosk-model-en-us-0.22"
manager_medium = SpeechRecognitionManager(engine="vosk", model_size="medium")
_ = SpeechRecognitionManager(engine="vosk", model_size="medium")
# Verify the medium model path is constructed correctly
mock_get_path.assert_called_with()
# Test large model size
mock_get_path.return_value = "/path/to/vosk-model-en-us-0.42"
manager_large = SpeechRecognitionManager(engine="vosk", model_size="large")
_ = SpeechRecognitionManager(engine="vosk", model_size="large")
# Verify the large model path is constructed correctly
mock_get_path.assert_called_with()
@@ -232,9 +229,7 @@ class TestSpeechRecognition(unittest.TestCase):
# Make os.path.exists return False to trigger download
with patch("os.path.exists", return_value=False):
# Instantiate manager with defer_download=False to trigger download
manager = SpeechRecognitionManager(
engine="vosk", model_size="small", defer_download=False
)
_ = SpeechRecognitionManager(engine="vosk", model_size="small", defer_download=False)
# Verify download was attempted
self.mock_download.assert_called_once()
@@ -352,8 +347,6 @@ class TestSpeechRecognition(unittest.TestCase):
# Setup side effect function that will be called when _process_final_buffer is called
def process_side_effect():
# Access transcription result directly from the model
result = {"text": "whisper transcription"}
# Simulate processing the result through command processor
processed_text, actions = self.mock_cmd.process_text("whisper transcription")
# Call callbacks as the real method would
tests/test_whisper_support.py
+7 -14
View File
@@ -3,9 +3,7 @@ Tests for Whisper speech recognition support.
"""
import sys
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import MagicMock, patch # noqa: F401
# Mock GTK and other dependencies before importing vocalinux
sys.modules["gi"] = MagicMock()
@@ -28,9 +26,7 @@ class TestWhisperSupport:
torch_mock.cuda.is_available.return_value = False
with patch.dict(sys.modules, {"whisper": whisper_mock, "torch": torch_mock}):
from vocalinux.speech_recognition.recognition_manager import (
SpeechRecognitionManager,
)
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
# Test that the method exists without actually calling it
assert hasattr(SpeechRecognitionManager, "_transcribe_with_whisper")
@@ -65,9 +61,7 @@ class TestWhisperSupport:
torch_mock.cuda.is_available.return_value = False
with patch.dict(sys.modules, {"whisper": whisper_mock, "torch": torch_mock}):
from vocalinux.speech_recognition.recognition_manager import (
SpeechRecognitionManager,
)
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
manager_class = SpeechRecognitionManager
@@ -95,11 +89,10 @@ class TestWhisperSupport:
# Patch modules and mock the download method to avoid network/file operations
with patch.dict(sys.modules, {"whisper": whisper_mock, "torch": torch_mock}), patch(
"vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager._download_whisper_model"
) as mock_download, patch("os.path.exists", return_value=True):
from vocalinux.speech_recognition.recognition_manager import (
SpeechRecognitionManager,
)
"vocalinux.speech_recognition.recognition_manager."
"SpeechRecognitionManager._download_whisper_model"
) as _mock_download, patch("os.path.exists", return_value=True):
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
# Create manager with Whisper engine and defer_download=False to trigger model loading
manager = SpeechRecognitionManager(