feat(models): memory-aware whisper.cpp ladder with conditional backend order

Free memory picks the load-time rung (large-v3-turbo-q8_0 default,
q5_0 under 1086 MiB free, small/small.en floor) for unpinned default
configs; thresholds derive from catalog sizes plus the protocol's
512 MiB headroom. A manual pin is never overridden, and a failed load
steps one rung down and retries exactly once before surfacing.

CUDA ranks before Vulkan only with a complete architecture-matching
toolkit (nvidia-smi GPU, nvcc, compute capability >= 7.0); otherwise
Vulkan, then CPU. The driver-only CUDA recommendation now demotes to
CPU logic instead of naming an unreachable VRAM ladder.

Refs #20
This commit is contained in:
soubarnak
2026-11-06 01:48:20 +05:30
parent 426f524145
commit df771d8fb1
9 changed files with 746 additions and 34 deletions
+4
View File
@@ -547,6 +547,9 @@ def main():
# loads the wrong model whenever the two disagree.
model_size = config_manager.get_model_size_for_engine(engine)
logger.info(f"Using model={model_size} (saved for engine {engine})")
# A CLI --model is an explicit pick; otherwise the variant pin state
# decides whether the loader may step the ladder down on failure (#20).
variant_pinned = cli_model_set or bool(config_manager.get_model_variant_for_engine(engine))
vad_sensitivity = saved_settings.get("vad_sensitivity", 3)
silence_timeout = saved_settings.get("silence_timeout", 1.2)
@@ -614,6 +617,7 @@ def main():
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),
whispercpp_variant_pinned=variant_pinned,
dictionary_manager=dictionary_manager,
remote_api_url=saved_settings.get("remote_api_url", ""),
remote_api_key=saved_settings.get("remote_api_key", ""),
@@ -0,0 +1,255 @@
"""Memory-aware whisper.cpp model ladder for Voculinux (ticket #20).
The default model is not one-size-fits-all on a 4 GB dGPU: the ladder picks
the rung that fits free memory at load time instead of failing halfway
through a Turbo load.
Rungs (quantized Turbo first, small floor last)::
large-v3-turbo-q8_0 -> large-v3-turbo-q5_0 -> small.en / small
Thresholds derive from the catalog on-disk sizes (874/574 MiB) plus the
512 MiB desktop headroom from the locked baseline protocol
(``scripts/measure_perf_baseline.py::recommend_thresholds``). Ticket #19
recorded no measured per-rung footprints, so these are size-plus-headroom
proxies: recalibrate them from hardware numbers when they land. The
derivation is pinned by ``tests/test_model_ladder.py`` so a catalog size
bump forces a threshold review.
Backend order follows the same readiness rule: CUDA ranks before Vulkan
only when a complete architecture-matching CUDA toolkit is installed
(``nvidia-smi`` sees a GPU, ``nvcc`` answers, compute capability >= 7.0);
otherwise the order is Vulkan, then CPU. A driver without the toolkit is
exactly what made the old VRAM-aware recommendation unreachable.
"""
import logging
import shutil
import subprocess
logger = logging.getLogger(__name__)
DEFAULT_RUNG: str = "large-v3-turbo-q8_0"
LOW_MEMORY_RUNG: str = "large-v3-turbo-q5_0"
FLOOR_RUNG_ENGLISH: str = "small.en"
FLOOR_RUNG_MULTILINGUAL: str = "small"
LADDER_RUNGS: tuple[str, ...] = (DEFAULT_RUNG, LOW_MEMORY_RUNG)
#: MiB of headroom kept for the desktop and other GPU clients. Same value
#: the baseline protocol adds to measured footprints.
HEADROOM_MIB: float = 512.0
#: Catalog on-disk sizes (MiB) the thresholds below are derived from.
_Q8_0_SIZE_MIB: float = 874.0
_Q5_0_SIZE_MIB: float = 574.0
#: Free memory at or above this keeps the default q8_0 rung.
PREFER_Q8_0_ABOVE_MIB: float = _Q8_0_SIZE_MIB + HEADROOM_MIB
#: Free memory at or above this keeps the q5_0 rung; below it is the floor.
PREFER_Q5_0_ABOVE_MIB: float = _Q5_0_SIZE_MIB + HEADROOM_MIB
#: Minimum CUDA compute capability (major, minor) the bundled backend
#: targets. Older architectures cannot run its kernels.
MIN_CUDA_COMPUTE_CAPABILITY: tuple[int, int] = (7, 0)
def select_ladder_rung(language_is_english: bool, free_mib: float) -> str:
"""Return the ladder rung that fits ``free_mib`` MiB of free memory.
Pure function of its inputs so the policy is testable without hardware.
"""
if free_mib >= PREFER_Q8_0_ABOVE_MIB:
return DEFAULT_RUNG
if free_mib >= PREFER_Q5_0_ABOVE_MIB:
return LOW_MEMORY_RUNG
return FLOOR_RUNG_ENGLISH if language_is_english else FLOOR_RUNG_MULTILINGUAL
def resolve_startup_rung(
pinned_variant: str,
language_is_english: bool,
free_mib: float,
) -> str:
"""Return the rung to load: a manual pin wins verbatim, otherwise the ladder.
A manual selection is never overridden here; only unpinned loads derive
their rung from free memory.
"""
if pinned_variant:
return pinned_variant
return select_ladder_rung(language_is_english, free_mib)
def next_rung_down(rung: str, language_is_english: bool) -> str | None:
"""Return the rung one step below ``rung``, or None at the floor.
The floor itself never steps down: a second consecutive failure is a
real error and must surface instead of looping.
"""
if rung == DEFAULT_RUNG:
return LOW_MEMORY_RUNG
if rung == LOW_MEMORY_RUNG:
return FLOOR_RUNG_ENGLISH if language_is_english else FLOOR_RUNG_MULTILINGUAL
return None
def _run_probe(argv: list[str], timeout: int = 10) -> str | None:
"""Run a probe command, returning stdout or None when unavailable.
Host binaries inherit a stripped host environment so probes keep working
inside the AppImage bundle (see ``utils/host_process.py``).
"""
from ..utils.host_process import host_env
if shutil.which(argv[0]) is None:
return None
try:
completed = subprocess.run(
argv, capture_output=True, text=True, timeout=timeout, env=host_env()
)
except (OSError, subprocess.SubprocessError) as error:
logger.debug("Probe %s failed: %s", argv[0], error)
return None
if completed.returncode != 0:
return None
return completed.stdout
def _parse_first_free_mib(smi_output: str) -> float | None:
"""Parse the first GPU's free memory (MiB) from nvidia-smi CSV output."""
for line in smi_output.splitlines():
fields = [field.strip() for field in line.split(",")]
if len(fields) >= 1:
try:
return float(fields[0])
except ValueError:
continue
return None
def _parse_first_compute_cap(smi_output: str) -> tuple[int, int] | None:
"""Parse the first GPU's compute capability from nvidia-smi CSV output."""
for line in smi_output.splitlines():
fields = [field.strip() for field in line.split(",")]
if len(fields) >= 2:
try:
major, _, minor = fields[1].partition(".")
return (int(major), int((minor or "0").strip()))
except ValueError:
continue
return None
def cuda_toolkit_ready_from_probe(
has_nvidia_gpu: bool,
nvcc_ok: bool,
compute_cap: tuple[int, int] | None,
) -> bool:
"""Return whether a complete architecture-matching CUDA toolkit is present.
Pure function of its inputs: an NVIDIA GPU alone (driver only) is not a
toolkit, and a toolkit that cannot target this architecture is not
usable either.
"""
if not has_nvidia_gpu or not nvcc_ok or compute_cap is None:
return False
return compute_cap >= MIN_CUDA_COMPUTE_CAPABILITY
def cuda_toolkit_ready() -> bool:
"""Probe the host for a complete architecture-matching CUDA toolkit."""
smi = _run_probe(
["nvidia-smi", "--query-gpu=memory.free,compute_cap", "--format=csv,noheader,nounits"]
)
if not smi:
return False
nvcc = _run_probe(["nvcc", "--version"])
return cuda_toolkit_ready_from_probe(
has_nvidia_gpu=True,
nvcc_ok=nvcc is not None,
compute_cap=_parse_first_compute_cap(smi),
)
def free_vram_mib() -> float | None:
"""Return the first NVIDIA GPU's free VRAM in MiB, or None without one."""
smi = _run_probe(["nvidia-smi", "--query-gpu=memory.free", "--format=csv,noheader,nounits"])
if not smi:
return None
return _parse_first_free_mib(smi)
def free_system_ram_mib() -> float | None:
"""Return free system RAM in MiB (psutil, falling back to /proc/meminfo)."""
try:
import psutil
return float(psutil.virtual_memory().available) / (1024**2)
except (ImportError, AttributeError, OSError) as error:
logger.debug("psutil free RAM unavailable: %s", error)
try:
with open("/proc/meminfo", encoding="utf-8") as handle:
for line in handle:
if line.startswith("MemAvailable:"):
return float(line.split()[1]) / 1024.0
except OSError as error:
logger.debug("/proc/meminfo unavailable: %s", error)
return None
def free_memory_mib() -> float | None:
"""Return free memory in MiB: discrete-GPU VRAM when present, else RAM."""
vram = free_vram_mib()
if vram is not None:
return vram
return free_system_ram_mib()
def preferred_backend_order() -> tuple[str, ...]:
"""Return the backend attempt order for this host.
CUDA ranks before Vulkan only when a complete architecture-matching
toolkit is installed; otherwise Vulkan, then CPU.
"""
if cuda_toolkit_ready():
return ("cuda", "vulkan", "cpu")
return ("vulkan", "cpu")
def describe_environment() -> str:
"""Return a one-line summary of the ladder inputs for logging."""
vram = free_vram_mib()
if vram is not None:
return f"free VRAM {vram:.0f} MiB, CUDA toolkit ready: {cuda_toolkit_ready()}"
ram = free_system_ram_mib()
if ram is not None:
return f"no dGPU, free RAM {ram:.0f} MiB"
return "memory unreadable; assuming the floor"
def is_ladder_rung(variant: str) -> bool:
"""Return whether ``variant`` is a rung the loader may step between."""
return variant in LADDER_RUNGS or variant in (FLOOR_RUNG_ENGLISH, FLOOR_RUNG_MULTILINGUAL)
__all__ = [
"DEFAULT_RUNG",
"LOW_MEMORY_RUNG",
"FLOOR_RUNG_ENGLISH",
"FLOOR_RUNG_MULTILINGUAL",
"LADDER_RUNGS",
"HEADROOM_MIB",
"PREFER_Q8_0_ABOVE_MIB",
"PREFER_Q5_0_ABOVE_MIB",
"MIN_CUDA_COMPUTE_CAPABILITY",
"select_ladder_rung",
"resolve_startup_rung",
"next_rung_down",
"is_ladder_rung",
"cuda_toolkit_ready_from_probe",
"cuda_toolkit_ready",
"free_memory_mib",
"preferred_backend_order",
"describe_environment",
]
@@ -506,6 +506,10 @@ class SpeechRecognitionManager:
# change a queued segment's language (#805).
self._session_language: Optional[str] = None
self.stop_sound_guard_ms = kwargs.get("stop_sound_guard_ms", 200)
# Whether the whisper.cpp variant was explicitly pinned in Settings.
# A manual selection is never overridden: neither the ladder nor the
# single step-down retry applies to it (ticket #20).
self._whispercpp_variant_pinned = bool(kwargs.get("whispercpp_variant_pinned", False))
self.dictionary_manager: Optional["CustomDictionaryManager"] = kwargs.get(
"dictionary_manager"
)
@@ -1509,7 +1513,43 @@ class SpeechRecognitionManager:
logger.error(f"Error in faster-whisper transcription: {e}", exc_info=True)
return ""
def _ladder_step_down_rung(self) -> Optional[str]:
"""Return the retry rung, or None when stepping down is not allowed.
Only ladder-selected (unpinned) loads step down, exactly one rung.
A manual pin and the small floor both fail loudly instead.
"""
if self._whispercpp_variant_pinned:
return None
from .model_ladder import is_ladder_rung, next_rung_down
if not is_ladder_rung(self.model_size):
return None
language_info = SUPPORTED_LANGUAGES.get(self.language, {})
language_is_english = (
isinstance(language_info, dict) and language_info.get("whisper") == "en"
)
return next_rung_down(self.model_size, language_is_english)
def _init_whispercpp(self) -> None:
"""Initialize the whisper.cpp engine, stepping one rung down on failure."""
fallback = self._ladder_step_down_rung()
try:
self._init_whispercpp_once()
return
except (RuntimeError, OSError) as first_error:
if fallback is None or fallback == self.model_size:
raise
logger.warning(
"whisper.cpp %s failed to load (%s); stepping down to %s once",
self.model_size,
first_error,
fallback,
)
self.model_size = fallback
self._init_whispercpp_once()
def _init_whispercpp_once(self) -> None:
"""Initialize the whisper.cpp speech recognition engine."""
try:
_preload_pywhispercpp_shared_libraries()
@@ -1739,8 +1779,9 @@ class SpeechRecognitionManager:
def _load_whispercpp_model(self, model_path: str):
"""Load the whisper.cpp model file and configure the compute backend.
Detects the optimal backend (Vulkan → CUDA → CPU), loads the model,
and falls back to CPU if the GPU backend is incompatible.
Detects the optimal backend (CUDA first only with a complete
toolkit, otherwise Vulkan → CPU), loads the model, and falls back
to CPU if the GPU backend is incompatible.
Args:
model_path: Filesystem path to the GGML model file.
@@ -1754,10 +1795,15 @@ class SpeechRecognitionManager:
get_backend_display_name,
)
from .model_ladder import preferred_backend_order
host_backend, host_info = detect_compute_backend()
actual_gpu_backend = self._detect_pywhispercpp_gpu_backend()
ggml_devices = detect_ggml_gpu_devices()
logger.info("whisper.cpp backend selection priority: Vulkan -> CUDA -> CPU")
logger.info(
"whisper.cpp backend selection priority: %s",
" -> ".join(preferred_backend_order()),
)
selected_gpu_device = None
use_gpu: Optional[bool] = None
+25 -1
View File
@@ -769,6 +769,30 @@ class ConfigManager:
if engine.lower() != "whisper_cpp":
return cast(str, saved)
pinned = self.get_model_variant_for_engine(engine)
# A fresh (or reset) install never chose a model: the saved value is
# still the "tiny" default and no variant is pinned. That default now
# resolves through the memory-aware ladder instead of the bare tiny
# id, so first run loads large-v3-turbo-q8_0 where it fits (ticket
# #20). Any explicit size or pin keeps the behavior below untouched.
if not pinned and str(saved or "tiny").lower() == "tiny":
from ..speech_recognition.model_ladder import free_memory_mib, resolve_startup_rung
language_id = str(sr_config.get("language", "auto"))
language_info = cast(dict[str, str], SUPPORTED_LANGUAGES.get(language_id, {}))
free_mib = free_memory_mib()
rung = resolve_startup_rung(
"", language_info.get("whisper") == "en", free_mib if free_mib is not None else 0.0
)
logger.info(
"whisper.cpp startup rung %s (free memory %.0f MiB)",
rung,
free_mib if free_mib is not None else -1.0,
)
return on_disk_stand_in(
rung, get_whispercpp_model_size(rung), language_info.get("whisper") == "en"
)
# Unpinned configs store the bare size; the engine still needs the
# language-derived loadable id (and leftover ``{size}.en`` must not
# block a later language change). The derived default stands down for
@@ -776,7 +800,7 @@ class ConfigManager:
# sibling download when a usable model is already on disk (#916).
return resolve_whispercpp_variant(
saved,
self.get_model_variant_for_engine(engine),
pinned,
sr_config.get("language", "auto"),
prefer_on_disk=True,
)
+16 -9
View File
@@ -581,15 +581,22 @@ def get_recommended_model() -> tuple[str, str]:
else:
return "base", f"Vulkan GPU with {ram_gb}GB RAM"
elif backend == ComputeBackend.CUDA:
vram_gb = _parse_cuda_vram_gib(backend_info or "")
if vram_gb is not None:
if vram_gb >= 8:
return "medium", f"CUDA GPU with {vram_gb:g}GB VRAM"
if vram_gb >= 4:
return "small", f"CUDA GPU with {vram_gb:g}GB VRAM"
return "base", "CUDA GPU with limited VRAM"
return "small", "CUDA GPU detected"
else:
# A driver without the complete toolkit cannot run the CUDA
# build, so the VRAM ladder below would be unreachable. Fall
# through to the Vulkan/CPU branches instead (ticket #20).
from ..speech_recognition.model_ladder import cuda_toolkit_ready
if cuda_toolkit_ready():
vram_gb = _parse_cuda_vram_gib(backend_info or "")
if vram_gb is not None:
if vram_gb >= 8:
return "medium", f"CUDA GPU with {vram_gb:g}GB VRAM"
if vram_gb >= 4:
return "small", f"CUDA GPU with {vram_gb:g}GB VRAM"
return "base", "CUDA GPU with limited VRAM"
return "small", "CUDA GPU detected"
backend = ComputeBackend.CPU
if backend == ComputeBackend.CPU:
# CPU-only recommendations based on RAM
if ram_gb >= 16:
return "base", f"{ram_gb}GB RAM - CPU inference"
+6 -1
View File
@@ -524,7 +524,12 @@ class TestConfigManager(unittest.TestCase):
config_manager = ConfigManager()
self.assertEqual(config_manager.get_model_size_for_engine("vosk"), "large")
self.assertEqual(config_manager.get_model_size_for_engine("whisper_cpp"), "tiny")
# The migrated whisper.cpp default ("tiny", unpinned) resolves through
# the memory-aware ladder instead of staying tiny (ticket #20): what
# matters here is that vosk's "large" does not leak across engines.
with patch("vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=100.0):
# "auto" language pairs the floor with the multilingual rung.
self.assertEqual(config_manager.get_model_size_for_engine("whisper_cpp"), "small")
self.assertEqual(config_manager.get_model_size_for_engine("whisper"), "tiny")
def test_resolver_prefers_the_engine_key_over_the_generic_one(self):
+1
View File
@@ -251,6 +251,7 @@ class TestMainModule(unittest.TestCase):
whispercpp_no_speech_thold=0.6,
whispercpp_n_threads=0,
whispercpp_gpu_device=None,
whispercpp_variant_pinned=True,
dictionary_manager=ANY,
remote_api_url="",
remote_api_key="",
+306
View File
@@ -0,0 +1,306 @@
"""Ticket #20: memory-aware model ladder, conditional backend order, retry.
All behavior is exercised with mocked probes and loaders: no GPU, no
network, no hardware. The threshold-derivation test pins the rung
thresholds to the catalog on-disk sizes plus the protocol headroom so a
catalog size bump forces a threshold review (ticket #19 recorded no
measured footprints, so these are size-plus-headroom proxies).
"""
from unittest.mock import MagicMock, patch
from vocalinux.speech_recognition import model_ladder
from vocalinux.speech_recognition.model_ladder import (
DEFAULT_RUNG,
FLOOR_RUNG_ENGLISH,
FLOOR_RUNG_MULTILINGUAL,
HEADROOM_MIB,
LOW_MEMORY_RUNG,
PREFER_Q5_0_ABOVE_MIB,
PREFER_Q8_0_ABOVE_MIB,
cuda_toolkit_ready_from_probe,
next_rung_down,
preferred_backend_order,
resolve_startup_rung,
select_ladder_rung,
)
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
from vocalinux.utils.whispercpp_model_info import WHISPERCPP_MODEL_INFO
def _size_mb(name: str) -> float:
return float(WHISPERCPP_MODEL_INFO[name]["size_mb"])
class TestThresholdDerivation:
"""Thresholds match the baseline ticket's recorded recommendation shape."""
def test_q8_threshold_is_catalog_size_plus_headroom(self) -> None:
assert PREFER_Q8_0_ABOVE_MIB == _size_mb(DEFAULT_RUNG) + HEADROOM_MIB
def test_q5_threshold_is_catalog_size_plus_headroom(self) -> None:
assert PREFER_Q5_0_ABOVE_MIB == _size_mb(LOW_MEMORY_RUNG) + HEADROOM_MIB
def test_rungs_are_ordered_by_size(self) -> None:
assert _size_mb(DEFAULT_RUNG) > _size_mb(LOW_MEMORY_RUNG)
class TestSelectLadderRung:
"""Free memory picks the rung; the floor pairs with the language."""
def test_plenty_of_memory_keeps_default(self) -> None:
assert select_ladder_rung(True, PREFER_Q8_0_ABOVE_MIB) == DEFAULT_RUNG
assert select_ladder_rung(False, 99999.0) == DEFAULT_RUNG
def test_between_thresholds_steps_to_q5(self) -> None:
assert select_ladder_rung(True, PREFER_Q8_0_ABOVE_MIB - 1) == LOW_MEMORY_RUNG
assert select_ladder_rung(True, PREFER_Q5_0_ABOVE_MIB) == LOW_MEMORY_RUNG
def test_below_q5_fits_the_floor_by_language(self) -> None:
assert select_ladder_rung(True, PREFER_Q5_0_ABOVE_MIB - 1) == FLOOR_RUNG_ENGLISH
assert select_ladder_rung(False, PREFER_Q5_0_ABOVE_MIB - 1) == FLOOR_RUNG_MULTILINGUAL
assert select_ladder_rung(True, 0.0) == FLOOR_RUNG_ENGLISH
assert select_ladder_rung(False, 0.0) == FLOOR_RUNG_MULTILINGUAL
class TestResolveStartupRung:
"""A manual pin wins verbatim; only unpinned loads use the ladder."""
def test_pinned_variant_never_overridden(self) -> None:
assert resolve_startup_rung("tiny", True, 99999.0) == "tiny"
assert resolve_startup_rung(LOW_MEMORY_RUNG, False, 0.0) == LOW_MEMORY_RUNG
assert resolve_startup_rung(FLOOR_RUNG_ENGLISH, True, 99999.0) == FLOOR_RUNG_ENGLISH
def test_unpinned_uses_ladder(self) -> None:
assert resolve_startup_rung("", True, 99999.0) == DEFAULT_RUNG
assert resolve_startup_rung("", False, 0.0) == FLOOR_RUNG_MULTILINGUAL
class TestNextRungDown:
"""One step down per failure; the floor fails loudly."""
def test_default_steps_to_low_memory(self) -> None:
assert next_rung_down(DEFAULT_RUNG, True) == LOW_MEMORY_RUNG
assert next_rung_down(DEFAULT_RUNG, False) == LOW_MEMORY_RUNG
def test_low_memory_steps_to_floor_by_language(self) -> None:
assert next_rung_down(LOW_MEMORY_RUNG, True) == FLOOR_RUNG_ENGLISH
assert next_rung_down(LOW_MEMORY_RUNG, False) == FLOOR_RUNG_MULTILINGUAL
def test_floor_has_no_step_down(self) -> None:
assert next_rung_down(FLOOR_RUNG_ENGLISH, True) is None
assert next_rung_down(FLOOR_RUNG_MULTILINGUAL, False) is None
def test_non_ladder_rungs_have_no_step_down(self) -> None:
assert next_rung_down("tiny", True) is None
assert next_rung_down("medium", False) is None
class TestCudaToolkitReadiness:
"""Driver alone is not a toolkit; arch must match too."""
def test_complete_matching_toolkit_is_ready(self) -> None:
assert cuda_toolkit_ready_from_probe(True, True, (8, 6)) is True
assert cuda_toolkit_ready_from_probe(True, True, (7, 0)) is True
def test_driver_without_toolkit_is_not_ready(self) -> None:
assert cuda_toolkit_ready_from_probe(True, False, (8, 6)) is False
def test_no_gpu_is_not_ready(self) -> None:
assert cuda_toolkit_ready_from_probe(False, True, (8, 6)) is False
assert cuda_toolkit_ready_from_probe(False, False, None) is False
def test_old_architecture_is_not_ready(self) -> None:
assert cuda_toolkit_ready_from_probe(True, True, (6, 1)) is False
assert cuda_toolkit_ready_from_probe(True, True, None) is False
class TestPreferredBackendOrder:
"""CUDA ranks first only with a complete toolkit."""
def test_cuda_first_with_toolkit(self) -> None:
with patch.object(model_ladder, "cuda_toolkit_ready", return_value=True):
assert preferred_backend_order() == ("cuda", "vulkan", "cpu")
def test_vulkan_then_cpu_without_toolkit(self) -> None:
with patch.object(model_ladder, "cuda_toolkit_ready", return_value=False):
assert preferred_backend_order() == ("vulkan", "cpu")
def _make_manager(**kwargs):
"""Create a whisper.cpp manager without running any initializer."""
with (
patch.object(SpeechRecognitionManager, "_init_vosk"),
patch.object(SpeechRecognitionManager, "_init_whisper"),
patch.object(SpeechRecognitionManager, "_init_whispercpp"),
):
return SpeechRecognitionManager(
engine="whisper_cpp",
model_size="small",
language="en-us",
defer_download=True,
**kwargs,
)
class TestStepDownRetry:
"""The loader steps one rung down and retries once, then surfaces."""
def test_first_failure_steps_down_and_retries_once(self) -> None:
manager = _make_manager()
manager.model_size = DEFAULT_RUNG
calls = []
def fail_once() -> None:
calls.append(manager.model_size)
if len(calls) == 1:
raise RuntimeError("CUDA out of memory")
with patch.object(SpeechRecognitionManager, "_init_whispercpp_once", side_effect=fail_once):
manager._init_whispercpp()
assert calls == [DEFAULT_RUNG, LOW_MEMORY_RUNG]
assert manager.model_size == LOW_MEMORY_RUNG
def test_second_failure_surfaces(self) -> None:
manager = _make_manager()
manager.model_size = DEFAULT_RUNG
with patch.object(
SpeechRecognitionManager, "_init_whispercpp_once", side_effect=RuntimeError("boom")
):
try:
manager._init_whispercpp()
except RuntimeError:
pass
else:
raise AssertionError("second failure must surface")
assert manager.model_size == LOW_MEMORY_RUNG
def test_manual_pin_never_steps_down(self) -> None:
manager = _make_manager(whispercpp_variant_pinned=True)
manager.model_size = DEFAULT_RUNG
loader = MagicMock(side_effect=RuntimeError("CUDA out of memory"))
with patch.object(SpeechRecognitionManager, "_init_whispercpp_once", loader):
try:
manager._init_whispercpp()
except RuntimeError:
pass
else:
raise AssertionError("manual pick must fail loudly")
assert loader.call_count == 1
assert manager.model_size == DEFAULT_RUNG
def test_floor_rung_fails_loudly(self) -> None:
manager = _make_manager()
manager.model_size = FLOOR_RUNG_ENGLISH
loader = MagicMock(side_effect=OSError("no space"))
with patch.object(SpeechRecognitionManager, "_init_whispercpp_once", loader):
try:
manager._init_whispercpp()
except OSError:
pass
else:
raise AssertionError("floor failure must surface")
assert loader.call_count == 1
def test_non_ladder_model_does_not_step_down(self) -> None:
manager = _make_manager()
manager.model_size = "medium"
loader = MagicMock(side_effect=RuntimeError("boom"))
with patch.object(SpeechRecognitionManager, "_init_whispercpp_once", loader):
try:
manager._init_whispercpp()
except RuntimeError:
pass
else:
raise AssertionError("non-ladder failure must surface")
assert loader.call_count == 1
class TestStartupRungSelection:
"""Unpinned defaults resolve through the ladder; explicit picks do not."""
def _manager_with(self, saved_size, pinned_variant, language="en-us"):
from vocalinux.ui.config_manager import ConfigManager
manager = ConfigManager.__new__(ConfigManager)
manager.config = {
"speech_recognition": {
"whisper_cpp_model_size": saved_size,
"whisper_cpp_model_variant": pinned_variant,
"language": language,
}
}
return manager
def test_unpinned_default_selects_rung_by_memory(self) -> None:
manager = self._manager_with("tiny", "", "en-us")
with patch(
"vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=99999.0
):
with patch(
"vocalinux.utils.whispercpp_model_info.is_model_downloaded", return_value=False
):
assert manager.get_model_size_for_engine("whisper_cpp") == DEFAULT_RUNG
def test_unpinned_default_low_memory_selects_q5(self) -> None:
manager = self._manager_with("tiny", "", "en-us")
with patch(
"vocalinux.speech_recognition.model_ladder.free_memory_mib",
return_value=PREFER_Q5_0_ABOVE_MIB,
):
with patch(
"vocalinux.utils.whispercpp_model_info.is_model_downloaded", return_value=False
):
assert manager.get_model_size_for_engine("whisper_cpp") == LOW_MEMORY_RUNG
def test_unpinned_default_tight_memory_selects_floor(self) -> None:
manager = self._manager_with("tiny", "", "pl")
with patch("vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=100.0):
with patch(
"vocalinux.utils.whispercpp_model_info.is_model_downloaded", return_value=False
):
assert manager.get_model_size_for_engine("whisper_cpp") == FLOOR_RUNG_MULTILINGUAL
def test_pinned_variant_never_uses_ladder(self) -> None:
manager = self._manager_with("tiny", LOW_MEMORY_RUNG, "en-us")
with patch(
"vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=99999.0
):
assert manager.get_model_size_for_engine("whisper_cpp") == LOW_MEMORY_RUNG
def test_explicit_size_keeps_legacy_derivation(self) -> None:
manager = self._manager_with("medium", "", "en-us")
with patch(
"vocalinux.speech_recognition.model_ladder.free_memory_mib", return_value=99999.0
):
with patch(
"vocalinux.utils.whispercpp_model_info.is_model_downloaded", return_value=False
):
assert manager.get_model_size_for_engine("whisper_cpp") == "medium.en"
class TestRecommendationGate:
"""Driver-only CUDA demotes to CPU logic; toolkit keeps the VRAM ladder."""
def _recommend(self, backend, toolkit):
from vocalinux.utils import whispercpp_model_info as info
mock_psutil = MagicMock()
mock_psutil.virtual_memory.return_value = MagicMock(total=16 * (1024**3))
with patch.dict("sys.modules", {"psutil": mock_psutil}):
with patch.object(info, "detect_compute_backend", return_value=backend):
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=toolkit,
):
return info.get_recommended_model()
def test_driver_only_cuda_demotes_to_cpu(self) -> None:
model, _reason = self._recommend(("cuda", "NVIDIA RTX 2050 (4GB)"), False)
assert model == "base"
def test_toolkit_cuda_keeps_vram_ladder(self) -> None:
model, reason = self._recommend(("cuda", "NVIDIA RTX 2050 (4GB)"), True)
assert model == "small"
assert "VRAM" in reason
+84 -20
View File
@@ -999,9 +999,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("vulkan", "Intel Arc A770")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "small")
@@ -1018,13 +1022,41 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("cuda", "NVIDIA RTX 4090 (24GB)")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
# High VRAM should recommend medium or higher
self.assertIn(model, ["medium", "large", "small"])
def test_get_recommended_model_with_driver_only_cuda_demotes(self):
"""A CUDA driver without the toolkit cannot run the build: CPU logic."""
import sys
mock_psutil = MagicMock()
mock_psutil.virtual_memory.return_value = MagicMock(total=16 * (1024**3))
with patch.dict("sys.modules", {"psutil": mock_psutil}):
with patch(
"vocalinux.utils.whispercpp_model_info.detect_compute_backend"
) as mock_backend:
mock_backend.return_value = ("cuda", "NVIDIA RTX 4090 (24GB)")
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=False,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, _reason = get_recommended_model()
# 16 GB system RAM under CPU logic recommends base.
self.assertEqual(model, "base")
def test_get_recommended_model_with_cpu_high_ram(self):
"""Test model recommendation with CPU and 16GB+ RAM."""
import sys
@@ -1038,9 +1070,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("cpu", "Intel Core i7")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "base")
@@ -1057,9 +1093,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("cpu", "AMD Ryzen 5")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "tiny")
@@ -1076,9 +1116,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("cpu", "Intel Atom")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "tiny")
@@ -1141,9 +1185,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("cuda", "NVIDIA GeForce RTX 3070 (8192 MiB)")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "medium")
self.assertIn("VRAM", reason)
@@ -1159,9 +1207,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("cuda", "NVIDIA GeForce RTX 3060 (6144 MiB)")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "small")
@@ -1176,9 +1228,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("cuda", "NVIDIA GeForce GTX 1050 (2048 MiB)")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "base")
@@ -1192,9 +1248,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("cuda", "NVIDIA GPU")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "small")
self.assertIn("CUDA GPU detected", reason)
@@ -1210,9 +1270,13 @@ class TestGetRecommendedModel(unittest.TestCase):
) as mock_backend:
mock_backend.return_value = ("vulkan", "Intel Arc A770")
from vocalinux.utils.whispercpp_model_info import get_recommended_model
with patch(
"vocalinux.speech_recognition.model_ladder.cuda_toolkit_ready",
return_value=True,
):
from vocalinux.utils.whispercpp_model_info import get_recommended_model
model, reason = get_recommended_model()
model, reason = get_recommended_model()
self.assertEqual(model, "small")
self.assertIn("8GB", reason)