soubarnak
df771d8fb1
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
307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""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
|