xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix(settings): make model downloads explicit and cancel the whole intent (#1006)

Browse Source 
* fix(settings): make model downloads explicit and cancel the whole intent

"Use it" on the recommendation card applied a different model than the one
shown, a cancelled or failed apply left the staged model behind so the next
Test Dictation downloaded it, and cancelling a download tore down and then
reloaded the previous model while the dialog waited.

- Recommendation is now passive text, compared by size tier, with a link to
  docs/MODEL_SELECTION.md. The "Use it" button is gone.
- Every Settings download asks first and shows size and source host.
  Declining, cancelling or failing drops the staged model spec and puts the
  pickers back on the saved config.
- Test Dictation refuses a model that is not on disk instead of applying it.
- reconfigure() fetches a missing model before releasing the loaded one, so
  a cancel restores state without reloading or rehashing the old model.
  Cancellation state is reset only when a download claim is taken or ended.
- Follow-layout mode shows the resolved language in the simple Main language
  control, read-only, and simple edits keep the "layout" setting.

Fixes #979
Fixes #980
Fixes #981
Fixes #982
Fixes #985
Fixes #988

* fix(settings): restore follow-layout mode when a model change is declined

* fix(settings): keep follow-layout and dictation consistent during model changes

* style(tests): format follow-layout regression tests

* fix(recognition): skip resume reinit while a model download is running

* fix(recognition): keep a lazy reload pending when resume is skipped during a download

---------

Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
This commit is contained in:
devin-ai-integration[bot]
2026-10-21 01:04:10 -07:00
committed by GitHub
co-authored by Jatin K Malik
parent f013b7ff88
commit 3e4230ffc2
14 changed files with 1143 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/MODEL_SELECTION.md
+61
View File
@@ -0,0 +1,61 @@
# Choosing a speech model
Vocalinux never downloads a speech model on its own. Changing the model pickers in
Settings → Speech Model only selects a model. If that model is not on disk yet, the
info card shows its download size and a **Download** button, and the transfer starts
only after you confirm it. Cancelling the download, or declining the confirmation,
puts the pickers back on the model you were already using.
This page is a static guide to help you pick. The numbers are approximate starting
points, not measurements of your machine. If dictation feels slow, go one size down;
if it makes too many mistakes and speed is fine, go one size up.
## whisper.cpp (default engine)
| Your hardware | Suggested size | Notes |
|---|---|---|
| Under 8 GB RAM, CPU only | Tiny | Fastest. Accuracy is noticeably lower. |
| 8 to 16 GB RAM, CPU only | Tiny or Base | Base is a good default on a recent CPU. |
| 16 GB RAM or more, CPU only | Base, or Small if you can wait | Small on CPU can take a few seconds per sentence. |
| Vulkan GPU (AMD, Intel, NVIDIA) with under 8 GB RAM | Base | |
| Vulkan GPU with 8 GB RAM or more | Small | Older or low power GPUs may still prefer Base. |
| NVIDIA GPU (CUDA) with under 4 GB VRAM | Base | |
| NVIDIA GPU (CUDA) with 4 to 8 GB VRAM | Small | |
| NVIDIA GPU (CUDA) with 8 GB VRAM or more | Medium, or Large v3 Turbo | Large v3 Turbo is close to Large in accuracy at lower cost. |
Settings → Speech Model may show a line such as "Suggested for this computer: Small".
That line is information only, based on the same rules as this table. It never changes
your model and never starts a download.
## Model sizes and variants
Each size comes in several variants. The quantized ones (`q5_0`, `q5_1`, `q8_0`) are
smaller files that use less memory, at a small cost in accuracy; `q8_0` is the
closest to the full model. The `.en` variants are English only and are a little
more accurate for English.
| Size | Full download | Q8_0 | Q5 | Parameters |
|---|---|---|---|---|
| Tiny | 74 MB | 32 MB | 15 MB (q5_1) | 39M |
| Base | 141 MB | 82 MB | 60 MB (q5_1) | 74M |
| Small | 465 MB | 190 MB | 163 MB (q5_1) | 244M |
| Medium | 1.4 GB | 823 MB | 568 MB (q5_0) | 769M |
| Large v3 Turbo | 1.6 GB | 874 MB | 574 MB (q5_0) | 809M |
| Large (v3) | 2.9 GB | | 1.1 GB (q5_0) | 1550M |
Sizes are the catalog values Vocalinux shows before a download; the file on disk can
differ slightly.
If you already have a quantized model of the size you want (for example Small Q8_0),
there is no reason to download the full model of the same size.
## Other engines
- **Vosk**: the small models (about 40 MB) run on any CPU and are the lightest
option.
- **Faster Whisper** and **OpenAI Whisper**: the same size guidance as the table above.
- **Parakeet**: one model of about 640 MB.
- **Remote API**: transcription runs on your server, nothing is downloaded locally.
Models are stored under `~/.local/share/vocalinux/models/` by default. Deleting a file there
removes the model; Vocalinux will show it as not downloaded again.
src/vocalinux/speech_recognition/recognition_manager.py
+102 -32
View File
@@ -68,6 +68,7 @@ from ..utils.pywhispercpp_loader import (
from ..utils.system_language import LANGUAGE_FOLLOWS_LAYOUT, language_for_active_layout
from ..utils.vosk_model_info import SUPPORTED_LANGUAGES, VOSK_MODEL_INFO
from ..utils.whisper_model_info import (
WHISPER_MODEL_SIZES,
migrate_legacy_checkpoint_names,
whisper_model_file,
whisper_model_url,
@@ -480,6 +481,7 @@ class SpeechRecognitionManager:
# Set by the UI layer to offer the download when a model is missing
self._model_missing_handler: Optional[Callable[[str], bool]] = None
self._download_cancelled = False
self._prefetching_model = False
# Held by whichever UI is downloading a model. The tray and the settings
# dialog both download in the background, and there is only one progress
# callback and one engine configuration between them.
@@ -654,9 +656,54 @@ class SpeechRecognitionManager:
else:
raise ValueError(f"Unsupported speech recognition engine: {self.engine}")
def _download_selected_model_if_missing(self) -> None:
"""Fetch a missing target model without disturbing the loaded engine."""
if self.engine == "whisper_cpp":
valid_models = [
name for name in WHISPERCPP_MODEL_INFO if name not in NON_DICTATION_MODELS
]
if self.model_size in valid_models and not os.path.exists(
get_model_path(self.model_size)
):
self._download_whispercpp_model()
elif self.engine == "whisper":
if self.model_size in WHISPER_MODEL_SIZES:
cache_dir, model_file = self._whisper_model_cache_paths()
if not os.path.exists(model_file):
self._download_whisper_model(cache_dir)
elif self.engine == "faster_whisper":
if not faster_whisper.is_model_downloaded(self.model_size):
self._download_faster_whisper_model()
elif self.engine == "parakeet":
if not parakeet.is_model_downloaded(self.model_size):
self._download_parakeet_model()
elif self.engine == "vosk":
self._configure_vosk_model_map()
if not os.path.exists(self._get_vosk_model_path()):
self._download_vosk_model()
def _whisper_model_cache_paths(self) -> tuple[str, str]:
"""Prepare the same Whisper cache and checkpoint path used by init."""
whisper_cache_dir = os.path.join(MODELS_DIR, "whisper")
os.makedirs(whisper_cache_dir, exist_ok=True)
migrate_legacy_checkpoint_names(whisper_cache_dir)
model_file = os.path.join(whisper_cache_dir, whisper_model_file(self.model_size))
return whisper_cache_dir, model_file
def _configure_vosk_model_map(self) -> None:
"""Build the VOSK size-to-language model mapping used by init and download."""
vosk_language = "en-us" if self.language == "auto" else self.language
model_info = cast(dict[str, dict[str, Any]], VOSK_MODEL_INFO)
self.vosk_model_map = {
"small": model_info["small"]["languages"].get(vosk_language),
"medium": model_info["medium"]["languages"].get(vosk_language),
"large": model_info["large"]["languages"].get(vosk_language),
}
#: Class-level default so a manager built without ``__init__`` still answers
#: "not following the layout" instead of raising on the dictation path.
language_preference: str = "auto"
_prefetching_model: bool = False
#: Same for the one-shot override state (#805): __new__-built stubs in tests
#: must not raise on the dictation path either.
@@ -930,14 +977,7 @@ class SpeechRecognitionManager:
def _init_vosk(self):
"""Initialize the VOSK speech recognition engine."""
# VOSK doesn't support auto-detect, so fall back to en-us for "auto"
vosk_language = "en-us" if self.language == "auto" else self.language
self.vosk_model_map = {
"small": VOSK_MODEL_INFO["small"]["languages"].get(vosk_language),
"medium": VOSK_MODEL_INFO["medium"]["languages"].get(vosk_language),
"large": VOSK_MODEL_INFO["large"]["languages"].get(vosk_language),
}
self._configure_vosk_model_map()
try:
from vosk import KaldiRecognizer, Model
@@ -1053,23 +1093,16 @@ class SpeechRecognitionManager:
import torch
# Validate model size for Whisper
valid_whisper_models = ["tiny", "base", "small", "medium", "large"]
if self.model_size not in valid_whisper_models:
if self.model_size not in WHISPER_MODEL_SIZES:
logger.warning(
f"Model size '{self.model_size}' not valid for Whisper. "
f"Valid options: {valid_whisper_models}. Using 'base' instead."
f"Valid options: {WHISPER_MODEL_SIZES}. Using 'base' instead."
)
self.model_size = "base"
# Only this directory counts: load_model() gets it as download_root
# and looks nowhere else, so a copy in ~/.cache/whisper saves nothing.
whisper_cache_dir = os.path.join(MODELS_DIR, "whisper")
os.makedirs(whisper_cache_dir, exist_ok=True)
# Before asking whether the model is here: an earlier release saved it
# under the catalog name, which for "large" is not what load_model
# looks for. Without this it refetches 2.9GB and orphans the old file.
migrate_legacy_checkpoint_names(whisper_cache_dir)
model_file = os.path.join(whisper_cache_dir, whisper_model_file(self.model_size))
whisper_cache_dir, model_file = self._whisper_model_cache_paths()
model_exists = os.path.exists(model_file)
if not model_exists and self._defer_download:
@@ -1187,8 +1220,6 @@ class SpeechRecognitionManager:
"""Download the faster-whisper model files with progress tracking."""
import requests
self._download_cancelled = False
model_dir = faster_whisper.get_model_path(self.model_size)
os.makedirs(model_dir, exist_ok=True)
logger.info(f"Downloading faster-whisper '{self.model_size}' model to {model_dir}")
@@ -2034,8 +2065,6 @@ class SpeechRecognitionManager:
"""Download the Parakeet model files with progress tracking."""
import requests
self._download_cancelled = False
model_dir = parakeet.get_model_path(self.model_size)
os.makedirs(model_dir, exist_ok=True)
logger.info(f"Downloading Parakeet '{self.model_size}' model to {model_dir}")
@@ -2682,7 +2711,6 @@ class SpeechRecognitionManager:
"""
import requests
self._download_cancelled = False
url = whisper_model_url(self.model_size)
model_file = os.path.join(cache_dir, whisper_model_file(self.model_size))
temp_file = model_file + ".tmp"
@@ -2708,8 +2736,6 @@ class SpeechRecognitionManager:
"""Download a whisper.cpp model with progress tracking."""
import requests
self._download_cancelled = False
model_name = model_name or self.model_size
model_info = WHISPERCPP_MODEL_INFO.get(model_name)
if not model_info:
@@ -2849,7 +2875,10 @@ class SpeechRecognitionManager:
Returns:
True if the claim was taken, False if a download is already running
"""
return self._download_claim.acquire(blocking=False)
claimed = self._download_claim.acquire(blocking=False)
if claimed:
self._download_cancelled = False
return claimed
def end_download(self):
"""Release the claim taken by try_begin_download()."""
@@ -2858,6 +2887,8 @@ class SpeechRecognitionManager:
except RuntimeError:
# Never claimed; nothing to release
pass
else:
self._download_cancelled = False
@property
def download_in_progress(self) -> bool:
@@ -2875,8 +2906,6 @@ class SpeechRecognitionManager:
import requests
self._download_cancelled = False
model_name = self.vosk_model_map.get(self.model_size)
if not model_name:
raise ValueError(f"Unknown model size: {self.model_size}")
@@ -3232,6 +3261,16 @@ class SpeechRecognitionManager:
logger.warning(f"Cannot start recognition in current state: {self.state}")
return False
if getattr(self, "_prefetching_model", False) is True:
logger.warning("Cannot start recognition while a speech model is downloading")
play_error_sound()
_show_notification(
"Speech model downloading",
"Dictation will be available when the new model has finished downloading.",
"dialog-information",
)
return False
if self._auto_paused:
logger.warning(
"Cannot start recognition: auto-paused while a configured app/game is running"
@@ -4284,10 +4323,32 @@ class SpeechRecognitionManager:
# When reconfiguring from UI, allow downloads
old_defer = self._defer_download
self._defer_download = not force_download
if force_download:
try:
self._prefetching_model = True
try:
self._download_selected_model_if_missing()
finally:
self._prefetching_model = False
except Exception:
self._restore_reconfigure_state(previous)
if language_changed:
self.command_processor.set_language(self.language)
self._restore_language_after_oneshot()
self._defer_download = old_defer
raise
# Lock model access during reinitialization to prevent race condition
# with transcription threads that may be using the model/recognizer
with self._model_lock:
# with transcription threads that may be using the model/recognizer.
# A download worker can be waiting here while Cancel is pressed, so
# poll instead of blocking indefinitely on the active transcription.
while not self._model_lock.acquire(timeout=0.1):
if self._download_claim.locked() and self._download_cancelled:
self._restore_reconfigure_state(previous)
self.command_processor.set_language(self.language)
self._defer_download = old_defer
raise RuntimeError("Download cancelled")
try:
# Release old resources explicitly if necessary (Python's GC might handle it)
self.model = None
self.recognizer = None
@@ -4330,8 +4391,9 @@ class SpeechRecognitionManager:
)
self._update_state(RecognitionState.ERROR)
raise
finally:
self._defer_download = old_defer
finally:
self._defer_download = old_defer
self._model_lock.release()
else:
# If only VOSK params changed, just log it
logger.info("Applied VAD/silence timeout changes.")
@@ -4521,6 +4583,14 @@ class SpeechRecognitionManager:
Also used after auto-pause / idle keep-alive to reload the model.
"""
self._cancel_reload_recording()
if getattr(self, "_prefetching_model", False) is True:
# reconfigure() is fetching the next model and will load it itself;
# initializing here would start a second transfer of the same files.
# Leave a lazy reload pending in case that download is cancelled.
logger.info("Skipping engine reinit: a speech model download is in progress")
self._auto_paused = False
self._idle_unloaded = not self.model_ready
return
logger.info("Reinitializing speech engine after system resume")
if self.state != RecognitionState.IDLE:
src/vocalinux/ui/settings_dialog.py
+161 -116
View File
@@ -27,6 +27,7 @@ import time
from contextlib import contextmanager
from functools import partial
from typing import TYPE_CHECKING, Any, Callable, Iterator, Literal, NamedTuple, Optional, cast
from urllib.parse import urlparse
import gi
@@ -43,6 +44,7 @@ from ..common_types import ( # noqa: E402
from ..custom_dictionary import normalize_corrections # noqa: E402
from ..gateway_embed import GatewayStatus, get_gateway_embed_manager # noqa: E402
from ..speech_recognition.silero_vad import is_silero_available # noqa: E402
from ..utils import faster_whisper_model_info as faster_whisper
from ..utils.faster_whisper_model_info import delete_model as delete_faster_whisper_model
from ..utils.faster_whisper_model_info import ( # noqa: E402
get_recommended_model as get_recommended_faster_whisper_model,
@@ -92,6 +94,7 @@ from ..utils.vosk_model_info import ( # noqa: E402
from ..utils.whisper_model_info import ( # noqa: E402
migrate_legacy_checkpoint_names,
whisper_model_file,
whisper_model_url,
)
from ..utils.whispercpp_model_info import delete_model as delete_whispercpp_model
from ..utils.whispercpp_model_info import (
@@ -285,6 +288,9 @@ def _engine_from_display(display_name: str) -> str:
return display_name.lower()
MODEL_SELECTION_DOC_URL = "https://github.com/VocaHQ/vocalinux/blob/main/docs/MODEL_SELECTION.md"
def _model_display_name(model_name: str) -> str:
"""Get a user-friendly model display name."""
if model_name == "large":
@@ -1209,6 +1215,11 @@ class SearchablePicker(Gtk.Box):
self._active_id = None
self._label.set_text("")
def clear_active(self, text: str = "") -> None:
"""Clear the selected id and show an optional readout."""
self._active_id = None
self._label.set_text(text)
def get_model(self) -> list:
"""The ComboBoxText-shaped store: [display text, id] per row."""
return self.base_model
@@ -2183,6 +2194,29 @@ def _undownloaded_model_info(language: str, engine: str, model_name: str) -> Opt
return None
def _model_download_source_url(
engine: str, model_name: str, language: str, info: Optional[dict] = None
) -> str:
"""Return the URL used to download a selected model."""
catalog_url = (info or {}).get("url")
if catalog_url:
return str(catalog_url)
if engine == "whisper":
return whisper_model_url(model_name)
if engine == "whisper_cpp":
return str(WHISPERCPP_MODEL_INFO.get(model_name, {}).get("url", ""))
if engine == "vosk":
model_dir = vosk_model_dirname(model_name, language)
return f"https://alphacephei.com/vosk/models/{model_dir}.zip" if model_dir else ""
if engine == "parakeet":
files = parakeet.model_files(model_name)
return parakeet.get_model_file_url(model_name, files[0]) if files else ""
if engine == "faster_whisper":
files = faster_whisper.model_files(model_name)
return faster_whisper.get_model_file_url(model_name, files[0]) if files else ""
return ""
def _make_download_dialog(
dialog: "SettingsDialog", model_name: str, size_mb: int, engine: str
) -> "ModelDownloadDialog":
@@ -4641,8 +4675,12 @@ class SettingsDialog(AboutSectionMixin):
self.model_download_button.connect("clicked", self._on_download_model_clicked)
self.model_info_card.pack_start(self.model_download_button, False, False, 0)
# The recommendation used to be a plain label, which left the panel stating
# the right answer while the pickers kept the wrong one (#778).
model_selection_link = Gtk.LinkButton.new_with_label(
MODEL_SELECTION_DOC_URL, "How to choose a model"
)
model_selection_link.set_halign(Gtk.Align.START)
self.simple_page.pack_start(model_selection_link, False, False, 0)
self.model_recommendation_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
self.model_recommendation_box.set_no_show_all(True)
@@ -4650,12 +4688,6 @@ class SettingsDialog(AboutSectionMixin):
self.model_recommendation.get_style_context().add_class("model-info-subtitle")
self.model_recommendation_box.pack_start(self.model_recommendation, True, True, 0)
self.model_recommendation_button = Gtk.Button(label="Use it")
self.model_recommendation_button.set_valign(Gtk.Align.CENTER)
self.model_recommendation_button.set_no_show_all(True)
self.model_recommendation_button.connect("clicked", self._on_apply_recommendation)
self.model_recommendation_box.pack_end(self.model_recommendation_button, False, False, 0)
self.model_info_card.pack_start(self.model_recommendation_box, False, False, 0)
# Language warning (e.g. auto-detect, English-only models) lives in
@@ -7461,75 +7493,11 @@ class SettingsDialog(AboutSectionMixin):
cast(str, language_id),
)
def _downloaded_alternative_for(self, recommended: str) -> Optional[str]:
"""Return a model already on disk that can stand in for ``recommended``.
Every entry in these pickers is a download decision priced in gigabytes, so
suggesting a fresh download while something equivalent is already on disk
wastes the user's bandwidth (#778). Only candidates that are not smaller than
the recommendation qualify, so this never quietly downgrades accuracy, and
English-only weights are only offered when English is actually selected.
"""
if recommended in WHISPERCPP_MODEL_INFO and is_whispercpp_model_downloaded(recommended):
return None
language_id = _language_for_whispercpp_variant(self)
wants_english = _language_is_english(cast(str, language_id))
recommended_info = WHISPERCPP_MODEL_INFO.get(recommended, {})
recommended_mb = cast(int, recommended_info.get("size_mb", 0))
# An English-only recommendation must not be answered with multilingual
# weights: they weigh the same and recognise English worse, which is the
# accuracy loss #776 is about.
needs_english_only = is_english_only_whispercpp_model(recommended)
best = None
best_mb = None
for model_name in list_downloaded_whispercpp_models():
if model_name in NON_DICTATION_MODELS:
continue
info = cast(Optional[dict[str, Any]], WHISPERCPP_MODEL_INFO.get(model_name))
if not info:
continue
if is_english_only_whispercpp_model(model_name) and not wants_english:
continue
if needs_english_only and not is_english_only_whispercpp_model(model_name):
continue
size_mb = cast(int, info["size_mb"])
if size_mb < recommended_mb:
continue
if best_mb is None or size_mb < best_mb:
best, best_mb = model_name, size_mb
return best
def _get_default_whispercpp_variant_for_size(self, model_size: str) -> Optional[str]:
"""Return the default specialization for a user-selected size."""
language_id = _language_for_whispercpp_variant(self)
return _default_whispercpp_variant_for_size(model_size, cast(str, language_id))
def _on_apply_recommendation(self, _button: Any) -> None:
"""Set both pickers to the model the card is offering.
Size and specialization must land under ``_populating_models`` so
``_on_model_changed`` does not auto-apply the default size variant
(e.g. full ``large``) before the intended target (e.g. turbo q5_0).
"""
target = getattr(self, "_recommended_target_model", None)
if not target or target not in WHISPERCPP_MODEL_INFO:
return
model_size = get_whispercpp_model_size(target)
with _suppressed(self, "_populating_models"):
self.model_combo.set_active_id(model_size)
self._populate_whispercpp_variant_options(model_size, target)
self.model_variant_combo.set_active_id(target)
self._sync_language_options_for_selected_model()
self._update_model_info()
self._refresh_unused_downloads()
self._auto_apply_settings()
def _is_selected_whispercpp_model_english_only(self) -> bool:
"""Return whether the selected model is a whisper.cpp English-only variant."""
if self._get_selected_engine() != "whisper_cpp":
@@ -8321,9 +8289,7 @@ class SettingsDialog(AboutSectionMixin):
def _on_language_changed(self, widget: Any) -> None:
"""Handle language selection change.
Repaint the recommendation card after the size list rebuilds so
``_recommended_target_model`` / ★ / "Use it" track EN→non-EN flips
instead of applying a stale English-only target.
Repaint the model information after the size list rebuilds.
"""
if self._processing_language_change:
return
@@ -8393,11 +8359,16 @@ class SettingsDialog(AboutSectionMixin):
language = self.language_combo.get_active_id() or self.language or "auto"
if _is_following_layout(self):
# Neither simple question can say "follow the layout", and writing an
# answer here would let the next simple edit silently replace the mode
# with whatever those two controls happen to read. Refresh only the
# part that does not depend on the language and leave them alone.
# Neither simple question can say "follow the layout", so show the
# resolved language as a disabled readout without changing the mode.
with _suppressed(self, "_simple_syncing"):
if not self.simple_language_combo.set_active_id(language):
self.simple_language_combo.clear_active("Detected each time")
self.simple_language_row.set_sensitive(False)
self.simple_language_row.set_tooltip_text(
"Follows your keyboard layout. Turn off Follow keyboard layout "
"in Advanced to pick a language."
)
recommended, _ = self._get_recommended_whispercpp_model_for_language()
current = self._get_selected_whispercpp_model()
self.simple_priority_combo.set_active_id(
@@ -8409,6 +8380,8 @@ class SettingsDialog(AboutSectionMixin):
self._update_simple_visibility()
return
self.simple_language_row.set_sensitive(True)
self.simple_language_row.set_tooltip_text(LANGUAGE_TOOLTIP)
is_auto = language == "auto"
stored_second = self.config_manager.get("speech_recognition", "simple_second_language", "")
@@ -8487,16 +8460,22 @@ class SettingsDialog(AboutSectionMixin):
self.engine_combo.set_active_id("whisper_cpp")
engine = "whisper_cpp"
self._set_combo_active_id_or_first(self.language_combo, language)
self.language = language
if not _is_following_layout(self):
self._set_combo_active_id_or_first(self.language_combo, language)
self.language = language
if engine != "whisper_cpp":
return
recommended, _ = self._get_recommended_whispercpp_model_for_language()
size = size_for_priority(get_whispercpp_model_size(recommended), priority)
variant = _default_whispercpp_variant_for_size(size, language) or size
variant = self._on_disk_stand_in(variant, size, language)
variant_language = (
cast(str, _language_for_whispercpp_variant(self))
if _is_following_layout(self)
else language
)
variant = _default_whispercpp_variant_for_size(size, variant_language) or size
variant = self._on_disk_stand_in(variant, size, variant_language)
self.model_combo.set_active_id(size)
self._populate_whispercpp_variant_options(size, variant)
@@ -8729,6 +8708,14 @@ class SettingsDialog(AboutSectionMixin):
model_display_name = _model_display_name(model_name)
recommended_display_name = _model_display_name(recommended)
recommended_info = {
"whisper": WHISPER_MODEL_INFO,
"whisper_cpp": WHISPERCPP_MODEL_INFO,
"vosk": VOSK_MODEL_INFO,
"parakeet": parakeet.PARAKEET_MODEL_INFO,
"faster_whisper": FASTER_WHISPER_MODEL_INFO,
}[engine].get(recommended, {})
recommended_size_mb = cast(int, recommended_info.get("size_mb", 0))
self.model_info_title.set_markup(f"<b>{model_display_name}</b>: {info['desc']}")
@@ -8741,29 +8728,19 @@ class SettingsDialog(AboutSectionMixin):
# names a model that is not on disk (#894).
self.model_download_button.set_visible(not is_downloaded)
target = recommended
message = None
if engine == "whisper_cpp":
already_have = self._downloaded_alternative_for(recommended)
if already_have and already_have != model_name:
target = already_have
message = (
f"You already have {_model_display_name(already_have)} on disk. "
"Using it needs no download."
)
if target == model_name:
self._recommended_target_model = None
recommendation_differs = (
get_whispercpp_model_size(recommended) != get_whispercpp_model_size(model_name)
if engine == "whisper_cpp"
else recommended != model_name
)
if not recommendation_differs:
self.model_recommendation_box.hide()
self.model_recommendation_button.hide()
else:
self._recommended_target_model = target
if message is None:
message = f"Recommended: {_model_display_name(target)} ({reason})"
self.model_recommendation.set_text(message)
self.model_recommendation.set_text(
f"Suggested for this computer: {recommended_display_name} "
f"({_format_size(recommended_size_mb)}, {reason})"
)
self.model_recommendation.show()
can_apply = engine == "whisper_cpp" and target in WHISPERCPP_MODEL_INFO
self.model_recommendation_button.set_visible(can_apply)
self.model_recommendation_box.show()
self._update_model_picker_tooltips()
@@ -8858,12 +8835,17 @@ class SettingsDialog(AboutSectionMixin):
return
if missing_model is not None:
if not self._confirm_model_download(
engine, model_name, int(missing_model["size_mb"])
):
self._drop_staged_model_spec_and_resync()
return
if not self.speech_engine.try_begin_download():
# The tray is already downloading a model; a second download
# would fight it over the engine's progress callback and
# configuration. Put the picker back on the saved model.
self._show_download_busy_dialog()
self._resync_model_ui_from_config()
self._drop_staged_model_spec_and_resync()
return
logger.info(f"Model {model_name} needs download, showing progress dialog")
download_dialog = _make_download_dialog(
@@ -8880,7 +8862,7 @@ class SettingsDialog(AboutSectionMixin):
def check_cancelled():
if download_dialog.cancelled:
self.speech_engine.cancel_download()
return not download_dialog.cancelled
return True
cancel_check_id = GLib.timeout_add(100, check_cancelled)
@@ -8896,7 +8878,7 @@ class SettingsDialog(AboutSectionMixin):
# False return must not be read as success: it
# means nothing was saved and the pickers still
# show settings the engine never took.
GLib.idle_add(self._idle_resync_model_ui_from_config)
GLib.idle_add(self._idle_drop_staged_model_spec_and_resync)
GLib.idle_add(
download_dialog.set_complete,
False,
@@ -8912,7 +8894,7 @@ class SettingsDialog(AboutSectionMixin):
# The settings were never saved, so the previously working
# model is still the configured one; put the pickers back on
# it so the UI matches the config and a retry is possible.
GLib.idle_add(self._idle_resync_model_ui_from_config)
GLib.idle_add(self._idle_drop_staged_model_spec_and_resync)
if "cancelled" in error_msg.lower():
GLib.idle_add(
download_dialog.set_complete,
@@ -8934,7 +8916,7 @@ class SettingsDialog(AboutSectionMixin):
download_dialog.destroy()
# Whatever ended the modal — cancel, failure, or a path not
# covered above — the pickers must not outlive the config.
self._resync_engine_ui_if_unapplied()
self._drop_staged_model_spec_and_resync()
# Repaint the info card too: after a success it still showed the
# pre-download render (amber size line, Download action) until
# some other change repainted it (#894 follow-up).
@@ -8991,6 +8973,16 @@ class SettingsDialog(AboutSectionMixin):
self._resync_model_ui_from_config()
return False
def _drop_staged_model_spec_and_resync(self) -> None:
"""Forget a model intent that was not applied and restore saved pickers."""
self._staged_model_spec = None
self._resync_model_ui_from_config()
def _idle_drop_staged_model_spec_and_resync(self) -> bool:
"""Drop an unapplied model intent before resyncing from a worker thread."""
self._staged_model_spec = None
return self._idle_resync_model_ui_from_config()
def _resync_model_ui_from_config(self):
"""Put the pickers back on the settings that are actually saved.
@@ -9018,9 +9010,21 @@ class SettingsDialog(AboutSectionMixin):
self._applying_settings = was_applying
self._populate_model_options()
saved_language = sr_config.get("language")
if saved_language:
self._sync_language_options_for_selected_model(saved_language)
follow = saved_language == LANGUAGE_FOLLOWS_LAYOUT
self._follow_layout_saved = follow
resolved_language = (
language_for_active_layout(SUPPORTED_LANGUAGES) or "auto"
if follow
else saved_language
)
if resolved_language:
self._sync_language_options_for_selected_model(resolved_language)
with _suppressed(self, "_processing_language_change"):
self.follow_layout_switch.set_active(follow and self._follow_layout_supported())
self._sync_follow_layout_controls()
self._update_language_warning()
self._update_model_info()
self._sync_simple_from_advanced()
except Exception as e: # pragma: no cover - UI resync must never mask the original error
logger.debug(f"Could not resync model pickers: {e}")
@@ -9154,6 +9158,15 @@ class SettingsDialog(AboutSectionMixin):
selected_settings = self.get_selected_settings()
selected_engine = selected_settings.get("engine")
selected_model = selected_settings.get("model_size")
if selected_engine and selected_model:
missing_model = _undownloaded_model_info(self.language, selected_engine, selected_model)
if missing_model is not None:
self.test_output_revealer.set_reveal_child(True)
self.test_buffer.set_text(
f"{_model_display_name(selected_model)} is not downloaded yet. "
"Press Download on the Speech Model page first, or pick a downloaded model."
)
return
# Compare to the live engine too: UI can match the file while the
# in-memory manager is still on another engine/size (sidecar, pending apply).
live_engine = getattr(self.speech_engine, "engine", None)
@@ -9486,14 +9499,18 @@ For now, the engine has been reverted to VOSK."""
missing_model = _undownloaded_model_info(self.language, engine, model_name)
if missing_model is not None:
if not self._confirm_model_download(engine, model_name, int(missing_model["size_mb"])):
self._drop_staged_model_spec_and_resync()
return False
if not self.speech_engine.try_begin_download():
# Same guard as in _auto_apply_settings: one download at a time.
self._show_download_busy_dialog()
self._resync_model_ui_from_config()
self._drop_staged_model_spec_and_resync()
return False
download_dialog = _make_download_dialog(
self, model_name, missing_model["size_mb"], engine
)
download_outcome = {"applied": False}
def progress_callback(fraction, speed, status):
GLib.idle_add(download_dialog.update_progress, fraction, speed, status)
@@ -9505,7 +9522,7 @@ For now, the engine has been reverted to VOSK."""
def check_cancelled():
if download_dialog.cancelled:
self.speech_engine.cancel_download()
return not download_dialog.cancelled
return True
cancel_check_id = GLib.timeout_add(100, check_cancelled)
@@ -9514,9 +9531,10 @@ For now, the engine has been reverted to VOSK."""
settings, raise_errors=True, force_reinit=True
)
if applied:
download_outcome["applied"] = True
GLib.idle_add(download_dialog.set_complete, True, "")
else:
GLib.idle_add(self._idle_resync_model_ui_from_config)
GLib.idle_add(self._idle_drop_staged_model_spec_and_resync)
GLib.idle_add(
download_dialog.set_complete,
False,
@@ -9531,7 +9549,7 @@ For now, the engine has been reverted to VOSK."""
error_msg = str(e)
# Nothing was saved, so the config still names the previous
# engine and model; put the pickers back on them.
GLib.idle_add(self._idle_resync_model_ui_from_config)
GLib.idle_add(self._idle_drop_staged_model_spec_and_resync)
if "cancelled" in error_msg.lower():
GLib.idle_add(download_dialog.set_complete, False, "Download cancelled")
elif engine == "whisper" and "no module named" in error_msg.lower():
@@ -9545,14 +9563,41 @@ For now, the engine has been reverted to VOSK."""
download_dialog.destroy()
self._populate_model_options()
self._resync_engine_ui_if_unapplied()
if not download_outcome["applied"]:
self._drop_staged_model_spec_and_resync()
# Same repaint as the auto-apply path: a finished download must not
# leave the card offering a Download for the model just fetched.
self._update_model_info()
return True
return bool(download_outcome["applied"])
return self._apply_settings_internal(settings)
def _confirm_model_download(self, engine: str, model_name: str, size_mb: int) -> bool:
"""Ask before downloading a model selected in Settings."""
info = _undownloaded_model_info(self.language, engine, model_name) or {}
url = _model_download_source_url(engine, model_name, self.language, info)
host = urlparse(url).netloc
secondary = f"About {_format_size(size_mb)}"
if host:
secondary += f" from {host}"
secondary += ". Nothing is downloaded until you confirm."
dialog = Gtk.MessageDialog(
transient_for=self,
flags=0,
message_type=Gtk.MessageType.QUESTION,
buttons=Gtk.ButtonsType.NONE,
text=f"Download {_model_display_name(model_name)}?",
)
dialog.format_secondary_text(secondary)
dialog.add_button("Cancel", Gtk.ResponseType.CANCEL)
dialog.add_button("Download", Gtk.ResponseType.OK)
dialog.set_default_response(Gtk.ResponseType.CANCEL)
try:
return cast(bool, dialog.run() == Gtk.ResponseType.OK)
finally:
dialog.destroy()
def _show_download_busy_dialog(self):
"""Tell the user a model download is already running elsewhere."""
dialog = Gtk.MessageDialog(
tests/test_language_list_filtering.py
+44
View File
@@ -181,6 +181,50 @@ def test_changed_fires_only_when_the_selection_actually_changes(settings_dialog)
assert picker._active_id == "ja"
def test_clear_active_shows_a_readout_without_emitting_changed(settings_dialog: Any) -> None:
picker = _picker_stub(settings_dialog, rows=LANGS, active="pl")
settings_dialog.SearchablePicker.clear_active(picker, "Detected each time")
assert picker._active_id is None
picker._label.set_text.assert_called_once_with("Detected each time")
picker.emit.assert_not_called()
def test_follow_layout_auto_uses_a_readout_without_applying_settings(
settings_dialog: Any,
) -> None:
picker = _picker_stub(settings_dialog, rows=LANGS, active="en-us")
label_text = {"value": "English (US)"}
picker._label.set_text.side_effect = lambda text: label_text.__setitem__("value", text)
picker.set_active_id.side_effect = (
lambda item_id: settings_dialog.SearchablePicker.set_active_id(picker, item_id)
)
picker.get_active_id.side_effect = lambda: settings_dialog.SearchablePicker.get_active_id(
picker
)
picker.clear_active.side_effect = lambda text="": settings_dialog.SearchablePicker.clear_active(
picker, text
)
dialog = Mock()
dialog._follow_layout_active = True
dialog.language = "auto"
dialog.language_combo.get_active_id.return_value = "auto"
dialog.simple_language_combo = picker
dialog._get_recommended_whispercpp_model_for_language.return_value = ("small", "reason")
dialog._get_selected_whispercpp_model.return_value = "small"
dialog._update_simple_visibility = Mock()
dialog._auto_apply_settings = Mock()
settings_dialog.SettingsDialog._sync_simple_from_advanced(dialog)
assert picker.get_active_id() is None
assert label_text["value"] == "Detected each time"
dialog.simple_language_row.set_sensitive.assert_called_once_with(False)
dialog._auto_apply_settings.assert_not_called()
picker.emit.assert_not_called()
def test_opening_starts_from_an_empty_search(settings_dialog):
"""Leftover text from last time would hide most of the list on open."""
picker = _picker_stub(settings_dialog, rows=LANGS)
tests/test_language_shortcuts.py
+1 -1
View File
@@ -665,7 +665,7 @@ def test_failed_reconfigure_settles_the_pending_oneshot_restore() -> None:
manager._init_selected_engine = MagicMock(side_effect=[RuntimeError("boom"), None])
with pytest.raises(RuntimeError):
manager.reconfigure(language="fr", model_size="tiny")
manager.reconfigure(language="fr", model_size="tiny", force_download=False)
assert manager.language == "en-us"
assert manager._oneshot_language_restore is None
tests/test_recognition_manager_advanced.py
+4 -3
View File
@@ -592,21 +592,21 @@ class TestReconfigure(unittest.TestCase):
mgr = _make_manager()
mgr.state = RecognitionState.IDLE
with patch.object(mgr, "_init_whispercpp"):
mgr.reconfigure(model_size="medium")
mgr.reconfigure(model_size="medium", force_download=False)
self.assertEqual(mgr.model_size, "medium")
def test_reconfigure_engine(self):
mgr = _make_manager()
mgr.state = RecognitionState.IDLE
with patch.object(mgr, "_init_vosk"):
mgr.reconfigure(engine="vosk")
mgr.reconfigure(engine="vosk", force_download=False)
self.assertEqual(mgr.engine, "vosk")
def test_reconfigure_language_candidates(self) -> None:
mgr = _make_manager()
mgr.state = RecognitionState.IDLE
with patch.object(mgr, "_init_whispercpp"):
mgr.reconfigure(whispercpp_language_candidates="en,es")
mgr.reconfigure(whispercpp_language_candidates="en,es", force_download=False)
self.assertEqual(mgr.whispercpp_language_candidates, ["en", "es"])
@@ -1069,6 +1069,7 @@ class TestWhispercppModelKwargs(unittest.TestCase):
whispercpp_temperature=0.5,
whispercpp_no_timestamps=False,
whispercpp_initial_prompt="hello",
force_download=False,
)
self.assertEqual(mgr.whispercpp_temperature, 0.5)
self.assertFalse(mgr.whispercpp_no_timestamps)
tests/test_recognition_manager_device_config.py
+1 -1
View File
@@ -2417,7 +2417,7 @@ class TestReconfiguration(unittest.TestCase):
with patch.object(manager, "_init_whisper"):
with patch.object(manager, "stop_recognition"):
manager.reconfigure(engine="whisper")
manager.reconfigure(engine="whisper", force_download=False)
assert manager.engine == "whisper"
tests/test_recognition_manager_download_buffer.py
+296
View File
@@ -16,6 +16,7 @@ import tempfile
import threading
import time
import unittest
from contextlib import ExitStack
from unittest.mock import MagicMock, Mock, PropertyMock, mock_open, patch
# Mock modules BEFORE importing anything from vocalinux
@@ -33,6 +34,7 @@ sys.modules["zipfile"] = MagicMock()
# Import after mocking
from conftest import mock_audio_feedback
import vocalinux.speech_recognition.recognition_manager as recognition_manager_module
from vocalinux.common_types import RecognitionState
from vocalinux.speech_recognition.command_processor import CommandProcessor
from vocalinux.speech_recognition.recognition_manager import (
@@ -97,6 +99,266 @@ class TestDownloadFunctions(unittest.TestCase):
manager._download_cancelled = False
assert manager._download_cancelled is False
def test_download_claim_resets_cancellation_for_the_next_attempt(self) -> None:
"""A stale cancel from one attempt must not cancel the next one."""
manager = self._create_manager(engine="whisper")
manager._download_cancelled = True
assert manager.try_begin_download() is True
assert manager._download_cancelled is False
manager.cancel_download()
manager.end_download()
assert manager._download_cancelled is False
def _assert_missing_model_dispatch(self, engine: str, model_size: str, downloader: str) -> None:
manager = self._create_manager()
manager.engine = engine
manager.model_size = model_size
with ExitStack() as stack:
availability = None
if engine == "whisper_cpp":
stack.enter_context(
patch.object(
recognition_manager_module,
"get_model_path",
return_value="/models/target",
)
)
availability = stack.enter_context(
patch.object(recognition_manager_module.os.path, "exists")
)
elif engine == "whisper":
stack.enter_context(
patch.object(
manager,
"_whisper_model_cache_paths",
return_value=("/models/whisper", "/models/whisper/target.pt"),
)
)
availability = stack.enter_context(
patch.object(recognition_manager_module.os.path, "exists")
)
elif engine == "faster_whisper":
availability = stack.enter_context(
patch.object(recognition_manager_module.faster_whisper, "is_model_downloaded")
)
elif engine == "parakeet":
availability = stack.enter_context(
patch.object(recognition_manager_module.parakeet, "is_model_downloaded")
)
else:
stack.enter_context(
patch.object(manager, "_get_vosk_model_path", return_value="/models/target")
)
availability = stack.enter_context(
patch.object(recognition_manager_module.os.path, "exists")
)
download = stack.enter_context(patch.object(manager, downloader))
for downloaded in (True, False):
with self.subTest(engine=engine, downloaded=downloaded):
availability.return_value = downloaded
manager._download_selected_model_if_missing()
if downloaded:
download.assert_not_called()
elif engine == "whisper":
download.assert_called_once_with("/models/whisper")
else:
download.assert_called_once_with()
download.reset_mock()
def test_download_selected_model_dispatches_whispercpp(self) -> None:
self._assert_missing_model_dispatch("whisper_cpp", "small", "_download_whispercpp_model")
def test_download_selected_model_dispatches_whisper(self) -> None:
self._assert_missing_model_dispatch("whisper", "small", "_download_whisper_model")
def test_download_selected_model_dispatches_faster_whisper(self) -> None:
self._assert_missing_model_dispatch(
"faster_whisper", "tiny", "_download_faster_whisper_model"
)
def test_download_selected_model_dispatches_parakeet(self) -> None:
self._assert_missing_model_dispatch("parakeet", "v3-european", "_download_parakeet_model")
def test_download_selected_model_dispatches_vosk(self) -> None:
self._assert_missing_model_dispatch("vosk", "small", "_download_vosk_model")
def test_reconfigure_download_cancellation_preserves_the_loaded_model(self) -> None:
manager = self._create_manager()
loaded_model = object()
manager.model = loaded_model
manager.engine = "whisper"
manager.model_size = "small"
manager.language = "en-us"
manager.command_processor = Mock()
restore_language = patch.object(manager, "_restore_language_after_oneshot")
restore_language_mock = restore_language.start()
self.patches.append(restore_language)
with (
patch.object(
recognition_manager_module,
"get_model_path",
return_value="/models/medium.bin",
),
patch.object(recognition_manager_module.os.path, "exists", return_value=False),
patch.object(
manager,
"_download_whispercpp_model",
side_effect=RuntimeError("Download cancelled"),
) as download,
patch.object(manager, "_init_selected_engine") as init_selected,
):
with self.assertRaisesRegex(RuntimeError, "Download cancelled"):
manager.reconfigure(
engine="whisper_cpp",
model_size="medium",
language="fr",
force_download=True,
)
assert manager.model is loaded_model
assert manager.engine == "whisper"
assert manager.model_size == "small"
assert manager.language == "en-us"
assert manager._defer_download is True
assert manager._model_lock.locked() is False
assert manager._prefetching_model is False
download.assert_called_once_with()
init_selected.assert_not_called()
manager.command_processor.set_language.assert_any_call("fr")
manager.command_processor.set_language.assert_any_call("en-us")
restore_language_mock.assert_called_once_with()
def test_reconfigure_prefetches_before_initializing_the_selected_engine(self) -> None:
manager = self._create_manager()
manager.engine = "whisper"
available = {"model": False}
loaded_model = object()
def model_exists(_path: str) -> bool:
return available["model"]
def download_model() -> None:
assert manager._prefetching_model is True
assert manager.start_recognition() is False
available["model"] = True
with (
patch.object(
recognition_manager_module,
"get_model_path",
return_value="/models/medium.bin",
),
patch.object(recognition_manager_module.os.path, "exists", side_effect=model_exists),
patch.object(recognition_manager_module, "_preload_pywhispercpp_shared_libraries"),
patch.object(manager, "_whispercpp_model_is_verified", return_value=True),
patch.object(
manager, "_download_whispercpp_model", side_effect=download_model
) as download,
patch.object(
manager,
"_load_whispercpp_model",
side_effect=lambda _path: setattr(manager, "model", loaded_model),
),
patch.object(
manager, "_init_selected_engine", wraps=manager._init_selected_engine
) as init,
patch.object(recognition_manager_module, "play_error_sound") as error_sound,
patch.object(recognition_manager_module, "_show_notification") as notification,
):
manager.reconfigure(
engine="whisper_cpp",
model_size="medium",
force_download=True,
)
assert available["model"] is True
download.assert_called_once_with()
init.assert_called_once_with()
assert manager.model is loaded_model
assert manager._prefetching_model is False
error_sound.assert_called_once_with()
notification.assert_called_once_with(
"Speech model downloading",
"Dictation will be available when the new model has finished downloading.",
"dialog-information",
)
def test_reconfigure_without_download_permission_skips_prefetch(self) -> None:
manager = self._create_manager()
manager.engine = "whisper"
with (
patch.object(manager, "_download_selected_model_if_missing") as prefetch,
patch.object(manager, "_init_selected_engine"),
):
manager.reconfigure(
engine="whisper_cpp",
model_size="medium",
force_download=False,
)
prefetch.assert_not_called()
def test_resume_skips_engine_reinit_while_model_is_prefetching(self) -> None:
manager = self._create_manager(engine="whisper")
manager.engine = "whisper_cpp"
manager._prefetching_model = True
manager._auto_paused = True
manager._idle_unloaded = True
with patch.object(manager, "_init_whispercpp") as init_whispercpp:
manager.reinitialize_after_resume()
init_whispercpp.assert_not_called()
self.assertFalse(manager._auto_paused)
self.assertFalse(manager._idle_unloaded)
def test_lazy_reload_follows_a_cancelled_prefetch_after_resume(self) -> None:
manager = self._create_manager(engine="whisper")
manager.engine = "whisper_cpp"
manager.model = None
manager._model_initialized = False
manager._prefetching_model = True
manager._auto_paused = True
manager.reinitialize_after_resume()
manager._prefetching_model = False
with patch.object(manager, "_init_whispercpp") as init_whispercpp:
manager.ensure_model_loaded()
init_whispercpp.assert_called_once()
def test_start_recognition_refuses_while_a_model_is_prefetching(self) -> None:
manager = self._create_manager()
manager.engine = "whisper_cpp"
manager._prefetching_model = True
uninitialized_manager = object.__new__(SpeechRecognitionManager)
assert uninitialized_manager._prefetching_model is False
with (
patch.object(recognition_manager_module, "play_error_sound") as error_sound,
patch.object(recognition_manager_module, "_show_notification") as notification,
):
assert manager.start_recognition() is False
manager.language = "pl"
with patch.object(
manager, "start_recognition", wraps=manager.start_recognition
) as start:
assert manager.start_recognition_with_language("pl") is False
start.assert_called_once_with(mode="toggle")
assert error_sound.call_count == 2
notification.assert_called_with(
"Speech model downloading",
"Dictation will be available when the new model has finished downloading.",
"dialog-information",
)
assert notification.call_count == 2
def test_download_vosk_model_with_progress(self):
"""Test VOSK model download with progress tracking."""
# Test that progress callback can be registered
@@ -378,6 +640,40 @@ class TestErrorHandling(unittest.TestCase):
assert manager.engine == "whisper"
assert manager.engine != original_engine
def test_reconfigure_aborts_when_download_cancelled_while_lock_is_busy(self) -> None:
"""Cancellation while waiting for a transcription lock restores state."""
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
manager._download_claim.acquire()
manager.engine = "vosk"
manager._model_lock.acquire()
manager._model_lock.release()
lock_held = threading.Event()
release_lock = threading.Event()
def hold_model_lock() -> None:
manager._model_lock.acquire()
lock_held.set()
release_lock.wait(timeout=2)
manager._model_lock.release()
holder = threading.Thread(target=hold_model_lock)
holder.start()
assert lock_held.wait(timeout=2)
manager.cancel_download()
try:
with patch.object(manager, "_download_selected_model_if_missing"):
with self.assertRaisesRegex(RuntimeError, "Download cancelled"):
manager.reconfigure(engine="whisper", force_download=True)
finally:
release_lock.set()
holder.join(timeout=2)
manager.end_download()
assert manager.engine == "vosk"
assert manager._defer_download is True
def test_reconfigure_language_change(self):
"""Test reconfiguring language triggers restart."""
patches = [
tests/test_recognition_manager_downloads.py
+24 -6
View File
@@ -1150,8 +1150,14 @@ class TestFailedReconfigureRestoresPreviousEngine:
manager._model_initialized = True
manager.state = RecognitionState.IDLE
with patch.object(
manager, "_init_faster_whisper", side_effect=RuntimeError("download failed")
with (
patch(
"vocalinux.speech_recognition.recognition_manager.faster_whisper.is_model_downloaded",
return_value=True,
),
patch.object(
manager, "_init_faster_whisper", side_effect=RuntimeError("download failed")
),
):
with patch.object(manager, "_init_whispercpp") as restore_init:
with pytest.raises(RuntimeError, match="download failed"):
@@ -1172,8 +1178,14 @@ class TestFailedReconfigureRestoresPreviousEngine:
manager.model_size = "tiny"
manager.state = RecognitionState.IDLE
with patch.object(
manager, "_init_faster_whisper", side_effect=RuntimeError("download failed")
with (
patch(
"vocalinux.speech_recognition.recognition_manager.faster_whisper.is_model_downloaded",
return_value=True,
),
patch.object(
manager, "_init_faster_whisper", side_effect=RuntimeError("download failed")
),
):
with patch.object(
manager, "_init_whispercpp", side_effect=RuntimeError("restore failed")
@@ -1210,8 +1222,14 @@ class TestFailedReconfigureRestoresPreviousEngine:
manager._model_initialized = True
manager.state = RecognitionState.IDLE
with patch.object(
manager, "_init_faster_whisper", side_effect=RuntimeError("download failed")
with (
patch(
"vocalinux.speech_recognition.recognition_manager.faster_whisper.is_model_downloaded",
return_value=True,
),
patch.object(
manager, "_init_faster_whisper", side_effect=RuntimeError("download failed")
),
):
with patch.object(manager, "_init_whispercpp") as restore_init:
with pytest.raises(RuntimeError, match="download failed"):
tests/test_recognition_manager_remote_api.py
+1 -1
View File
@@ -747,7 +747,7 @@ class TestHTTPSession(unittest.TestCase):
session = manager._http_session
with patch.object(manager, "_init_vosk"):
manager.reconfigure(engine="vosk")
manager.reconfigure(engine="vosk", force_download=False)
session.close.assert_called_once()
self.assertIsNone(manager._http_session)
tests/test_recommendation_is_actionable.py
+79 -147
View File
@@ -1,12 +1,13 @@
"""The recommendation must be applicable, and must not ignore the disk (#778)."""
"""Model recommendations are informational and never change the selection."""
from __future__ import annotations
import importlib
import inspect
import sys
from collections.abc import Iterator
from typing import Any
from unittest.mock import MagicMock, Mock, call, patch
from unittest.mock import MagicMock, Mock, patch
import pytest
@@ -15,12 +16,7 @@ import vocalinux.ui
@pytest.fixture(scope="module")
def settings_dialog() -> Iterator[Any]:
"""Import settings_dialog with real base classes for its GTK subclasses.
Same reasoning as tests/test_settings_model_state.py; both sys.modules and the
package attribute are restored so later tests patching the module by name do
not reach a second module object (#686).
"""
"""Import settings_dialog with concrete GTK base classes."""
repository = sys.modules["gi.repository"]
bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")}
saved_module = sys.modules.pop("vocalinux.ui.settings_dialog", None)
@@ -44,160 +40,96 @@ def dialog_class(settings_dialog: Any) -> Any:
return settings_dialog.SettingsDialog
def _dialog_stub(language: str = "en-us") -> Mock:
def _dialog_stub() -> Mock:
dialog = Mock()
dialog.language = language
dialog.language_combo.get_active_id.return_value = language
dialog._populating_models = False
dialog._processing_language_change = False
# Plain Mock attrs are truthy; these guards must be real False or the
# language-changed handler returns before updating dialog.language.
dialog._initializing = False
dialog._applying_settings = False
dialog._simple_driving = False
dialog.engine_combo.get_active_text.return_value = "whisper.cpp"
dialog.model_variant_combo.get_active_id.return_value = "small.en"
dialog.model_combo.get_active_id.return_value = "small"
dialog.language_combo.get_active_id.return_value = "en-us"
dialog._get_selected_whispercpp_model.return_value = "small.en"
dialog._get_recommended_whispercpp_model_for_language.return_value = (
"small",
"8GB RAM - good balance",
)
dialog._refresh_simple_readout = Mock()
return dialog
def _with_disk(settings_dialog: Any, downloaded: list[str]) -> tuple[Any, Any]:
"""Patch the disk so only ``downloaded`` counts as present."""
return (
patch.object(settings_dialog, "list_downloaded_whispercpp_models", return_value=downloaded),
def test_recommendation_has_no_action_button(settings_dialog: Any, dialog_class: Any) -> None:
source = inspect.getsource(dialog_class.__init__)
assert "Use it" not in source
assert "model_recommendation_button" not in source
def test_same_size_whispercpp_variant_is_not_shown(settings_dialog: Any, dialog_class: Any) -> None:
dialog = _dialog_stub()
with patch.object(settings_dialog, "is_whispercpp_model_downloaded", return_value=True):
dialog_class._update_model_info(dialog)
dialog.model_recommendation_box.hide.assert_called_once_with()
dialog.model_recommendation.set_text.assert_not_called()
def test_different_size_recommendation_shows_size_and_reason(
settings_dialog: Any, dialog_class: Any
) -> None:
dialog = _dialog_stub()
dialog._get_selected_whispercpp_model.return_value = "tiny"
with patch.object(settings_dialog, "is_whispercpp_model_downloaded", return_value=True):
dialog_class._update_model_info(dialog)
message = dialog.model_recommendation.set_text.call_args.args[0]
assert message.startswith("Suggested for this computer: Small")
assert "MB" in message
assert "8GB RAM - good balance" in message
dialog.model_recommendation_box.show.assert_called_once_with()
def test_other_engine_recommendation_uses_recommended_model_size(
settings_dialog: Any, dialog_class: Any
) -> None:
dialog = _dialog_stub()
dialog.engine_combo.get_active_text.return_value = "Vosk"
dialog.model_combo.get_active_id.return_value = "small"
dialog.language = "en-us"
with (
patch.object(
settings_dialog,
"is_whispercpp_model_downloaded",
side_effect=lambda name: name in downloaded,
"VOSK_MODEL_INFO",
{
"small": {"size_mb": 50, "desc": "Selected model"},
"large": {"size_mb": 250, "desc": "Recommended model"},
},
),
)
patch.object(settings_dialog, "_is_vosk_model_downloaded", return_value=True),
patch.object(settings_dialog, "_get_recommended_vosk_model", return_value=("large", "RAM")),
):
dialog_class._update_model_info(dialog)
assert "(250 MB, RAM)" in dialog.model_recommendation.set_text.call_args.args[0]
def test_a_bigger_model_on_disk_is_offered_instead_of_a_download(
settings_dialog: Any, dialog_class: Any
) -> None:
"""The reported case: recommended small.en, medium.en already paid for."""
dialog = _dialog_stub()
listed, downloaded = _with_disk(settings_dialog, ["medium.en", "small", "tiny"])
with listed, downloaded:
alternative = dialog_class._downloaded_alternative_for(dialog, "small.en")
assert alternative == "medium.en"
def test_nothing_is_offered_when_the_recommendation_is_already_on_disk(
def test_recommendation_does_not_change_picker_or_config_state(
settings_dialog: Any, dialog_class: Any
) -> None:
dialog = _dialog_stub()
listed, downloaded = _with_disk(settings_dialog, ["small.en", "medium.en"])
dialog.config_manager = Mock()
with listed, downloaded:
assert dialog_class._downloaded_alternative_for(dialog, "small.en") is None
def test_a_smaller_model_is_never_offered(settings_dialog: Any, dialog_class: Any) -> None:
"""Reusing a download must not quietly cost accuracy."""
dialog = _dialog_stub()
listed, downloaded = _with_disk(settings_dialog, ["tiny", "tiny.en"])
with listed, downloaded:
assert dialog_class._downloaded_alternative_for(dialog, "medium.en") is None
def test_english_only_weights_are_not_offered_for_another_language(
settings_dialog: Any, dialog_class: Any
) -> None:
dialog = _dialog_stub(language="pl")
listed, downloaded = _with_disk(settings_dialog, ["medium.en"])
with listed, downloaded:
assert dialog_class._downloaded_alternative_for(dialog, "small") is None
def test_the_smallest_qualifying_model_wins(settings_dialog: Any, dialog_class: Any) -> None:
"""Between two usable downloads, take the cheaper one to run."""
dialog = _dialog_stub()
listed, downloaded = _with_disk(settings_dialog, ["large", "medium.en"])
with listed, downloaded:
assert dialog_class._downloaded_alternative_for(dialog, "small.en") == "medium.en"
def test_applying_the_recommendation_sets_both_pickers(
settings_dialog: Any, dialog_class: Any
) -> None:
"""Clicking must move size and specialization together, not just one."""
dialog = _dialog_stub()
dialog._recommended_target_model = "small.en"
dialog_class._on_apply_recommendation(dialog, None)
dialog.model_combo.set_active_id.assert_called_once_with("small")
dialog._populate_whispercpp_variant_options.assert_called_once_with("small", "small.en")
dialog.model_variant_combo.set_active_id.assert_called_once_with("small.en")
dialog._sync_language_options_for_selected_model.assert_called_once_with()
dialog._update_model_info.assert_called_once_with()
dialog._refresh_unused_downloads.assert_called_once_with()
dialog._auto_apply_settings.assert_called_once_with()
assert dialog._populating_models is False
def test_applying_does_nothing_without_a_target(settings_dialog: Any, dialog_class: Any) -> None:
dialog = _dialog_stub()
dialog._recommended_target_model = None
dialog_class._on_apply_recommendation(dialog, None)
with patch.object(settings_dialog, "is_whispercpp_model_downloaded", return_value=True):
dialog_class._update_model_info(dialog)
dialog.model_combo.set_active_id.assert_not_called()
dialog._auto_apply_settings.assert_not_called()
dialog.model_variant_combo.set_active_id.assert_not_called()
dialog.config_manager.set.assert_not_called()
dialog.config_manager.save_settings.assert_not_called()
def test_applying_recommendation_suppresses_model_changed_auto_apply(
settings_dialog: Any, dialog_class: Any
) -> None:
"""Size change must not briefly auto-apply the default size before the target.
Without ``_populating_models``, ``set_active_id(large)`` would fire
``_on_model_changed`` and download full large before turbo-q5_0 lands.
"""
dialog = _dialog_stub()
dialog._recommended_target_model = "large-v3-turbo-q5_0"
seen_while_setting_size: list[bool] = []
def capture_flag(_model_size: str) -> bool:
seen_while_setting_size.append(dialog._populating_models)
return True
dialog.model_combo.set_active_id.side_effect = capture_flag
dialog_class._on_apply_recommendation(dialog, None)
assert seen_while_setting_size == [True]
dialog._populate_whispercpp_variant_options.assert_called_once_with(
"large", "large-v3-turbo-q5_0"
def test_model_selection_docs_url_is_stable(settings_dialog: Any) -> None:
assert (
settings_dialog.MODEL_SELECTION_DOC_URL
== "https://github.com/VocaHQ/vocalinux/blob/main/docs/MODEL_SELECTION.md"
)
dialog.model_variant_combo.set_active_id.assert_called_once_with("large-v3-turbo-q5_0")
# One apply after both pickers settle — not during the size change.
dialog._auto_apply_settings.assert_called_once_with()
assert dialog._populating_models is False
def test_language_change_refreshes_recommendation_ui(
settings_dialog: Any, dialog_class: Any
) -> None:
"""EN→PL must refresh recommendation target / label so Use it is not stale."""
dialog = _dialog_stub(language="en-us")
dialog.language_combo.get_active_id.return_value = "pl"
dialog.engine_combo.get_active_text.return_value = "whisper.cpp"
dialog._recommended_target_model = "small.en"
dialog_class._on_language_changed(dialog, None)
assert dialog.language == "pl"
dialog._populate_model_options.assert_called_once_with()
dialog._update_language_warning.assert_called_once_with()
dialog._update_model_info.assert_called_once_with()
dialog._auto_apply_settings.assert_called_once_with()
# Refresh happens before auto-apply so the card is current when settings land.
assert dialog.mock_calls.index(call._update_model_info()) < dialog.mock_calls.index(
call._auto_apply_settings()
)
assert dialog._processing_language_change is False
tests/test_settings_model_state.py
+267 -6
View File
@@ -13,6 +13,7 @@ import pytest
import vocalinux.ui
from vocalinux.common_types import RecognitionState
from vocalinux.utils.model_choice import FASTEST, priority_for_size
@pytest.fixture(scope="module")
@@ -201,6 +202,112 @@ def test_failed_auto_apply_resyncs_the_pickers_with_the_config(settings_dialog,
assert dialog._applying_settings is False
def test_auto_apply_confirms_before_starting_a_download(
settings_dialog: Any, dialog_class: type[Any]
) -> None:
dialog = _dialog_stub()
dialog.get_selected_settings.return_value = {
"engine": "whisper_cpp",
"model_size": "small",
"model_variant": "small",
}
dialog._confirm_model_download = Mock(return_value=False)
dialog._drop_staged_model_spec_and_resync = Mock()
with patch.object(
settings_dialog,
"_undownloaded_model_info",
return_value={"size_mb": 100, "url": "https://example.test/model.bin"},
):
dialog_class._auto_apply_settings(dialog, allow_download=True)
dialog._confirm_model_download.assert_called_once_with("whisper_cpp", "small", 100)
dialog.speech_engine.try_begin_download.assert_not_called()
dialog._drop_staged_model_spec_and_resync.assert_called_once_with()
def test_apply_confirms_before_starting_a_download(
settings_dialog: Any, dialog_class: type[Any]
) -> None:
dialog = _dialog_stub()
dialog.get_selected_settings.return_value = {
"engine": "whisper_cpp",
"model_size": "small",
"model_variant": "small",
}
dialog._confirm_model_download = Mock(return_value=False)
dialog._drop_staged_model_spec_and_resync = Mock()
with patch.object(
settings_dialog,
"_undownloaded_model_info",
return_value={"size_mb": 100, "url": "https://example.test/model.bin"},
):
assert dialog_class.apply_settings(dialog) is False
dialog._confirm_model_download.assert_called_once_with("whisper_cpp", "small", 100)
dialog.speech_engine.try_begin_download.assert_not_called()
dialog._drop_staged_model_spec_and_resync.assert_called_once_with()
def test_download_confirmation_includes_catalog_host_and_cancel_default(
settings_dialog: Any, dialog_class: type[Any]
) -> None:
gtk = settings_dialog.Gtk
gtk.MessageDialog.reset_mock()
message_dialog = gtk.MessageDialog.return_value
message_dialog.run.return_value = gtk.ResponseType.CANCEL
with patch.object(
settings_dialog,
"_undownloaded_model_info",
return_value={"url": "https://models.example.test/model.bin"},
):
assert dialog_class._confirm_model_download(Mock(), "whisper_cpp", "small", 125) is False
gtk.MessageDialog.assert_called_once()
kwargs = gtk.MessageDialog.call_args.kwargs
assert kwargs["message_type"] == gtk.MessageType.QUESTION
assert kwargs["text"] == "Download Small?"
message_dialog.format_secondary_text.assert_called_once_with(
"About 125 MB from models.example.test. Nothing is downloaded until you confirm."
)
message_dialog.set_default_response.assert_called_once_with(gtk.ResponseType.CANCEL)
@pytest.mark.parametrize(
("engine", "model_name", "language", "host"),
[
("whisper", "small", "en-us", "openaipublic.azureedge.net"),
("whisper_cpp", "small", "en-us", "huggingface.co"),
("vosk", "small", "en-us", "alphacephei.com"),
("parakeet", "v3-european", "en-us", "huggingface.co"),
("faster_whisper", "tiny", "en-us", "huggingface.co"),
],
)
def test_model_download_source_url_covers_each_downloadable_engine(
settings_dialog: Any, engine: str, model_name: str, language: str, host: str
) -> None:
url = settings_dialog._model_download_source_url(engine, model_name, language)
assert settings_dialog.urlparse(url).netloc == host
def test_download_confirmation_derives_host_when_catalog_has_no_url(
settings_dialog: Any, dialog_class: type[Any]
) -> None:
gtk = settings_dialog.Gtk
gtk.MessageDialog.reset_mock()
gtk.MessageDialog.return_value.run.return_value = gtk.ResponseType.CANCEL
with patch.object(settings_dialog, "_undownloaded_model_info", return_value={"size_mb": 125}):
assert dialog_class._confirm_model_download(Mock(), "whisper", "small", 125) is False
gtk.MessageDialog.return_value.format_secondary_text.assert_called_once_with(
"About 125 MB from openaipublic.azureedge.net. Nothing is downloaded until you confirm."
)
def test_already_downloaded_auto_apply_runs_on_a_worker(settings_dialog, dialog_class):
"""A model already on disk must not reconfigure on the GTK main loop."""
dialog = _dialog_stub()
@@ -430,7 +537,7 @@ def test_download_path_resyncs_when_the_apply_reports_failure(settings_dialog, d
modal = modal_class.return_value
scheduled = [(func, args) for func, args in idle_calls]
assert (dialog._idle_resync_model_ui_from_config, ()) in scheduled
assert (dialog._idle_drop_staged_model_spec_and_resync, ()) in scheduled
assert (modal.set_complete, (True, "")) not in scheduled
assert any(func is modal.set_complete and args[0] is False for func, args in scheduled)
dialog._save_selected_settings.assert_not_called()
@@ -456,7 +563,7 @@ def test_download_path_resyncs_when_the_download_is_cancelled(settings_dialog, d
dialog_class._auto_apply_settings(dialog, allow_download=True)
modal = modal_class.return_value
assert (dialog._idle_resync_model_ui_from_config, ()) in idle_calls
assert (dialog._idle_drop_staged_model_spec_and_resync, ()) in idle_calls
assert (modal.set_complete, (False, "Download cancelled")) in idle_calls
@@ -478,7 +585,7 @@ def test_modal_close_resyncs_an_engine_that_never_applied(settings_dialog, dialo
):
dialog_class._auto_apply_settings(dialog, allow_download=True)
dialog._resync_engine_ui_if_unapplied.assert_called_once()
dialog._drop_staged_model_spec_and_resync.assert_called_once_with()
def test_idle_resync_skips_a_destroyed_dialog(dialog_class):
@@ -556,6 +663,107 @@ def test_resync_restores_language_from_saved_config(dialog_class):
dialog.engine_combo.set_active_id.assert_not_called()
def _resync_dialog_for_language(
dialog_class: type[Any], saved_language: str
) -> tuple[Mock, dict[str, Any]]:
dialog = _dialog_stub()
state: dict[str, Any] = {
"language": "auto",
"follow": False,
"priority": FASTEST,
}
dialog.config_manager.get_settings.return_value = {
"speech_recognition": {
"engine": "whisper_cpp",
"model_size": "medium",
"language": saved_language,
}
}
dialog.config_manager.get.return_value = ""
dialog.engine_combo.get_active_text.return_value = "whisper.cpp"
dialog.language_combo.get_active_id.side_effect = lambda: state["language"]
dialog.language_combo.set_active_id.side_effect = lambda value: state.__setitem__(
"language", value
)
dialog.follow_layout_switch.get_active.side_effect = lambda: state["follow"]
dialog.follow_layout_switch.set_active.side_effect = lambda value: state.__setitem__(
"follow", value
)
dialog.simple_priority_combo.get_active_id.side_effect = lambda: state["priority"]
dialog.simple_priority_combo.set_active_id.side_effect = lambda value: state.__setitem__(
"priority", value
)
dialog._get_selected_engine.return_value = "whisper_cpp"
dialog._get_selected_whispercpp_model.return_value = "medium"
dialog._get_recommended_whispercpp_model_for_language.return_value = ("small", "reason")
dialog._follow_layout_supported.return_value = True
dialog._is_selected_whispercpp_model_english_only.return_value = False
def set_combo_active_id(combo: Any, value: str) -> bool:
combo.set_active_id(value)
return True
dialog._set_combo_active_id_or_first.side_effect = set_combo_active_id
dialog._populate_model_options = Mock()
dialog._populate_language_options = Mock()
dialog._update_model_info = Mock()
dialog._update_model_picker_tooltips = Mock()
dialog._update_simple_visibility = Mock()
dialog._auto_apply_settings = Mock()
dialog._confirm_model_download = Mock()
dialog.config_manager.update_speech_recognition_settings = Mock()
dialog.speech_engine.try_begin_download = Mock()
dialog._simple_syncing = False
dialog._staged_model_spec = {"engine": "whisper_cpp", "model_size": "large"}
_bind_real(
dialog,
dialog_class,
"_drop_staged_model_spec_and_resync",
"_resync_model_ui_from_config",
"_sync_language_options_for_selected_model",
"_sync_follow_layout_controls",
"_update_language_warning",
"_sync_simple_from_advanced",
)
return dialog, state
def test_declined_model_resync_restores_follow_layout_and_saved_simple_priority(
settings_dialog: Any, dialog_class: type[Any]
) -> None:
dialog, state = _resync_dialog_for_language(dialog_class, "layout")
with patch.object(settings_dialog, "language_for_active_layout", return_value="en-us"):
dialog._drop_staged_model_spec_and_resync()
expected_priority = priority_for_size("small", "medium")
assert dialog._staged_model_spec is None
assert dialog._follow_layout_saved is True
assert dialog.follow_layout_switch.get_active() is True
assert dialog._follow_layout_active is True
assert dialog.language_combo.get_active_id() == "en-us"
assert dialog.language == "en-us"
assert state["priority"] == expected_priority
dialog.config_manager.update_speech_recognition_settings.assert_not_called()
dialog.speech_engine.try_begin_download.assert_not_called()
dialog._confirm_model_download.assert_not_called()
dialog._auto_apply_settings.assert_not_called()
def test_resync_with_saved_language_disables_follow_layout(
dialog_class: type[Any],
) -> None:
dialog, _state = _resync_dialog_for_language(dialog_class, "de")
dialog._drop_staged_model_spec_and_resync()
assert dialog._follow_layout_saved is False
assert dialog.follow_layout_switch.get_active() is False
assert dialog._follow_layout_active is False
assert dialog.language_combo.get_active_id() == "de"
assert dialog.language == "de"
def test_picker_handlers_early_return_while_applying(dialog_class: type[Any]) -> None:
"""Size, specialization, and language handlers must not rebuild or apply mid-apply."""
dialog = _dialog_stub()
@@ -744,6 +952,7 @@ def _download_setup(dialog: Mock) -> None:
"language": "auto",
}
dialog._apply_settings_internal.return_value = True
dialog._confirm_model_download.return_value = True
@pytest.mark.parametrize("entry", ["_auto_apply_settings", "apply_settings"])
@@ -772,7 +981,7 @@ def test_settings_refuses_a_download_while_the_tray_holds_the_engine(
modal_class.assert_not_called()
dialog._apply_settings_internal.assert_not_called()
dialog._show_download_busy_dialog.assert_called_once_with()
dialog._resync_model_ui_from_config.assert_called_once_with()
dialog._drop_staged_model_spec_and_resync.assert_called_once_with()
@pytest.mark.parametrize("entry", ["_auto_apply_settings", "apply_settings"])
@@ -796,6 +1005,43 @@ def test_settings_releases_the_engine_once_the_download_is_over(
dialog.speech_engine.end_download.assert_called_once_with()
@pytest.mark.parametrize("entry", ["_auto_apply_settings", "apply_settings"])
def test_cancel_poll_stays_registered_until_download_worker_removes_it(
settings_dialog: Any, dialog_class: type[Any], entry: str
) -> None:
dialog = _dialog_stub()
_download_setup(dialog)
source_callbacks = []
callback_results = []
glib = MagicMock()
def timeout_add(_interval: int, callback: Callable[[], bool]) -> int:
source_callbacks.append(callback)
callback_results.append(callback())
return 74
glib.timeout_add.side_effect = timeout_add
with (
patch.object(settings_dialog, "_undownloaded_model_info", return_value={"size_mb": 100}),
patch.object(settings_dialog, "ModelDownloadDialog") as modal_class,
patch.object(settings_dialog, "GLib", glib),
patch.object(settings_dialog.threading, "Thread", _InlineThread),
):
modal_class.return_value.cancelled = True
if entry == "_auto_apply_settings":
dialog_class._auto_apply_settings(dialog, allow_download=True)
else:
dialog_class.apply_settings(dialog)
dialog.speech_engine.try_begin_download.assert_called_once_with()
assert len(source_callbacks) == 1
assert callback_results == [True]
glib.source_remove.assert_called_once_with(74)
dialog.speech_engine.cancel_download.assert_called_once_with()
dialog.speech_engine.end_download.assert_called_once_with()
def test_a_picker_change_only_stages_a_missing_model(
settings_dialog: Any, dialog_class: type[Any]
) -> None:
@@ -921,7 +1167,6 @@ def _update_model_info_stub(dialog_class: type[Any]) -> Mock:
dialog.engine_combo.get_active_text.return_value = "whisper.cpp"
dialog._get_selected_whispercpp_model.return_value = "small"
dialog._get_recommended_whispercpp_model_for_language.return_value = ("small", "fits")
dialog._downloaded_alternative_for.return_value = None
return dialog
@@ -1185,6 +1430,17 @@ def test_whisper_selected_settings_keep_combo_language(dialog_class):
assert settings["language"] == "fr"
def test_follow_layout_selected_settings_keep_the_layout_sentinel(
settings_dialog: Any, dialog_class: type[Any]
) -> None:
dialog = _dialog_for_selected_settings("whisper.cpp", "de")
with patch.object(settings_dialog, "_is_following_layout", return_value=True):
settings = dialog_class.get_selected_settings(dialog)
assert settings["language"] == settings_dialog.LANGUAGE_FOLLOWS_LAYOUT
def test_parakeet_engine_change_forces_language_auto(dialog_class):
"""A leftover catalog language from another engine must not stick in memory."""
dialog = _dialog_stub()
@@ -1458,7 +1714,12 @@ def test_speech_manager_reconfigure_to_parakeet_normalizes_language():
defer_download=True,
)
assert manager.language == "en-us"
manager.reconfigure(engine="parakeet", model_size="v3-european", language="fr")
manager.reconfigure(
engine="parakeet",
model_size="v3-european",
language="fr",
force_download=False,
)
assert manager.engine == "parakeet"
assert manager.language == "auto"
assert manager.command_processor.language == "auto"
tests/test_settings_test_dictation.py
+40 -6
View File
@@ -118,17 +118,42 @@ def test_test_dictation_missing_model_skips_listen_timer() -> None:
thread_cls.assert_not_called()
assert dialog._test_active is False
assert "Speech Model page" in dialog.test_buffer._text
assert (
dialog.test_buffer._text
== "Tiny is not downloaded yet. Press Download on the Speech Model page first, "
"or pick a downloaded model."
)
assert "No speech detected" not in dialog.test_buffer._text
dialog.speech_engine.set_text_callbacks.assert_any_call([])
dialog.apply_settings.assert_not_called()
dialog.test_button.set_label.assert_not_called()
dialog.test_output_revealer.set_reveal_child.assert_called_with(True)
def test_cancelled_settings_apply_does_not_start_test_dictation() -> None:
dialog = _dialog_for_test(start_return=True)
dialog.get_selected_settings.return_value["model_size"] = "small"
dialog.apply_settings.return_value = False
with (
patch.object(settings_dialog, "_undownloaded_model_info", return_value=None),
patch.object(settings_dialog.threading, "Thread") as thread_cls,
):
SettingsDialog._on_test_clicked(dialog, None)
dialog.apply_settings.assert_called_once()
dialog.speech_engine.start_recognition.assert_not_called()
thread_cls.assert_not_called()
assert dialog._test_active is False
assert dialog.test_buffer._text == "Failed to apply settings. Please try again."
def test_test_dictation_auto_paused_message() -> None:
dialog = _dialog_for_test(start_return=False, is_auto_paused=True)
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
with (
patch.object(settings_dialog, "_undownloaded_model_info", return_value=None),
patch.object(settings_dialog.threading, "Thread") as thread_cls,
):
SettingsDialog._on_test_clicked(dialog, None)
thread_cls.assert_not_called()
@@ -139,7 +164,10 @@ def test_test_dictation_auto_paused_message() -> None:
def test_test_dictation_start_failure_generic_message() -> None:
dialog = _dialog_for_test(start_return=False, model_ready=True, is_auto_paused=False)
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
with (
patch.object(settings_dialog, "_undownloaded_model_info", return_value=None),
patch.object(settings_dialog.threading, "Thread") as thread_cls,
):
SettingsDialog._on_test_clicked(dialog, None)
thread_cls.assert_not_called()
@@ -149,7 +177,10 @@ def test_test_dictation_start_failure_generic_message() -> None:
def test_test_dictation_started_arms_listen_timer() -> None:
dialog = _dialog_for_test(start_return=True)
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
with (
patch.object(settings_dialog, "_undownloaded_model_info", return_value=None),
patch.object(settings_dialog.threading, "Thread") as thread_cls,
):
SettingsDialog._on_test_clicked(dialog, None)
thread_cls.assert_called_once()
@@ -174,7 +205,10 @@ def test_test_dictation_reconfigures_when_live_engine_differs_from_ui() -> None:
dialog.apply_settings.side_effect = _apply_and_sync
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
with (
patch.object(settings_dialog, "_undownloaded_model_info", return_value=None),
patch.object(settings_dialog.threading, "Thread") as thread_cls,
):
SettingsDialog._on_test_clicked(dialog, None)
dialog.apply_settings.assert_called_once()
tests/test_simple_model_settings.py
+62
View File
@@ -2,6 +2,7 @@
import importlib
import sys
from typing import Any
from unittest.mock import MagicMock, Mock, patch
import pytest
@@ -387,6 +388,67 @@ def test_auto_detect_shows_up_as_the_other_languages_switch(settings_dialog, dia
dialog.simple_multi_switch.set_active.assert_called_once_with(True)
def test_follow_layout_shows_and_disables_the_resolved_simple_language(
settings_dialog: Any, dialog_class: Any
) -> None:
dialog = _dialog_stub(language="de")
dialog.language_combo.get_active_id.return_value = "de"
dialog._get_selected_whispercpp_model.return_value = "small"
dialog._get_recommended_whispercpp_model_for_language.return_value = ("small", "reason")
with patch.object(settings_dialog, "_is_following_layout", return_value=True):
dialog_class._sync_simple_from_advanced(dialog)
dialog.simple_language_combo.set_active_id.assert_called_once_with("de")
dialog.simple_language_row.set_sensitive.assert_called_once_with(False)
dialog.simple_language_row.set_tooltip_text.assert_called_once_with(
"Follows your keyboard layout. Turn off Follow keyboard layout "
"in Advanced to pick a language."
)
def test_follow_layout_priority_change_does_not_write_resolved_language(
settings_dialog: Any, dialog_class: Any
) -> None:
dialog = _dialog_stub(language="de", priority=ACCURATE, recommended="small")
dialog._get_selected_engine.return_value = "whisper_cpp"
with patch.object(settings_dialog, "_is_following_layout", return_value=True):
dialog_class._apply_simple_choice(dialog)
dialog._set_combo_active_id_or_first.assert_not_called()
assert dialog.language == "de"
dialog.model_combo.set_active_id.assert_called_once_with("medium")
def test_follow_layout_priority_change_keeps_multilingual_variant_and_mode(
settings_dialog: Any, dialog_class: Any
) -> None:
dialog = _dialog_stub(language="en-us", priority=FASTEST, recommended="small")
dialog._follow_layout_active = True
dialog.follow_layout_switch.get_active.return_value = True
dialog._follow_layout_supported.return_value = True
selected_variant: dict[str, Any] = {"id": None}
dialog.model_variant_combo.set_active_id.side_effect = lambda variant: selected_variant.update(
id=variant
)
dialog._get_selected_whispercpp_model.side_effect = lambda: selected_variant["id"] or "small"
dialog._is_selected_whispercpp_model_english_only.side_effect = lambda: (
settings_dialog.is_english_only_whispercpp_model(dialog._get_selected_whispercpp_model())
)
dialog_class._apply_simple_choice(dialog)
dialog_class._sync_follow_layout_controls(dialog)
assert selected_variant["id"] == "base"
dialog.model_variant_combo.set_active_id.assert_called_once_with("base")
dialog._on_disk_stand_in.assert_called_once_with(
"base", "base", settings_dialog.LANGUAGE_FOLLOWS_LAYOUT
)
assert dialog.follow_layout_switch.get_active() is True
assert dialog._follow_layout_active is True
@pytest.mark.parametrize("stale_second", ["auto", "en-us"])
def test_pinned_advanced_language_plus_stale_stored_second_does_not_replace_the_pin(
settings_dialog, dialog_class, stale_second