diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 3c39c32..4a3ecb0 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -974,7 +974,13 @@ class SpeechRecognitionManager: # Download progress tracking self._download_progress_callback: Optional[Callable[[float, float, str], None]] = None + # 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 + # 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. + self._download_claim = threading.Lock() self._defer_download = defer_download self._model_initialized = False # True while auto-pause has unloaded the model for a configured app/game @@ -2184,6 +2190,57 @@ class SpeechRecognitionManager: """ self._download_progress_callback = callback + def set_model_missing_handler(self, handler: Optional[Callable[[str], bool]]): + """ + Set a handler for dictation attempted without a model on disk. + + The UI layer registers one to offer the download directly; without a + handler, or when it declines, a plain notification is shown instead. + + Args: + handler: Function(engine) -> True if it took care of the situation, + or None to clear + """ + self._model_missing_handler = handler + + def _notify_model_missing(self) -> bool: + """Let the registered handler take over the missing model, if it can.""" + handler = self._model_missing_handler + if handler is None: + return False + try: + return bool(handler(self.engine)) + except Exception as e: + logger.error(f"Model-missing handler failed: {e}", exc_info=True) + return False + + def try_begin_download(self) -> bool: + """ + Claim the engine for a model download. + + The tray and the settings dialog can each start one, and they would + otherwise overwrite each other's progress callback and reconfigure the + engine underneath one another. Whoever claims first downloads; the + other is told to wait. + + Returns: + True if the claim was taken, False if a download is already running + """ + return self._download_claim.acquire(blocking=False) + + def end_download(self): + """Release the claim taken by try_begin_download().""" + try: + self._download_claim.release() + except RuntimeError: + # Never claimed; nothing to release + pass + + @property + def download_in_progress(self) -> bool: + """Whether some UI currently holds the download claim.""" + return self._download_claim.locked() + def cancel_download(self): """Request cancellation of the current download.""" self._download_cancelled = True @@ -2468,12 +2525,13 @@ class SpeechRecognitionManager: "Please download via Settings." ) play_error_sound() - _show_notification( - "No Speech Model", - "Please open Settings and download a speech recognition model " - "to use dictation.", - "dialog-warning", - ) + if not self._notify_model_missing(): + _show_notification( + "No Speech Model", + "Please open Settings and download a speech recognition model " + "to use dictation.", + "dialog-warning", + ) return False logger.info("Starting speech recognition") @@ -3066,6 +3124,7 @@ class SpeechRecognitionManager: audio_device_index: Optional[int] = None, audio_device_name: Optional[str] = None, force_download: bool = True, + force_reinit: bool = False, **kwargs, # Allow for future expansion ): """ @@ -3080,6 +3139,9 @@ class SpeechRecognitionManager: audio_device_index: Audio input device index (None for default, -1 to clear). audio_device_name: Audio device name for stable re-resolution. force_download: If True, download missing models (default: True for UI-triggered reconfigures). + force_reinit: Re-initialize even when nothing changed. force_download + only takes effect during a re-init, so a caller asking for the + model it is already configured for needs this to fetch it. """ logger.info( f"Reconfiguring speech engine. New settings: engine={engine}, model_size={model_size}, " @@ -3087,7 +3149,7 @@ class SpeechRecognitionManager: f"audio_device={audio_device_index}, audio_device_name={audio_device_name}" ) - restart_needed = False + restart_needed = force_reinit old_engine = self.engine if engine is not None and engine != self.engine: self.engine = engine diff --git a/src/vocalinux/ui/notifications.py b/src/vocalinux/ui/notifications.py new file mode 100644 index 0000000..c986860 --- /dev/null +++ b/src/vocalinux/ui/notifications.py @@ -0,0 +1,151 @@ +"""Desktop notifications, with a clickable action where the server offers one. + +libnotify is used when it is available and the notification server advertises +the "actions" capability, so a notification can carry a button and be updated +in place. Everywhere else this falls back to spawning ``notify-send``, which is +what the rest of the app has always used. +""" + +import logging +import subprocess +from typing import Callable, Optional + +logger = logging.getLogger(__name__) + +APP_NAME = "Vocalinux" + +# Notifications waiting for their action to be clicked. Without a reference the +# notification can be garbage collected before the user answers it, and the +# callback then never runs. +_pending: set = set() + +_notify_module = None +_notify_initialized = False + + +def _libnotify(): + """Return an initialized Notify module, or None when unavailable.""" + global _notify_module, _notify_initialized + + if _notify_initialized: + return _notify_module + + _notify_initialized = True + try: + import gi + + gi.require_version("Notify", "0.7") + from gi.repository import Notify + + if not Notify.is_initted() and not Notify.init(APP_NAME): + logger.debug("libnotify could not be initialized") + return None + _notify_module = Notify + except Exception as e: + logger.debug(f"libnotify unavailable, falling back to notify-send: {e}") + _notify_module = None + + return _notify_module + + +def supports_actions() -> bool: + """Report whether the notification server can show action buttons.""" + notify = _libnotify() + if notify is None: + return False + try: + return "actions" in (notify.get_server_caps() or []) + except Exception as e: + logger.debug(f"Could not read notification server capabilities: {e}") + return False + + +def _notify_send(title: str, message: str, icon: str) -> None: + """Show a notification through the notify-send binary.""" + try: + subprocess.Popen( + ["notify-send", "-i", icon, "-a", APP_NAME, title, message], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except (FileNotFoundError, OSError) as e: + logger.debug(f"Could not show notification: {e}") + + +def notify( + title: str, + message: str, + icon: str = "dialog-information", + action: Optional[tuple[str, Callable[[], None]]] = None, +): + """Show a notification, optionally carrying a single action button. + + Args: + title: Notification summary. + message: Notification body. + icon: Themed icon name. + action: ``(label, callback)`` for the button. The callback runs on the + main loop when the user clicks it. + + Returns: + A handle that :func:`update` can refresh, or None when the notification + was shown through the fallback and cannot be updated. + """ + notify_module = _libnotify() + if notify_module is None or (action is not None and not supports_actions()): + # An unclickable button would strand the user, so drop to plain text and + # keep the instructions that tell them where to go instead. + _notify_send(title, message, icon) + return None + + try: + notification = notify_module.Notification.new(title, message, icon) + if action is not None: + label, callback = action + + def _on_action(_notification, _action_id, *_user_data): + _pending.discard(notification) + try: + callback() + except Exception as e: + logger.error(f"Notification action failed: {e}", exc_info=True) + + notification.add_action("default-action", label, _on_action, None) + notification.connect("closed", lambda _n: _pending.discard(notification)) + _pending.add(notification) + + notification.show() + return notification + except Exception as e: + logger.debug(f"libnotify failed, falling back to notify-send: {e}") + _notify_send(title, message, icon) + return None + + +def update(notification, title: str, message: str, icon: str = "dialog-information") -> bool: + """Refresh an existing notification in place. + + Returns: + True when the notification was updated, False when there is nothing to + update (fallback notifications cannot be changed after the fact). + """ + if notification is None: + return False + try: + notification.update(title, message, icon) + notification.show() + return True + except Exception as e: + logger.debug(f"Could not update notification: {e}") + return False + + +def close(notification) -> None: + """Withdraw a notification that is still on screen.""" + if notification is None: + return + try: + _pending.discard(notification) + notification.close() + except Exception as e: + logger.debug(f"Could not close notification: {e}") diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index 8f52a37..cb02013 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -22,7 +22,7 @@ import os import re import threading import time -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, NamedTuple, Optional import gi @@ -1011,6 +1011,49 @@ def _get_recommended_vosk_model() -> tuple: return "small", "Default recommendation" +class ModelRecommendation(NamedTuple): + """The model this system should use for an engine, ready to present.""" + + model_id: str + reason: str + display_name: str + size_label: str + + +def recommended_model_for_engine( + engine: str, language: str = "auto" +) -> Optional[ModelRecommendation]: + """Return the model to download for an engine, or None if it needs none. + + This is the recommendation the model picker marks with a star, exposed for + callers outside the dialog (the tray offers it when dictation is attempted + without a model). + """ + if engine == "whisper_cpp": + recommended_model, reason = get_recommended_whispercpp_model() + model_id, reason = _recommended_whispercpp_variant_for_language( + recommended_model, reason, language + ) + size_mb = WHISPERCPP_MODEL_INFO.get(model_id, {}).get("size_mb", 0) + elif engine == "whisper": + model_id, reason = _get_recommended_whisper_model() + size_mb = WHISPER_MODEL_INFO.get(model_id, {}).get("size_mb", 0) + elif engine == "vosk": + model_id, reason = _get_recommended_vosk_model() + size_mb = VOSK_MODEL_INFO.get(model_id, {}).get("size_mb", 0) + else: + # Remote API transcribes server-side; there is nothing to download. + return None + + size_mb = size_mb if isinstance(size_mb, int) else 0 + return ModelRecommendation( + model_id=model_id, + reason=reason, + display_name=_model_display_name(model_id), + size_label=_format_size(size_mb), + ) + + # GDK key-symbol names that are themselves modifiers (skipped while recording). _GDK_MODIFIER_KEYNAMES = { "Control_L", @@ -5131,6 +5174,13 @@ class SettingsDialog(Gtk.Dialog): model_info = VOSK_MODEL_INFO.get(model_name, {"size_mb": 50}) if needs_download: + 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() + return logger.info(f"Model {model_name} needs download, showing progress dialog") download_dialog = ModelDownloadDialog( self, @@ -5173,6 +5223,7 @@ class SettingsDialog(Gtk.Dialog): finally: GLib.source_remove(cancel_check_id) self.speech_engine.set_download_progress_callback(None) + self.speech_engine.end_download() except Exception as e: error_msg = str(e) @@ -5518,6 +5569,11 @@ For now, the engine has been reverted to VOSK.""" model_info = VOSK_MODEL_INFO.get(model_name, {"size_mb": 50}) if needs_download: + 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() + return download_dialog = ModelDownloadDialog( self, model_name, @@ -5554,6 +5610,7 @@ For now, the engine has been reverted to VOSK.""" finally: GLib.source_remove(cancel_check_id) self.speech_engine.set_download_progress_callback(None) + self.speech_engine.end_download() except Exception as e: error_msg = str(e) @@ -5578,6 +5635,19 @@ For now, the engine has been reverted to VOSK.""" return self._apply_settings_internal(settings) + def _show_download_busy_dialog(self): + """Tell the user a model download is already running elsewhere.""" + dialog = Gtk.MessageDialog( + transient_for=self, + flags=0, + message_type=Gtk.MessageType.INFO, + buttons=Gtk.ButtonsType.OK, + text="A model download is already running", + ) + dialog.format_secondary_text("Wait for it to finish, then pick the model again.") + dialog.run() + dialog.destroy() + def _apply_settings_internal(self, settings: dict, raise_errors: bool = False) -> bool: """Internal method to apply settings. diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index 597fe6f..f412b2a 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -8,6 +8,8 @@ recognition process and displaying its status. import logging import os import signal +import threading +import time from typing import Callable, Optional import gi @@ -40,10 +42,11 @@ from ..suspend_handler import SuspendHandler from ..utils.resource_manager import ResourceManager from ..utils.update_checker import ReleaseInfo from ..utils.update_monitor import UpdateMonitor +from . import notifications from .config_manager import get_shared_config_manager from .keyboard_backends import DEFAULT_SHORTCUT, DEFAULT_SHORTCUT_MODE from .keyboard_shortcuts import KeyboardShortcutManager -from .settings_dialog import SettingsDialog +from .settings_dialog import SettingsDialog, recommended_model_for_engine logger = logging.getLogger(__name__) @@ -55,6 +58,25 @@ APP_ID = FLATPAK_ID or "vocalinux" _resource_manager = ResourceManager() ICON_DIR = _resource_manager.icons_dir +# How often a running model download refreshes its notification. +_DOWNLOAD_NOTIFY_INTERVAL_SECONDS = 5.0 + + +def _idle_once(func, *args): + """Schedule func on the GTK main loop for exactly one run. + + GLib.idle_add repeats a callback for as long as it returns a truthy value, + and the notification helpers return handles and success flags, so they must + not be handed to idle_add directly. + """ + + def _call(): + func(*args) + return False + + GLib.idle_add(_call) + + # Bundled icon file names (paths under ICON_DIR). DEFAULT_ICON = "vocalinux-microphone-off" ACTIVE_ICON = "vocalinux-microphone" @@ -142,6 +164,11 @@ class TrayIndicator: # Register for speech recognition state changes self.speech_engine.register_state_callback(self._on_recognition_state_changed) + # Offer the download instead of only telling the user a model is missing + self._model_download_active = False + if hasattr(self.speech_engine, "set_model_missing_handler"): + self.speech_engine.set_model_missing_handler(self._offer_recommended_model) + # Initialize the icon files and validate resources self._init_icons() self._validate_resources() @@ -510,6 +537,159 @@ class TrayIndicator: # Update the UI in the GTK main thread GLib.idle_add(self._update_ui, state) + def _offer_recommended_model(self, engine: str) -> bool: + """Offer the recommended model when dictation finds none installed. + + Args: + engine: The engine that has no model on disk. + + Returns: + True when the user was told what to do, False to let the caller + fall back to its own notification. + """ + if self._model_download_active: + return True + + language = self.config_manager.get_str("speech_recognition", "language", "auto") + try: + recommendation = recommended_model_for_engine(engine, language) + except Exception as e: + logger.error(f"Could not work out a recommended model for {engine}: {e}") + return False + + if recommendation is None: + # Remote API transcribes server-side; there is nothing to download. + return False + + # Dictation can be triggered from the shortcut thread, so hand the + # notification to the main loop that owns the UI. + GLib.idle_add(self._show_model_offer, engine, recommendation) + return True + + def _show_model_offer(self, engine: str, recommendation) -> bool: + """Show the offer notification. Runs on the main loop.""" + action = None + if notifications.supports_actions(): + action = ( + f"Download {recommendation.display_name} ({recommendation.size_label})", + lambda: self._download_recommended_model(engine, recommendation), + ) + + if action is not None: + body = ( + f"{recommendation.display_name} ({recommendation.size_label}) is recommended " + "for your system. Download it from this notification, or pick another one " + "in Settings." + ) + else: + # No action button on this server, so pointing at one would strand + # the user; send them where they can actually do something. + body = ( + f"{recommendation.display_name} ({recommendation.size_label}) is recommended " + "for your system. Open Settings to download a speech recognition model." + ) + + notifications.notify("No Speech Model", body, "dialog-warning", action=action) + return False + + def _download_recommended_model(self, engine: str, recommendation) -> None: + """Start downloading the offered model without blocking the main loop. + + Runs on the main loop: libnotify dispatches the action callback there. + """ + if self._model_download_active: + return + if not self.speech_engine.try_begin_download(): + # Settings is already downloading a model. A second download would + # fight it over the engine's single progress callback and leave + # whichever finished last as the configured model. + notifications.notify( + "Download already in progress", + "Another speech model is being downloaded. Wait for it to finish, " + "then try again.", + "dialog-information", + ) + return + self._model_download_active = True + # libnotify and GObject are not thread-safe, so the notification is + # created here and only its handle travels to the worker, which routes + # every later update back through _idle_once. + progress = notifications.notify( + "Downloading speech model", + f"{recommendation.display_name} ({recommendation.size_label}) — starting...", + "folder-download", + ) + threading.Thread( + target=self._run_recommended_model_download, + args=(engine, recommendation, progress), + daemon=True, + name="model-download", + ).start() + + def _run_recommended_model_download(self, engine: str, recommendation, progress) -> None: + """Download the offered model, then report how it went.""" + last_notified = 0.0 + + def on_progress(fraction, speed_mbps, status): + nonlocal last_notified + now = time.monotonic() + if now - last_notified < _DOWNLOAD_NOTIFY_INTERVAL_SECONDS: + return + last_notified = now + _idle_once( + notifications.update, + progress, + "Downloading speech model", + f"{recommendation.display_name}: {status}", + "folder-download", + ) + + try: + self.speech_engine.set_download_progress_callback(on_progress) + self.speech_engine.reconfigure( + engine=engine, + model_size=recommendation.model_id, + force_download=True, + # reconfigure() only re-initializes when engine, model or + # language change, and force_download is read during that + # re-init. The recommendation is often what is already + # configured (first run is whisper_cpp + tiny, and tiny is what + # a CPU-only machine is recommended), so without this the call + # is a no-op that downloads nothing. + force_reinit=True, + ) + if not self.speech_engine.model_ready: + raise RuntimeError("the engine did not load the model after downloading it") + + # Persist only now: the model is on disk and the engine loaded it. + self.config_manager.set_model_size_for_engine(engine, recommendation.model_id) + self.config_manager.save_config() + + shortcut = self.config_manager.get_str( + "shortcuts", "toggle_recognition", DEFAULT_SHORTCUT + ) + _idle_once(notifications.close, progress) + _idle_once( + notifications.notify, + "Speech model ready", + f"{recommendation.display_name} is installed. Press {shortcut} to dictate.", + "emblem-ok", + ) + except Exception as e: + logger.error(f"Could not download {recommendation.model_id}: {e}", exc_info=True) + _idle_once(notifications.close, progress) + _idle_once( + notifications.notify, + "Model download failed", + f"{recommendation.display_name} was not installed: {e}. " + "Check your connection and try again from Settings.", + "dialog-error", + ) + finally: + self.speech_engine.set_download_progress_callback(None) + self.speech_engine.end_download() + self._model_download_active = False + def _update_ui(self, state: RecognitionState): """ Update the UI based on the recognition state. diff --git a/tests/test_model_missing_offer.py b/tests/test_model_missing_offer.py new file mode 100644 index 0000000..316a7d4 --- /dev/null +++ b/tests/test_model_missing_offer.py @@ -0,0 +1,486 @@ +"""Tests for offering the recommended model when dictation finds none.""" + +import importlib +import sys +from unittest.mock import MagicMock, Mock, patch + +import pytest + +import vocalinux.ui + + +def _load(module_name, extra_bases=()): + """Import a GTK module with real base classes so its methods stay callable. + + conftest swaps gi for a MagicMock, which would leave every ``class X(Gtk.Y)`` + as a mock and make its methods unreachable. + + Reimporting rebinds the module both in ``sys.modules`` and as an attribute + on its package. Leaving those two pointing at different objects breaks any + later test that patches the module by name, so the caller restores both; + doing this at collection time is what desynced test_model_deletion in #686. + """ + repository = sys.modules["gi.repository"] + names = ("Box", "ListBoxRow", "Dialog") + tuple(extra_bases) + bases = {name: type(name, (), {}) for name in names} + attribute = module_name.rsplit(".", 1)[1] + saved_module = sys.modules.pop(module_name, None) + saved_attribute = getattr(vocalinux.ui, attribute, None) + try: + with patch.object(repository, "Gtk", MagicMock(**bases)): + yield importlib.import_module(module_name) + finally: + sys.modules.pop(module_name, None) + if saved_module is not None: + sys.modules[module_name] = saved_module + if saved_attribute is not None: + setattr(vocalinux.ui, attribute, saved_attribute) + elif hasattr(vocalinux.ui, attribute): + delattr(vocalinux.ui, attribute) + + +@pytest.fixture(scope="module") +def settings_dialog(): + yield from _load("vocalinux.ui.settings_dialog") + + +@pytest.fixture(scope="module") +def tray_indicator(settings_dialog): + yield from _load("vocalinux.ui.tray_indicator") + + +@pytest.fixture +def TrayIndicator(tray_indicator): + return tray_indicator.TrayIndicator + + +@pytest.fixture +def recommendation(settings_dialog): + return settings_dialog.ModelRecommendation( + model_id="small.en", reason="Vulkan GPU", display_name="Small EN", size_label="466 MB" + ) + + +def _tray_stub(): + tray = Mock() + tray._model_download_active = False + tray.config_manager.get_str.return_value = "en-us" + tray.speech_engine.model_ready = True + return tray + + +class TestRecommendedModelForEngine: + """The offer must name the model the picker stars, not a fixed one.""" + + def test_whisper_cpp_follows_the_hardware_recommendation(self, settings_dialog): + with patch.object( + settings_dialog, + "get_recommended_whispercpp_model", + return_value=("small", "Vulkan GPU"), + ): + result = settings_dialog.recommended_model_for_engine("whisper_cpp", "de") + + assert result.model_id == "small" + assert "Vulkan GPU" in result.reason + assert result.size_label + + def test_english_selection_gets_the_english_only_variant(self, settings_dialog): + with patch.object( + settings_dialog, + "get_recommended_whispercpp_model", + return_value=("small", "Vulkan GPU"), + ): + result = settings_dialog.recommended_model_for_engine("whisper_cpp", "en-us") + + assert result.model_id == "small.en" + + def test_vosk_uses_its_own_recommendation(self, settings_dialog): + with patch.object( + settings_dialog, "_get_recommended_vosk_model", return_value=("medium", "8GB RAM") + ): + result = settings_dialog.recommended_model_for_engine("vosk", "en-us") + + assert result.model_id == "medium" + + def test_remote_api_has_nothing_to_download(self, settings_dialog): + assert settings_dialog.recommended_model_for_engine("remote_api", "auto") is None + + +class TestOffer: + """What the notification carries.""" + + def _show(self, tray_indicator, tray, engine, rec, supports_actions): + """Show the offer notification, return the notifications mock.""" + with patch.object(tray_indicator, "notifications") as mock_notifications: + mock_notifications.supports_actions.return_value = supports_actions + tray_indicator.TrayIndicator._show_model_offer(tray, engine, rec) + return mock_notifications + + def test_offer_is_handed_to_the_main_loop(self, tray_indicator, TrayIndicator, recommendation): + """Dictation can start on the shortcut thread, which must not touch the UI.""" + tray = _tray_stub() + + with patch.object(tray_indicator, "GLib") as glib: + with patch.object( + tray_indicator, "recommended_model_for_engine", return_value=recommendation + ): + handled = TrayIndicator._offer_recommended_model(tray, "whisper_cpp") + + assert handled is True + glib.idle_add.assert_called_once_with(tray._show_model_offer, "whisper_cpp", recommendation) + + def test_offer_carries_a_download_button(self, tray_indicator, recommendation): + tray = _tray_stub() + + mock_notifications = self._show( + tray_indicator, tray, "whisper_cpp", recommendation, supports_actions=True + ) + + action = mock_notifications.notify.call_args.kwargs["action"] + assert action[0] == "Download Small EN (466 MB)" + assert "from this notification" in mock_notifications.notify.call_args.args[1] + + # The button downloads exactly what was offered. + action[1]() + tray._download_recommended_model.assert_called_once_with("whisper_cpp", recommendation) + + def test_offer_without_server_action_support_is_still_shown( + self, tray_indicator, settings_dialog + ): + """No button means the body must not tell the user to click one.""" + tray = _tray_stub() + rec = settings_dialog.ModelRecommendation( + model_id="small", reason="8GB RAM", display_name="Small", size_label="466 MB" + ) + + mock_notifications = self._show(tray_indicator, tray, "vosk", rec, supports_actions=False) + + body = mock_notifications.notify.call_args.args[1] + assert mock_notifications.notify.call_args.kwargs["action"] is None + assert "Settings" in body + assert "from this notification" not in body + + +class _RestartGatedEngine: + """A speech engine that gates re-init the way reconfigure() really does. + + Only engine, model and language changes set restart_needed, and + force_download is read during that re-init — so a caller asking for the + model it is already configured for downloads nothing unless it also asks + for the re-init. + """ + + def __init__(self, engine, model_size, language): + self.engine = engine + self.model_size = model_size + self.language = language + self.model_ready = False + self.reinits = 0 + self.state = None + + def set_download_progress_callback(self, callback): + pass + + def try_begin_download(self): + return True + + def end_download(self): + pass + + def reconfigure( + self, + engine=None, + model_size=None, + language=None, + force_download=True, + force_reinit=False, + **kwargs, + ): + restart_needed = force_reinit + if engine is not None and engine != self.engine: + self.engine = engine + restart_needed = True + if model_size is not None and model_size != self.model_size: + self.model_size = model_size + restart_needed = True + if language is not None and language != self.language: + self.language = language + restart_needed = True + + if restart_needed and force_download: + self.reinits += 1 + self.model_ready = True + + +class TestDownload: + """What the button does once clicked.""" + + def _run(self, tray_indicator, tray, rec, progress=None): + with patch.object(tray_indicator, "notifications"): + with patch.object(tray_indicator, "GLib") as glib: + glib.idle_add.side_effect = lambda func, *args: func(*args) + tray_indicator.TrayIndicator._run_recommended_model_download( + tray, "whisper_cpp", rec, progress or object() + ) + + def test_successful_download_is_saved_for_that_engine( + self, tray_indicator, tray, recommendation + ): + self._run(tray_indicator, tray, recommendation) + + tray.speech_engine.reconfigure.assert_called_once_with( + engine="whisper_cpp", + model_size="small.en", + force_download=True, + force_reinit=True, + ) + tray.config_manager.set_model_size_for_engine.assert_called_once_with( + "whisper_cpp", "small.en" + ) + tray.config_manager.save_config.assert_called_once() + assert tray._model_download_active is False + + def test_the_model_already_configured_is_still_fetched( + self, tray_indicator, tray, recommendation + ): + """The case that breaks the feature: recommendation == configured model. + + First-run defaults are whisper_cpp + tiny, and tiny is exactly what a + CPU-only machine is recommended, so nothing about the request differs + from what the engine already holds. Without asking for the re-init the + click downloads nothing and still reports the model as ready. + """ + tray.speech_engine = _RestartGatedEngine("whisper_cpp", recommendation.model_id, "en-us") + + self._run(tray_indicator, tray, recommendation) + + assert tray.speech_engine.reinits == 1 + assert tray.speech_engine.model_ready is True + tray.config_manager.set_model_size_for_engine.assert_called_once_with( + "whisper_cpp", recommendation.model_id + ) + tray.config_manager.save_config.assert_called_once() + + def test_a_model_that_did_not_load_is_not_reported_as_ready( + self, tray_indicator, tray, recommendation + ): + """reconfigure() returning without raising is not proof of a model.""" + tray.speech_engine.model_ready = False + + with patch.object(tray_indicator, "notifications") as mock_notifications: + with patch.object(tray_indicator, "GLib") as glib: + glib.idle_add.side_effect = lambda func, *args: func(*args) + tray_indicator.TrayIndicator._run_recommended_model_download( + tray, "whisper_cpp", recommendation, object() + ) + + tray.config_manager.set_model_size_for_engine.assert_not_called() + tray.config_manager.save_config.assert_not_called() + titles = [call.args[0] for call in mock_notifications.notify.call_args_list] + assert "Model download failed" in titles + assert "Speech model ready" not in titles + + def test_failed_download_is_not_saved(self, tray_indicator, tray, recommendation): + tray.speech_engine.reconfigure.side_effect = RuntimeError("network down") + + self._run(tray_indicator, tray, recommendation) + + tray.config_manager.set_model_size_for_engine.assert_not_called() + # The engine must not keep reporting progress to a finished download. + tray.speech_engine.set_download_progress_callback.assert_called_with(None) + assert tray._model_download_active is False + + def test_the_first_notification_is_created_on_the_main_loop( + self, tray_indicator, tray, recommendation + ): + """libnotify is not thread-safe, so the worker only gets a handle.""" + with patch.object(tray_indicator, "notifications") as mock_notifications: + with patch.object(tray_indicator.threading, "Thread") as thread: + tray_indicator.TrayIndicator._download_recommended_model( + tray, "whisper_cpp", recommendation + ) + + mock_notifications.notify.assert_called_once() + assert mock_notifications.notify.call_args.args[0] == "Downloading speech model" + handle = mock_notifications.notify.return_value + assert thread.call_args.kwargs["args"] == ("whisper_cpp", recommendation, handle) + + def test_main_loop_callbacks_do_not_ask_to_run_again( + self, tray_indicator, tray, recommendation + ): + """GLib.idle_add repeats a callback that returns truthy, so ours must not. + + The notification helpers return handles and success flags; handing them + to idle_add directly would re-show the notification on every idle cycle. + """ + results = [] + + def fake_reconfigure(**kwargs): + # Progress arrives while the download inside reconfigure() runs. + on_progress = tray.speech_engine.set_download_progress_callback.call_args[0][0] + on_progress(0.5, 10.0, "50%") + + tray.speech_engine.reconfigure.side_effect = fake_reconfigure + + with patch.object(tray_indicator, "notifications") as mock_notifications: + mock_notifications.update.return_value = True + with patch.object(tray_indicator, "GLib") as glib: + glib.idle_add.side_effect = lambda func, *args: results.append(func(*args)) + tray_indicator.TrayIndicator._run_recommended_model_download( + tray, "whisper_cpp", recommendation, object() + ) + + assert results, "expected callbacks scheduled on the main loop" + assert all(not result for result in results) + + def test_a_second_click_does_not_start_a_second_download(self, tray_indicator): + tray = _tray_stub() + tray._model_download_active = True + + with patch.object(tray_indicator.threading, "Thread") as thread: + tray_indicator.TrayIndicator._download_recommended_model(tray, "vosk", Mock()) + + thread.assert_not_called() + + +@pytest.fixture +def tray(): + return _tray_stub() + + +@pytest.mark.parametrize("handler_result", [True, False]) +def test_engine_defers_to_the_registered_handler(handler_result): + """The engine only shows its own notification when nobody handled it.""" + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + manager = Mock() + manager._model_missing_handler = Mock(return_value=handler_result) + manager.engine = "whisper_cpp" + + assert SpeechRecognitionManager._notify_model_missing(manager) is handler_result + manager._model_missing_handler.assert_called_once_with("whisper_cpp") + + +def test_engine_without_a_handler_keeps_its_own_notification(): + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + manager = Mock() + manager._model_missing_handler = None + + assert SpeechRecognitionManager._notify_model_missing(manager) is False + + +def _reconfigure_manager(): + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + manager = MagicMock() + manager.engine = "whisper_cpp" + manager.model_size = "tiny" + manager.language = "en-us" + manager._defer_download = True + return SpeechRecognitionManager, manager + + +def test_force_reinit_reinitializes_an_engine_that_did_not_change(): + """The production contract behind the offer: force_download needs a re-init. + + Downloads only happen while the engine re-initializes, and that is gated on + engine, model or language actually changing. + """ + manager_class, manager = _reconfigure_manager() + seen = {} + manager._init_whispercpp.side_effect = lambda: seen.update(defer=manager._defer_download) + + manager_class.reconfigure( + manager, + engine="whisper_cpp", + model_size="tiny", + language="en-us", + force_download=True, + force_reinit=True, + ) + + manager._init_whispercpp.assert_called_once() + assert seen["defer"] is False + + +def test_an_unchanged_reconfigure_still_does_nothing_by_default(): + """Callers that only tweak VAD must not pay for a re-init.""" + manager_class, manager = _reconfigure_manager() + + manager_class.reconfigure( + manager, + engine="whisper_cpp", + model_size="tiny", + language="en-us", + force_download=True, + ) + + manager._init_whispercpp.assert_not_called() + + +class TestOneDownloadAtATime: + """The tray and Settings share one engine, so only one of them downloads.""" + + def _run(self, tray_indicator, tray, rec): + with patch.object(tray_indicator, "notifications"): + with patch.object(tray_indicator, "GLib") as glib: + glib.idle_add.side_effect = lambda func, *args: func(*args) + tray_indicator.TrayIndicator._run_recommended_model_download( + tray, "whisper_cpp", rec, object() + ) + + def test_tray_refuses_when_another_download_holds_the_engine(self, tray_indicator): + tray = _tray_stub() + tray.speech_engine.try_begin_download.return_value = False + + with patch.object(tray_indicator, "notifications") as mock_notifications: + with patch.object(tray_indicator.threading, "Thread") as thread: + tray_indicator.TrayIndicator._download_recommended_model(tray, "vosk", Mock()) + + thread.assert_not_called() + assert tray._model_download_active is False + assert mock_notifications.notify.call_args.args[0] == "Download already in progress" + + def test_tray_releases_the_engine_when_the_download_ends( + self, tray_indicator, tray, recommendation + ): + self._run(tray_indicator, tray, recommendation) + + tray.speech_engine.end_download.assert_called_once_with() + + def test_tray_releases_the_engine_when_the_download_fails( + self, tray_indicator, tray, recommendation + ): + tray.speech_engine.reconfigure.side_effect = RuntimeError("boom") + + self._run(tray_indicator, tray, recommendation) + + tray.speech_engine.end_download.assert_called_once_with() + + def test_download_claim_is_exclusive_until_released(self): + import threading + + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + manager = MagicMock() + manager._download_claim = threading.Lock() + + assert SpeechRecognitionManager.try_begin_download(manager) is True + assert SpeechRecognitionManager.try_begin_download(manager) is False + assert SpeechRecognitionManager.download_in_progress.fget(manager) is True + + SpeechRecognitionManager.end_download(manager) + + assert SpeechRecognitionManager.download_in_progress.fget(manager) is False + assert SpeechRecognitionManager.try_begin_download(manager) is True + + def test_releasing_an_unclaimed_engine_is_harmless(self): + import threading + + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + manager = MagicMock() + manager._download_claim = threading.Lock() + + SpeechRecognitionManager.end_download(manager) # must not raise diff --git a/tests/test_notifications.py b/tests/test_notifications.py new file mode 100644 index 0000000..5fd6977 --- /dev/null +++ b/tests/test_notifications.py @@ -0,0 +1,152 @@ +"""Tests for vocalinux.ui.notifications against a fake libnotify. + +Every other test patches the whole ``notifications`` module away, so the +libnotify/notify-send boundary itself was never exercised. These drive the +module directly: ``conftest`` already installs ``gi.repository`` as a +MagicMock, so the fake ``Notify`` is steered through that. +""" + +import sys +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture +def notifications(): + """The real module with its lazy state reset, restored afterwards.""" + from vocalinux.ui import notifications as module + + saved = (module._notify_module, module._notify_initialized, set(module._pending)) + module._notify_module = None + module._notify_initialized = False + module._pending.clear() + yield module + module._notify_module, module._notify_initialized = saved[0], saved[1] + module._pending.clear() + module._pending.update(saved[2]) + + +@pytest.fixture +def fake_notify(notifications): + """A libnotify that advertises action buttons, handed to the module directly. + + Installed through the module's own lazy cache rather than by patching + ``gi.repository.Notify``: that global is shared with every other test in + the process, and background threads left behind by earlier tests (the + single-instance checks in test_main, for one) keep calling into it. A fake + that only this module can reach is immune to them. + """ + notify = MagicMock() + notify.get_server_caps.return_value = ["body", "actions"] + notifications._notify_module = notify + notifications._notify_initialized = True + return notify + + +@pytest.fixture +def notify_send(notifications): + with patch.object(notifications.subprocess, "Popen") as popen: + yield popen + + +def _registered_callback(notification, signal_name): + """The callback the module connected to ``signal_name`` on the fake.""" + for call in notification.connect.call_args_list: + if call.args[0] == signal_name: + return call.args[1] + raise AssertionError(f"nothing connected to {signal_name!r}") + + +class TestFallbackToNotifySend: + def test_action_without_server_support_falls_back_to_notify_send( + self, notifications, fake_notify, notify_send + ): + fake_notify.get_server_caps.return_value = ["body"] + + handle = notifications.notify( + "Title", "Body", "dialog-warning", action=("Go", lambda: None) + ) + + assert handle is None + notify_send.assert_called_once() + argv = notify_send.call_args.args[0] + assert argv[0] == "notify-send" + assert argv[-2:] == ["Title", "Body"] + + def test_failed_libnotify_init_falls_back(self, notifications, notify_send): + broken = MagicMock() + broken.is_initted.return_value = False + broken.init.return_value = False + with patch.object(sys.modules["gi.repository"], "Notify", broken): + handle = notifications.notify("Title", "Body") + + assert handle is None + assert notifications._notify_module is None + notify_send.assert_called_once() + + def test_plain_notification_needs_no_action_support( + self, notifications, fake_notify, notify_send + ): + fake_notify.get_server_caps.return_value = ["body"] + + handle = notifications.notify("Title", "Body") + + assert handle is fake_notify.Notification.new.return_value + notify_send.assert_not_called() + + +class TestPendingHandles: + def test_action_keeps_the_notification_alive_until_answered(self, notifications, fake_notify): + handle = notifications.notify("Title", "Body", action=("Go", lambda: None)) + + assert handle in notifications._pending + + _registered_callback(handle, "closed")(handle) + + assert handle not in notifications._pending + + def test_action_callback_runs_and_releases_the_handle(self, notifications, fake_notify): + callback = MagicMock() + handle = notifications.notify("Title", "Body", action=("Go", callback)) + on_action = handle.add_action.call_args.args[2] + + on_action(handle, "default-action", None) + + callback.assert_called_once_with() + assert handle not in notifications._pending + + def test_a_failing_action_callback_is_logged_not_raised(self, notifications, fake_notify): + handle = notifications.notify( + "Title", "Body", action=("Go", MagicMock(side_effect=RuntimeError("boom"))) + ) + on_action = handle.add_action.call_args.args[2] + + on_action(handle, "default-action", None) # must not raise + + assert handle not in notifications._pending + + def test_close_releases_the_handle(self, notifications, fake_notify): + handle = notifications.notify("Title", "Body", action=("Go", lambda: None)) + + notifications.close(handle) + + assert handle not in notifications._pending + handle.close.assert_called_once_with() + + +class TestUpdate: + def test_update_returns_false_for_fallback_handles(self, notifications): + assert notifications.update(None, "Title", "Body") is False + + def test_update_refreshes_a_real_handle_in_place(self, notifications, fake_notify): + # A handle of our own: the fake's auto-created children are not safe to + # count on, since other tests leave calls behind on shared mocks. + handle = MagicMock() + fake_notify.Notification.new.return_value = handle + assert notifications.notify("Title", "Body") is handle + handle.show.assert_called_once_with() + + assert notifications.update(handle, "New", "Text", "folder-download") is True + handle.update.assert_called_once_with("New", "Text", "folder-download") + assert handle.show.call_count == 2 diff --git a/tests/test_settings_model_state.py b/tests/test_settings_model_state.py index 3257820..70c406d 100644 --- a/tests/test_settings_model_state.py +++ b/tests/test_settings_model_state.py @@ -286,3 +286,57 @@ def test_closing_the_dialog_resyncs_an_engine_that_was_never_applied(settings_di dialog_class._on_settings_dialog_response(dialog, dialog, gtk.ResponseType.CLOSE) dialog._resync_engine_ui_if_unapplied.assert_called_once() + + +def _download_setup(dialog): + """A model that is not on disk, so the apply goes down the download path.""" + dialog.get_selected_settings.return_value = { + "engine": "whisper_cpp", + "model_size": "small", + "language": "auto", + } + dialog._apply_settings_internal.return_value = True + + +@pytest.mark.parametrize("entry", ["_auto_apply_settings", "apply_settings"]) +def test_settings_refuses_a_download_while_the_tray_holds_the_engine( + settings_dialog, dialog_class, entry +): + """Both ways into a download stop at the engine's claim. + + The tray downloads in the background too; a second download would fight it + over the one progress callback and the one engine configuration. + """ + dialog = _dialog_stub() + _download_setup(dialog) + dialog.speech_engine.try_begin_download.return_value = False + with ( + patch.object(settings_dialog, "is_whispercpp_model_downloaded", return_value=False), + patch.object(settings_dialog, "ModelDownloadDialog") as modal_class, + patch.object(settings_dialog, "GLib", MagicMock()), + patch.object(settings_dialog.threading, "Thread", _InlineThread), + ): + getattr(dialog_class, entry)(dialog) + + 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() + + +@pytest.mark.parametrize("entry", ["_auto_apply_settings", "apply_settings"]) +def test_settings_releases_the_engine_once_the_download_is_over( + settings_dialog, dialog_class, entry +): + dialog = _dialog_stub() + _download_setup(dialog) + with ( + patch.object(settings_dialog, "is_whispercpp_model_downloaded", return_value=False), + patch.object(settings_dialog, "ModelDownloadDialog"), + patch.object(settings_dialog, "GLib", MagicMock()), + patch.object(settings_dialog.threading, "Thread", _InlineThread), + ): + getattr(dialog_class, entry)(dialog) + + dialog.speech_engine.try_begin_download.assert_called_once_with() + dialog.speech_engine.end_download.assert_called_once_with()