xavierk
180cd4d512
Model recommendation keyed to the installed build and free memory, fork-sourced Void templates, signed xbps release workflow on Gitea with the pinned repository key, niri injection guard and wtype-first order, tightened naming sweep, web copy fixes, and CI security and duplication fixes.
328 lines
14 KiB
Python
328 lines
14 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 provisional transcribe peaks (1132 MiB for q8_0, 846 MiB
|
|
for q5_0, ticket #19, not yet recorded in the repo) plus the 512 MiB
|
|
protocol headroom. Replacing them with a recorded measurement is a
|
|
deliberate change to this test.
|
|
"""
|
|
|
|
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."""
|
|
|
|
#: Ticket #19 measured transcribe peaks (MiB) per rung.
|
|
Q8_0_MEASURED_MIB: float = 1132.0
|
|
Q5_0_MEASURED_MIB: float = 846.0
|
|
|
|
def test_q8_threshold_is_measured_footprint_plus_headroom(self) -> None:
|
|
assert PREFER_Q8_0_ABOVE_MIB == self.Q8_0_MEASURED_MIB + HEADROOM_MIB
|
|
assert PREFER_Q8_0_ABOVE_MIB == 1644.0
|
|
|
|
def test_q5_threshold_is_measured_footprint_plus_headroom(self) -> None:
|
|
assert PREFER_Q5_0_ABOVE_MIB == self.Q5_0_MEASURED_MIB + HEADROOM_MIB
|
|
assert PREFER_Q5_0_ABOVE_MIB == 1358.0
|
|
|
|
def test_rungs_are_ordered(self) -> None:
|
|
assert PREFER_Q8_0_ABOVE_MIB > PREFER_Q5_0_ABOVE_MIB
|
|
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
|
|
|
|
def test_recorded_boundaries_select_recorded_rungs(self) -> None:
|
|
assert select_ladder_rung(True, 1644.0) == DEFAULT_RUNG
|
|
assert select_ladder_rung(True, 1500.0) == LOW_MEMORY_RUNG
|
|
assert select_ladder_rung(True, 1358.0) == LOW_MEMORY_RUNG
|
|
assert select_ladder_rung(True, 1200.0) == FLOOR_RUNG_ENGLISH
|
|
assert select_ladder_rung(False, 1200.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 TestFreeMemoryForBackend:
|
|
"""The memory reading follows the installed build (spec #16, ticket #20)."""
|
|
|
|
def test_vulkan_build_reads_discrete_vram_when_present(self) -> None:
|
|
with (
|
|
patch.object(model_ladder, "free_vram_mib", return_value=2600.0),
|
|
patch.object(model_ladder, "free_system_ram_mib", return_value=9000.0),
|
|
):
|
|
assert model_ladder.free_memory_for_backend("vulkan") == (2600.0, "VRAM")
|
|
|
|
def test_cuda_build_reads_discrete_vram_when_present(self) -> None:
|
|
with patch.object(model_ladder, "free_vram_mib", return_value=1500.0):
|
|
assert model_ladder.free_memory_for_backend("cuda") == (1500.0, "VRAM")
|
|
|
|
def test_gpu_build_without_an_nvidia_reading_reads_ram(self) -> None:
|
|
"""Integrated graphics: no discrete VRAM reading, so system RAM decides."""
|
|
with (
|
|
patch.object(model_ladder, "free_vram_mib", return_value=None),
|
|
patch.object(model_ladder, "free_system_ram_mib", return_value=6000.0),
|
|
):
|
|
assert model_ladder.free_memory_for_backend("vulkan") == (6000.0, "RAM")
|
|
|
|
def test_cpu_build_never_reads_vram(self) -> None:
|
|
vram = MagicMock(return_value=2600.0)
|
|
with (
|
|
patch.object(model_ladder, "free_vram_mib", vram),
|
|
patch.object(model_ladder, "free_system_ram_mib", return_value=1200.0),
|
|
):
|
|
assert model_ladder.free_memory_for_backend("cpu") == (1200.0, "RAM")
|
|
vram.assert_not_called()
|