diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index 3fe74bf..0cafaaa 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -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", ""), diff --git a/src/vocalinux/speech_recognition/model_ladder.py b/src/vocalinux/speech_recognition/model_ladder.py new file mode 100644 index 0000000..68a090a --- /dev/null +++ b/src/vocalinux/speech_recognition/model_ladder.py @@ -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", +] diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index facb095..d341227 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -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 diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index e7e8156..3ca03f0 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -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, ) diff --git a/src/vocalinux/utils/whispercpp_model_info.py b/src/vocalinux/utils/whispercpp_model_info.py index 6b12ab2..d4172e6 100644 --- a/src/vocalinux/utils/whispercpp_model_info.py +++ b/src/vocalinux/utils/whispercpp_model_info.py @@ -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" diff --git a/tests/test_config_manager.py b/tests/test_config_manager.py index dc59ed4..1a1523d 100644 --- a/tests/test_config_manager.py +++ b/tests/test_config_manager.py @@ -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): diff --git a/tests/test_main.py b/tests/test_main.py index a092b9e..a93a5aa 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -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="", diff --git a/tests/test_model_ladder.py b/tests/test_model_ladder.py new file mode 100644 index 0000000..84e7565 --- /dev/null +++ b/tests/test_model_ladder.py @@ -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 diff --git a/tests/test_whispercpp_model_info_ext.py b/tests/test_whispercpp_model_info_ext.py index 6613e06..b409528 100644 --- a/tests/test_whispercpp_model_info_ext.py +++ b/tests/test_whispercpp_model_info_ext.py @@ -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)