fix(vulkan): auto-select discrete GPU and add manual device selection (#590)
* fix(vulkan): auto-select discrete GPU and add manual device selection On hybrid GPU systems (Intel iGPU + NVIDIA dGPU), whisper.cpp's Vulkan backend always selected device 0 (the integrated GPU), resulting in significantly slower performance. Changes: - Add detect_vulkan_devices() to enumerate all Vulkan GPUs and identify discrete vs integrated devices - Set GGML_VULKAN_DEVICE env var before model init to select the correct GPU (auto-selects discrete GPU when available) - Add whispercpp_gpu_device config option for manual device selection - Add GPU device dropdown in Settings > Advanced for users with multiple Vulkan GPUs Fixes #589 * test: add coverage for Vulkan GPU device selection - Test detect_vulkan_devices() with single/multiple GPUs - Test _prefer_discrete_vulkan_device() preference logic - Test GPU device selection in recognition_manager - Verify GGML_VULKAN_DEVICE env var is set correctly * fix(speech_recognition): pass whispercpp GPU device to pywhispercpp Model context_params The Vulkan device-selection logic from PR #590 set the GGML_VULKAN_DEVICE environment variable, but pywhispercpp did not pick it up. Pass the selected GPU device through the pywhispercpp Model context_params so whisper.cpp actually initializes the requested Vulkan device. Fixes #589 * fix(speech_recognition): fallback when pywhispercpp lacks context_params Older pywhispercpp releases (< ~1.4.0) reject the context_params keyword and raise TypeError. Detect that specific failure and retry without GPU device selection so the engine still loads on those versions. Fixes #589 * refactor(speech_recognition): inline context_params assignment Simplify the GPU device selection path and remove the temporary dict. * fix(vulkan): parse real vulkaninfo GPU0: headers detect_vulkan_devices() only recognized fictional "GPU id = N" headers, so real vulkaninfo --summary output (GPU0:/GPU1:) returned no devices and hybrid NVIDIA systems fell through to CUDA. Also resolve preferred device names by GPU id instead of list position. Fixes #589 * refactor(whispercpp): drop ineffective GGML_VULKAN_DEVICE env var pywhispercpp ignores GGML_VULKAN_DEVICE; device selection already goes through context_params={"gpu_device": N}. Remove the misleading env write so maintainers do not think the env var is the selection path.
This commit is contained in:
Jatin K Malik
@@ -399,6 +399,7 @@ def main():
|
||||
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),
|
||||
whispercpp_gpu_device=advanced_settings.get("whispercpp_gpu_device", None),
|
||||
remote_api_url=saved_settings.get("remote_api_url", ""),
|
||||
remote_api_key=saved_settings.get("remote_api_key", ""),
|
||||
remote_api_endpoint=saved_settings.get("remote_api_endpoint", "/inference"),
|
||||
|
||||
@@ -842,6 +842,7 @@ class SpeechRecognitionManager:
|
||||
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)
|
||||
self.whispercpp_gpu_device = kwargs.get("whispercpp_gpu_device", None)
|
||||
|
||||
# Remote API settings
|
||||
self.remote_api_url = kwargs.get("remote_api_url", "")
|
||||
@@ -1198,11 +1199,31 @@ class SpeechRecognitionManager:
|
||||
)
|
||||
return compatible_kwargs
|
||||
|
||||
def _load_model_with_compatible_params(self, model_path: str, model_kwargs: dict):
|
||||
def _load_model_with_compatible_params(
|
||||
self, model_path: str, model_kwargs: dict, gpu_device: Optional[int] = None
|
||||
):
|
||||
from pywhispercpp.model import Model
|
||||
|
||||
compatible_kwargs = self._filter_whispercpp_model_kwargs(model_kwargs)
|
||||
return Model(model_path, **compatible_kwargs)
|
||||
if gpu_device is not None and gpu_device >= 0:
|
||||
compatible_kwargs["context_params"] = {"gpu_device": gpu_device}
|
||||
|
||||
try:
|
||||
return Model(model_path, **compatible_kwargs)
|
||||
except TypeError as exc:
|
||||
# Older pywhispercpp releases (< ~1.4.0) do not accept context_params.
|
||||
# Retry without GPU device selection so the engine still loads.
|
||||
if "context_params" in compatible_kwargs and (
|
||||
"context_params" in str(exc) or "unexpected keyword" in str(exc)
|
||||
):
|
||||
logger.warning(
|
||||
"pywhispercpp does not support context_params; "
|
||||
"upgrade pywhispercpp to enable GPU device selection. "
|
||||
f"Falling back to default device. Error: {exc}"
|
||||
)
|
||||
compatible_kwargs.pop("context_params")
|
||||
return Model(model_path, **compatible_kwargs)
|
||||
raise
|
||||
|
||||
def _detect_pywhispercpp_gpu_backend(self) -> str:
|
||||
"""Detect whether pywhispercpp's native library actually has GPU support."""
|
||||
@@ -1235,7 +1256,9 @@ class SpeechRecognitionManager:
|
||||
|
||||
from ..utils.whispercpp_model_info import (
|
||||
ComputeBackend,
|
||||
_prefer_discrete_vulkan_device,
|
||||
detect_compute_backend,
|
||||
detect_vulkan_devices,
|
||||
get_backend_display_name,
|
||||
)
|
||||
|
||||
@@ -1246,6 +1269,22 @@ class SpeechRecognitionManager:
|
||||
f"whisper.cpp using {get_backend_display_name(backend)} backend: {backend_info}"
|
||||
)
|
||||
|
||||
# Select Vulkan GPU via pywhispercpp context_params (GGML_VULKAN_DEVICE is ignored).
|
||||
selected_gpu_device = None
|
||||
if backend == ComputeBackend.VULKAN:
|
||||
gpu_device_index = self.whispercpp_gpu_device
|
||||
if gpu_device_index is None or gpu_device_index < 0:
|
||||
gpu_device_index = _prefer_discrete_vulkan_device()
|
||||
|
||||
if gpu_device_index is not None:
|
||||
devices = detect_vulkan_devices()
|
||||
device_name = next(
|
||||
(d["name"] for d in devices if d["index"] == gpu_device_index),
|
||||
"unknown",
|
||||
)
|
||||
logger.info(f"Using Vulkan GPU [{gpu_device_index}]: {device_name}")
|
||||
selected_gpu_device = gpu_device_index
|
||||
|
||||
# Log hardware summary
|
||||
import psutil
|
||||
|
||||
@@ -1279,7 +1318,9 @@ class SpeechRecognitionManager:
|
||||
|
||||
# Attempt to load model; filter unsupported params and fall back to CPU if needed
|
||||
try:
|
||||
self.model = self._load_model_with_compatible_params(model_path, model_kwargs)
|
||||
self.model = self._load_model_with_compatible_params(
|
||||
model_path, model_kwargs, gpu_device=selected_gpu_device
|
||||
)
|
||||
except RuntimeError as model_error:
|
||||
loaded_backend = self._handle_gpu_fallback(
|
||||
model_error, model_path, model_kwargs, ComputeBackend.CPU
|
||||
@@ -2895,6 +2936,7 @@ class SpeechRecognitionManager:
|
||||
"whispercpp_logprob_thold",
|
||||
"whispercpp_no_speech_thold",
|
||||
"whispercpp_n_threads",
|
||||
"whispercpp_gpu_device",
|
||||
):
|
||||
if param_name in kwargs:
|
||||
setattr(self, param_name, kwargs[param_name])
|
||||
|
||||
@@ -75,6 +75,7 @@ DEFAULT_CONFIG = {
|
||||
"whispercpp_logprob_thold": -1.0,
|
||||
"whispercpp_no_speech_thold": 0.6,
|
||||
"whispercpp_n_threads": 0, # 0 = auto-detect optimal thread count; set to override
|
||||
"whispercpp_gpu_device": None, # None = auto-select discrete GPU; int = specific device index
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@@ -36,6 +36,7 @@ from ..utils.whispercpp_model_info import MODEL_SIZES as WHISPERCPP_MODEL_SIZES
|
||||
from ..utils.whispercpp_model_info import (
|
||||
WHISPERCPP_MODEL_INFO,
|
||||
detect_compute_backend,
|
||||
detect_vulkan_devices,
|
||||
get_backend_display_name,
|
||||
)
|
||||
from ..utils.whispercpp_model_info import get_model_size as get_whispercpp_model_size
|
||||
@@ -2243,6 +2244,22 @@ class SettingsDialog(Gtk.Dialog):
|
||||
controls_box.pack_start(info_box, False, False, 0)
|
||||
controls_box.pack_start(group, False, False, 0)
|
||||
|
||||
gpu_group = PreferencesGroup(title="GPU Device")
|
||||
self.gpu_device_combo = Gtk.ComboBoxText()
|
||||
self.gpu_device_combo.set_size_request(250, -1)
|
||||
self.gpu_device_combo.set_tooltip_text(
|
||||
"Select which GPU to use for whisper.cpp Vulkan acceleration"
|
||||
)
|
||||
_prevent_scroll_on_hover(self.gpu_device_combo)
|
||||
self._populate_gpu_devices()
|
||||
gpu_row = PreferenceRow(
|
||||
title="Vulkan GPU",
|
||||
subtitle="Device for hardware acceleration",
|
||||
widget=self.gpu_device_combo,
|
||||
)
|
||||
gpu_group.add_row(gpu_row)
|
||||
controls_box.pack_start(gpu_group, False, False, 0)
|
||||
|
||||
separator = Gtk.Separator(orientation=Gtk.Orientation.HORIZONTAL)
|
||||
separator.set_margin_top(8)
|
||||
separator.set_margin_bottom(8)
|
||||
@@ -2262,6 +2279,8 @@ class SettingsDialog(Gtk.Dialog):
|
||||
self.advanced_initial_prompt_buffer = self.advanced_initial_prompt_textview.get_buffer()
|
||||
self.advanced_initial_prompt_buffer.connect("changed", self._on_advanced_prompt_changed)
|
||||
|
||||
self.gpu_device_combo.connect("changed", self._on_advanced_param_changed)
|
||||
|
||||
self.power_user_switch.connect("state-set", self._on_power_user_toggled)
|
||||
|
||||
def _build_remote_server_section(self):
|
||||
@@ -2465,6 +2484,7 @@ class SettingsDialog(Gtk.Dialog):
|
||||
self.advanced_entropy_thold_spin.set_value(defaults["whispercpp_entropy_thold"])
|
||||
self.advanced_logprob_thold_spin.set_value(defaults["whispercpp_logprob_thold"])
|
||||
self.advanced_no_speech_thold_spin.set_value(defaults["whispercpp_no_speech_thold"])
|
||||
self.gpu_device_combo.set_active_id("-1")
|
||||
finally:
|
||||
self._applying_settings = False
|
||||
|
||||
@@ -3284,6 +3304,11 @@ class SettingsDialog(Gtk.Dialog):
|
||||
"whispercpp_no_speech_thold": self.advanced_no_speech_thold_spin.get_value(),
|
||||
}
|
||||
|
||||
gpu_device_id = self.gpu_device_combo.get_active_id()
|
||||
if gpu_device_id is not None:
|
||||
gpu_device_val = int(gpu_device_id)
|
||||
settings["whispercpp_gpu_device"] = None if gpu_device_val == -1 else gpu_device_val
|
||||
|
||||
# Remote API additional settings
|
||||
if engine == "remote_api":
|
||||
settings["remote_api_url"] = self.remote_api_url_entry.get_text().strip()
|
||||
@@ -3534,6 +3559,24 @@ For now, the engine has been reverted to VOSK."""
|
||||
error_dialog.destroy()
|
||||
return False
|
||||
|
||||
def _populate_gpu_devices(self):
|
||||
"""Populate the GPU device dropdown with available Vulkan devices."""
|
||||
self.gpu_device_combo.remove_all()
|
||||
self.gpu_device_combo.append("-1", "Auto (prefer discrete GPU)")
|
||||
|
||||
devices = detect_vulkan_devices()
|
||||
for device in devices:
|
||||
type_label = device["device_type"].capitalize()
|
||||
label = f"[{device['index']}] {device['name']} ({type_label})"
|
||||
self.gpu_device_combo.append(str(device["index"]), label)
|
||||
|
||||
saved_device = self.config_manager.get("advanced", "whispercpp_gpu_device", None)
|
||||
if saved_device is None:
|
||||
self.gpu_device_combo.set_active_id("-1")
|
||||
else:
|
||||
if not self.gpu_device_combo.set_active_id(str(saved_device)):
|
||||
self.gpu_device_combo.set_active_id("-1")
|
||||
|
||||
def _populate_audio_devices(self):
|
||||
"""Populate the audio device dropdown with available input devices."""
|
||||
from ..speech_recognition.recognition_manager import get_audio_input_devices
|
||||
|
||||
@@ -7,6 +7,7 @@ for whisper.cpp, supporting Vulkan, CUDA, and CPU backends.
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
from functools import lru_cache
|
||||
from typing import Optional
|
||||
@@ -125,41 +126,131 @@ class ComputeBackend:
|
||||
CPU = "cpu"
|
||||
|
||||
|
||||
# Real vulkaninfo --summary headers look like "GPU0:"; some older/alternate
|
||||
# tools print "GPU id = 0". Accept both so hybrid detection does not silently
|
||||
# fall through to CUDA when Vulkan is present.
|
||||
_VULKANINFO_GPU_HEADER_RE = re.compile(
|
||||
r"^(?:GPU(\d+)\s*:|GPU\s+id\s*[:=]\s*(\d+))\s*$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _parse_vulkaninfo_gpu_header(line: str) -> Optional[int]:
|
||||
"""Return a GPU index from a vulkaninfo device header line, if present."""
|
||||
match = _VULKANINFO_GPU_HEADER_RE.match(line.strip())
|
||||
if not match:
|
||||
return None
|
||||
return int(match.group(1) or match.group(2))
|
||||
|
||||
|
||||
def _append_vulkan_device(
|
||||
devices: list[dict],
|
||||
current_index: Optional[int],
|
||||
current_name: Optional[str],
|
||||
current_type: str,
|
||||
) -> None:
|
||||
"""Append a parsed Vulkan device when index and name are both known."""
|
||||
if current_index is None or current_name is None:
|
||||
return
|
||||
devices.append(
|
||||
{
|
||||
"index": current_index,
|
||||
"name": current_name,
|
||||
"device_type": current_type,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def detect_vulkan_support() -> tuple[bool, Optional[str]]:
|
||||
"""
|
||||
Detect if Vulkan is available and get device info.
|
||||
def detect_vulkan_devices() -> list[dict]:
|
||||
"""Enumerate all Vulkan-capable GPU devices.
|
||||
|
||||
Returns:
|
||||
Tuple of (is_available, device_name)
|
||||
"""
|
||||
"""
|
||||
Detect if Vulkan is available and get device info.
|
||||
|
||||
Returns:
|
||||
Tuple of (is_available, device_name)
|
||||
List of dicts with keys: index (int), name (str),
|
||||
device_type (str: "discrete" or "integrated" or "other").
|
||||
"""
|
||||
devices: list[dict] = []
|
||||
try:
|
||||
# Check for vulkaninfo command
|
||||
result = subprocess.run(
|
||||
["vulkaninfo", "--summary"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
if result.returncode == 0:
|
||||
# Try to extract GPU name from output
|
||||
for line in result.stdout.split("\n"):
|
||||
if "deviceName" in line or "GPU" in line:
|
||||
device_name = line.split(":")[-1].strip()
|
||||
if device_name:
|
||||
logger.info(f"Vulkan support detected: {device_name}")
|
||||
return True, device_name
|
||||
logger.info("Vulkan support detected")
|
||||
return True, "Vulkan GPU"
|
||||
except (subprocess.TimeoutExpired, FileNotFoundError, Exception) as e:
|
||||
logger.debug(f"Vulkan detection failed: {e}")
|
||||
if result.returncode != 0:
|
||||
return devices
|
||||
|
||||
current_index = None
|
||||
current_name = None
|
||||
current_type = "other"
|
||||
|
||||
for line in result.stdout.split("\n"):
|
||||
stripped = line.strip()
|
||||
header_index = _parse_vulkaninfo_gpu_header(stripped)
|
||||
|
||||
if header_index is not None:
|
||||
_append_vulkan_device(devices, current_index, current_name, current_type)
|
||||
current_index = header_index
|
||||
current_name = None
|
||||
current_type = "other"
|
||||
continue
|
||||
|
||||
if "deviceName" in stripped and "=" in stripped:
|
||||
current_name = stripped.split("=", 1)[-1].strip()
|
||||
|
||||
if "deviceType" in stripped and "=" in stripped:
|
||||
type_val = stripped.split("=", 1)[-1].strip().upper()
|
||||
if "DISCRETE" in type_val:
|
||||
current_type = "discrete"
|
||||
elif "INTEGRATED" in type_val:
|
||||
current_type = "integrated"
|
||||
|
||||
_append_vulkan_device(devices, current_index, current_name, current_type)
|
||||
|
||||
except (subprocess.TimeoutExpired, FileNotFoundError, Exception) as e:
|
||||
logger.debug(f"Vulkan device enumeration failed: {e}")
|
||||
|
||||
return devices
|
||||
|
||||
|
||||
def _prefer_discrete_vulkan_device() -> Optional[int]:
|
||||
"""Return the index of the preferred Vulkan GPU (discrete if available)."""
|
||||
devices = detect_vulkan_devices()
|
||||
for device in devices:
|
||||
if device["device_type"] == "discrete":
|
||||
return device["index"]
|
||||
return devices[0]["index"] if devices else None
|
||||
|
||||
|
||||
def _vulkan_device_name_by_index(devices: list[dict], device_index: Optional[int]) -> Optional[str]:
|
||||
"""Resolve a Vulkan device name by GPU index (not list position)."""
|
||||
if not devices:
|
||||
return None
|
||||
if device_index is None:
|
||||
return devices[0]["name"]
|
||||
for device in devices:
|
||||
if device["index"] == device_index:
|
||||
return device["name"]
|
||||
return devices[0]["name"]
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def detect_vulkan_support() -> tuple[bool, Optional[str]]:
|
||||
"""Detect if Vulkan is available and get device info.
|
||||
|
||||
When multiple Vulkan GPUs exist, prefers the discrete GPU.
|
||||
|
||||
Returns:
|
||||
Tuple of (is_available, device_name)
|
||||
"""
|
||||
devices = detect_vulkan_devices()
|
||||
if devices:
|
||||
preferred_idx = _prefer_discrete_vulkan_device()
|
||||
device_name = _vulkan_device_name_by_index(devices, preferred_idx)
|
||||
logger.info(f"Vulkan support detected: {device_name}")
|
||||
return True, device_name
|
||||
|
||||
logger.debug("Vulkan detection failed")
|
||||
return False, None
|
||||
|
||||
|
||||
|
||||
@@ -241,6 +241,7 @@ class TestMainModule(unittest.TestCase):
|
||||
whispercpp_logprob_thold=-1.0,
|
||||
whispercpp_no_speech_thold=0.6,
|
||||
whispercpp_n_threads=0,
|
||||
whispercpp_gpu_device=None,
|
||||
remote_api_url="",
|
||||
remote_api_key="",
|
||||
remote_api_endpoint="/inference",
|
||||
|
||||
@@ -501,6 +501,207 @@ class TestWhispercppInitialization(unittest.TestCase):
|
||||
manager._init_whispercpp()
|
||||
|
||||
|
||||
class TestWhispercppGpuDeviceSelection(unittest.TestCase):
|
||||
"""Test whisper.cpp GPU device selection logic."""
|
||||
|
||||
def test_gpu_device_auto_select_discrete(self):
|
||||
"""Test auto-selection of discrete GPU when configured as None."""
|
||||
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=None)
|
||||
manager.model_size = "tiny"
|
||||
|
||||
mock_pywhispercpp = MagicMock()
|
||||
mock_model = MagicMock()
|
||||
mock_pywhispercpp.Model = MagicMock(return_value=mock_model)
|
||||
mock_pywhispercpp.model.Model = MagicMock(return_value=mock_model)
|
||||
|
||||
mock_psutil = MagicMock()
|
||||
mock_psutil.virtual_memory.return_value.total = 8 * 1024 * 1024 * 1024
|
||||
|
||||
vulkan_devices = [
|
||||
{"index": 0, "name": "Intel UHD 630", "device_type": "integrated"},
|
||||
{"index": 1, "name": "NVIDIA RTX 2060", "device_type": "discrete"},
|
||||
]
|
||||
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO", {"tiny": {}}
|
||||
):
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
||||
return_value="/fake/model.bin",
|
||||
):
|
||||
with patch("os.path.getsize", return_value=100000000):
|
||||
with patch("os.path.exists", return_value=True):
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{
|
||||
"pywhispercpp": mock_pywhispercpp,
|
||||
"pywhispercpp.model": mock_pywhispercpp,
|
||||
"psutil": mock_psutil,
|
||||
},
|
||||
):
|
||||
import vocalinux.utils.whispercpp_model_info as whispercpp_info
|
||||
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"detect_compute_backend",
|
||||
return_value=(whispercpp_info.ComputeBackend.VULKAN, "Vulkan GPU"),
|
||||
):
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"get_backend_display_name",
|
||||
return_value="Vulkan",
|
||||
):
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"detect_vulkan_devices",
|
||||
return_value=vulkan_devices,
|
||||
):
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"_prefer_discrete_vulkan_device",
|
||||
return_value=1,
|
||||
):
|
||||
manager._init_whispercpp()
|
||||
assert mock_pywhispercpp.Model.call_args.kwargs.get(
|
||||
"context_params"
|
||||
) == {"gpu_device": 1}
|
||||
|
||||
def test_gpu_device_explicit_index(self):
|
||||
"""Test explicit GPU device index selection."""
|
||||
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=0)
|
||||
manager.model_size = "tiny"
|
||||
|
||||
mock_pywhispercpp = MagicMock()
|
||||
mock_model = MagicMock()
|
||||
mock_pywhispercpp.Model = MagicMock(return_value=mock_model)
|
||||
mock_pywhispercpp.model.Model = MagicMock(return_value=mock_model)
|
||||
|
||||
mock_psutil = MagicMock()
|
||||
mock_psutil.virtual_memory.return_value.total = 8 * 1024 * 1024 * 1024
|
||||
|
||||
vulkan_devices = [
|
||||
{"index": 0, "name": "Intel UHD 630", "device_type": "integrated"},
|
||||
{"index": 1, "name": "NVIDIA RTX 2060", "device_type": "discrete"},
|
||||
]
|
||||
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO", {"tiny": {}}
|
||||
):
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
||||
return_value="/fake/model.bin",
|
||||
):
|
||||
with patch("os.path.getsize", return_value=100000000):
|
||||
with patch("os.path.exists", return_value=True):
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{
|
||||
"pywhispercpp": mock_pywhispercpp,
|
||||
"pywhispercpp.model": mock_pywhispercpp,
|
||||
"psutil": mock_psutil,
|
||||
},
|
||||
):
|
||||
import vocalinux.utils.whispercpp_model_info as whispercpp_info
|
||||
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"detect_compute_backend",
|
||||
return_value=(whispercpp_info.ComputeBackend.VULKAN, "Vulkan GPU"),
|
||||
):
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"get_backend_display_name",
|
||||
return_value="Vulkan",
|
||||
):
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"detect_vulkan_devices",
|
||||
return_value=vulkan_devices,
|
||||
):
|
||||
manager._init_whispercpp()
|
||||
assert mock_pywhispercpp.Model.call_args.kwargs.get(
|
||||
"context_params"
|
||||
) == {"gpu_device": 0}
|
||||
|
||||
def test_gpu_device_context_params_fallback_on_old_pywhispercpp(self):
|
||||
"""Test graceful fallback when pywhispercpp rejects context_params."""
|
||||
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=1)
|
||||
manager.model_size = "tiny"
|
||||
|
||||
mock_pywhispercpp = MagicMock()
|
||||
mock_model = MagicMock()
|
||||
|
||||
def _model_side_effect(*args, **kwargs):
|
||||
if kwargs.get("context_params"):
|
||||
raise TypeError("Model() got an unexpected keyword argument 'context_params'")
|
||||
return mock_model
|
||||
|
||||
mock_pywhispercpp.Model.side_effect = _model_side_effect
|
||||
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{
|
||||
"pywhispercpp": mock_pywhispercpp,
|
||||
"pywhispercpp.model": mock_pywhispercpp,
|
||||
},
|
||||
):
|
||||
result = manager._load_model_with_compatible_params("/fake/model.bin", {}, gpu_device=1)
|
||||
assert result is mock_model
|
||||
assert mock_pywhispercpp.Model.call_count == 2
|
||||
assert mock_pywhispercpp.Model.call_args_list[0].kwargs.get("context_params") == {
|
||||
"gpu_device": 1
|
||||
}
|
||||
assert "context_params" not in mock_pywhispercpp.Model.call_args_list[1].kwargs
|
||||
|
||||
def test_gpu_device_not_set_for_cpu_backend(self):
|
||||
"""Test that GPU device selection is skipped on CPU backend."""
|
||||
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=None)
|
||||
manager.model_size = "tiny"
|
||||
|
||||
mock_pywhispercpp = MagicMock()
|
||||
mock_model = MagicMock()
|
||||
mock_pywhispercpp.Model = MagicMock(return_value=mock_model)
|
||||
mock_pywhispercpp.model.Model = MagicMock(return_value=mock_model)
|
||||
|
||||
mock_psutil = MagicMock()
|
||||
mock_psutil.virtual_memory.return_value.total = 8 * 1024 * 1024 * 1024
|
||||
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO", {"tiny": {}}
|
||||
):
|
||||
with patch(
|
||||
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
||||
return_value="/fake/model.bin",
|
||||
):
|
||||
with patch("os.path.getsize", return_value=100000000):
|
||||
with patch("os.path.exists", return_value=True):
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{
|
||||
"pywhispercpp": mock_pywhispercpp,
|
||||
"pywhispercpp.model": mock_pywhispercpp,
|
||||
"psutil": mock_psutil,
|
||||
},
|
||||
):
|
||||
import vocalinux.utils.whispercpp_model_info as whispercpp_info
|
||||
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"detect_compute_backend",
|
||||
return_value=(whispercpp_info.ComputeBackend.CPU, "CPU"),
|
||||
):
|
||||
with patch.object(
|
||||
whispercpp_info,
|
||||
"get_backend_display_name",
|
||||
return_value="CPU",
|
||||
):
|
||||
manager._init_whispercpp()
|
||||
assert (
|
||||
"context_params"
|
||||
not in mock_pywhispercpp.Model.call_args.kwargs
|
||||
)
|
||||
|
||||
|
||||
class TestTranscription(unittest.TestCase):
|
||||
"""Test transcription methods."""
|
||||
|
||||
|
||||
@@ -205,9 +205,221 @@ class TestModelInfo(unittest.TestCase):
|
||||
self.assertFalse(is_english_only_model("large-v3-turbo"))
|
||||
|
||||
|
||||
class TestDetectVulkanDevices(unittest.TestCase):
|
||||
"""Test cases for detect_vulkan_devices function."""
|
||||
|
||||
def setUp(self):
|
||||
"""Clear lru_cache before each test."""
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_devices
|
||||
|
||||
detect_vulkan_devices.cache_clear()
|
||||
|
||||
def test_detect_vulkan_devices_empty_when_unavailable(self):
|
||||
"""Test that detect_vulkan_devices returns empty list when vulkaninfo unavailable."""
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.side_effect = FileNotFoundError("vulkaninfo not found")
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_devices
|
||||
|
||||
devices = detect_vulkan_devices()
|
||||
self.assertEqual(devices, [])
|
||||
|
||||
def test_detect_vulkan_devices_single_discrete(self):
|
||||
"""Test detection of a single discrete GPU from real vulkaninfo --summary."""
|
||||
vulkaninfo_output = (
|
||||
"Devices:\n"
|
||||
"========\n"
|
||||
"GPU0:\n"
|
||||
"\tapiVersion = 1.3.296\n"
|
||||
"\tdriverVersion = 550.54.14\n"
|
||||
"\tvendorID = 0x10de\n"
|
||||
"\tdeviceID = 0x2206\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_DISCRETE_GPU\n"
|
||||
"\tdeviceName = NVIDIA RTX 3080\n"
|
||||
)
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_devices
|
||||
|
||||
devices = detect_vulkan_devices()
|
||||
self.assertEqual(len(devices), 1)
|
||||
self.assertEqual(devices[0]["index"], 0)
|
||||
self.assertEqual(devices[0]["name"], "NVIDIA RTX 3080")
|
||||
self.assertEqual(devices[0]["device_type"], "discrete")
|
||||
|
||||
def test_detect_vulkan_devices_hybrid_system(self):
|
||||
"""Test detection on hybrid system with real vulkaninfo --summary output."""
|
||||
vulkaninfo_output = (
|
||||
"Devices:\n"
|
||||
"========\n"
|
||||
"GPU0:\n"
|
||||
"\tapiVersion = 1.3.296\n"
|
||||
"\tdriverVersion = 24.2.8\n"
|
||||
"\tvendorID = 0x8086\n"
|
||||
"\tdeviceID = 0x3e92\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_INTEGRATED_GPU\n"
|
||||
"\tdeviceName = Intel(R) UHD Graphics 630 (CFL GT2)\n"
|
||||
"GPU1:\n"
|
||||
"\tapiVersion = 1.3.242\n"
|
||||
"\tdriverVersion = 550.54.14\n"
|
||||
"\tvendorID = 0x10de\n"
|
||||
"\tdeviceID = 0x1f08\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_DISCRETE_GPU\n"
|
||||
"\tdeviceName = NVIDIA GeForce RTX 2060\n"
|
||||
)
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_devices
|
||||
|
||||
devices = detect_vulkan_devices()
|
||||
self.assertEqual(len(devices), 2)
|
||||
self.assertEqual(devices[0]["index"], 0)
|
||||
self.assertEqual(devices[0]["name"], "Intel(R) UHD Graphics 630 (CFL GT2)")
|
||||
self.assertEqual(devices[0]["device_type"], "integrated")
|
||||
self.assertEqual(devices[1]["index"], 1)
|
||||
self.assertEqual(devices[1]["name"], "NVIDIA GeForce RTX 2060")
|
||||
self.assertEqual(devices[1]["device_type"], "discrete")
|
||||
|
||||
def test_detect_vulkan_devices_legacy_gpu_id_format(self):
|
||||
"""Test compatibility with alternate 'GPU id = N' vulkaninfo headers."""
|
||||
vulkaninfo_output = (
|
||||
"GPU id = 0\n"
|
||||
"deviceName = Intel UHD 630\n"
|
||||
"deviceType = PHYSICAL_DEVICE_TYPE_INTEGRATED_GPU\n"
|
||||
"GPU id = 1\n"
|
||||
"deviceName = NVIDIA RTX 2060\n"
|
||||
"deviceType = PHYSICAL_DEVICE_TYPE_DISCRETE_GPU\n"
|
||||
)
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_devices
|
||||
|
||||
devices = detect_vulkan_devices()
|
||||
self.assertEqual(len(devices), 2)
|
||||
self.assertEqual(devices[0]["index"], 0)
|
||||
self.assertEqual(devices[1]["index"], 1)
|
||||
self.assertEqual(devices[1]["device_type"], "discrete")
|
||||
|
||||
def test_detect_vulkan_devices_on_timeout(self):
|
||||
"""Test that timeout returns empty list."""
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.side_effect = subprocess.TimeoutExpired("vulkaninfo", 5)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_devices
|
||||
|
||||
devices = detect_vulkan_devices()
|
||||
self.assertEqual(devices, [])
|
||||
|
||||
def test_detect_vulkan_devices_on_error(self):
|
||||
"""Test that error returncode returns empty list."""
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=1, stdout="")
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_devices
|
||||
|
||||
devices = detect_vulkan_devices()
|
||||
self.assertEqual(devices, [])
|
||||
|
||||
|
||||
class TestPreferDiscreteVulkanDevice(unittest.TestCase):
|
||||
"""Test cases for _prefer_discrete_vulkan_device function."""
|
||||
|
||||
def setUp(self):
|
||||
"""Clear lru_cache before each test."""
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_devices
|
||||
|
||||
detect_vulkan_devices.cache_clear()
|
||||
|
||||
def test_prefer_discrete_when_available(self):
|
||||
"""Test that discrete GPU is preferred when available."""
|
||||
vulkaninfo_output = (
|
||||
"GPU0:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_INTEGRATED_GPU\n"
|
||||
"\tdeviceName = Intel UHD 630\n"
|
||||
"GPU1:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_DISCRETE_GPU\n"
|
||||
"\tdeviceName = NVIDIA RTX 2060\n"
|
||||
)
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import _prefer_discrete_vulkan_device
|
||||
|
||||
preferred_idx = _prefer_discrete_vulkan_device()
|
||||
self.assertEqual(preferred_idx, 1)
|
||||
|
||||
def test_prefer_discrete_falls_back_to_first(self):
|
||||
"""Test that first device is used when no discrete GPU available."""
|
||||
vulkaninfo_output = (
|
||||
"GPU0:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_INTEGRATED_GPU\n"
|
||||
"\tdeviceName = Intel UHD 630\n"
|
||||
)
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import _prefer_discrete_vulkan_device
|
||||
|
||||
preferred_idx = _prefer_discrete_vulkan_device()
|
||||
self.assertEqual(preferred_idx, 0)
|
||||
|
||||
def test_prefer_discrete_returns_none_when_empty(self):
|
||||
"""Test that None is returned when no devices found."""
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.side_effect = FileNotFoundError("vulkaninfo not found")
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import _prefer_discrete_vulkan_device
|
||||
|
||||
preferred_idx = _prefer_discrete_vulkan_device()
|
||||
self.assertIsNone(preferred_idx)
|
||||
|
||||
def test_prefer_discrete_with_noncontiguous_gpu_ids(self):
|
||||
"""Preferred index is the Vulkan GPU id, not the list position."""
|
||||
vulkaninfo_output = (
|
||||
"GPU0:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_INTEGRATED_GPU\n"
|
||||
"\tdeviceName = Intel UHD 630\n"
|
||||
"GPU2:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_DISCRETE_GPU\n"
|
||||
"\tdeviceName = NVIDIA RTX 2060\n"
|
||||
)
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import (
|
||||
_prefer_discrete_vulkan_device,
|
||||
detect_vulkan_support,
|
||||
)
|
||||
|
||||
preferred_idx = _prefer_discrete_vulkan_device()
|
||||
self.assertEqual(preferred_idx, 2)
|
||||
|
||||
detect_vulkan_support.cache_clear()
|
||||
is_available, device_name = detect_vulkan_support()
|
||||
self.assertTrue(is_available)
|
||||
self.assertEqual(device_name, "NVIDIA RTX 2060")
|
||||
|
||||
|
||||
class TestDetectVulkanSupport(unittest.TestCase):
|
||||
"""Test cases for detect_vulkan_support function."""
|
||||
|
||||
def setUp(self):
|
||||
"""Clear lru_cache before each test."""
|
||||
from vocalinux.utils.whispercpp_model_info import (
|
||||
detect_compute_backend,
|
||||
detect_cuda_support,
|
||||
detect_vulkan_devices,
|
||||
detect_vulkan_support,
|
||||
)
|
||||
|
||||
detect_vulkan_support.cache_clear()
|
||||
detect_vulkan_devices.cache_clear()
|
||||
detect_cuda_support.cache_clear()
|
||||
detect_compute_backend.cache_clear()
|
||||
|
||||
def test_detect_vulkan_support_returns_tuple(self):
|
||||
"""Test that detect_vulkan_support returns a tuple."""
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_support
|
||||
@@ -218,15 +430,70 @@ class TestDetectVulkanSupport(unittest.TestCase):
|
||||
|
||||
def test_detect_vulkan_support_when_available(self):
|
||||
"""Test Vulkan detection when vulkaninfo is available."""
|
||||
vulkaninfo_output = (
|
||||
"GPU0:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_DISCRETE_GPU\n"
|
||||
"\tdeviceName = Intel Arc\n"
|
||||
)
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout="deviceName: Intel Arc\n")
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_support
|
||||
|
||||
is_available, device_name = detect_vulkan_support()
|
||||
|
||||
self.assertTrue(is_available)
|
||||
self.assertIsNotNone(device_name)
|
||||
self.assertEqual(device_name, "Intel Arc")
|
||||
|
||||
def test_detect_vulkan_support_prefers_discrete_on_hybrid(self):
|
||||
"""Hybrid systems should report the discrete GPU name."""
|
||||
vulkaninfo_output = (
|
||||
"GPU0:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_INTEGRATED_GPU\n"
|
||||
"\tdeviceName = Intel(R) UHD Graphics 630 (CFL GT2)\n"
|
||||
"GPU1:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_DISCRETE_GPU\n"
|
||||
"\tdeviceName = NVIDIA GeForce RTX 2060\n"
|
||||
)
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
|
||||
from vocalinux.utils.whispercpp_model_info import detect_vulkan_support
|
||||
|
||||
is_available, device_name = detect_vulkan_support()
|
||||
|
||||
self.assertTrue(is_available)
|
||||
self.assertEqual(device_name, "NVIDIA GeForce RTX 2060")
|
||||
|
||||
def test_compute_backend_stays_vulkan_with_real_summary_and_nvidia(self):
|
||||
"""Regression: real GPU0: summary must not fall through to CUDA."""
|
||||
vulkaninfo_output = (
|
||||
"Devices:\n"
|
||||
"========\n"
|
||||
"GPU0:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_INTEGRATED_GPU\n"
|
||||
"\tdeviceName = Intel(R) UHD Graphics 630 (CFL GT2)\n"
|
||||
"GPU1:\n"
|
||||
"\tdeviceType = PHYSICAL_DEVICE_TYPE_DISCRETE_GPU\n"
|
||||
"\tdeviceName = NVIDIA GeForce RTX 2060\n"
|
||||
)
|
||||
|
||||
def run_side_effect(cmd, **kwargs):
|
||||
if cmd and cmd[0] == "vulkaninfo":
|
||||
return MagicMock(returncode=0, stdout=vulkaninfo_output)
|
||||
if cmd and cmd[0] == "nvidia-smi":
|
||||
return MagicMock(returncode=0, stdout="NVIDIA GeForce RTX 2060, 6144 MiB")
|
||||
return MagicMock(returncode=1, stdout="")
|
||||
|
||||
with patch("subprocess.run", side_effect=run_side_effect):
|
||||
from vocalinux.utils.whispercpp_model_info import (
|
||||
ComputeBackend,
|
||||
detect_compute_backend,
|
||||
)
|
||||
|
||||
backend, backend_info = detect_compute_backend()
|
||||
self.assertEqual(backend, ComputeBackend.VULKAN)
|
||||
self.assertEqual(backend_info, "NVIDIA GeForce RTX 2060")
|
||||
|
||||
def test_detect_vulkan_support_when_unavailable(self):
|
||||
"""Test Vulkan detection when vulkaninfo is not available."""
|
||||
|
||||
Reference in New Issue
Block a user