feat(vad): add Silero VAD and drop silence-only buffers (#447)
* feat(vad): replace amplitude-based VAD with Silero neural network Replace the simple amplitude threshold VAD with Silero VAD, a lightweight neural network model that classifies audio as speech/non-speech. This addresses the root cause of #302 (silence not stopping recording) and - Add silero_vad.py wrapper using ONNX Runtime with bundled Silero v5 model - Accumulate 512-sample chunks with 64-sample context window per official API - Map existing vad_sensitivity (1-5) to Silero thresholds (0.8-0.3) - Use max() across sub-chunks to avoid missing speech onsets - Fall back to amplitude VAD when onnxruntime is unavailable - Add onnxruntime as optional dependency (pip install vocalinux[vad]) Closes #344 * address review: add tests, LICENSE, logging, validation fixes - Add unit tests for SileroVAD (process, reset, fallback, sensitivity mapping) - Add MIT LICENSE for bundled Silero VAD model in data/ - Log which VAD backend is active on init - Validate chunk length in SileroVAD.process() - Skip is_speech check until a full Silero chunk is processed - Add thread safety note to SileroVAD docstring - Document resampling rate effect on VAD resolution - Comment explaining why load_silero_vad() catches broad Exception * address review: bundle license, document [vad] extra, surface backend in UI - Pack data/LICENSE into the wheel (pyproject package-data) so attribution travels with the model - README: short note on enabling neural VAD via vocalinux[vad] - Settings: VAD Sensitivity row subtitle shows the active backend; if onnxruntime is missing, an info-row beneath the group spells out the install command and the restart requirement - silero_vad.is_silero_available() probes onnxruntime + model presence w/o instantiating an InferenceSession - CI installs .[dev,vad] so Silero paths get exercised - Integration test for _record_audio Silero branch (mocked pyaudio + scripted SileroVAD), covering silence flush, speech reset, and sensitivity threshold mapping at sens=1 vs sens=5 - Tests no longer skip when other modules globally mock numpy: pop the top-level numpy ref so a fresh import gives the real package, and swap sys.modules["numpy"] inside the _record_audio call so the lazy import inside the function picks up real numpy too * fix(vad): drop silence-only recording buffers * test(vad): align stop buffer expectations * test(vad): preserve real numpy module * fix(install): install vad support opportunistically * test(vad): cover recognition manager vad edges --------- Co-authored-by: Sbenazar <127324689+Sbenazar@users.noreply.github.com>
This commit is contained in:
co-authored by
Sbenazar
parent
6dff5b2296
commit
3f5a5effff
@@ -143,7 +143,7 @@ jobs:
|
||||
python -m pip install --upgrade pip setuptools wheel
|
||||
# Install PyGObject from pip (with build deps installed above)
|
||||
pip install PyGObject
|
||||
pip install -e ".[dev]"
|
||||
pip install -e ".[dev,vad]"
|
||||
echo "VIRTUAL_ENV=$VIRTUAL_ENV" >> $GITHUB_ENV
|
||||
echo "PATH=$VIRTUAL_ENV/bin:$PATH" >> $GITHUB_ENV
|
||||
|
||||
|
||||
@@ -146,6 +146,7 @@ The installer will:
|
||||
- **Auto-detect your hardware** (GPU, RAM, Vulkan support)
|
||||
- **Recommend the best engine** for your system
|
||||
- **Download the appropriate model** (~39MB for whisper.cpp tiny)
|
||||
- **Install neural VAD support** when ONNX Runtime is available
|
||||
- **Install in ~1-2 minutes** (vs 5-10 min with old Whisper)
|
||||
|
||||
> **Note**: Always installs the latest release. For a specific version, check [GitHub Releases](https://github.com/jatinkrmalik/vocalinux/releases).
|
||||
@@ -294,6 +295,18 @@ Configuration is stored in `~/.config/vocalinux/config.json`:
|
||||
|
||||
You can also configure settings through the graphical Settings dialog (right-click the tray icon).
|
||||
|
||||
### Neural Voice Activity Detection
|
||||
|
||||
Vocalinux ships with a Silero VAD model and uses it automatically when `onnxruntime` is available. The official installer attempts to install this support automatically. Without it, recording falls back to the simpler amplitude-threshold VAD.
|
||||
|
||||
For manual or PyPI installs, enable neural VAD with:
|
||||
|
||||
```bash
|
||||
pip install "vocalinux[vad]"
|
||||
```
|
||||
|
||||
Restart Vocalinux after install. The Recognition tab in Settings shows which backend is active. The same `vad_sensitivity` (1-5) works for both -- it's mapped to a Silero probability threshold internally (1 = 0.8, 5 = 0.3).
|
||||
|
||||
## 🔧 Development Setup
|
||||
|
||||
```bash
|
||||
|
||||
+9
-4
@@ -15,6 +15,7 @@ curl -fsSL https://raw.githubusercontent.com/jatinkrmalik/vocalinux/main/install
|
||||
That's it! The installer handles everything automatically:
|
||||
- ✅ Installs whisper.cpp (~1-2 minutes, no heavy dependencies!)
|
||||
- ✅ Auto-detects your GPU (AMD, Intel, NVIDIA all supported)
|
||||
- ✅ Installs neural VAD support when ONNX Runtime is available
|
||||
- ✅ Downloads the tiny model (~39MB)
|
||||
- ✅ Configures everything automatically
|
||||
|
||||
@@ -202,12 +203,13 @@ Examples:
|
||||
- **whisper.cpp** (default): High-performance C++ engine with Vulkan GPU support
|
||||
- **Whisper**: OpenAI's PyTorch-based engine (NVIDIA GPU only)
|
||||
- **VOSK**: Lightweight engine for older systems
|
||||
5. **Downloads speech recognition models**:
|
||||
5. **Installs neural VAD support** when ONNX Runtime is available, with a safe amplitude-VAD fallback otherwise
|
||||
6. **Downloads speech recognition models**:
|
||||
- whisper.cpp: ~39MB tiny model (or larger if selected)
|
||||
- Whisper: ~75MB tiny model + PyTorch dependencies (~2.3GB with CUDA)
|
||||
- VOSK: ~40MB small model
|
||||
6. **Installs desktop integration** (icons, .desktop file)
|
||||
7. **Creates activation script** for easy environment activation
|
||||
7. **Installs desktop integration** (icons, .desktop file)
|
||||
8. **Creates activation script** for easy environment activation
|
||||
|
||||
### Installation Time Comparison
|
||||
|
||||
@@ -397,8 +399,11 @@ pip install .
|
||||
# With Whisper support
|
||||
pip install ".[whisper]"
|
||||
|
||||
# With neural VAD support
|
||||
pip install ".[vad]"
|
||||
|
||||
# Development mode
|
||||
pip install -e ".[dev]"
|
||||
pip install -e ".[dev,vad]"
|
||||
```
|
||||
|
||||
For PyPI instead of a local checkout, use `pip install vocalinux` after installing
|
||||
|
||||
@@ -205,6 +205,7 @@ The installer will:
|
||||
- ✅ Update your existing installation
|
||||
- ✅ Preserve your configuration and models
|
||||
- ✅ Install any new dependencies
|
||||
- ✅ Attempt to install neural VAD support, falling back safely if ONNX Runtime is unavailable
|
||||
|
||||
### If You Installed from Source
|
||||
|
||||
|
||||
+40
-2
@@ -1659,7 +1659,7 @@ install_system_dependencies() {
|
||||
print_info "4. Or install from source in a virtual environment:"
|
||||
print_info " python3 -m venv venv"
|
||||
print_info " source venv/bin/activate"
|
||||
print_info " pip install -e .[whisper]"
|
||||
print_info " pip install -e .[whisper,vad]"
|
||||
print_info ""
|
||||
print_info "For more information, see the project wiki:"
|
||||
print_info " https://github.com/jatinkrmalik/vocalinux/wiki"
|
||||
@@ -2402,6 +2402,40 @@ install_python_package() {
|
||||
return $?
|
||||
}
|
||||
|
||||
# Silero/ONNX Runtime gives much better speech/silence decisions, but
|
||||
# onnxruntime wheels are not guaranteed for every Python/platform combo.
|
||||
# Install it opportunistically so fresh installs and rerun-updates get the
|
||||
# neural VAD when available without blocking the amplitude fallback path.
|
||||
install_vad_support() {
|
||||
local PIP_LOG_FILE="$1"
|
||||
local EDITABLE_MODE="${2:-no}"
|
||||
|
||||
print_info "Installing neural VAD support (Silero / ONNX Runtime)..."
|
||||
local VAD_INSTALL_SUCCESS=false
|
||||
if [[ "$EDITABLE_MODE" == "yes" ]]; then
|
||||
pip install -e ".[vad]" --log "$PIP_LOG_FILE" && VAD_INSTALL_SUCCESS=true
|
||||
else
|
||||
pip install ".[vad]" --log "$PIP_LOG_FILE" && VAD_INSTALL_SUCCESS=true
|
||||
fi
|
||||
|
||||
if [[ "$VAD_INSTALL_SUCCESS" == "true" ]]; then
|
||||
if "$VENV_DIR/bin/python" - <<'PY' 2>/dev/null
|
||||
from vocalinux.speech_recognition.silero_vad import is_silero_available
|
||||
raise SystemExit(0 if is_silero_available() else 1)
|
||||
PY
|
||||
then
|
||||
print_success "Neural VAD support installed and verified successfully."
|
||||
else
|
||||
print_warning "Neural VAD dependencies installed, but Silero VAD could not be verified."
|
||||
print_warning "Vocalinux will use amplitude-based VAD until this is resolved."
|
||||
fi
|
||||
else
|
||||
print_warning "Failed to install neural VAD support."
|
||||
print_warning "Vocalinux will still work using amplitude-based VAD."
|
||||
print_warning "Check the pip log for details: $PIP_LOG_FILE"
|
||||
fi
|
||||
}
|
||||
|
||||
if [[ "$DEV_MODE" == "yes" ]]; then
|
||||
print_info "Installing Vocalinux in development mode..."
|
||||
|
||||
@@ -2420,11 +2454,13 @@ install_python_package() {
|
||||
|
||||
# Install all optional dependencies for development
|
||||
print_info "Installing all optional dependencies for development..."
|
||||
pip install ".[whisper,dev]" --log "$PIP_LOG_FILE" || {
|
||||
pip install -e ".[whisper,dev]" --log "$PIP_LOG_FILE" || {
|
||||
print_warning "Failed to install some optional dependencies."
|
||||
print_warning "Some features may not work correctly."
|
||||
}
|
||||
|
||||
install_vad_support "$PIP_LOG_FILE" yes
|
||||
|
||||
if [[ "${SELECTED_ENGINE:-whisper_cpp}" == "whisper_cpp" ]]; then
|
||||
install_whispercpp_with_gpu_support "$PIP_LOG_FILE"
|
||||
fi
|
||||
@@ -2438,6 +2474,8 @@ install_python_package() {
|
||||
return 1
|
||||
}
|
||||
|
||||
install_vad_support "$PIP_LOG_FILE"
|
||||
|
||||
# Engine installation logic:
|
||||
# - SELECTED_ENGINE is set by interactive mode or --engine flag
|
||||
# - WHISPERCPP_BACKEND is set by interactive mode ("gpu" or "cpu")
|
||||
|
||||
+5
-1
@@ -59,6 +59,9 @@ whisper = [
|
||||
"torch>=2.0.0",
|
||||
"torchaudio>=2.0.0",
|
||||
]
|
||||
vad = [
|
||||
"onnxruntime>=1.16.0",
|
||||
]
|
||||
dev = [
|
||||
"pytest>=7.0.0",
|
||||
"pytest-cov>=4.0.0",
|
||||
@@ -96,8 +99,9 @@ package-dir = {"" = "src"}
|
||||
version = {attr = "vocalinux.version.__version__"}
|
||||
|
||||
[tool.setuptools.package-data]
|
||||
"*" = ["*.svg", "*.wav"]
|
||||
"*" = ["*.svg", "*.wav", "*.onnx"]
|
||||
"vocalinux" = ["resources/**/*"]
|
||||
"vocalinux.speech_recognition" = ["data/*.onnx", "data/LICENSE"]
|
||||
|
||||
[tool.black]
|
||||
line-length = 100
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
The silero_vad.onnx model is from the Silero VAD project:
|
||||
|
||||
https://github.com/snakers4/silero-vad
|
||||
|
||||
Version: v5 (ONNX export)
|
||||
License: MIT
|
||||
Copyright (c) Silero Team
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
Binary file not shown.
@@ -22,6 +22,7 @@ from ..ui.audio_feedback import play_error_sound, play_start_sound, play_stop_so
|
||||
from ..utils.vosk_model_info import VOSK_MODEL_INFO
|
||||
from ..utils.whispercpp_model_info import WHISPERCPP_MODEL_INFO, get_model_path, is_model_downloaded
|
||||
from .command_processor import CommandProcessor
|
||||
from .silero_vad import SILERO_CHUNK_SIZE, load_silero_vad
|
||||
|
||||
|
||||
# ALSA error handler to suppress warnings during PyAudio initialization
|
||||
@@ -742,6 +743,13 @@ class SpeechRecognitionManager:
|
||||
self.vad_sensitivity = kwargs.get("vad_sensitivity", 3)
|
||||
self.silence_timeout = kwargs.get("silence_timeout", 2.0)
|
||||
|
||||
# Silero VAD (neural-network-based, falls back to amplitude if unavailable)
|
||||
self._silero_vad = load_silero_vad()
|
||||
if self._silero_vad is not None:
|
||||
logger.info("Using Silero neural VAD")
|
||||
else:
|
||||
logger.info("Using amplitude-based VAD (install vocalinux[vad] for neural VAD)")
|
||||
|
||||
# Audio device selection (None means use system default)
|
||||
self.audio_device_index = kwargs.get("audio_device_index", None)
|
||||
|
||||
@@ -764,6 +772,7 @@ class SpeechRecognitionManager:
|
||||
self.should_record = False
|
||||
self._recognition_mode = "toggle" # "toggle" or "push_to_talk"
|
||||
self.audio_buffer = []
|
||||
self._recording_segment_has_speech = False
|
||||
self._buffer_lock = threading.Lock() # Thread safety for audio_buffer
|
||||
self._model_lock = threading.Lock() # Thread safety for model/recognizer access
|
||||
self._segment_queue = queue.Queue(maxsize=32)
|
||||
@@ -1895,10 +1904,17 @@ class SpeechRecognitionManager:
|
||||
self.stop_sound_guard_ms,
|
||||
)
|
||||
|
||||
if self.audio_buffer:
|
||||
logger.debug(f"Enqueuing final buffer with {len(self.audio_buffer)} chunks")
|
||||
if self.audio_buffer and self._recording_segment_has_speech:
|
||||
logger.debug(f"Enqueuing final speech buffer with {len(self.audio_buffer)} chunks")
|
||||
self._enqueue_audio_segment(self.audio_buffer)
|
||||
self.audio_buffer = []
|
||||
elif self.audio_buffer:
|
||||
logger.debug(
|
||||
"Dropping final audio buffer with no detected speech "
|
||||
f"({len(self.audio_buffer)} chunks)"
|
||||
)
|
||||
self.audio_buffer = []
|
||||
self._recording_segment_has_speech = False
|
||||
|
||||
# Wake up recognition thread so it can drain queued segments and stop
|
||||
self._signal_recognition_stop()
|
||||
@@ -2016,8 +2032,20 @@ class SpeechRecognitionManager:
|
||||
# Record audio while should_record is True
|
||||
silence_counter = 0
|
||||
speech_detected_in_session = False
|
||||
self._recording_segment_has_speech = False
|
||||
log_level_interval = 0 # Counter for periodic level logging
|
||||
max_level_seen = 0.0
|
||||
# Accumulator for 512-sample Silero chunks. When the capture rate
|
||||
# is higher than 16 kHz (e.g. 48 kHz), resampling produces fewer
|
||||
# than 1024 samples per read (~341 at 48 kHz), so the buffer may
|
||||
# need several reads to fill a full 512-sample chunk. This is
|
||||
# expected -- VAD decisions simply arrive less frequently (every
|
||||
# ~128 ms instead of ~64 ms) with no impact on accuracy.
|
||||
silero_chunk_buf = np.array([], dtype=np.int16)
|
||||
|
||||
# Reset Silero VAD state for this recording session
|
||||
if self._silero_vad is not None:
|
||||
self._silero_vad.reset()
|
||||
|
||||
while self.should_record:
|
||||
try:
|
||||
@@ -2057,7 +2085,7 @@ class SpeechRecognitionManager:
|
||||
|
||||
self.audio_buffer.append(data)
|
||||
|
||||
# Simple Voice Activity Detection (VAD)
|
||||
# Voice Activity Detection (VAD)
|
||||
audio_data = np.frombuffer(data, dtype=np.int16)
|
||||
volume = np.abs(audio_data).mean()
|
||||
|
||||
@@ -2082,22 +2110,54 @@ class SpeechRecognitionManager:
|
||||
)
|
||||
log_level_interval = 0
|
||||
|
||||
# Threshold based on sensitivity (1-5)
|
||||
# Ensure vad_sensitivity is treated as integer for calculation
|
||||
try:
|
||||
vad_sens = int(self.vad_sensitivity)
|
||||
threshold = 500 / max(1, min(5, vad_sens)) # Use self.vad_sensitivity
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
f"Invalid VAD sensitivity value: {self.vad_sensitivity}. Using default 3."
|
||||
)
|
||||
threshold = 500 / 3
|
||||
# Determine if current chunk contains speech
|
||||
is_speech = False
|
||||
if self._silero_vad is not None:
|
||||
# Silero VAD: accumulate samples into 512-sample chunks
|
||||
speech_prob = 0.0
|
||||
chunk_processed = False
|
||||
silero_chunk_buf = np.concatenate([silero_chunk_buf, audio_data])
|
||||
while len(silero_chunk_buf) >= SILERO_CHUNK_SIZE:
|
||||
chunk_512 = silero_chunk_buf[:SILERO_CHUNK_SIZE]
|
||||
silero_chunk_buf = silero_chunk_buf[SILERO_CHUNK_SIZE:]
|
||||
speech_prob = max(speech_prob, self._silero_vad.process(chunk_512))
|
||||
chunk_processed = True
|
||||
|
||||
if volume < threshold: # Silence
|
||||
# Map vad_sensitivity (1-5) to threshold:
|
||||
# 1 (least sensitive) -> 0.8, 5 (most sensitive) -> 0.3
|
||||
try:
|
||||
vad_sens = int(self.vad_sensitivity)
|
||||
vad_sens = max(1, min(5, vad_sens))
|
||||
except ValueError:
|
||||
vad_sens = 3
|
||||
silero_threshold = 0.8 - (vad_sens - 1) * 0.125
|
||||
|
||||
# Skip speech decision until at least one full chunk
|
||||
# has been processed to avoid false silence detection
|
||||
if chunk_processed:
|
||||
is_speech = speech_prob >= silero_threshold
|
||||
else:
|
||||
# Amplitude fallback when Silero is unavailable
|
||||
try:
|
||||
vad_sens = int(self.vad_sensitivity)
|
||||
threshold = 500 / max(1, min(5, vad_sens))
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
f"Invalid VAD sensitivity value: {self.vad_sensitivity}. Using default 3."
|
||||
)
|
||||
threshold = 500 / 3
|
||||
is_speech = volume >= threshold
|
||||
|
||||
if not is_speech: # Silence
|
||||
silence_counter += CHUNK / RATE # Convert chunks to seconds
|
||||
if silence_counter > self.silence_timeout: # Use self.silence_timeout
|
||||
if silence_counter > self.silence_timeout:
|
||||
if len(self.audio_buffer) > 0:
|
||||
if self._recognition_mode == "push_to_talk":
|
||||
if not self._recording_segment_has_speech:
|
||||
logger.debug(
|
||||
"Silence detected with no speech, dropping audio buffer"
|
||||
)
|
||||
self.audio_buffer = []
|
||||
elif self._recognition_mode == "push_to_talk":
|
||||
logger.debug(
|
||||
"Silence detected in push-to-talk mode, "
|
||||
"deferring transcription until key release"
|
||||
@@ -2106,13 +2166,21 @@ class SpeechRecognitionManager:
|
||||
logger.debug("Silence detected, queueing audio segment")
|
||||
self._enqueue_audio_segment(self.audio_buffer)
|
||||
self.audio_buffer = []
|
||||
self._recording_segment_has_speech = False
|
||||
silence_counter = 0
|
||||
else: # Speech
|
||||
self._recording_segment_has_speech = True
|
||||
if not speech_detected_in_session:
|
||||
logger.debug(
|
||||
f"Speech detected (level={normalized_level:.1f}%, "
|
||||
f"threshold={500 / max(1, min(5, int(self.vad_sensitivity))):.0f})"
|
||||
)
|
||||
if self._silero_vad is not None:
|
||||
logger.debug(
|
||||
f"Speech detected (silero_prob={speech_prob:.2f}, "
|
||||
f"threshold={silero_threshold:.3f})"
|
||||
)
|
||||
else:
|
||||
logger.debug(
|
||||
f"Speech detected (level={normalized_level:.1f}%, "
|
||||
f"threshold={500 / max(1, min(5, int(self.vad_sensitivity))):.0f})"
|
||||
)
|
||||
speech_detected_in_session = True
|
||||
silence_counter = 0
|
||||
except (IOError, OSError) as e:
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
"""
|
||||
Silero VAD wrapper for Vocalinux.
|
||||
|
||||
Provides neural-network-based Voice Activity Detection using the Silero VAD
|
||||
ONNX model. Falls back gracefully when onnxruntime is unavailable.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
import numpy as np
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Path to the bundled ONNX model
|
||||
_MODEL_PATH = os.path.join(os.path.dirname(__file__), "data", "silero_vad.onnx")
|
||||
|
||||
# Silero VAD expects 512 samples per chunk at 16kHz (32ms)
|
||||
SILERO_CHUNK_SIZE = 512
|
||||
SILERO_SAMPLE_RATE = 16000
|
||||
|
||||
# Context window size (prepended to each chunk for cross-chunk continuity)
|
||||
_CONTEXT_SIZE = 64
|
||||
|
||||
|
||||
class SileroVAD:
|
||||
"""Silero VAD wrapper using ONNX Runtime inference.
|
||||
|
||||
Not thread-safe -- process() and reset() mutate internal LSTM state.
|
||||
Currently only called from the recording thread in _record_audio().
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
import onnxruntime as ort
|
||||
|
||||
opts = ort.SessionOptions()
|
||||
opts.inter_op_num_threads = 1
|
||||
opts.intra_op_num_threads = 1
|
||||
opts.log_severity_level = 3 # suppress ort warnings
|
||||
self._session = ort.InferenceSession(
|
||||
_MODEL_PATH, sess_options=opts, providers=["CPUExecutionProvider"]
|
||||
)
|
||||
self._state = np.zeros((2, 1, 128), dtype=np.float32)
|
||||
self._context = np.zeros((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
self._sr = np.array(SILERO_SAMPLE_RATE, dtype=np.int64)
|
||||
logger.info("Silero VAD model loaded")
|
||||
|
||||
def reset(self):
|
||||
"""Reset internal LSTM state and context (call between utterances)."""
|
||||
self._state = np.zeros((2, 1, 128), dtype=np.float32)
|
||||
self._context = np.zeros((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
|
||||
def process(self, audio_int16: np.ndarray) -> float:
|
||||
"""Return speech probability for a 512-sample int16 chunk.
|
||||
|
||||
Args:
|
||||
audio_int16: numpy array of int16 PCM samples, length 512.
|
||||
|
||||
Returns:
|
||||
Speech probability in [0, 1].
|
||||
|
||||
Raises:
|
||||
ValueError: If audio_int16 length is not SILERO_CHUNK_SIZE.
|
||||
"""
|
||||
if len(audio_int16) != SILERO_CHUNK_SIZE:
|
||||
raise ValueError(f"Expected {SILERO_CHUNK_SIZE} samples, got {len(audio_int16)}")
|
||||
# Normalize int16 -> float32 [-1, 1]
|
||||
audio_f32 = audio_int16.astype(np.float32) / 32768.0
|
||||
audio_f32 = audio_f32.reshape(1, -1)
|
||||
|
||||
# Prepend context from previous chunk (64 samples) for cross-chunk continuity
|
||||
input_with_context = np.concatenate([self._context, audio_f32], axis=1)
|
||||
self._context = audio_f32[:, -_CONTEXT_SIZE:]
|
||||
|
||||
output, self._state = self._session.run(
|
||||
None, {"input": input_with_context, "sr": self._sr, "state": self._state}
|
||||
)
|
||||
return float(output[0][0])
|
||||
|
||||
|
||||
def is_silero_available() -> bool:
|
||||
"""Cheap probe for whether neural VAD can be loaded.
|
||||
|
||||
Checks onnxruntime importability and model file presence without
|
||||
instantiating an InferenceSession.
|
||||
"""
|
||||
try:
|
||||
import onnxruntime # noqa: F401
|
||||
except ImportError:
|
||||
return False
|
||||
return os.path.exists(_MODEL_PATH)
|
||||
|
||||
|
||||
def load_silero_vad():
|
||||
"""Try to load Silero VAD, return None on failure."""
|
||||
try:
|
||||
vad = SileroVAD()
|
||||
return vad
|
||||
except Exception as e:
|
||||
# Catches ImportError (missing onnxruntime), FileNotFoundError,
|
||||
# onnxruntime.capi.onnxruntime_pybind11_state.NoSuchFile (missing model),
|
||||
# and RuntimeError (corrupted model / ONNX errors).
|
||||
logger.warning(f"Silero VAD unavailable, falling back to amplitude VAD: {e}")
|
||||
return None
|
||||
@@ -28,6 +28,7 @@ gi.require_version("Gdk", "3.0")
|
||||
from gi.repository import Gdk, GLib, Gtk, Pango # noqa: E402
|
||||
|
||||
from ..common_types import RecognitionState # noqa: E402
|
||||
from ..speech_recognition.silero_vad import is_silero_available # noqa: E402
|
||||
from ..utils.vosk_model_info import SUPPORTED_LANGUAGES, VOSK_MODEL_INFO # noqa: E402
|
||||
from ..utils.whispercpp_model_info import (
|
||||
WHISPERCPP_MODEL_INFO,
|
||||
@@ -1162,12 +1163,18 @@ class SettingsDialog(Gtk.Dialog):
|
||||
self.vad_spin = Gtk.SpinButton.new_with_range(1, 5, 1)
|
||||
self.vad_spin.set_tooltip_text("Higher = more sensitive to quiet speech")
|
||||
_prevent_scroll_on_hover(self.vad_spin)
|
||||
vad_row = PreferenceRow(
|
||||
silero_active = is_silero_available()
|
||||
vad_subtitle = (
|
||||
"Sensitivity (1-5) -- backend: Silero neural VAD"
|
||||
if silero_active
|
||||
else "Sensitivity (1-5) -- backend: amplitude (install vocalinux[vad] for neural)"
|
||||
)
|
||||
self.vad_row = PreferenceRow(
|
||||
title="VAD Sensitivity",
|
||||
subtitle="Voice Activity Detection sensitivity (1-5)",
|
||||
subtitle=vad_subtitle,
|
||||
widget=self.vad_spin,
|
||||
)
|
||||
group.add_row(vad_row)
|
||||
group.add_row(self.vad_row)
|
||||
|
||||
# Silence Timeout
|
||||
self.silence_spin = Gtk.SpinButton.new_with_range(0.5, 5.0, 0.1)
|
||||
@@ -1196,6 +1203,33 @@ class SettingsDialog(Gtk.Dialog):
|
||||
|
||||
self.recognition_settings_tab.pack_start(group, False, False, 0)
|
||||
|
||||
if not silero_active:
|
||||
vad_info_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||||
vad_info_box.get_style_context().add_class("info-box")
|
||||
vad_info_box.set_margin_start(4)
|
||||
vad_info_box.set_margin_end(4)
|
||||
vad_info_box.set_margin_top(4)
|
||||
|
||||
info_icon = Gtk.Image.new_from_icon_name(
|
||||
"dialog-information-symbolic", Gtk.IconSize.MENU
|
||||
)
|
||||
vad_info_box.pack_start(info_icon, False, False, 0)
|
||||
|
||||
vad_info_label = Gtk.Label(
|
||||
label=(
|
||||
"Neural VAD is more accurate at separating speech from background "
|
||||
"noise. To enable it, install onnxruntime and restart Vocalinux:\n"
|
||||
' pip install "vocalinux[vad]"'
|
||||
),
|
||||
xalign=0,
|
||||
wrap=True,
|
||||
selectable=True,
|
||||
)
|
||||
vad_info_label.get_style_context().add_class("tip-label")
|
||||
vad_info_box.pack_start(vad_info_label, True, True, 0)
|
||||
|
||||
self.recognition_settings_tab.pack_start(vad_info_box, False, False, 0)
|
||||
|
||||
# Connect signals
|
||||
self.vad_spin.connect("value-changed", self._on_vad_changed)
|
||||
self.silence_spin.connect("value-changed", self._on_silence_changed)
|
||||
|
||||
@@ -14,6 +14,16 @@ import pytest
|
||||
# Set PYTEST_RUNNING early so audio_feedback module can detect it
|
||||
os.environ["PYTEST_RUNNING"] = "1"
|
||||
|
||||
# Several legacy tests replace sys.modules["numpy"] with MagicMock at module
|
||||
# import time. Preserve the real module once so newer tests can restore it
|
||||
# without unloading/re-importing NumPy's compiled extensions.
|
||||
try:
|
||||
import numpy as _real_numpy
|
||||
except ImportError:
|
||||
_real_numpy = None
|
||||
else:
|
||||
sys._vocalinux_real_numpy = _real_numpy
|
||||
|
||||
# Prevent specific known-blocking daemon threads from starting during tests.
|
||||
# Source code in ibus_engine.py and evdev_backend.py spawns daemon threads
|
||||
# (socket server, device monitor) that block on socket.accept() or
|
||||
|
||||
@@ -643,6 +643,7 @@ class TestStartStopRecognition(unittest.TestCase):
|
||||
manager.state = RecognitionState.LISTENING
|
||||
manager.should_record = True
|
||||
manager.audio_buffer = [b"\x00\x00" for _ in range(20)]
|
||||
manager._recording_segment_has_speech = True
|
||||
|
||||
# Create dummy threads
|
||||
manager.audio_thread = MagicMock()
|
||||
@@ -656,6 +657,27 @@ class TestStartStopRecognition(unittest.TestCase):
|
||||
manager.stop_recognition()
|
||||
assert manager.should_record is False
|
||||
assert len(enqueue_mock.call_args.args[0]) == 17
|
||||
assert manager._recording_segment_has_speech is False
|
||||
|
||||
def test_stop_recognition_drops_final_buffer_without_speech(self):
|
||||
"""Final silence-only buffers should not be queued for transcription."""
|
||||
manager = _make_manager()
|
||||
manager.state = RecognitionState.LISTENING
|
||||
manager.should_record = True
|
||||
manager.audio_buffer = [b"\x00\x00" for _ in range(20)]
|
||||
manager._recording_segment_has_speech = False
|
||||
|
||||
manager.audio_thread = MagicMock()
|
||||
manager.audio_thread.is_alive.return_value = False
|
||||
manager.recognition_thread = MagicMock()
|
||||
manager.recognition_thread.is_alive.return_value = False
|
||||
|
||||
with patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"):
|
||||
with patch.object(manager, "_signal_recognition_stop"):
|
||||
with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
|
||||
manager.stop_recognition()
|
||||
enqueue_mock.assert_not_called()
|
||||
assert manager.audio_buffer == []
|
||||
|
||||
|
||||
class TestDownloads(unittest.TestCase):
|
||||
|
||||
@@ -339,6 +339,7 @@ class TestBufferManagement(unittest.TestCase):
|
||||
manager.state = RecognitionState.LISTENING
|
||||
manager.should_record = True
|
||||
manager.audio_buffer = [b"small"]
|
||||
manager._recording_segment_has_speech = True
|
||||
manager.audio_thread = MagicMock()
|
||||
manager.audio_thread.is_alive.return_value = False
|
||||
manager.recognition_thread = MagicMock()
|
||||
@@ -357,6 +358,7 @@ class TestBufferManagement(unittest.TestCase):
|
||||
manager.state = RecognitionState.LISTENING
|
||||
manager.should_record = True
|
||||
manager.audio_buffer = [b"x" * 100 for _ in range(20)]
|
||||
manager._recording_segment_has_speech = True
|
||||
manager.audio_thread = MagicMock()
|
||||
manager.audio_thread.is_alive.return_value = False
|
||||
manager.recognition_thread = MagicMock()
|
||||
|
||||
@@ -0,0 +1,388 @@
|
||||
"""
|
||||
Integration-style tests for the Silero VAD path in _record_audio.
|
||||
|
||||
Drives the recording loop with a mocked PyAudio stream and a Mock SileroVAD
|
||||
returning scripted speech probabilities. Verifies:
|
||||
|
||||
- silence-only input -> no transcription segment is enqueued
|
||||
- speech followed by silence -> _enqueue_audio_segment is called after silence_timeout
|
||||
- speech -> silence_counter resets, no premature flush
|
||||
- vad_sensitivity threshold mapping kicks in (sens=1 vs sens=5 give different decisions)
|
||||
- Silero state is reset at the start of every recording session
|
||||
"""
|
||||
|
||||
import sys
|
||||
import unittest # noqa: E402
|
||||
from unittest.mock import MagicMock, patch # noqa: E402
|
||||
|
||||
# Earlier test modules install `sys.modules["numpy"] = MagicMock()` at module
|
||||
# load and don't restore it. Reuse the real module cached by conftest instead
|
||||
# of unloading/re-importing NumPy's compiled extensions.
|
||||
if isinstance(sys.modules.get("numpy"), MagicMock):
|
||||
_real_numpy = getattr(sys, "_vocalinux_real_numpy", None)
|
||||
if _real_numpy is not None:
|
||||
sys.modules["numpy"] = _real_numpy
|
||||
|
||||
import numpy as np # noqa: E402
|
||||
|
||||
# ---- Mock external deps that recognition_manager imports ----
|
||||
_MOCK_KEYS = ["vosk", "whisper", "torch", "pyaudio", "pywhispercpp", "pywhispercpp.model"]
|
||||
_ORIG_MODULES = {}
|
||||
for _k in _MOCK_KEYS:
|
||||
_ORIG_MODULES[_k] = sys.modules.get(_k)
|
||||
if _k not in sys.modules:
|
||||
sys.modules[_k] = MagicMock()
|
||||
|
||||
if "gi" not in sys.modules:
|
||||
sys.modules["gi"] = MagicMock()
|
||||
if "gi.repository" not in sys.modules:
|
||||
sys.modules["gi.repository"] = MagicMock()
|
||||
|
||||
from vocalinux.common_types import RecognitionState # noqa: E402
|
||||
from vocalinux.speech_recognition.recognition_manager import ( # noqa: E402
|
||||
SpeechRecognitionManager,
|
||||
)
|
||||
|
||||
for _k, _v in _ORIG_MODULES.items():
|
||||
if _v is not None:
|
||||
sys.modules[_k] = _v
|
||||
elif _k in sys.modules and isinstance(sys.modules[_k], MagicMock):
|
||||
del sys.modules[_k]
|
||||
|
||||
|
||||
CHUNK_BYTES = 1024 * 2 # 1024 int16 samples
|
||||
SILENT_CHUNK = b"\x00" * (1024 * 2) # 1024 int16 zero samples
|
||||
|
||||
|
||||
def _make_manager():
|
||||
"""Create a SpeechRecognitionManager with init paths stubbed out."""
|
||||
with (
|
||||
patch.object(SpeechRecognitionManager, "_init_vosk"),
|
||||
patch.object(SpeechRecognitionManager, "_init_whisper"),
|
||||
patch.object(SpeechRecognitionManager, "_init_whispercpp"),
|
||||
):
|
||||
mgr = SpeechRecognitionManager(
|
||||
engine="whisper_cpp",
|
||||
model_size="small",
|
||||
language="en-us",
|
||||
defer_download=True,
|
||||
)
|
||||
return mgr
|
||||
|
||||
|
||||
class TestManagerSileroInitialization(unittest.TestCase):
|
||||
def test_logs_when_silero_vad_loads(self):
|
||||
"""Manager initialization should report when neural VAD is active."""
|
||||
vad = MagicMock()
|
||||
with (
|
||||
patch.object(SpeechRecognitionManager, "_init_vosk"),
|
||||
patch.object(SpeechRecognitionManager, "_init_whisper"),
|
||||
patch.object(SpeechRecognitionManager, "_init_whispercpp"),
|
||||
patch(
|
||||
"vocalinux.speech_recognition.recognition_manager.load_silero_vad",
|
||||
return_value=vad,
|
||||
),
|
||||
patch("vocalinux.speech_recognition.recognition_manager.logger") as logger_mock,
|
||||
):
|
||||
mgr = SpeechRecognitionManager(
|
||||
engine="whisper_cpp",
|
||||
model_size="small",
|
||||
language="en-us",
|
||||
defer_download=True,
|
||||
)
|
||||
|
||||
self.assertIs(mgr._silero_vad, vad)
|
||||
logger_mock.info.assert_any_call("Using Silero neural VAD")
|
||||
|
||||
|
||||
def _make_pyaudio_module(stream):
|
||||
"""Build a fake pyaudio module that returns the given stream from audio.open()."""
|
||||
audio = MagicMock()
|
||||
audio.get_device_count.return_value = 1
|
||||
audio.get_default_input_device_info.return_value = {
|
||||
"index": 0,
|
||||
"name": "mock",
|
||||
"maxInputChannels": 1,
|
||||
"defaultSampleRate": 16000,
|
||||
}
|
||||
audio.get_device_info_by_index.return_value = {
|
||||
"index": 0,
|
||||
"name": "mock",
|
||||
"maxInputChannels": 1,
|
||||
"defaultSampleRate": 16000,
|
||||
}
|
||||
audio.is_format_supported.return_value = True
|
||||
audio.open.return_value = stream
|
||||
|
||||
pyaudio_mod = MagicMock(paInt16=8, PyAudio=MagicMock(return_value=audio))
|
||||
return pyaudio_mod, audio
|
||||
|
||||
|
||||
def _scripted_stream(probs_per_chunk, manager, max_iters=None, chunk_bytes=None):
|
||||
"""Return a Mock stream whose read() drives `manager.should_record` to False
|
||||
once we've consumed `max_iters` (or len(probs_per_chunk)) chunks.
|
||||
|
||||
Each read() returns silent bytes; the speech/silence decision comes from
|
||||
the scripted probabilities fed via the mocked SileroVAD. `chunk_bytes`
|
||||
overrides the default 1024-int16 size (e.g. stereo at 16 kHz needs 4096
|
||||
bytes per CHUNK frame).
|
||||
"""
|
||||
stream = MagicMock()
|
||||
counter = {"n": 0}
|
||||
limit = max_iters if max_iters is not None else len(probs_per_chunk)
|
||||
payload = b"\x00" * (chunk_bytes if chunk_bytes is not None else 1024 * 2)
|
||||
|
||||
def _read(*args, **kwargs):
|
||||
counter["n"] += 1
|
||||
if counter["n"] >= limit:
|
||||
manager.should_record = False
|
||||
return payload
|
||||
|
||||
stream.read.side_effect = _read
|
||||
return stream
|
||||
|
||||
|
||||
def _make_silero(probs):
|
||||
"""Mock SileroVAD: process() returns scripted probs, repeating the last
|
||||
value forever once the list is exhausted (each outer read() drives 2 silero
|
||||
chunks of 512 samples, so we need plenty of values).
|
||||
"""
|
||||
vad = MagicMock()
|
||||
state = {"i": 0, "last": probs[-1] if probs else 0.0}
|
||||
|
||||
def _process(_chunk):
|
||||
i = state["i"]
|
||||
if i < len(probs):
|
||||
state["i"] = i + 1
|
||||
return probs[i]
|
||||
return state["last"]
|
||||
|
||||
vad.process.side_effect = _process
|
||||
return vad
|
||||
|
||||
|
||||
class TestRecordAudioSileroPath(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mgr = _make_manager()
|
||||
self.mgr.state = RecognitionState.LISTENING
|
||||
self.mgr.should_record = True
|
||||
self.mgr.audio_buffer = []
|
||||
self.mgr.silence_timeout = 0.5 # short timeout for fast tests
|
||||
self.mgr.vad_sensitivity = 3
|
||||
|
||||
# Capture flushes triggered by the silence-detection branch
|
||||
self.enqueued = []
|
||||
self.mgr._enqueue_audio_segment = lambda buf: self.enqueued.append(list(buf))
|
||||
|
||||
def _drive(
|
||||
self,
|
||||
probs,
|
||||
vad_sensitivity=3,
|
||||
max_iters=None,
|
||||
channels=1,
|
||||
rate=16000,
|
||||
chunk_bytes=None,
|
||||
):
|
||||
"""Run _record_audio with the given scripted Silero probabilities."""
|
||||
self.mgr.vad_sensitivity = vad_sensitivity
|
||||
self.mgr._silero_vad = _make_silero(probs)
|
||||
|
||||
stream = _scripted_stream(
|
||||
probs,
|
||||
self.mgr,
|
||||
max_iters=max_iters or len(probs),
|
||||
chunk_bytes=chunk_bytes,
|
||||
)
|
||||
pyaudio_mod, _audio = _make_pyaudio_module(stream)
|
||||
|
||||
# _record_audio does `import numpy as np` lazily inside the function;
|
||||
# earlier test modules may have left sys.modules["numpy"] as a
|
||||
# MagicMock, so put real numpy back for the duration of the call.
|
||||
with (
|
||||
patch.dict(sys.modules, {"pyaudio": pyaudio_mod, "numpy": np}),
|
||||
patch(
|
||||
"vocalinux.speech_recognition.recognition_manager._get_supported_channels",
|
||||
return_value=channels,
|
||||
),
|
||||
patch(
|
||||
"vocalinux.speech_recognition.recognition_manager._get_supported_sample_rate",
|
||||
return_value=rate,
|
||||
),
|
||||
):
|
||||
self.mgr._record_audio()
|
||||
|
||||
def test_reset_called_at_session_start(self):
|
||||
"""Silero state must be cleared between recording sessions."""
|
||||
self._drive(probs=[0.1] * 5)
|
||||
self.assertTrue(self.mgr._silero_vad.reset.called)
|
||||
|
||||
def test_silence_only_drops_buffer_without_enqueue(self):
|
||||
"""Sustained silence past silence_timeout should not reach transcription."""
|
||||
# Each chunk = 1024/16000 = 0.064s. silence_timeout=0.5s, so ~8 chunks
|
||||
# of silence crosses the timeout. Run 20 chunks to ensure the threshold fires.
|
||||
self._drive(probs=[0.05] * 20, vad_sensitivity=3)
|
||||
self.assertEqual(len(self.enqueued), 0, "silence-only input should be dropped")
|
||||
|
||||
def test_speech_then_silence_triggers_segment_enqueue(self):
|
||||
"""Speech followed by sustained silence should flush the speech segment."""
|
||||
probs = [0.95] * 6 + [0.05] * 20
|
||||
self._drive(probs=probs, vad_sensitivity=3)
|
||||
self.assertGreaterEqual(
|
||||
len(self.enqueued), 1, "expected speech-followed-by-silence to enqueue"
|
||||
)
|
||||
|
||||
def test_speech_prevents_premature_flush(self):
|
||||
"""Speech probs above threshold should reset the silence counter."""
|
||||
# Continuous speech (prob well above threshold for sens=3 -> 0.55).
|
||||
# 20 chunks * 0.064s = 1.28s, more than silence_timeout=0.5s,
|
||||
# but speech_counter never accumulates -> no flush.
|
||||
self._drive(probs=[0.95] * 20, vad_sensitivity=3)
|
||||
self.assertEqual(len(self.enqueued), 0, "speech should keep silence counter at 0, no flush")
|
||||
|
||||
def test_sensitivity_lowest_blocks_borderline_speech(self):
|
||||
"""At sensitivity=1 (threshold 0.8), prob=0.6 is silence."""
|
||||
# 20 chunks of borderline prob; with strict threshold these read as silence
|
||||
# and the silence-only buffer gets dropped.
|
||||
self._drive(probs=[0.6] * 20, vad_sensitivity=1)
|
||||
self.assertEqual(len(self.enqueued), 0)
|
||||
|
||||
def test_sensitivity_highest_accepts_borderline_speech(self):
|
||||
"""At sensitivity=5 (threshold 0.3), prob=0.6 is speech, no flush."""
|
||||
self._drive(probs=[0.6] * 20, vad_sensitivity=5)
|
||||
self.assertEqual(len(self.enqueued), 0)
|
||||
|
||||
def test_invalid_sensitivity_uses_default_silero_threshold(self):
|
||||
"""Bad sensitivity values should fall back to the default Silero threshold."""
|
||||
self._drive(probs=[0.95] * 20, vad_sensitivity="bad")
|
||||
self.assertEqual(len(self.enqueued), 0)
|
||||
|
||||
def test_push_to_talk_defers_speech_segment_until_release(self):
|
||||
"""Push-to-talk should keep a spoken segment buffered after silence."""
|
||||
self.mgr._recognition_mode = "push_to_talk"
|
||||
|
||||
self._drive(probs=[0.95] * 6 + [0.05] * 20, vad_sensitivity=3)
|
||||
|
||||
self.assertEqual(len(self.enqueued), 0)
|
||||
self.assertGreater(len(self.mgr.audio_buffer), 0)
|
||||
self.assertTrue(self.mgr._recording_segment_has_speech)
|
||||
|
||||
def test_stereo_capture_is_downmixed_without_error(self):
|
||||
"""CHANNELS=2 -> the stereo->mono branch runs and the loop completes."""
|
||||
self._drive(probs=[0.05] * 12, vad_sensitivity=3, channels=2, chunk_bytes=1024 * 2 * 2)
|
||||
# No assertion on enqueue count -- just verify the path runs cleanly.
|
||||
|
||||
def test_resample_branch_runs_at_48khz(self):
|
||||
"""rate=48000 -> the resample branch in _record_audio runs."""
|
||||
self._drive(probs=[0.05] * 12, vad_sensitivity=3, rate=48000)
|
||||
|
||||
|
||||
class TestRecordAudioAmplitudeFallback(unittest.TestCase):
|
||||
"""When onnxruntime is unavailable, _record_audio falls back to an amplitude
|
||||
threshold. These tests drive that path with `_silero_vad = None`.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.mgr = _make_manager()
|
||||
self.mgr.state = RecognitionState.LISTENING
|
||||
self.mgr.should_record = True
|
||||
self.mgr.audio_buffer = []
|
||||
self.mgr.silence_timeout = 0.5
|
||||
self.mgr.vad_sensitivity = 3
|
||||
self.mgr._silero_vad = None
|
||||
|
||||
self.enqueued = []
|
||||
self.mgr._enqueue_audio_segment = lambda buf: self.enqueued.append(list(buf))
|
||||
|
||||
def _drive(self, payload, max_iters):
|
||||
stream = MagicMock()
|
||||
counter = {"n": 0}
|
||||
|
||||
def _read(*a, **kw):
|
||||
counter["n"] += 1
|
||||
if counter["n"] >= max_iters:
|
||||
self.mgr.should_record = False
|
||||
return payload
|
||||
|
||||
stream.read.side_effect = _read
|
||||
pyaudio_mod, _ = _make_pyaudio_module(stream)
|
||||
with (
|
||||
patch.dict(sys.modules, {"pyaudio": pyaudio_mod, "numpy": np}),
|
||||
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,
|
||||
),
|
||||
):
|
||||
self.mgr._record_audio()
|
||||
|
||||
def test_silence_drops_buffer(self):
|
||||
"""All-zero samples -> volume below threshold -> no transcription segment."""
|
||||
self._drive(payload=b"\x00" * (1024 * 2), max_iters=20)
|
||||
self.assertEqual(len(self.enqueued), 0)
|
||||
|
||||
def test_invalid_sensitivity_uses_default_amplitude_threshold(self):
|
||||
"""Bad sensitivity values should fall back to the default amplitude threshold."""
|
||||
self.mgr.vad_sensitivity = "bad"
|
||||
self._drive(payload=b"\x00" * (1024 * 2), max_iters=20)
|
||||
self.assertEqual(len(self.enqueued), 0)
|
||||
|
||||
def test_loud_audio_is_speech(self):
|
||||
"""Loud samples -> volume above threshold -> speech, no flush."""
|
||||
loud = (np.full(1024, 20000, dtype=np.int16)).tobytes()
|
||||
self._drive(payload=loud, max_iters=20)
|
||||
self.assertEqual(len(self.enqueued), 0)
|
||||
|
||||
|
||||
class TestAudioLevelCallback(unittest.TestCase):
|
||||
"""The audio-level callback list is iterated every chunk with normalized
|
||||
level in [0, 100]. Verify a registered callback receives values.
|
||||
"""
|
||||
|
||||
def test_callback_receives_levels(self):
|
||||
mgr = _make_manager()
|
||||
mgr.state = RecognitionState.LISTENING
|
||||
mgr.should_record = True
|
||||
mgr.audio_buffer = []
|
||||
mgr.silence_timeout = 0.5
|
||||
mgr.vad_sensitivity = 3
|
||||
mgr._silero_vad = None
|
||||
mgr._enqueue_audio_segment = lambda buf: None
|
||||
|
||||
levels = []
|
||||
mgr._audio_level_callbacks = [levels.append]
|
||||
|
||||
stream = MagicMock()
|
||||
counter = {"n": 0}
|
||||
|
||||
def _read(*a, **kw):
|
||||
counter["n"] += 1
|
||||
if counter["n"] >= 5:
|
||||
mgr.should_record = False
|
||||
return b"\x00" * (1024 * 2)
|
||||
|
||||
stream.read.side_effect = _read
|
||||
pyaudio_mod, _ = _make_pyaudio_module(stream)
|
||||
with (
|
||||
patch.dict(sys.modules, {"pyaudio": pyaudio_mod, "numpy": np}),
|
||||
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,
|
||||
),
|
||||
):
|
||||
mgr._record_audio()
|
||||
|
||||
self.assertGreaterEqual(len(levels), 1)
|
||||
for v in levels:
|
||||
self.assertGreaterEqual(v, 0.0)
|
||||
self.assertLessEqual(v, 100.0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,243 @@
|
||||
"""
|
||||
Tests for Silero VAD wrapper (silero_vad.py).
|
||||
|
||||
Covers:
|
||||
- SileroVAD.process() return type and validation
|
||||
- SileroVAD.reset() state clearing
|
||||
- load_silero_vad() success and graceful fallback
|
||||
- Sensitivity-to-threshold mapping
|
||||
"""
|
||||
|
||||
import sys
|
||||
from unittest.mock import MagicMock, patch # noqa: E402
|
||||
|
||||
# Earlier test modules install `sys.modules["numpy"] = MagicMock()` at module
|
||||
# load and don't restore it. Reuse the real module cached by conftest instead
|
||||
# of unloading/re-importing NumPy's compiled extensions.
|
||||
if isinstance(sys.modules.get("numpy"), MagicMock):
|
||||
_real_numpy = getattr(sys, "_vocalinux_real_numpy", None)
|
||||
if _real_numpy is not None:
|
||||
sys.modules["numpy"] = _real_numpy
|
||||
|
||||
import numpy as np # noqa: E402
|
||||
import pytest # noqa: E402
|
||||
|
||||
from vocalinux.speech_recognition import silero_vad as _sv_mod # noqa: E402
|
||||
from vocalinux.speech_recognition.silero_vad import ( # noqa: E402
|
||||
_CONTEXT_SIZE,
|
||||
SILERO_CHUNK_SIZE,
|
||||
SILERO_SAMPLE_RATE,
|
||||
SileroVAD,
|
||||
is_silero_available,
|
||||
load_silero_vad,
|
||||
)
|
||||
|
||||
# silero_vad imported numpy when first loaded (likely with a mocked numpy
|
||||
# reference). Point its module-level `np` at the real numpy so SileroVAD
|
||||
# methods use real array operations.
|
||||
_sv_mod.np = np
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_mock_session(prob=0.05):
|
||||
"""Create a mock ONNX InferenceSession mimicking Silero VAD output."""
|
||||
session = MagicMock()
|
||||
state = np.zeros((2, 1, 128), dtype=np.float32)
|
||||
session.run.return_value = ([[prob]], state)
|
||||
return session
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSileroVADProcess:
|
||||
"""Test the process() method."""
|
||||
|
||||
def test_returns_float(self):
|
||||
vad = SileroVAD.__new__(SileroVAD)
|
||||
vad._session = _make_mock_session()
|
||||
vad._state = np.zeros((2, 1, 128), dtype=np.float32)
|
||||
vad._context = np.zeros((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
vad._sr = np.array(SILERO_SAMPLE_RATE, dtype=np.int64)
|
||||
|
||||
chunk = np.zeros(SILERO_CHUNK_SIZE, dtype=np.int16)
|
||||
prob = vad.process(chunk)
|
||||
assert isinstance(prob, float)
|
||||
|
||||
def test_prob_from_session(self):
|
||||
vad = SileroVAD.__new__(SileroVAD)
|
||||
vad._session = _make_mock_session(prob=0.42)
|
||||
vad._state = np.zeros((2, 1, 128), dtype=np.float32)
|
||||
vad._context = np.zeros((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
vad._sr = np.array(SILERO_SAMPLE_RATE, dtype=np.int64)
|
||||
|
||||
chunk = np.zeros(SILERO_CHUNK_SIZE, dtype=np.int16)
|
||||
prob = vad.process(chunk)
|
||||
assert abs(prob - 0.42) < 1e-6
|
||||
|
||||
def test_session_receives_correct_input_shape(self):
|
||||
"""Session should receive (1, 512+64) float32 input."""
|
||||
session = _make_mock_session()
|
||||
vad = SileroVAD.__new__(SileroVAD)
|
||||
vad._session = session
|
||||
vad._state = np.zeros((2, 1, 128), dtype=np.float32)
|
||||
vad._context = np.zeros((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
vad._sr = np.array(SILERO_SAMPLE_RATE, dtype=np.int64)
|
||||
|
||||
chunk = np.zeros(SILERO_CHUNK_SIZE, dtype=np.int16)
|
||||
vad.process(chunk)
|
||||
|
||||
call_kwargs = session.run.call_args
|
||||
input_tensor = call_kwargs[1]["input"] if call_kwargs[1] else call_kwargs[0][1]["input"]
|
||||
assert input_tensor.shape == (1, SILERO_CHUNK_SIZE + _CONTEXT_SIZE)
|
||||
assert input_tensor.dtype == np.float32
|
||||
|
||||
def test_wrong_length_raises_value_error(self):
|
||||
vad = SileroVAD.__new__(SileroVAD)
|
||||
vad._session = _make_mock_session()
|
||||
vad._state = np.zeros((2, 1, 128), dtype=np.float32)
|
||||
vad._context = np.zeros((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
vad._sr = np.array(SILERO_SAMPLE_RATE, dtype=np.int64)
|
||||
|
||||
with pytest.raises(ValueError, match="Expected 512"):
|
||||
vad.process(np.zeros(256, dtype=np.int16))
|
||||
with pytest.raises(ValueError, match="Expected 512"):
|
||||
vad.process(np.zeros(1024, dtype=np.int16))
|
||||
|
||||
def test_context_updated_after_process(self):
|
||||
"""Context should contain last 64 samples of the processed chunk."""
|
||||
vad = SileroVAD.__new__(SileroVAD)
|
||||
vad._session = _make_mock_session()
|
||||
vad._state = np.zeros((2, 1, 128), dtype=np.float32)
|
||||
vad._context = np.zeros((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
vad._sr = np.array(SILERO_SAMPLE_RATE, dtype=np.int64)
|
||||
|
||||
chunk = np.arange(SILERO_CHUNK_SIZE, dtype=np.int16)
|
||||
vad.process(chunk)
|
||||
|
||||
expected = chunk[-_CONTEXT_SIZE:].astype(np.float32) / 32768.0
|
||||
assert np.allclose(vad._context[0], expected)
|
||||
|
||||
|
||||
class TestSileroVADReset:
|
||||
"""Test state reset between sessions."""
|
||||
|
||||
def test_reset_clears_state(self):
|
||||
vad = SileroVAD.__new__(SileroVAD)
|
||||
vad._state = np.ones((2, 1, 128), dtype=np.float32)
|
||||
vad._context = np.ones((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
|
||||
vad.reset()
|
||||
|
||||
assert np.allclose(vad._state, 0.0)
|
||||
assert np.allclose(vad._context, 0.0)
|
||||
|
||||
def test_reset_preserves_correct_shapes(self):
|
||||
vad = SileroVAD.__new__(SileroVAD)
|
||||
vad._state = np.ones((2, 1, 128), dtype=np.float32)
|
||||
vad._context = np.ones((1, _CONTEXT_SIZE), dtype=np.float32)
|
||||
|
||||
vad.reset()
|
||||
|
||||
assert vad._state.shape == (2, 1, 128)
|
||||
assert vad._context.shape == (1, _CONTEXT_SIZE)
|
||||
|
||||
|
||||
class TestSensitivityMapping:
|
||||
"""Test the sensitivity -> threshold mapping used in recognition_manager."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"sensitivity,expected_threshold",
|
||||
[
|
||||
(1, 0.800),
|
||||
(2, 0.675),
|
||||
(3, 0.550),
|
||||
(4, 0.425),
|
||||
(5, 0.300),
|
||||
],
|
||||
)
|
||||
def test_mapping(self, sensitivity, expected_threshold):
|
||||
threshold = 0.8 - (sensitivity - 1) * 0.125
|
||||
assert abs(threshold - expected_threshold) < 1e-10
|
||||
|
||||
|
||||
class TestConstants:
|
||||
"""Test module constants."""
|
||||
|
||||
def test_chunk_size(self):
|
||||
assert SILERO_CHUNK_SIZE == 512
|
||||
|
||||
def test_sample_rate(self):
|
||||
assert SILERO_SAMPLE_RATE == 16000
|
||||
|
||||
def test_context_size(self):
|
||||
assert _CONTEXT_SIZE == 64
|
||||
|
||||
|
||||
class TestLoadFallback:
|
||||
"""Test graceful fallback when onnxruntime is unavailable."""
|
||||
|
||||
def test_load_returns_none_on_import_error(self):
|
||||
"""If SileroVAD() raises ImportError, load_silero_vad returns None."""
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.silero_vad.SileroVAD",
|
||||
side_effect=ImportError("no onnxruntime"),
|
||||
):
|
||||
result = load_silero_vad()
|
||||
assert result is None
|
||||
|
||||
def test_load_returns_none_on_runtime_error(self):
|
||||
"""If SileroVAD() raises RuntimeError (bad model), load returns None."""
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.silero_vad.SileroVAD",
|
||||
side_effect=RuntimeError("corrupted model"),
|
||||
):
|
||||
result = load_silero_vad()
|
||||
assert result is None
|
||||
|
||||
def test_load_returns_vad_on_success(self):
|
||||
"""load_silero_vad() returns a SileroVAD when init succeeds."""
|
||||
mock_vad = MagicMock(spec=SileroVAD)
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.silero_vad.SileroVAD",
|
||||
return_value=mock_vad,
|
||||
):
|
||||
result = load_silero_vad()
|
||||
assert result is mock_vad
|
||||
|
||||
|
||||
class TestIsSileroAvailable:
|
||||
"""Cheap probe used by Settings UI to decide whether to show the install hint."""
|
||||
|
||||
def test_returns_false_without_onnxruntime(self):
|
||||
with patch.dict(sys.modules, {"onnxruntime": None}):
|
||||
assert is_silero_available() is False
|
||||
|
||||
def test_returns_false_when_model_missing(self):
|
||||
# onnxruntime importable, but model file gone
|
||||
ort_mod = MagicMock()
|
||||
with (
|
||||
patch.dict(sys.modules, {"onnxruntime": ort_mod}),
|
||||
patch(
|
||||
"vocalinux.speech_recognition.silero_vad.os.path.exists",
|
||||
return_value=False,
|
||||
),
|
||||
):
|
||||
assert is_silero_available() is False
|
||||
|
||||
def test_returns_true_when_both_present(self):
|
||||
ort_mod = MagicMock()
|
||||
with (
|
||||
patch.dict(sys.modules, {"onnxruntime": ort_mod}),
|
||||
patch(
|
||||
"vocalinux.speech_recognition.silero_vad.os.path.exists",
|
||||
return_value=True,
|
||||
),
|
||||
):
|
||||
assert is_silero_available() is True
|
||||
Reference in New Issue
Block a user