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

fix(whispercpp): reduce CPU threads and ensure GPU backend builds in dev mode (#439)

Browse Source 
* fix(whispercpp): reduce CPU threads and ensure GPU backend builds in dev mode

- Reduce default n_threads from cpu_count() (16) to max 8 (CPU) / max 4 (GPU)
- Add runtime pywhispercpp GPU capability detection (_detect_pywhispercpp_gpu_backend)
- Make n_threads configurable via whispercpp_n_threads config option
- Fix install.sh regression: GPU source build was skipped in --dev mode
- Extract GPU install logic into install_whispercpp_with_gpu_support()
- Add is_pywhispercpp_gpu_capable() to detect GPU libs in installed pywhispercpp
- Force rebuild pywhispercpp when GPU is detected but lacks GPU support

Fixes CPU spiking to 100% on GPU systems due to excessive thread count
and ensures dev installs get GPU acceleration.

* fix(audio): skip output-only HDMI devices when selecting input device

Add _resolve_valid_input_device() that filters devices by
maxInputChannels > 0, falling back preferred → default → first valid.

Previously, HDMI output devices (0 input channels) could be selected
as the default audio input, causing 'Invalid number of channels' errors.
Now both _record_audio() and _attempt_audio_reconnection() validate the
device has input channels before probing or opening the stream.

* fix(whispercpp): treat n_threads=0 as auto-detect, not literal zero

Config default whispercpp_n_threads=0 was being passed directly
to whisper.cpp as n_threads=0, causing 'cannot create std::vector
larger than max_size()' transcription errors. Now 0 and negative
values are treated as auto-detect, falling through to the smart
thread count logic (max 4 with GPU, max 8 on CPU).

* test: add whispercpp_n_threads to mock assertions and config defaults

Update test_main.py mock to include the new whispercpp_n_threads=0 kwarg
and test_config_manager.py to assert the default config value.

* fix(audio): harden device enumeration against test mocks

The new HDMI-filter helper and GPU-backend detector crashed unit tests
that pass MagicMock() for the PyAudio/pathlib surfaces:

- _find_pywhispercpp_shared_library_dirs hit TypeError when tests
  monkey-patch os.stat (Path.is_dir internally calls S_ISDIR on a
  non-int st_mode). Widen the catch to OSError/TypeError/ValueError so
  unconfigured stat results just skip the candidate dir.

- _resolve_valid_input_device crashed with 'MagicMock > int' on
  audio.get_device_count() / info.get('maxInputChannels'). Coerce the
  count, treat zero/unknown as 'use system default', and include
  non-dict info objects rather than filtering everything out.

No production behavior changes: real PyAudio always returns int counts
and dict device info, so the HDMI filter still works as designed.

* test(recognition): cover new HDMI filter and GPU backend detector

- TestResolveValidInputDevice: 9 cases for the HDMI-filter resolver
  (preferred passthrough, HDMI fallback, no-default fallback, no-input
  None return, and three mock-tolerance escape hatches).
- TestDetectPywhispercppGpuBackend: 4 cases for runtime GPU library
  detection (cpu/vulkan/cuda/empty dirs).
- TestFindPywhispercppSharedLibraryDirsResilience: proves the widened
  exception catch survives a TypeError from pathlib.Path.is_dir.

* test(text-injection): wait for IBus init thread to avoid race on slow runners

The IBus environment promotion runs in a background daemon thread spawned
by TextInjector.__init__. The two TestIBusRuntimeFallback cases asserted
on injector.environment immediately after construction, which raced the
thread on Python 3.10 GHA runners and produced intermittent failures.

Join the init thread (with a generous timeout) before asserting, so the
test exercises the post-warmup state deterministically.
This commit is contained in:
Jatin K Malik
2026-05-30 16:44:53 -07:00
committed by GitHub
parent 86766d0ea0
commit c65cc14a47
8 changed files with 502 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
install.sh
+183 -150
View File
@@ -2124,6 +2124,20 @@ print(":".join(dirs))
PY
}
is_pywhispercpp_gpu_capable() {
local LIB_DIRS
LIB_DIRS=$(get_pywhispercpp_library_path || true)
[ -n "$LIB_DIRS" ] || return 1
local IFS=:
for dir in $LIB_DIRS; do
if [ -f "$dir/libggml-vulkan.so" ] || [ -f "$dir/libggml-cuda.so" ]; then
return 0
fi
done
return 1
}
with_pywhispercpp_library_path() {
local PYWHISPERCPP_LIBRARY_PATH
PYWHISPERCPP_LIBRARY_PATH=$(get_pywhispercpp_library_path || true)
@@ -2183,11 +2197,20 @@ should_rebuild_whispercpp() {
esac
if [[ "$NON_INTERACTIVE" == "yes" ]]; then
if [[ "$HAS_NVIDIA_GPU" == "yes" || "$HAS_VULKAN" == "yes" ]] && ! is_pywhispercpp_gpu_capable; then
print_info "Non-interactive mode: GPU detected but pywhispercpp lacks GPU support. Rebuilding..."
return 0
fi
print_info "Non-interactive mode: reusing existing pywhispercpp installation."
print_info "Use --rebuild-whispercpp to force a rebuild."
return 1
fi
if [[ "$HAS_NVIDIA_GPU" == "yes" || "$HAS_VULKAN" == "yes" ]] && ! is_pywhispercpp_gpu_capable; then
print_info "GPU detected but pywhispercpp lacks GPU support. Rebuilding..."
return 0
fi
read -p "Rebuild/reinstall pywhispercpp? This can take several minutes. (y/N) " -n 1 -r
echo
if [[ $REPLY =~ ^[Yy]$ ]]; then
@@ -2198,6 +2221,161 @@ should_rebuild_whispercpp() {
return 1
}
install_whispercpp_with_gpu_support() {
local PIP_LOG_FILE="$1"
print_info ""
print_info "╔════════════════════════════════════════════════════════╗"
print_info "║ Installing WHISPER.CPP (Recommended) ║"
print_info "╠════════════════════════════════════════════════════════╣"
print_info "║ • Fastest speech recognition ║"
print_info "║ • Works with any GPU: NVIDIA, AMD, Intel ║"
print_info "║ • Uses Vulkan for GPU acceleration ║"
print_info "║ • CPU-only mode available ║"
print_info "╚════════════════════════════════════════════════════════╝"
print_info ""
# Detect GPU and install pywhispercpp with appropriate GPU support
detect_nvidia_gpu || true
detect_vulkan || true
local GPU_BACKEND="CPU"
local GPU_INSTALL_SUCCESS=false
local SKIP_WHISPERCPP_INSTALL=false
local PYWHISPERCPP_CMAKE_ARGS
PYWHISPERCPP_CMAKE_ARGS=$(get_pywhispercpp_cmake_args)
if [[ "$WHISPERCPP_ALREADY_INSTALLED" == "true" ]]; then
GPU_BACKEND="existing"
if should_rebuild_whispercpp; then
print_info "Existing pywhispercpp will be replaced."
else
SKIP_WHISPERCPP_INSTALL=true
fi
fi
# Check if user explicitly chose CPU backend in interactive mode
if [[ "$SKIP_WHISPERCPP_INSTALL" == "true" ]]; then
print_info "Skipping pywhispercpp reinstall; existing compiled bindings remain in place."
elif [[ "${WHISPERCPP_BACKEND}" == "cpu" ]]; then
print_info "ℹ Installing CPU-only version (as requested)..."
GPU_BACKEND="CPU"
else
# Try Vulkan first (works with all GPUs: NVIDIA, AMD, Intel)
if [[ "$HAS_VULKAN" == "yes" ]]; then
print_info "✓ Vulkan detected: $VULKAN_DEVICE"
print_info " Installing pywhispercpp with Vulkan support..."
GPU_BACKEND="Vulkan"
print_info "Installing pywhispercpp ($GPU_BACKEND backend)..."
if CMAKE_ARGS="${CMAKE_ARGS:+$CMAKE_ARGS }$PYWHISPERCPP_CMAKE_ARGS" GGML_VULKAN=1 pip install --force-reinstall --no-cache-dir git+https://github.com/absadiki/pywhispercpp --log "$PIP_LOG_FILE" 2>&1; then
GPU_INSTALL_SUCCESS=true
else
print_warning "Vulkan build failed - checking for NVIDIA GPU to try CUDA..."
fi
fi
# If Vulkan failed or not available, try CUDA for NVIDIA GPUs
if [[ "$GPU_INSTALL_SUCCESS" != "true" && "$HAS_NVIDIA_GPU" == "yes" ]]; then
print_info "✓ NVIDIA GPU detected: $GPU_NAME"
print_info " Installing pywhispercpp with CUDA support..."
GPU_BACKEND="CUDA"
print_info "Installing pywhispercpp ($GPU_BACKEND backend)..."
if CMAKE_ARGS="${CMAKE_ARGS:+$CMAKE_ARGS }$PYWHISPERCPP_CMAKE_ARGS" GGML_CUDA=1 pip install --force-reinstall --no-cache-dir git+https://github.com/absadiki/pywhispercpp --log "$PIP_LOG_FILE" 2>&1; then
GPU_INSTALL_SUCCESS=true
fi
fi
fi
# Fall back to CPU version if GPU install failed or no GPU detected
if [[ "$SKIP_WHISPERCPP_INSTALL" != "true" && "$GPU_INSTALL_SUCCESS" != "true" ]]; then
if [[ "$GPU_BACKEND" != "CPU" ]]; then
print_warning "Failed to install pywhispercpp with $GPU_BACKEND support, falling back to CPU version..."
# Provide helpful error messages for common issues
if [[ "$GPU_BACKEND" == "Vulkan" ]]; then
print_info " To use Vulkan GPU acceleration, please install Vulkan development libraries:"
print_info " Ubuntu/Debian: sudo apt install libvulkan-dev vulkan-tools glslc || glslang-tools"
print_info " Fedora: sudo dnf install vulkan-loader-devel vulkan-tools glslang"
print_info " Arch: sudo pacman -S vulkan-headers vulkan-tools glslang"
print_info " openSUSE: sudo zypper install vulkan-devel vulkan-tools shaderc"
elif [[ "$GPU_BACKEND" == "CUDA" ]]; then
print_info " To use CUDA GPU acceleration, please install CUDA toolkit:"
print_info " Visit: https://developer.nvidia.com/cuda-downloads"
fi
elif [[ "${WHISPERCPP_BACKEND}" != "cpu" ]]; then
print_info "ℹ No GPU detected - installing CPU-only version"
print_info " CPU mode is still very fast!"
fi
GPU_BACKEND="CPU"
print_info "Installing pywhispercpp ($GPU_BACKEND backend)..."
install_cpu_pywhispercpp "$PIP_LOG_FILE" || {
print_error "Failed to install pywhispercpp"
return 1
}
fi
if [[ "$SKIP_WHISPERCPP_INSTALL" == "true" ]]; then
print_success "pywhispercpp reused from existing installation"
else
print_success "pywhispercpp installed with $GPU_BACKEND backend"
fi
if ! is_pywhispercpp_installed; then
print_warning "pywhispercpp installed but import verification failed."
print_warning "This can happen when libwhisper.so is installed beside the Python extension without a runtime library path."
if [[ "$SKIP_WHISPERCPP_INSTALL" != "true" ]]; then
print_warning "Trying RPATH-aware CPU pywhispercpp fallback..."
GPU_BACKEND="CPU"
if install_cpu_pywhispercpp "$PIP_LOG_FILE"; then
print_success "CPU pywhispercpp fallback installed"
else
print_warning "CPU pywhispercpp fallback installation failed"
fi
fi
fi
if ! is_pywhispercpp_installed; then
print_warning "pywhispercpp is still unavailable; setting VOSK as the default engine so Vocalinux can start."
print_warning "You can switch back to whisper.cpp from Settings after reinstalling pywhispercpp."
SELECTED_ENGINE="vosk"
local FALLBACK_VOSK_CONFIG="$CONFIG_DIR/config.json"
if [ ! -f "$FALLBACK_VOSK_CONFIG" ]; then
mkdir -p "$CONFIG_DIR"
cat > "$FALLBACK_VOSK_CONFIG" << 'FALLBACK_VOSK_CONFIG'
{
"speech_recognition": {
"engine": "vosk",
"model_size": "small",
"vosk_model_size": "small",
"whisper_model_size": "tiny",
"whisper_cpp_model_size": "tiny",
"vad_sensitivity": 3,
"silence_timeout": 2.0
},
"audio": {
"device_index": null,
"device_name": null
},
"shortcuts": {
"toggle_recognition": "ctrl+ctrl"
},
"ui": {
"start_minimized": false,
"show_notifications": true
},
"advanced": {
"debug_logging": false,
"wayland_mode": false
}
}
FALLBACK_VOSK_CONFIG
fi
fi
echo ""
}
# Function to install Python package with error handling and verification
install_python_package() {
# Create a temporary directory for pip logs
@@ -2246,6 +2424,10 @@ install_python_package() {
print_warning "Failed to install some optional dependencies."
print_warning "Some features may not work correctly."
}
if [[ "${SELECTED_ENGINE:-whisper_cpp}" == "whisper_cpp" ]]; then
install_whispercpp_with_gpu_support "$PIP_LOG_FILE"
fi
else
print_info "Installing Vocalinux..."
@@ -2262,156 +2444,7 @@ install_python_package() {
# - Default is whisper_cpp for best performance
case "${SELECTED_ENGINE:-whisper_cpp}" in
whisper_cpp)
print_info ""
print_info "╔════════════════════════════════════════════════════════╗"
print_info "║ Installing WHISPER.CPP (Recommended) ║"
print_info "╠════════════════════════════════════════════════════════╣"
print_info "║ • Fastest speech recognition ║"
print_info "║ • Works with any GPU: NVIDIA, AMD, Intel ║"
print_info "║ • Uses Vulkan for GPU acceleration ║"
print_info "║ • CPU-only mode available ║"
print_info "╚════════════════════════════════════════════════════════╝"
print_info ""
# Detect GPU and install pywhispercpp with appropriate GPU support
detect_nvidia_gpu || true
detect_vulkan || true
local GPU_BACKEND="CPU"
local GPU_INSTALL_SUCCESS=false
local SKIP_WHISPERCPP_INSTALL=false
local PYWHISPERCPP_CMAKE_ARGS
PYWHISPERCPP_CMAKE_ARGS=$(get_pywhispercpp_cmake_args)
if [[ "$WHISPERCPP_ALREADY_INSTALLED" == "true" ]]; then
GPU_BACKEND="existing"
if should_rebuild_whispercpp; then
print_info "Existing pywhispercpp will be replaced."
else
SKIP_WHISPERCPP_INSTALL=true
fi
fi
# Check if user explicitly chose CPU backend in interactive mode
if [[ "$SKIP_WHISPERCPP_INSTALL" == "true" ]]; then
print_info "Skipping pywhispercpp reinstall; existing compiled bindings remain in place."
elif [[ "${WHISPERCPP_BACKEND}" == "cpu" ]]; then
print_info "ℹ Installing CPU-only version (as requested)..."
GPU_BACKEND="CPU"
else
# Try Vulkan first (works with all GPUs: NVIDIA, AMD, Intel)
if [[ "$HAS_VULKAN" == "yes" ]]; then
print_info "✓ Vulkan detected: $VULKAN_DEVICE"
print_info " Installing pywhispercpp with Vulkan support..."
GPU_BACKEND="Vulkan"
print_info "Installing pywhispercpp ($GPU_BACKEND backend)..."
if CMAKE_ARGS="${CMAKE_ARGS:+$CMAKE_ARGS }$PYWHISPERCPP_CMAKE_ARGS" GGML_VULKAN=1 pip install --force-reinstall --no-cache-dir git+https://github.com/absadiki/pywhispercpp --log "$PIP_LOG_FILE" 2>&1; then
GPU_INSTALL_SUCCESS=true
else
print_warning "Vulkan build failed - checking for NVIDIA GPU to try CUDA..."
fi
fi
# If Vulkan failed or not available, try CUDA for NVIDIA GPUs
if [[ "$GPU_INSTALL_SUCCESS" != "true" && "$HAS_NVIDIA_GPU" == "yes" ]]; then
print_info "✓ NVIDIA GPU detected: $GPU_NAME"
print_info " Installing pywhispercpp with CUDA support..."
GPU_BACKEND="CUDA"
print_info "Installing pywhispercpp ($GPU_BACKEND backend)..."
if CMAKE_ARGS="${CMAKE_ARGS:+$CMAKE_ARGS }$PYWHISPERCPP_CMAKE_ARGS" GGML_CUDA=1 pip install --force-reinstall --no-cache-dir git+https://github.com/absadiki/pywhispercpp --log "$PIP_LOG_FILE" 2>&1; then
GPU_INSTALL_SUCCESS=true
fi
fi
fi
# Fall back to CPU version if GPU install failed or no GPU detected
if [[ "$SKIP_WHISPERCPP_INSTALL" != "true" && "$GPU_INSTALL_SUCCESS" != "true" ]]; then
if [[ "$GPU_BACKEND" != "CPU" ]]; then
print_warning "Failed to install pywhispercpp with $GPU_BACKEND support, falling back to CPU version..."
# Provide helpful error messages for common issues
if [[ "$GPU_BACKEND" == "Vulkan" ]]; then
print_info " To use Vulkan GPU acceleration, please install Vulkan development libraries:"
print_info " Ubuntu/Debian: sudo apt install libvulkan-dev vulkan-tools glslc || glslang-tools"
print_info " Fedora: sudo dnf install vulkan-loader-devel vulkan-tools glslang"
print_info " Arch: sudo pacman -S vulkan-headers vulkan-tools glslang"
print_info " openSUSE: sudo zypper install vulkan-devel vulkan-tools shaderc"
elif [[ "$GPU_BACKEND" == "CUDA" ]]; then
print_info " To use CUDA GPU acceleration, please install CUDA toolkit:"
print_info " Visit: https://developer.nvidia.com/cuda-downloads"
fi
elif [[ "${WHISPERCPP_BACKEND}" != "cpu" ]]; then
print_info "ℹ No GPU detected - installing CPU-only version"
print_info " CPU mode is still very fast!"
fi
GPU_BACKEND="CPU"
print_info "Installing pywhispercpp ($GPU_BACKEND backend)..."
install_cpu_pywhispercpp "$PIP_LOG_FILE" || {
print_error "Failed to install pywhispercpp"
return 1
}
fi
if [[ "$SKIP_WHISPERCPP_INSTALL" == "true" ]]; then
print_success "pywhispercpp reused from existing installation"
else
print_success "pywhispercpp installed with $GPU_BACKEND backend"
fi
if ! is_pywhispercpp_installed; then
print_warning "pywhispercpp installed but import verification failed."
print_warning "This can happen when libwhisper.so is installed beside the Python extension without a runtime library path."
if [[ "$SKIP_WHISPERCPP_INSTALL" != "true" ]]; then
print_warning "Trying RPATH-aware CPU pywhispercpp fallback..."
GPU_BACKEND="CPU"
if install_cpu_pywhispercpp "$PIP_LOG_FILE"; then
print_success "CPU pywhispercpp fallback installed"
else
print_warning "CPU pywhispercpp fallback installation failed"
fi
fi
fi
if ! is_pywhispercpp_installed; then
print_warning "pywhispercpp is still unavailable; setting VOSK as the default engine so Vocalinux can start."
print_warning "You can switch back to whisper.cpp from Settings after reinstalling pywhispercpp."
SELECTED_ENGINE="vosk"
local FALLBACK_VOSK_CONFIG="$CONFIG_DIR/config.json"
if [ ! -f "$FALLBACK_VOSK_CONFIG" ]; then
mkdir -p "$CONFIG_DIR"
cat > "$FALLBACK_VOSK_CONFIG" << 'FALLBACK_VOSK_CONFIG'
{
"speech_recognition": {
"engine": "vosk",
"model_size": "small",
"vosk_model_size": "small",
"whisper_model_size": "tiny",
"whisper_cpp_model_size": "tiny",
"vad_sensitivity": 3,
"silence_timeout": 2.0
},
"audio": {
"device_index": null,
"device_name": null
},
"shortcuts": {
"toggle_recognition": "ctrl+ctrl"
},
"ui": {
"start_minimized": false,
"show_notifications": true
},
"advanced": {
"debug_logging": false,
"wayland_mode": false
}
}
FALLBACK_VOSK_CONFIG
fi
fi
echo ""
install_whispercpp_with_gpu_support "$PIP_LOG_FILE"
;;
whisper)
src/vocalinux/main.py
+1
View File
@@ -386,6 +386,7 @@ def main():
whispercpp_entropy_thold=advanced_settings.get("whispercpp_entropy_thold", 2.4),
whispercpp_logprob_thold=advanced_settings.get("whispercpp_logprob_thold", -1.0),
whispercpp_no_speech_thold=advanced_settings.get("whispercpp_no_speech_thold", 0.6),
whispercpp_n_threads=advanced_settings.get("whispercpp_n_threads", 0),
)
# Initialize text injection system
src/vocalinux/speech_recognition/recognition_manager.py
+164 -37
View File
@@ -113,15 +113,17 @@ def _find_pywhispercpp_shared_library_dirs() -> list[str]:
for candidate_dir in candidate_dirs:
try:
resolved_dir = str(candidate_dir.resolve())
except OSError:
if resolved_dir in seen or not candidate_dir.is_dir():
continue
has_native_lib = any(candidate_dir.glob("libwhisper*.so*")) or any(
candidate_dir.glob("libggml*.so*")
)
except (OSError, TypeError, ValueError):
# TypeError/ValueError can surface when tests monkey-patch os.stat or
# when pathlib internals receive unexpected types from mocks.
continue
if resolved_dir in seen or not candidate_dir.is_dir():
continue
has_native_lib = any(candidate_dir.glob("libwhisper*.so*")) or any(
candidate_dir.glob("libggml*.so*")
)
if has_native_lib:
seen.add(resolved_dir)
library_dirs.append(resolved_dir)
@@ -217,6 +219,78 @@ def get_audio_input_devices() -> list:
return devices
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).
Checks that the device has maxInputChannels > 0. Falls back from
preferred_index → system default → first available input device.
Args:
audio: PyAudio instance
preferred_index: User-configured device index (or None for system default)
Returns:
A valid device index with input channels, or None if none found.
"""
input_device_indices = []
default_input_index = None
try:
default_info = audio.get_default_input_device_info()
default_input_index = default_info.get("index")
except (IOError, OSError, TypeError, ValueError, AttributeError):
pass
try:
device_count = int(audio.get_device_count())
except (IOError, OSError, TypeError, ValueError, AttributeError):
# MagicMock-based tests or misbehaving drivers can yield non-int counts.
return preferred_index
if device_count <= 0:
# No enumeration available; let PyAudio fall back to system default.
return preferred_index
for i in range(device_count):
try:
info = audio.get_device_info_by_index(i)
except (IOError, OSError, TypeError, ValueError, AttributeError):
continue
if not isinstance(info, dict):
# Non-dict result (e.g. MagicMock in tests) — can't filter by channels,
# so include the device rather than excluding all of them.
input_device_indices.append(i)
continue
channels = info.get("maxInputChannels", 0)
if isinstance(channels, (int, float)) and channels > 0:
input_device_indices.append(i)
if not input_device_indices:
return None
if preferred_index is not None and preferred_index in input_device_indices:
return preferred_index
if preferred_index is not None:
try:
device_name = audio.get_device_info_by_index(preferred_index).get("name", "unknown")
except (IOError, OSError):
device_name = "unknown"
logger.warning(
"Configured audio device [%s] (%s) has no input channels. "
"Falling back to a valid input device.",
preferred_index,
device_name,
)
if default_input_index is not None and default_input_index in input_device_indices:
return default_input_index
return input_device_indices[0]
def _get_supported_channels(audio, device_index: Optional[int] = None) -> int:
"""
Detect the supported number of channels for the audio device.
@@ -680,6 +754,7 @@ class SpeechRecognitionManager:
self.whispercpp_entropy_thold = kwargs.get("whispercpp_entropy_thold", 2.4)
self.whispercpp_logprob_thold = kwargs.get("whispercpp_logprob_thold", -1.0)
self.whispercpp_no_speech_thold = kwargs.get("whispercpp_no_speech_thold", 0.6)
self.whispercpp_n_threads = kwargs.get("whispercpp_n_threads", None)
# Audio diagnostics tracking
self._last_audio_level = 0.0
@@ -1032,6 +1107,21 @@ class SpeechRecognitionManager:
compatible_kwargs = self._filter_whispercpp_model_kwargs(model_kwargs)
return Model(model_path, **compatible_kwargs)
def _detect_pywhispercpp_gpu_backend(self) -> str:
"""Detect whether pywhispercpp's native library actually has GPU support."""
_preload_pywhispercpp_shared_libraries()
for library_dir in _find_pywhispercpp_shared_library_dirs():
root = Path(library_dir)
for pattern in ("libggml-vulkan*.so*", "libggml-cuda*.so*"):
matches = list(root.glob(pattern))
if matches:
lib_name = matches[0].name.lower()
if "vulkan" in lib_name:
return "vulkan"
if "cuda" in lib_name:
return "cuda"
return "cpu"
def _load_whispercpp_model(self, model_path: str):
"""Load the whisper.cpp model file and configure the compute backend.
@@ -1075,7 +1165,16 @@ class SpeechRecognitionManager:
logger.info(f"Loading whisper.cpp '{self.model_size}' model...")
self.model = None # Release previous model if re-initializing
n_threads = multiprocessing.cpu_count()
actual_gpu_backend = self._detect_pywhispercpp_gpu_backend()
has_gpu_libs = actual_gpu_backend in ("vulkan", "cuda")
if self.whispercpp_n_threads is not None and self.whispercpp_n_threads > 0:
n_threads = self.whispercpp_n_threads
elif has_gpu_libs:
n_threads = max(1, multiprocessing.cpu_count() // 4)
else:
n_threads = min(multiprocessing.cpu_count(), 8)
load_start_time = time.time()
loaded_backend = backend
@@ -1090,10 +1189,16 @@ class SpeechRecognitionManager:
)
load_duration = time.time() - load_start_time
logger.info(
f"whisper.cpp configured with n_threads={n_threads} "
f"(detected {multiprocessing.cpu_count()} CPUs)"
)
if has_gpu_libs:
logger.info(
f"whisper.cpp configured with n_threads={n_threads} "
f"(GPU backend: {actual_gpu_backend})"
)
else:
logger.info(
f"whisper.cpp configured with n_threads={n_threads} "
f"(CPU-only; pywhispercpp lacks GPU libraries)"
)
logger.info(f"whisper.cpp model loaded in {load_duration:.2f}s ({loaded_backend} backend)")
self._model_initialized = True
@@ -1832,6 +1937,18 @@ class SpeechRecognitionManager:
self._pyaudio_instance = pyaudio.PyAudio()
audio = self._pyaudio_instance
# Resolve a valid input device — skip output-only devices (e.g. HDMI)
resolved_device_index = _resolve_valid_input_device(audio, self.audio_device_index)
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."
)
play_error_sound()
audio.terminate()
self._update_state(RecognitionState.ERROR)
return
# Log available devices for debugging
logger.debug("Available audio input devices:")
for i in range(audio.get_device_count()):
@@ -1845,11 +1962,11 @@ class SpeechRecognitionManager:
continue
# Detect supported channel count first (some devices require stereo)
CHANNELS = _get_supported_channels(audio, self.audio_device_index)
CHANNELS = _get_supported_channels(audio, resolved_device_index)
logger.info(f"Using {CHANNELS} channel(s) for recording")
# Detect supported sample rate for the selected device
RATE = _get_supported_sample_rate(audio, self.audio_device_index, CHANNELS)
RATE = _get_supported_sample_rate(audio, resolved_device_index, CHANNELS)
self._capture_sample_rate = RATE
logger.info(f"Using sample rate: {RATE}Hz")
@@ -1862,24 +1979,21 @@ class SpeechRecognitionManager:
"frames_per_buffer": CHUNK,
}
# Use specified device if set, otherwise use system default
if self.audio_device_index is not None:
stream_kwargs["input_device_index"] = self.audio_device_index
try:
device_info = audio.get_device_info_by_index(self.audio_device_index)
logger.info(
f"Using audio device [{self.audio_device_index}]: {device_info.get('name')}"
)
except (IOError, OSError):
logger.warning(f"Could not get info for device index {self.audio_device_index}")
else:
try:
default_device = audio.get_default_input_device_info()
logger.info(
f"Using default audio device [{default_device.get('index')}]: {default_device.get('name')}"
)
except (IOError, OSError):
logger.warning("Could not get default input device info")
# Use the resolved device (skip if already system default)
try:
default_idx = audio.get_default_input_device_info().get("index")
except (IOError, OSError):
default_idx = None
if resolved_device_index != default_idx:
stream_kwargs["input_device_index"] = resolved_device_index
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')}"
)
except (IOError, OSError):
logger.warning(f"Could not get info for device index {resolved_device_index}")
try:
self._audio_stream = audio.open(**stream_kwargs)
@@ -2279,6 +2393,7 @@ class SpeechRecognitionManager:
"whispercpp_entropy_thold",
"whispercpp_logprob_thold",
"whispercpp_no_speech_thold",
"whispercpp_n_threads",
):
if param_name in kwargs:
setattr(self, param_name, kwargs[param_name])
@@ -2366,16 +2481,24 @@ class SpeechRecognitionManager:
except Exception as e:
logger.debug(f"Error closing old audio stream: {e}")
# Resolve a valid input device — skip output-only devices (e.g. HDMI)
resolved_device_index = _resolve_valid_input_device(
audio_instance, self.audio_device_index
)
if resolved_device_index is None:
logger.error("Reconnection failed: no input devices available.")
return False
# Stream configuration
CHUNK = 1024
FORMAT = pyaudio.paInt16
# Detect supported channel count first (some devices require stereo)
CHANNELS = _get_supported_channels(audio_instance, self.audio_device_index)
CHANNELS = _get_supported_channels(audio_instance, resolved_device_index)
logger.debug(f"Reconnecting with {CHANNELS} channel(s)")
# Detect supported sample rate for the device
RATE = _get_supported_sample_rate(audio_instance, self.audio_device_index, CHANNELS)
RATE = _get_supported_sample_rate(audio_instance, resolved_device_index, CHANNELS)
self._capture_sample_rate = RATE
logger.debug(f"Reconnecting with sample rate: {RATE}Hz")
@@ -2387,9 +2510,13 @@ class SpeechRecognitionManager:
"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
# Use resolved device (skip if already system default)
try:
default_idx = audio_instance.get_default_input_device_info().get("index")
except (IOError, OSError):
default_idx = None
if resolved_device_index != default_idx:
stream_kwargs["input_device_index"] = resolved_device_index
# Attempt to open new stream
new_stream = audio_instance.open(**stream_kwargs)
src/vocalinux/ui/config_manager.py
+1
View File
@@ -66,6 +66,7 @@ DEFAULT_CONFIG = {
"whispercpp_entropy_thold": 2.4,
"whispercpp_logprob_thold": -1.0,
"whispercpp_no_speech_thold": 0.6,
"whispercpp_n_threads": 0, # 0 = auto-detect optimal thread count; set to override
},
}
tests/test_config_manager.py
+1
View File
@@ -543,6 +543,7 @@ class TestTypedAccessors(unittest.TestCase):
self.assertEqual(advanced["whispercpp_entropy_thold"], 2.4)
self.assertEqual(advanced["whispercpp_logprob_thold"], -1.0)
self.assertEqual(advanced["whispercpp_no_speech_thold"], 0.6)
self.assertEqual(advanced["whispercpp_n_threads"], 0)
def test_whispercpp_advanced_persistence(self):
self.config_manager.set("advanced", "whispercpp_temperature", 0.5)
tests/test_main.py
+1
View File
@@ -230,6 +230,7 @@ class TestMainModule(unittest.TestCase):
whispercpp_entropy_thold=2.4,
whispercpp_logprob_thold=-1.0,
whispercpp_no_speech_thold=0.6,
whispercpp_n_threads=0,
)
mock_text.assert_called_once_with(wayland_mode=True)
mock_action_handler.assert_called_once_with(mock_text_instance)
tests/test_recognition_manager_helpers.py
+144
View File
@@ -41,6 +41,7 @@ from vocalinux.speech_recognition.recognition_manager import (
_find_pywhispercpp_shared_library_dirs,
_get_system_model_paths,
_preload_pywhispercpp_shared_libraries,
_resolve_valid_input_device,
_show_notification,
)
from vocalinux.speech_recognition.recognition_manager import ( # noqa: E402
@@ -292,3 +293,146 @@ class TestTestAudioInput(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
def _make_audio_mock(devices, default_index=None):
"""Build a PyAudio-like mock returning the given device dicts."""
audio = MagicMock()
audio.get_device_count.return_value = len(devices)
audio.get_device_info_by_index.side_effect = lambda i: devices[i]
if default_index is not None:
audio.get_default_input_device_info.return_value = devices[default_index]
else:
audio.get_default_input_device_info.side_effect = IOError("no default")
return audio
class TestResolveValidInputDevice:
"""Coverage for the HDMI-filtering audio device resolver."""
def test_preferred_index_with_input_channels_is_returned(self):
devices = [
{"index": 0, "name": "HDMI", "maxInputChannels": 0},
{"index": 1, "name": "USB Mic", "maxInputChannels": 2},
]
audio = _make_audio_mock(devices, default_index=1)
assert _resolve_valid_input_device(audio, preferred_index=1) == 1
def test_preferred_index_without_input_falls_back_to_default(self):
devices = [
{"index": 0, "name": "HDMI", "maxInputChannels": 0},
{"index": 1, "name": "USB Mic", "maxInputChannels": 2},
]
audio = _make_audio_mock(devices, default_index=1)
assert _resolve_valid_input_device(audio, preferred_index=0) == 1
def test_preferred_index_none_uses_default(self):
devices = [
{"index": 0, "name": "USB Mic", "maxInputChannels": 1},
{"index": 1, "name": "HDMI", "maxInputChannels": 0},
]
audio = _make_audio_mock(devices, default_index=0)
assert _resolve_valid_input_device(audio, preferred_index=None) == 0
def test_no_default_falls_back_to_first_input(self):
devices = [
{"index": 0, "name": "HDMI", "maxInputChannels": 0},
{"index": 1, "name": "USB Mic", "maxInputChannels": 1},
]
audio = _make_audio_mock(devices, default_index=None)
assert _resolve_valid_input_device(audio, preferred_index=None) == 1
def test_no_input_devices_returns_none(self):
devices = [{"index": 0, "name": "HDMI", "maxInputChannels": 0}]
audio = _make_audio_mock(devices, default_index=None)
assert _resolve_valid_input_device(audio, preferred_index=None) is None
def test_device_count_exception_returns_preferred(self):
audio = MagicMock()
audio.get_default_input_device_info.side_effect = IOError("nope")
audio.get_device_count.side_effect = OSError("driver dead")
assert _resolve_valid_input_device(audio, preferred_index=3) == 3
def test_zero_device_count_returns_preferred(self):
audio = MagicMock()
audio.get_default_input_device_info.side_effect = IOError("nope")
audio.get_device_count.return_value = 0
assert _resolve_valid_input_device(audio, preferred_index=7) == 7
def test_non_dict_info_is_treated_as_valid(self):
audio = MagicMock()
audio.get_default_input_device_info.side_effect = IOError("nope")
audio.get_device_count.return_value = 1
audio.get_device_info_by_index.return_value = MagicMock()
assert _resolve_valid_input_device(audio, preferred_index=0) == 0
def test_per_device_info_error_is_skipped(self):
audio = MagicMock()
audio.get_default_input_device_info.side_effect = IOError("nope")
audio.get_device_count.return_value = 2
def info(i):
if i == 0:
raise OSError("bad device")
return {"index": 1, "name": "USB Mic", "maxInputChannels": 1}
audio.get_device_info_by_index.side_effect = info
assert _resolve_valid_input_device(audio, preferred_index=None) == 1
class TestDetectPywhispercppGpuBackend:
"""Coverage for the runtime GPU library detector."""
def _make_manager(self):
manager = SpeechRecognitionManager.__new__(SpeechRecognitionManager)
return manager
def test_returns_cpu_when_no_gpu_libs(self, tmp_path, monkeypatch):
libs = tmp_path / "pywhispercpp.libs"
libs.mkdir()
(libs / "libwhisper.so.1").write_text("")
monkeypatch.setattr(rm, "_preload_pywhispercpp_shared_libraries", lambda: None)
monkeypatch.setattr(rm, "_find_pywhispercpp_shared_library_dirs", lambda: [str(libs)])
assert self._make_manager()._detect_pywhispercpp_gpu_backend() == "cpu"
def test_detects_vulkan(self, tmp_path, monkeypatch):
libs = tmp_path / "pywhispercpp.libs"
libs.mkdir()
(libs / "libggml-vulkan.so").write_text("")
monkeypatch.setattr(rm, "_preload_pywhispercpp_shared_libraries", lambda: None)
monkeypatch.setattr(rm, "_find_pywhispercpp_shared_library_dirs", lambda: [str(libs)])
assert self._make_manager()._detect_pywhispercpp_gpu_backend() == "vulkan"
def test_detects_cuda(self, tmp_path, monkeypatch):
libs = tmp_path / "pywhispercpp.libs"
libs.mkdir()
(libs / "libggml-cuda.so").write_text("")
monkeypatch.setattr(rm, "_preload_pywhispercpp_shared_libraries", lambda: None)
monkeypatch.setattr(rm, "_find_pywhispercpp_shared_library_dirs", lambda: [str(libs)])
assert self._make_manager()._detect_pywhispercpp_gpu_backend() == "cuda"
def test_returns_cpu_when_no_library_dirs(self, monkeypatch):
monkeypatch.setattr(rm, "_preload_pywhispercpp_shared_libraries", lambda: None)
monkeypatch.setattr(rm, "_find_pywhispercpp_shared_library_dirs", lambda: [])
assert self._make_manager()._detect_pywhispercpp_gpu_backend() == "cpu"
class TestFindPywhispercppSharedLibraryDirsResilience:
"""Coverage for the widened exception handling in library discovery."""
def test_typeerror_from_pathlib_is_swallowed(self, tmp_path, monkeypatch):
# Simulate tests that monkey-patch os.stat globally and yield non-int st_mode,
# causing pathlib.Path.is_dir to raise TypeError mid-iteration.
import pathlib
original_is_dir = pathlib.Path.is_dir
def broken_is_dir(self, *args, **kwargs):
raise TypeError("an integer is required")
monkeypatch.setattr(pathlib.Path, "is_dir", broken_is_dir)
try:
result = _find_pywhispercpp_shared_library_dirs()
finally:
monkeypatch.setattr(pathlib.Path, "is_dir", original_is_dir)
assert isinstance(result, list)
tests/test_text_injector.py
+7
View File
@@ -1455,6 +1455,11 @@ class TestIBusRuntimeFallback(unittest.TestCase):
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "DISPLAY": ":0"}):
injector = TextInjector()
# IBus environment promotion happens in a background daemon thread
# (see TextInjector._initialize_ibus_in_background). Wait for it to
# finish so the assertion below is not subject to scheduler timing.
if injector._ibus_init_thread is not None:
injector._ibus_init_thread.join(timeout=5)
self.assertEqual(injector.environment, DesktopEnvironment.X11_IBUS)
result = injector.inject_text("Hello from fallback")
@@ -1485,6 +1490,8 @@ class TestIBusRuntimeFallback(unittest.TestCase):
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
injector = TextInjector()
if injector._ibus_init_thread is not None:
injector._ibus_init_thread.join(timeout=5)
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS)
result = injector.inject_text("Hello via wayland fallback")