From 4c070ded463ae0732c8bfd977abdebcbc4dc01ab Mon Sep 17 00:00:00 2001 From: Jatin K Malik Date: Sun, 26 Jul 2026 22:29:00 -0700 Subject: [PATCH] feat(offload): auto-pause apps + idle keep-alive model unload (#592) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(auto-pause): unload speech model while selected apps run Add Settings toggle + process list so Vocalinux frees CPU/GPU/RAM while configured games/apps are running, then reloads the model when they exit. Closes the process-list half of #445 (window-focus pause is separate). * test(auto-pause): isolate tray wiring to avoid stale GLib mocks Move TrayIndicator auto-pause tests into test_suspend_handler so test_auto_pause_monitor never imports tray_indicator under a collection-time gi MagicMock. That import pinned GLib.SOURCE_REMOVE and broke resume tests later in the suite (CI on #517). * test(auto-pause): raise patch coverage for codecov target Cover process snapshot I/O, GLib poll scheduling, config coercion, unload edge cases, and tray fallback paths so codecov/patch clears 80%. * feat(keepalive): idle model offload for GPU/battery saving Add opt-in keep-alive that unloads the speech model after inactivity so hybrid-GPU laptops can release VRAM/Vulkan. Next dictation lazy-reloads. Shares unload_model with auto-pause (#445) using distinct reasons so app-pause still blocks dictation while idle unload does not. Fixes #591 (with #445 via shared offload infrastructure). * test(keepalive): mock gi.repository.GLib via sys.modules Match the auto-pause GLib test pattern so CI on Python 3.9/3.10 sees the late `from gi.repository import GLib` import. * test(keepalive): drop post-fire source_remove assertion After the GLib timeout callback runs, _timeout_id is already cleared, so shutdown does not call source_remove. * fix(settings): hide auto-pause empty-state label when apps are listed The dialog-wide show_all() after _load_and_apply_settings() re-showed the 'No apps listed yet' row that _refresh_auto_pause_list had hidden. Mark the row no_show_all and manage visibility explicitly. Co-authored-by: Jatin K Malik * feat(settings): move auto-pause and idle unload to dedicated Power tab - New Power tab between Shortcuts and General so resource management features are easy to find; General returns to app-behavior toggles only - Rename 'Model Keep-Alive (Battery)' to 'Unload When Idle' — the old name read as the opposite of what the feature does - Clearer copy throughout: row titles, descriptions, tooltips, 'From running…' becomes 'Choose Running App…', empty-state hint - Gray out app-list and idle-timeout controls while their feature toggle is off Co-authored-by: Jatin K Malik * fix(settings): show which notebook tab is selected The tab CSS targeted :selected, but GTK3 marks the active notebook tab with :checked, so no selected styling ever applied — all tabs rendered identically. Use :checked and add an accent underline + bold label. Co-authored-by: Jatin K Malik --------- Co-authored-by: Cursor Agent Co-authored-by: Jatin K Malik --- src/vocalinux/auto_pause_monitor.py | 255 +++++++ src/vocalinux/model_keepalive.py | 185 +++++ .../speech_recognition/recognition_manager.py | 125 +++- src/vocalinux/ui/config_manager.py | 13 + src/vocalinux/ui/settings_dialog.py | 388 +++++++++- src/vocalinux/ui/tray_indicator.py | 113 ++- tests/test_auto_pause_monitor.py | 696 ++++++++++++++++++ tests/test_model_keepalive.py | 156 ++++ tests/test_suspend_handler.py | 186 +++++ 9 files changed, 2107 insertions(+), 10 deletions(-) create mode 100644 src/vocalinux/auto_pause_monitor.py create mode 100644 src/vocalinux/model_keepalive.py create mode 100644 tests/test_auto_pause_monitor.py create mode 100644 tests/test_model_keepalive.py diff --git a/src/vocalinux/auto_pause_monitor.py b/src/vocalinux/auto_pause_monitor.py new file mode 100644 index 0000000..baced9a --- /dev/null +++ b/src/vocalinux/auto_pause_monitor.py @@ -0,0 +1,255 @@ +""" +Auto-pause monitor: unload speech model while selected apps/games are running. + +Polls the process table for configured executable basenames. When any match +is found, invokes an on_pause callback (caller should stop dictation and +unload the model). When no matches remain, invokes on_resume (reload model). + +Matching is intentionally pure and unit-testable; I/O lives in helpers used +by the poller. +""" + +from __future__ import annotations + +import logging +import os +from typing import Callable, Iterable, Optional, Sequence + +logger = logging.getLogger(__name__) + +# Default poll interval when config does not specify one +DEFAULT_POLL_INTERVAL_SECONDS = 5 + +# Reasonable bounds for poll interval (seconds) +_MIN_POLL_INTERVAL = 1 +_MAX_POLL_INTERVAL = 60 + + +def normalize_process_name(name: str) -> str: + """Normalize a process or configured app name for comparison. + + Uses lowercase basename and strips a trailing ``.exe`` so wine/Proton + games match the same configured name as native processes. + """ + if not name: + return "" + base = os.path.basename(name.strip()).lower() + if base.endswith(".exe"): + base = base[:-4] + return base + + +def configured_names_set(apps: Sequence[str]) -> set[str]: + """Return the set of normalized configured process names (empty entries dropped).""" + return {normalize_process_name(a) for a in apps if a and str(a).strip()} + + +def any_configured_process_running( + configured_apps: Sequence[str], + running_names: Iterable[str], +) -> bool: + """Return True if any configured app name matches a running process name. + + Matching is case-insensitive basename equality after ``normalize_process_name``. + An empty configured list never matches. + """ + targets = configured_names_set(configured_apps) + if not targets: + return False + + for raw in running_names: + if normalize_process_name(raw) in targets: + return True + return False + + +def collect_running_process_names() -> set[str]: + """Snapshot names and exe basenames of currently running processes via psutil.""" + import psutil + + names: set[str] = set() + for proc in psutil.process_iter(["name", "exe"]): + try: + info = proc.info + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + name = info.get("name") + if name: + names.add(name) + exe = info.get("exe") + if exe: + try: + names.add(os.path.basename(exe)) + except Exception: + pass + return names + + +class AutoPauseMonitor: + """Polls for configured processes and fires pause/resume callbacks. + + Designed like :class:`~vocalinux.suspend_handler.SuspendHandler`: the tray + owns lifecycle (start with the app, stop on quit). Callbacks run on the + GLib main loop when a timeout source is used. + + Args: + get_config: Callable returning ``(enabled, apps, poll_interval_seconds)``. + Re-read every poll so Settings changes apply without restart. + on_pause: Called once when the match set becomes non-empty. + on_resume: Called once when the match set becomes empty after a pause. + process_snapshot: Optional override for :func:`collect_running_process_names` + (used in tests). + use_glib: If True (default), schedule polls with GLib.timeout_add_seconds. + If False, only ``check_once`` / manual ``poll`` is available (tests). + """ + + def __init__( + self, + get_config: Callable[[], tuple[bool, Sequence[str], float]], + on_pause: Optional[Callable[[], None]] = None, + on_resume: Optional[Callable[[], None]] = None, + process_snapshot: Optional[Callable[[], Iterable[str]]] = None, + use_glib: bool = True, + ): + self._get_config = get_config + self._on_pause = on_pause + self._on_resume = on_resume + self._process_snapshot = process_snapshot or collect_running_process_names + self._use_glib = use_glib + self._paused = False + self._timeout_id: Optional[int] = None + self._running = False + self._last_poll_interval = DEFAULT_POLL_INTERVAL_SECONDS + + @property + def paused(self) -> bool: + """True while a configured process is considered active (after on_pause).""" + return self._paused + + @property + def active(self) -> bool: + """True if the poller is scheduled / running.""" + return self._running + + def start(self) -> None: + """Start periodic polling. Safe to call if already started.""" + if self._running: + return + self._running = True + if self._use_glib: + self._schedule_next_poll(immediate=True) + logger.info("Auto-pause monitor started") + + def stop(self) -> None: + """Stop polling and clear the GLib timeout source.""" + self._running = False + self._cancel_timeout() + logger.info("Auto-pause monitor stopped") + + def shutdown(self) -> None: + """Alias for :meth:`stop` (mirrors SuspendHandler.shutdown).""" + self.stop() + + def check_once(self) -> bool: + """Run a single poll cycle. Returns whether a configured process matched. + + Safe for unit tests without GLib. Updates paused state and may invoke + on_pause / on_resume. + """ + enabled, apps, _interval = self._read_config() + if not enabled: + self._clear_pause_if_needed() + return False + + running = self._process_snapshot() + matched = any_configured_process_running(apps, running) + if matched: + self._enter_pause_if_needed() + else: + self._clear_pause_if_needed() + return matched + + def _read_config(self) -> tuple[bool, Sequence[str], float]: + try: + enabled, apps, interval = self._get_config() + except Exception: + logger.error("Failed to read auto-pause config", exc_info=True) + return False, [], DEFAULT_POLL_INTERVAL_SECONDS + + if not isinstance(apps, (list, tuple)): + apps = [] + try: + interval_f = float(interval) + except (TypeError, ValueError): + interval_f = DEFAULT_POLL_INTERVAL_SECONDS + interval_f = max(_MIN_POLL_INTERVAL, min(_MAX_POLL_INTERVAL, interval_f)) + return bool(enabled), apps, interval_f + + def _enter_pause_if_needed(self) -> None: + if self._paused: + return + self._paused = True + logger.info("Auto-pause: configured app detected — pausing / unloading model") + if self._on_pause: + try: + self._on_pause() + except Exception: + logger.error("Error in auto-pause on_pause callback", exc_info=True) + + def _clear_pause_if_needed(self) -> None: + if not self._paused: + return + self._paused = False + logger.info("Auto-pause: no configured apps running — resuming / reloading model") + if self._on_resume: + try: + self._on_resume() + except Exception: + logger.error("Error in auto-pause on_resume callback", exc_info=True) + + def _schedule_next_poll(self, immediate: bool = False) -> None: + if not self._running or not self._use_glib: + return + self._cancel_timeout() + _, _, interval = self._read_config() + self._last_poll_interval = interval + try: + from gi.repository import GLib + except Exception: + logger.warning("GLib unavailable; auto-pause poller not scheduled") + return + + delay = 0 if immediate else int(interval) + # GLib.timeout_add_seconds requires >= 1 for seconds API when non-zero; + # use timeout_add(0) equivalent via idle for immediate first check. + if delay <= 0: + self._timeout_id = GLib.idle_add(self._glib_poll) + else: + self._timeout_id = GLib.timeout_add_seconds(delay, self._glib_poll) + + def _cancel_timeout(self) -> None: + if self._timeout_id is None: + return + try: + from gi.repository import GLib + + GLib.source_remove(self._timeout_id) + except Exception: + pass + self._timeout_id = None + + def _glib_poll(self) -> bool: + """GLib source callback: poll once then reschedule with current interval.""" + self._timeout_id = None + if not self._running: + return False # SOURCE_REMOVE + + try: + self.check_once() + except Exception: + logger.error("Auto-pause poll failed", exc_info=True) + + # Reschedule so interval changes from Settings take effect + if self._running: + self._schedule_next_poll(immediate=False) + return False # always remove this source; we reschedule explicitly diff --git a/src/vocalinux/model_keepalive.py b/src/vocalinux/model_keepalive.py new file mode 100644 index 0000000..005c2b6 --- /dev/null +++ b/src/vocalinux/model_keepalive.py @@ -0,0 +1,185 @@ +""" +Model keep-alive: unload the speech model after a configurable idle timeout. + +Mirrors :class:`~vocalinux.auto_pause_monitor.AutoPauseMonitor` / SuspendHandler +style — the tray owns lifecycle, callbacks run on the GLib main loop when used +with ``use_glib=True``. Designed so unit tests can drive the timer without GTK. +""" + +from __future__ import annotations + +import logging +from typing import Callable, Optional + +logger = logging.getLogger(__name__) + +# Default idle timeout when keep-alive is enabled but config omits a value +DEFAULT_IDLE_TIMEOUT_SECONDS = 300 + +# Reasonable bounds when enabled (seconds) +_MIN_IDLE_TIMEOUT = 60 +_MAX_IDLE_TIMEOUT = 3600 + + +def clamp_idle_timeout(seconds: float) -> float: + """Clamp an idle timeout to the supported range.""" + try: + value = float(seconds) + except (TypeError, ValueError): + return float(DEFAULT_IDLE_TIMEOUT_SECONDS) + return max(_MIN_IDLE_TIMEOUT, min(_MAX_IDLE_TIMEOUT, value)) + + +class ModelKeepAlive: + """Schedules idle unload of the speech model after inactivity. + + Args: + get_config: Callable returning ``(enabled, idle_timeout_seconds)``. + Re-read on each bump so Settings changes apply without restart. + on_idle_unload: Called once when the idle timeout elapses. + is_safe_to_unload: Optional predicate; unload only runs when this + returns True (default: always). Use to require IDLE + not auto-paused. + use_glib: If True (default), schedule with GLib.timeout_add_seconds. + If False, only ``fire_if_due`` / manual ``tick`` is available (tests). + """ + + def __init__( + self, + get_config: Callable[[], tuple[bool, float]], + on_idle_unload: Optional[Callable[[], None]] = None, + is_safe_to_unload: Optional[Callable[[], bool]] = None, + use_glib: bool = True, + ): + self._get_config = get_config + self._on_idle_unload = on_idle_unload + self._is_safe_to_unload = is_safe_to_unload + self._use_glib = use_glib + self._timeout_id: Optional[int] = None + self._running = False + self._armed = False + + @property + def active(self) -> bool: + """True if the keep-alive helper has been started.""" + return self._running + + @property + def armed(self) -> bool: + """True while an idle timer is scheduled.""" + return self._armed + + def start(self) -> None: + """Mark the helper as running. Safe to call if already started.""" + if self._running: + return + self._running = True + logger.info("Model keep-alive started") + + def stop(self) -> None: + """Cancel any pending timer and mark stopped.""" + self._running = False + self.cancel() + logger.info("Model keep-alive stopped") + + def shutdown(self) -> None: + """Alias for :meth:`stop` (mirrors SuspendHandler.shutdown).""" + self.stop() + + def bump(self) -> None: + """Reset the idle deadline from now (call when recognition becomes IDLE).""" + if not self._running: + return + + enabled, timeout = self._read_config() + if not enabled: + self.cancel() + return + + self._schedule(timeout) + + def cancel(self) -> None: + """Cancel any pending idle unload (call while LISTENING/PROCESSING).""" + self._cancel_timeout() + self._armed = False + + def fire_if_due(self) -> bool: + """Manually run the idle unload path (tests / non-GLib). Returns True if fired.""" + if not self._running or not self._armed: + return False + return self._fire_unload() + + def _read_config(self) -> tuple[bool, float]: + try: + enabled, timeout = self._get_config() + except Exception: + logger.error("Failed to read model keep-alive config", exc_info=True) + return False, float(DEFAULT_IDLE_TIMEOUT_SECONDS) + + timeout_f = clamp_idle_timeout(timeout) + return bool(enabled), timeout_f + + def _schedule(self, timeout_seconds: float) -> None: + self._cancel_timeout() + self._armed = True + + if not self._use_glib: + # Tests drive fire_if_due explicitly; still mark armed. + return + + try: + from gi.repository import GLib + except Exception: + logger.warning("GLib unavailable; model keep-alive timer not scheduled") + self._armed = False + return + + delay = max(1, int(timeout_seconds)) + self._timeout_id = GLib.timeout_add_seconds(delay, self._glib_timeout) + + def _cancel_timeout(self) -> None: + if self._timeout_id is None: + return + try: + from gi.repository import GLib + + GLib.source_remove(self._timeout_id) + except Exception: + pass + self._timeout_id = None + + def _glib_timeout(self) -> bool: + """GLib source callback: attempt idle unload once.""" + self._timeout_id = None + self._fire_unload() + return False # SOURCE_REMOVE — bump() reschedules after next activity + + def _fire_unload(self) -> bool: + if not self._running: + self._armed = False + return False + + enabled, _timeout = self._read_config() + if not enabled: + self._armed = False + return False + + if self._is_safe_to_unload is not None: + try: + if not self._is_safe_to_unload(): + logger.debug("Keep-alive unload skipped: not safe to unload") + self._armed = False + return False + except Exception: + logger.error("Keep-alive is_safe_to_unload failed", exc_info=True) + self._armed = False + return False + + self._armed = False + logger.info("Model keep-alive idle timeout reached — unloading model") + if self._on_idle_unload: + try: + self._on_idle_unload() + except Exception: + logger.error("Error in keep-alive on_idle_unload callback", exc_info=True) + return False + return True diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 26cb2ed..cf34b41 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -816,6 +816,10 @@ class SpeechRecognitionManager: self._download_cancelled = False self._defer_download = defer_download self._model_initialized = False + # True while auto-pause has unloaded the model for a configured app/game + self._auto_paused = False + # True while idle keep-alive has unloaded the model (dictation may lazy-reload) + self._idle_unloaded = False # Speech detection parameters (load defaults, will be overridden by configure) self.vad_sensitivity = kwargs.get("vad_sensitivity", 3) @@ -2282,19 +2286,46 @@ class SpeechRecognitionManager: logger.warning(f"Cannot start recognition in current state: {self.state}") return - # Check if model is ready - if not self.model_ready: + if self._auto_paused: logger.warning( - "Cannot start recognition: model not downloaded. " "Please download via Settings." + "Cannot start recognition: auto-paused while a configured app/game is running" ) play_error_sound() _show_notification( - "No Speech Model", - "Please open Settings and download a speech recognition model " "to use dictation.", - "dialog-warning", + "Dictation Paused", + "Vocalinux is paused because a listed game or app is running. " + "Close that app or remove it from Auto-Pause settings to resume.", + "dialog-information", ) return + # Check if model is ready (lazy-reload after idle keep-alive unload) + if not self.model_ready: + if self._idle_unloaded: + logger.info("Model was unloaded by keep-alive; reloading before dictation") + if not self.ensure_model_loaded(): + play_error_sound() + _show_notification( + "Model Reload Failed", + "Vocalinux could not reload the speech model. " + "Open Settings to check your engine and try again.", + "dialog-warning", + ) + return + else: + logger.warning( + "Cannot start recognition: model not downloaded. " + "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", + ) + return + logger.info("Starting speech recognition") self._update_state(RecognitionState.LISTENING) @@ -3109,12 +3140,92 @@ class SpeechRecognitionManager: logger.error(f"Unexpected error during audio reconnection: {e}") return False + def unload_model(self, reason: str = "auto_pause") -> None: + """Stop recognition (if active) and release model resources from memory. + + Args: + reason: Why the model is being unloaded: + - ``"auto_pause"``: a configured game/app is running; dictation is blocked + until :meth:`reinitialize_after_resume` (or auto-pause resume). + - ``"idle_keepalive"``: idle keep-alive timeout; next dictation may + lazy-reload via :meth:`ensure_model_loaded`. + - ``"manual"`` / other: unload only without setting pause/idle flags. + """ + logger.info("Unloading speech model (reason=%s)", reason) + + if self.state != RecognitionState.IDLE: + logger.info("Stopping active recognition before model unload") + self.stop_recognition() + + with self._model_lock: + self.model = None + self.recognizer = None + if self._http_session is not None: + try: + self._http_session.close() + except Exception: + logger.debug("Error closing HTTP session during unload", exc_info=True) + self._http_session = None + self._model_initialized = False + + if reason == "auto_pause": + self._auto_paused = True + self._idle_unloaded = False + logger.info("Speech model unloaded; dictation blocked until auto-pause clears") + elif reason == "idle_keepalive": + # Auto-pause owns lifecycle if both could apply + if not self._auto_paused: + self._idle_unloaded = True + logger.info("Speech model unloaded after idle keep-alive timeout") + else: + logger.info("Speech model unloaded") + + try: + import gc + + gc.collect() + except Exception: + pass + + @property + def is_auto_paused(self) -> bool: + """True when the model was unloaded for auto-pause and not yet reloaded.""" + return bool(self._auto_paused) + + @property + def is_idle_unloaded(self) -> bool: + """True when the model was unloaded by idle keep-alive and not yet reloaded.""" + return bool(self._idle_unloaded) + + def ensure_model_loaded(self) -> bool: + """Ensure the speech model is loaded and ready for dictation. + + Returns: + True if the model is ready after this call. Returns False when + auto-pause still owns the unload, or when reload fails. + """ + if self.model_ready: + return True + if self._auto_paused: + logger.warning("Cannot load model: auto-pause is still active") + return False + + try: + self.reinitialize_after_resume() + except Exception: + logger.error("ensure_model_loaded failed", exc_info=True) + return False + + return self.model_ready + def reinitialize_after_resume(self): """Reinitialize the speech engine after system resume from suspend. Stops any active recognition, releases stale model resources, and re-creates the engine so that the audio pipeline and model are in a clean state for new dictation. + + Also used after auto-pause / idle keep-alive to reload the model. """ logger.info("Reinitializing speech engine after system resume") @@ -3143,6 +3254,8 @@ class SpeechRecognitionManager: logger.error("Cannot reinitialize: unknown engine '%s'", self.engine) return + self._auto_paused = False + self._idle_unloaded = False logger.info("Speech engine reinitialized after resume") except Exception: logger.error("Failed to reinitialize speech engine after resume", exc_info=True) diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index fa44f02..0cbb208 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -59,6 +59,19 @@ DEFAULT_CONFIG = { "autostart": False, "first_run": True, }, + "auto_pause": { + # When enabled, unload the speech model while any listed process is running + # so games/apps can use full CPU/GPU/RAM. Model reloads when they exit. + "enabled": False, + "apps": [], # Process/executable basenames, e.g. ["overwatch", "steam"] + "poll_interval_seconds": 5, # How often to scan running processes + }, + "model_keepalive": { + # Opt-in idle unload so hybrid-GPU laptops can release VRAM/Vulkan and + # let the dGPU sleep. Next dictation lazy-reloads the model (cold start). + "enabled": False, + "idle_timeout_seconds": 300, # 5 minutes when enabled + }, "text_injection": { "copy_to_clipboard": False, # Disabled by default; users can enable in Settings }, diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index d95b84a..0feae7e 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -346,9 +346,11 @@ SETTINGS_CSS = """ background-color: alpha(@theme_selected_bg_color, 0.2); } -.notebook tab:selected { +/* GTK3 marks the selected notebook tab with :checked (not :selected) */ +.notebook tab:checked { background-color: transparent; color: @theme_fg_color; + box-shadow: inset 0 -3px @theme_selected_bg_color; } .notebook tab label { @@ -356,8 +358,8 @@ SETTINGS_CSS = """ font-size: 0.95em; } -.notebook tab:selected label { - font-weight: 600; +.notebook tab:checked label { + font-weight: 700; } .notebook header { @@ -1077,6 +1079,12 @@ class SettingsDialog(Gtk.Dialog): self.shortcuts_tab.set_margin_start(16) self.shortcuts_tab.set_margin_end(16) + self.power_tab = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12) + self.power_tab.set_margin_top(16) + self.power_tab.set_margin_bottom(16) + self.power_tab.set_margin_start(16) + self.power_tab.set_margin_end(16) + self.general_tab = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12) self.general_tab.set_margin_top(16) self.general_tab.set_margin_bottom(16) @@ -1120,6 +1128,11 @@ class SettingsDialog(Gtk.Dialog): shortcuts_label.set_tooltip_text("Keyboard shortcuts") notebook.append_page(_scrollable(self.shortcuts_tab), shortcuts_label) + # Power tab - resource management (auto-pause, idle unload) + power_label = Gtk.Label(label="Power") + power_label.set_tooltip_text("Free CPU, GPU, and memory while gaming or idle") + notebook.append_page(_scrollable(self.power_tab), power_label) + # General tab - least important (application behavior) general_label = Gtk.Label(label="General") general_label.set_tooltip_text("General settings") @@ -1142,6 +1155,8 @@ class SettingsDialog(Gtk.Dialog): # Build UI sections into appropriate tabs self._build_general_section() + self._build_auto_pause_section() + self._build_model_keepalive_section() self._build_audio_section() self._build_engine_section() self._build_remote_server_section() @@ -1287,6 +1302,359 @@ class SettingsDialog(Gtk.Dialog): self.start_minimized_switch.connect("state-set", self._on_start_minimized_toggled) self.copy_to_clipboard_switch.connect("state-set", self._on_copy_to_clipboard_toggled) + def _build_auto_pause_section(self): + """Build Auto-Pause settings: enable toggle + process name list.""" + group = PreferencesGroup( + title="Auto-Pause for Games & Apps", + description=( + "While any app in the list is running, Vocalinux pauses dictation and " + "unloads the speech model to free CPU, GPU, and memory. Everything " + "resumes automatically when the app closes." + ), + ) + + self.auto_pause_switch = Gtk.Switch() + self.auto_pause_switch.set_tooltip_text( + "Pause dictation and unload the speech model while any listed app is running" + ) + enable_row = PreferenceRow( + title="Pause for Listed Apps", + subtitle="Free resources while any app below is running", + widget=self.auto_pause_switch, + ) + group.add_row(enable_row) + + # Add process name: entry + Add button + add_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + add_box.set_margin_top(8) + add_box.set_margin_bottom(4) + add_box.set_margin_start(16) + add_box.set_margin_end(16) + + self.auto_pause_entry = Gtk.Entry() + self.auto_pause_entry.set_placeholder_text("Process name (e.g. overwatch, steam)") + self.auto_pause_entry.set_hexpand(True) + self.auto_pause_entry.set_tooltip_text( + "Process name, matched case-insensitively. " + "For Windows games under Wine/Proton, the .exe suffix is optional." + ) + add_box.pack_start(self.auto_pause_entry, True, True, 0) + + self.auto_pause_add_btn = Gtk.Button(label="Add") + self.auto_pause_add_btn.set_tooltip_text("Add this process name to the list") + self.auto_pause_add_btn.connect("clicked", self._on_auto_pause_add_clicked) + self.auto_pause_entry.connect("activate", self._on_auto_pause_add_clicked) + add_box.pack_start(self.auto_pause_add_btn, False, False, 0) + + self.auto_pause_pick_btn = Gtk.Button(label="Choose Running App…") + self.auto_pause_pick_btn.set_tooltip_text("Pick from apps that are running right now") + self.auto_pause_pick_btn.connect("clicked", self._on_auto_pause_pick_running) + add_box.pack_start(self.auto_pause_pick_btn, False, False, 0) + + # Custom row for the add controls (not a PreferenceRow) + add_row = Gtk.ListBoxRow() + add_row.set_activatable(False) + add_row.add(add_box) + group.add_row(add_row) + + # List of configured apps + self.auto_pause_listbox = Gtk.ListBox() + self.auto_pause_listbox.set_selection_mode(Gtk.SelectionMode.NONE) + self.auto_pause_listbox.set_activate_on_single_click(False) + + list_scrolled = Gtk.ScrolledWindow() + list_scrolled.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) + list_scrolled.set_min_content_height(80) + list_scrolled.set_max_content_height(160) + list_scrolled.set_margin_start(8) + list_scrolled.set_margin_end(8) + list_scrolled.set_margin_bottom(8) + list_scrolled.add(self.auto_pause_listbox) + + list_row = Gtk.ListBoxRow() + list_row.set_activatable(False) + list_row.add(list_scrolled) + group.add_row(list_row) + + self.auto_pause_empty_label = Gtk.Label( + label="No apps in the list yet. Type a process name or choose a running app above.", + xalign=0, + ) + self.auto_pause_empty_label.get_style_context().add_class("preference-row-subtitle") + self.auto_pause_empty_label.set_margin_start(16) + self.auto_pause_empty_label.set_margin_end(16) + self.auto_pause_empty_label.set_margin_bottom(12) + + empty_row = Gtk.ListBoxRow() + empty_row.set_activatable(False) + empty_row.add(self.auto_pause_empty_label) + group.add_row(empty_row) + # Keep the dialog-wide show_all() from re-showing this row when the + # list is non-empty; visibility is managed in _refresh_auto_pause_list. + empty_row.set_no_show_all(True) + self._auto_pause_empty_row = empty_row + + self.power_tab.pack_start(group, False, False, 0) + + self.auto_pause_switch.connect("state-set", self._on_auto_pause_enabled_toggled) + + def _update_auto_pause_sensitivity(self, enabled: bool) -> None: + """Gray out the app-list controls while auto-pause is disabled.""" + for widget in ( + self.auto_pause_entry, + self.auto_pause_add_btn, + self.auto_pause_pick_btn, + self.auto_pause_listbox, + ): + widget.set_sensitive(enabled) + + def _get_auto_pause_apps(self) -> list: + """Return a clean list of configured auto-pause process names.""" + apps = self.config_manager.get("auto_pause", "apps", []) or [] + if not isinstance(apps, list): + return [] + return [str(a).strip() for a in apps if a and str(a).strip()] + + def _save_auto_pause_apps(self, apps: list) -> None: + # Deduplicate case-insensitively while preserving first-seen casing + seen = set() + cleaned = [] + for name in apps: + key = name.strip().lower() + if not key or key in seen: + continue + seen.add(key) + cleaned.append(name.strip()) + self.config_manager.set("auto_pause", "apps", cleaned) + self.config_manager.save_settings() + self._refresh_auto_pause_list() + + def _refresh_auto_pause_list(self) -> None: + """Rebuild the auto-pause app list UI from config.""" + if not hasattr(self, "auto_pause_listbox"): + return + + for child in list(self.auto_pause_listbox.get_children()): + self.auto_pause_listbox.remove(child) + + apps = self._get_auto_pause_apps() + if hasattr(self, "_auto_pause_empty_row"): + if len(apps) == 0: + self.auto_pause_empty_label.show() + self._auto_pause_empty_row.show() + else: + self._auto_pause_empty_row.hide() + + for name in apps: + row = Gtk.ListBoxRow() + row.set_activatable(False) + hbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + hbox.set_margin_top(6) + hbox.set_margin_bottom(6) + hbox.set_margin_start(16) + hbox.set_margin_end(16) + + label = Gtk.Label(label=name, xalign=0) + label.set_hexpand(True) + hbox.pack_start(label, True, True, 0) + + remove_btn = Gtk.Button(label="Remove") + remove_btn.set_tooltip_text(f"Remove {name} from auto-pause list") + remove_btn.connect("clicked", self._on_auto_pause_remove_clicked, name) + hbox.pack_start(remove_btn, False, False, 0) + + row.add(hbox) + self.auto_pause_listbox.add(row) + + self.auto_pause_listbox.show_all() + + def _on_auto_pause_enabled_toggled(self, widget, state): + enabled = bool(state) + self._update_auto_pause_sensitivity(enabled) + if self._initializing or self._applying_settings: + return False + logger.info("Auto-pause enabled toggled: %s", enabled) + self.config_manager.set("auto_pause", "enabled", enabled) + self.config_manager.save_settings() + return False + + def _on_auto_pause_add_clicked(self, widget): + if self._initializing or self._applying_settings: + return + name = self.auto_pause_entry.get_text().strip() + if not name: + return + # Strip path if user pasted a full path + name = os.path.basename(name) + if name.lower().endswith(".exe"): + name = name[:-4] + apps = self._get_auto_pause_apps() + if name.lower() in {a.lower() for a in apps}: + self.auto_pause_entry.set_text("") + return + apps.append(name) + self._save_auto_pause_apps(apps) + self.auto_pause_entry.set_text("") + logger.info("Added auto-pause app: %s", name) + + def _on_auto_pause_remove_clicked(self, widget, name: str): + if self._initializing or self._applying_settings: + return + apps = [a for a in self._get_auto_pause_apps() if a.lower() != name.lower()] + self._save_auto_pause_apps(apps) + logger.info("Removed auto-pause app: %s", name) + + def _on_auto_pause_pick_running(self, widget): + """Show a simple dialog listing running process names to add.""" + if self._initializing or self._applying_settings: + return + + try: + import psutil + except ImportError: + logger.warning("psutil not available for process picker") + return + + names: set[str] = set() + for proc in psutil.process_iter(["name"]): + try: + n = proc.info.get("name") + if n: + base = os.path.basename(n) + if base.lower().endswith(".exe"): + base = base[:-4] + names.add(base) + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + + if not names: + return + + dialog = Gtk.Dialog(title="Choose a Running App", transient_for=self, flags=0) + dialog.add_button("Cancel", Gtk.ResponseType.CANCEL) + dialog.add_button("Add", Gtk.ResponseType.OK) + dialog.set_default_size(360, 400) + + scrolled = Gtk.ScrolledWindow() + scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC) + scrolled.set_margin_top(8) + scrolled.set_margin_bottom(8) + scrolled.set_margin_start(8) + scrolled.set_margin_end(8) + + listbox = Gtk.ListBox() + listbox.set_selection_mode(Gtk.SelectionMode.MULTIPLE) + already = {a.lower() for a in self._get_auto_pause_apps()} + for name in sorted(names, key=str.lower): + if name.lower() in already: + continue + row = Gtk.ListBoxRow() + label = Gtk.Label(label=name, xalign=0) + label.set_margin_start(8) + label.set_margin_end(8) + label.set_margin_top(4) + label.set_margin_bottom(4) + row.add(label) + listbox.add(row) + scrolled.add(listbox) + dialog.get_content_area().pack_start(scrolled, True, True, 0) + dialog.show_all() + + response = dialog.run() + if response == Gtk.ResponseType.OK: + selected = listbox.get_selected_rows() + apps = self._get_auto_pause_apps() + existing = {a.lower() for a in apps} + for row in selected: + label = row.get_child() + if isinstance(label, Gtk.Label): + name = label.get_text().strip() + if name and name.lower() not in existing: + apps.append(name) + existing.add(name.lower()) + self._save_auto_pause_apps(apps) + dialog.destroy() + + def _build_model_keepalive_section(self): + """Build idle unload settings: enable toggle + idle timeout.""" + group = PreferencesGroup( + title="Unload When Idle", + description=( + "After a period of inactivity, unload the speech model to save memory " + "and battery — on hybrid-GPU laptops (NVIDIA Optimus) this lets the " + "GPU sleep. The next dictation reloads the model automatically, which " + "may take a moment on larger models." + ), + ) + + self.model_keepalive_switch = Gtk.Switch() + self.model_keepalive_switch.set_tooltip_text( + "Unload the speech model after the idle timeout to save memory and battery" + ) + enable_row = PreferenceRow( + title="Unload Model When Idle", + subtitle="Free memory and GPU power after a period of inactivity", + widget=self.model_keepalive_switch, + ) + group.add_row(enable_row) + + self.model_keepalive_timeout_combo = Gtk.ComboBoxText() + self.model_keepalive_timeout_combo.set_size_request(160, -1) + # id = seconds as string + for seconds, label in ( + (60, "1 minute"), + (300, "5 minutes"), + (600, "10 minutes"), + (900, "15 minutes"), + (1800, "30 minutes"), + ): + self.model_keepalive_timeout_combo.append(str(seconds), label) + self.model_keepalive_timeout_combo.set_tooltip_text( + "How long to wait after the last dictation before unloading the model" + ) + _prevent_scroll_on_hover(self.model_keepalive_timeout_combo) + timeout_row = PreferenceRow( + title="Idle Timeout", + subtitle="Unload the model after this much inactivity", + widget=self.model_keepalive_timeout_combo, + ) + group.add_row(timeout_row) + + self.power_tab.pack_start(group, False, False, 0) + + self.model_keepalive_switch.connect("state-set", self._on_model_keepalive_enabled_toggled) + self.model_keepalive_timeout_combo.connect( + "changed", self._on_model_keepalive_timeout_changed + ) + + def _update_model_keepalive_sensitivity(self, enabled: bool) -> None: + """Gray out the idle timeout selector while idle unload is disabled.""" + self.model_keepalive_timeout_combo.set_sensitive(enabled) + + def _on_model_keepalive_enabled_toggled(self, widget, state): + enabled = bool(state) + self._update_model_keepalive_sensitivity(enabled) + if self._initializing or self._applying_settings: + return False + logger.info("Model keep-alive enabled toggled: %s", enabled) + self.config_manager.set("model_keepalive", "enabled", enabled) + self.config_manager.save_settings() + return False + + def _on_model_keepalive_timeout_changed(self, widget): + if self._initializing or self._applying_settings: + return + active_id = self.model_keepalive_timeout_combo.get_active_id() + if not active_id: + return + try: + seconds = int(active_id) + except (TypeError, ValueError): + return + logger.info("Model keep-alive timeout set to %s seconds", seconds) + self.config_manager.set("model_keepalive", "idle_timeout_seconds", seconds) + self.config_manager.save_settings() + def _on_autostart_toggled(self, widget, state): """Handle toggle of the autostart switch.""" if self._initializing or self._applying_settings: @@ -2516,6 +2884,20 @@ class SettingsDialog(Gtk.Dialog): self.copy_to_clipboard_switch.set_active(copy_to_clipboard) self.sound_effects_switch.set_active(self.config_manager.is_sound_effects_enabled()) + auto_pause_settings = self.config_manager.get_settings().get("auto_pause", {}) + auto_pause_enabled = bool(auto_pause_settings.get("enabled", False)) + self.auto_pause_switch.set_active(auto_pause_enabled) + self._update_auto_pause_sensitivity(auto_pause_enabled) + self._refresh_auto_pause_list() + + keepalive_settings = self.config_manager.get_settings().get("model_keepalive", {}) + keepalive_enabled = bool(keepalive_settings.get("enabled", False)) + self.model_keepalive_switch.set_active(keepalive_enabled) + self._update_model_keepalive_sensitivity(keepalive_enabled) + timeout_seconds = int(keepalive_settings.get("idle_timeout_seconds", 300) or 300) + if not self.model_keepalive_timeout_combo.set_active_id(str(timeout_seconds)): + self.model_keepalive_timeout_combo.set_active_id("300") + available_engines = get_available_engines() available_count = 0 diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index 3feb8c8..246f2f8 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -28,7 +28,9 @@ except (ImportError, ValueError): from gi.repository import GdkPixbuf, Gio, GLib, GObject, Gtk # Import local modules - Use protocols to avoid circular imports +from ..auto_pause_monitor import DEFAULT_POLL_INTERVAL_SECONDS, AutoPauseMonitor from ..common_types import RecognitionState, SpeechRecognitionManagerProtocol, TextInjectorProtocol +from ..model_keepalive import DEFAULT_IDLE_TIMEOUT_SECONDS, ModelKeepAlive from ..suspend_handler import SuspendHandler from ..utils.resource_manager import ResourceManager from .config_manager import ConfigManager @@ -130,6 +132,23 @@ class TrayIndicator: on_resume=self._on_system_resume, ) + # Auto-pause: unload model while configured games/apps are running + self._auto_pause_monitor = AutoPauseMonitor( + get_config=self._get_auto_pause_config, + on_pause=self._on_auto_pause, + on_resume=self._on_auto_resume, + ) + self._auto_pause_monitor.start() + + # Idle keep-alive: unload model after inactivity (battery / Optimus) + self._model_keepalive = ModelKeepAlive( + get_config=self._get_model_keepalive_config, + on_idle_unload=self._on_keepalive_idle_unload, + is_safe_to_unload=self._is_safe_for_keepalive_unload, + ) + self._model_keepalive.start() + self._model_keepalive.bump() + # Set up keyboard shortcuts with mode support self._setup_keyboard_shortcuts() @@ -423,6 +442,13 @@ class TrayIndicator: Args: state: The new recognition state """ + keepalive = getattr(self, "_model_keepalive", None) + if keepalive is not None: + if state == RecognitionState.IDLE: + keepalive.bump() + else: + keepalive.cancel() + # Update the UI in the GTK main thread GLib.idle_add(self._update_ui, state) @@ -560,8 +586,77 @@ class TrayIndicator: logger.debug("About clicked") show_about_dialog(parent=None) + def _get_auto_pause_config(self): + """Return (enabled, apps, poll_interval_seconds) for AutoPauseMonitor.""" + enabled = self.config_manager.get_bool("auto_pause", "enabled", False) + apps = self.config_manager.get("auto_pause", "apps", []) or [] + if not isinstance(apps, list): + apps = [] + interval = self.config_manager.get_float( + "auto_pause", "poll_interval_seconds", float(DEFAULT_POLL_INTERVAL_SECONDS) + ) + return enabled, apps, interval + + def _get_model_keepalive_config(self): + """Return (enabled, idle_timeout_seconds) for ModelKeepAlive.""" + enabled = self.config_manager.get_bool("model_keepalive", "enabled", False) + timeout = self.config_manager.get_float( + "model_keepalive", + "idle_timeout_seconds", + float(DEFAULT_IDLE_TIMEOUT_SECONDS), + ) + return enabled, timeout + + def _is_safe_for_keepalive_unload(self) -> bool: + """Keep-alive may unload only when idle and not under auto-pause.""" + if getattr(self.speech_engine, "is_auto_paused", False) is True: + return False + if ( + getattr(self, "_auto_pause_monitor", None) is not None + and self._auto_pause_monitor.paused + ): + return False + return self.speech_engine.state == RecognitionState.IDLE + + def _on_keepalive_idle_unload(self): + """Unload speech model after idle keep-alive timeout.""" + logger.info("Keep-alive idle timeout — unloading speech model") + unload = getattr(self.speech_engine, "unload_model", None) + if callable(unload): + unload(reason="idle_keepalive") + + def _on_auto_pause(self): + """Unload speech model when a configured game/app is detected.""" + logger.info("Auto-pause triggered — unloading speech model") + keepalive = getattr(self, "_model_keepalive", None) + if keepalive is not None: + keepalive.cancel() + unload = getattr(self.speech_engine, "unload_model", None) + if callable(unload): + unload(reason="auto_pause") + else: + # Fallback for engines that only expose reinit/stop + if self.speech_engine.state != RecognitionState.IDLE: + self.speech_engine.stop_recognition() + + def _on_auto_resume(self): + """Reload speech model after configured games/apps have exited.""" + logger.info("Auto-pause cleared — reloading speech model") + reinit = getattr(self.speech_engine, "reinitialize_after_resume", None) + if callable(reinit): + try: + reinit() + except Exception: + logger.error("Failed to reload speech engine after auto-pause", exc_info=True) + keepalive = getattr(self, "_model_keepalive", None) + if keepalive is not None: + keepalive.bump() + def _on_system_suspend(self): """Stop active recognition before the system goes to sleep.""" + keepalive = getattr(self, "_model_keepalive", None) + if keepalive is not None: + keepalive.cancel() if self.speech_engine.state != RecognitionState.IDLE: logger.info("System suspending — stopping active recognition") self.speech_engine.stop_recognition() @@ -572,9 +667,16 @@ class TrayIndicator: Speech engine reinitializes after 2s (audio hardware recovers fast). Keyboard backend waits for /dev/input to settle using inotify-backed directory monitoring, with a timer fallback if monitoring unavailable. + + If auto-pause still has a configured app running, skip speech reinit — + the auto-pause monitor owns model lifecycle until that app exits. """ logger.info("System resumed — scheduling reinit") - GLib.timeout_add_seconds(2, self._reinit_speech_after_resume) + # Use `is True` so duck-typed mocks without a real bool flag still reinit. + if getattr(self.speech_engine, "is_auto_paused", False) is True: + logger.info("Skipping resume reinit: auto-pause still active") + else: + GLib.timeout_add_seconds(2, self._reinit_speech_after_resume) GLib.timeout_add_seconds(2, self._start_input_device_monitor) def _reinit_speech_after_resume(self): @@ -582,6 +684,9 @@ class TrayIndicator: self.speech_engine.reinitialize_after_resume() except Exception: logger.error("Failed to reinitialize after resume", exc_info=True) + keepalive = getattr(self, "_model_keepalive", None) + if keepalive is not None: + keepalive.bump() return GLib.SOURCE_REMOVE def _start_input_device_monitor(self): @@ -652,6 +757,12 @@ class TrayIndicator: if self._suspend_handler is not None: self._suspend_handler.shutdown() + if getattr(self, "_auto_pause_monitor", None) is not None: + self._auto_pause_monitor.shutdown() + + if getattr(self, "_model_keepalive", None) is not None: + self._model_keepalive.shutdown() + self._cleanup_input_monitor() # Stop the keyboard shortcut manager diff --git a/tests/test_auto_pause_monitor.py b/tests/test_auto_pause_monitor.py new file mode 100644 index 0000000..628ea9f --- /dev/null +++ b/tests/test_auto_pause_monitor.py @@ -0,0 +1,696 @@ +""" +Tests for auto-pause process matching, monitor pause/resume, and model unload. + +Exercises the shipped matching helpers and AutoPauseMonitor against real +callbacks — not a parallel reimplementation of the match logic. +""" + +import os +import sys +import unittest +from unittest.mock import MagicMock, patch + +# Heavy deps used transitively when importing recognition manager +mock_pywhispercpp = MagicMock() +mock_pywhispercpp.model = MagicMock() +mock_pywhispercpp.model.Model = MagicMock() + +sys.modules.setdefault("vosk", MagicMock()) +sys.modules.setdefault("whisper", MagicMock()) +sys.modules.setdefault("pywhispercpp", mock_pywhispercpp) +sys.modules.setdefault("pywhispercpp.model", mock_pywhispercpp.model) +sys.modules.setdefault("requests", MagicMock()) +sys.modules.setdefault("pyaudio", MagicMock()) +sys.modules.setdefault("wave", MagicMock()) +sys.modules.setdefault("tqdm", MagicMock()) +sys.modules.setdefault("numpy", MagicMock()) +sys.modules.setdefault("torch", MagicMock()) +sys.modules.setdefault("psutil", MagicMock()) + +from vocalinux.auto_pause_monitor import ( # noqa: E402 + AutoPauseMonitor, + any_configured_process_running, + configured_names_set, + normalize_process_name, +) +from vocalinux.common_types import RecognitionState # noqa: E402 +from vocalinux.ui.config_manager import DEFAULT_CONFIG # noqa: E402 + + +class TestNormalizeAndMatch(unittest.TestCase): + """Pure matching helpers (no I/O).""" + + def test_normalize_basename_and_case(self): + self.assertEqual(normalize_process_name("Overwatch"), "overwatch") + self.assertEqual(normalize_process_name("/usr/bin/Steam"), "steam") + self.assertEqual(normalize_process_name("game.EXE"), "game") + self.assertEqual(normalize_process_name(" chrome "), "chrome") + self.assertEqual(normalize_process_name(""), "") + + def test_configured_names_set_drops_empty(self): + self.assertEqual( + configured_names_set(["Overwatch", "", " ", "steam"]), + {"overwatch", "steam"}, + ) + + def test_match_when_process_running(self): + self.assertTrue( + any_configured_process_running( + ["overwatch", "steam"], + ["bash", "Overwatch.exe", "Xorg"], + ) + ) + + def test_no_match_when_absent(self): + self.assertFalse( + any_configured_process_running( + ["overwatch"], + ["bash", "chrome", "Xorg"], + ) + ) + + def test_empty_config_never_matches(self): + self.assertFalse(any_configured_process_running([], ["overwatch", "steam"])) + self.assertFalse(any_configured_process_running(["", " "], ["overwatch"])) + + def test_match_via_exe_basename_path(self): + self.assertTrue( + any_configured_process_running( + ["firefox"], + ["/usr/lib/firefox/firefox"], + ) + ) + + +class TestAutoPauseMonitor(unittest.TestCase): + """Poller state machine: pause once, resume once, no double-fire.""" + + def _make_monitor(self, enabled=True, apps=None, snapshot=None): + apps = apps if apps is not None else ["overwatch"] + config = {"enabled": enabled, "apps": list(apps), "interval": 5.0} + pause_cb = MagicMock() + resume_cb = MagicMock() + running = snapshot if snapshot is not None else (lambda: set()) + + def get_config(): + return config["enabled"], config["apps"], config["interval"] + + monitor = AutoPauseMonitor( + get_config=get_config, + on_pause=pause_cb, + on_resume=resume_cb, + process_snapshot=running if callable(running) else (lambda: running), + use_glib=False, + ) + return monitor, pause_cb, resume_cb, config + + def test_pause_when_match_appears(self): + running = {"bash", "overwatch"} + monitor, pause_cb, resume_cb, _ = self._make_monitor( + snapshot=lambda: running, + ) + + matched = monitor.check_once() + + self.assertTrue(matched) + self.assertTrue(monitor.paused) + pause_cb.assert_called_once() + resume_cb.assert_not_called() + + def test_no_pause_when_disabled(self): + monitor, pause_cb, resume_cb, _ = self._make_monitor( + enabled=False, + snapshot=lambda: {"overwatch"}, + ) + + matched = monitor.check_once() + + self.assertFalse(matched) + self.assertFalse(monitor.paused) + pause_cb.assert_not_called() + resume_cb.assert_not_called() + + def test_no_pause_when_no_match(self): + monitor, pause_cb, resume_cb, _ = self._make_monitor( + snapshot=lambda: {"bash", "chrome"}, + ) + + matched = monitor.check_once() + + self.assertFalse(matched) + pause_cb.assert_not_called() + + def test_resume_when_match_clears(self): + state = {"running": {"overwatch"}} + monitor, pause_cb, resume_cb, _ = self._make_monitor( + snapshot=lambda: state["running"], + ) + + monitor.check_once() + pause_cb.assert_called_once() + + state["running"] = {"bash"} + matched = monitor.check_once() + + self.assertFalse(matched) + self.assertFalse(monitor.paused) + resume_cb.assert_called_once() + + def test_no_double_pause(self): + monitor, pause_cb, resume_cb, _ = self._make_monitor( + snapshot=lambda: {"overwatch"}, + ) + + monitor.check_once() + monitor.check_once() + + pause_cb.assert_called_once() + resume_cb.assert_not_called() + + def test_no_double_resume(self): + state = {"running": {"overwatch"}} + monitor, pause_cb, resume_cb, _ = self._make_monitor( + snapshot=lambda: state["running"], + ) + monitor.check_once() + state["running"] = set() + monitor.check_once() + monitor.check_once() + + resume_cb.assert_called_once() + + def test_disabling_while_paused_resumes(self): + state = {"enabled": True, "running": {"overwatch"}} + pause_cb = MagicMock() + resume_cb = MagicMock() + + def get_config(): + return state["enabled"], ["overwatch"], 5.0 + + monitor = AutoPauseMonitor( + get_config=get_config, + on_pause=pause_cb, + on_resume=resume_cb, + process_snapshot=lambda: state["running"], + use_glib=False, + ) + monitor.check_once() + pause_cb.assert_called_once() + + state["enabled"] = False + monitor.check_once() + resume_cb.assert_called_once() + self.assertFalse(monitor.paused) + + def test_pause_callback_exception_does_not_raise(self): + pause_cb = MagicMock(side_effect=RuntimeError("boom")) + monitor = AutoPauseMonitor( + get_config=lambda: (True, ["x"], 5.0), + on_pause=pause_cb, + process_snapshot=lambda: {"x"}, + use_glib=False, + ) + # Should not raise + monitor.check_once() + self.assertTrue(monitor.paused) + + def test_start_stop_without_glib(self): + monitor, _, _, _ = self._make_monitor() + monitor.start() + self.assertTrue(monitor.active) + monitor.stop() + self.assertFalse(monitor.active) + monitor.shutdown() + self.assertFalse(monitor.active) + + def test_start_is_idempotent(self): + monitor, _, _, _ = self._make_monitor() + monitor.start() + monitor.start() + self.assertTrue(monitor.active) + monitor.stop() + + def test_resume_callback_exception_does_not_raise(self): + state = {"running": {"x"}} + resume_cb = MagicMock(side_effect=RuntimeError("resume boom")) + monitor = AutoPauseMonitor( + get_config=lambda: (True, ["x"], 5.0), + on_pause=MagicMock(), + on_resume=resume_cb, + process_snapshot=lambda: state["running"], + use_glib=False, + ) + monitor.check_once() + state["running"] = set() + monitor.check_once() # should not raise + resume_cb.assert_called_once() + self.assertFalse(monitor.paused) + + def test_read_config_handles_get_config_failure(self): + monitor = AutoPauseMonitor( + get_config=MagicMock(side_effect=RuntimeError("config broken")), + use_glib=False, + ) + matched = monitor.check_once() + self.assertFalse(matched) + self.assertFalse(monitor.paused) + + def test_read_config_coerces_invalid_apps_and_interval(self): + monitor = AutoPauseMonitor( + get_config=lambda: (True, "not-a-list", "bad"), + process_snapshot=lambda: {"overwatch"}, + use_glib=False, + ) + # invalid apps → treated as empty → no match even if overwatch runs + self.assertFalse(monitor.check_once()) + + monitor2 = AutoPauseMonitor( + get_config=lambda: (True, ["overwatch"], 999), # clamps to max 60 + process_snapshot=lambda: {"overwatch"}, + use_glib=False, + ) + enabled, apps, interval = monitor2._read_config() + self.assertTrue(enabled) + self.assertEqual(list(apps), ["overwatch"]) + self.assertEqual(interval, 60.0) + + monitor3 = AutoPauseMonitor( + get_config=lambda: (True, ["x"], 0), # clamps to min 1 + use_glib=False, + ) + _, _, interval3 = monitor3._read_config() + self.assertEqual(interval3, 1.0) + + @patch("vocalinux.auto_pause_monitor.GLib", create=True) + def test_start_with_glib_schedules_idle_then_timeout(self, _unused): + """Drive real GLib scheduling helpers via mocked gi.repository.GLib.""" + mock_glib = MagicMock() + mock_glib.idle_add.return_value = 11 + mock_glib.timeout_add_seconds.return_value = 22 + + with patch.dict( + "sys.modules", + {"gi": MagicMock(), "gi.repository": MagicMock(GLib=mock_glib)}, + ): + # Force re-import path used inside methods + import vocalinux.auto_pause_monitor as apm + + monitor = apm.AutoPauseMonitor( + get_config=lambda: (False, [], 5.0), + process_snapshot=lambda: set(), + use_glib=True, + ) + monitor.start() + self.assertTrue(monitor.active) + mock_glib.idle_add.assert_called() + # Simulate the idle callback (immediate first poll) + result = monitor._glib_poll() + self.assertFalse(result) # always SOURCE_REMOVE style + mock_glib.timeout_add_seconds.assert_called() + monitor.stop() + mock_glib.source_remove.assert_called() + + def test_glib_poll_when_stopped_returns_false(self): + monitor = AutoPauseMonitor( + get_config=lambda: (False, [], 5.0), + use_glib=False, + ) + monitor._running = False + self.assertFalse(monitor._glib_poll()) + + def test_glib_poll_swallows_check_once_errors(self): + monitor = AutoPauseMonitor( + get_config=lambda: (True, ["x"], 5.0), + process_snapshot=MagicMock(side_effect=RuntimeError("psutil down")), + use_glib=False, + ) + monitor._running = True + # use_glib False so reschedule is a no-op after poll + self.assertFalse(monitor._glib_poll()) + + def test_cancel_timeout_noop_when_none(self): + monitor = AutoPauseMonitor( + get_config=lambda: (False, [], 5.0), + use_glib=False, + ) + monitor._timeout_id = None + monitor._cancel_timeout() # no raise + self.assertIsNone(monitor._timeout_id) + + def test_cancel_timeout_handles_source_remove_failure(self): + mock_glib = MagicMock() + mock_glib.source_remove.side_effect = RuntimeError("bad id") + with patch.dict( + "sys.modules", + {"gi": MagicMock(), "gi.repository": MagicMock(GLib=mock_glib)}, + ): + monitor = AutoPauseMonitor( + get_config=lambda: (False, [], 5.0), + use_glib=False, + ) + monitor._timeout_id = 99 + monitor._cancel_timeout() + self.assertIsNone(monitor._timeout_id) + + def test_schedule_skipped_when_not_running(self): + monitor = AutoPauseMonitor( + get_config=lambda: (False, [], 5.0), + use_glib=True, + ) + monitor._running = False + monitor._schedule_next_poll(immediate=True) + self.assertIsNone(monitor._timeout_id) + + def test_schedule_handles_glib_import_failure(self): + monitor = AutoPauseMonitor( + get_config=lambda: (False, [], 5.0), + use_glib=True, + ) + monitor._running = True + real_import = __import__ + + def fake_import(name, globals=None, locals=None, fromlist=(), level=0): + if name == "gi" or name.startswith("gi."): + raise ImportError("no gi") + return real_import(name, globals, locals, fromlist, level) + + with patch("builtins.__import__", side_effect=fake_import): + monitor._schedule_next_poll(immediate=False) + self.assertIsNone(monitor._timeout_id) + + +class TestCollectRunningProcessNames(unittest.TestCase): + """collect_running_process_names uses real helper with mocked psutil.""" + + def _psutil_mock(self): + mock_psutil = MagicMock() + mock_psutil.NoSuchProcess = type("NoSuchProcess", (Exception,), {}) + mock_psutil.AccessDenied = type("AccessDenied", (Exception,), {}) + mock_psutil.ZombieProcess = type("ZombieProcess", (Exception,), {}) + return mock_psutil + + def test_collects_name_and_exe_basename(self): + import vocalinux.auto_pause_monitor as apm + + mock_psutil = self._psutil_mock() + proc_ok = MagicMock() + proc_ok.info = {"name": "steam", "exe": "/usr/bin/steam"} + proc_no_name = MagicMock() + proc_no_name.info = {"name": None, "exe": "/opt/game/Game.exe"} + + class Denied: + @property + def info(self): + raise mock_psutil.AccessDenied() + + mock_psutil.process_iter.return_value = [proc_ok, Denied(), proc_no_name] + + with patch.dict("sys.modules", {"psutil": mock_psutil}): + names = apm.collect_running_process_names() + + self.assertIn("steam", names) + self.assertIn("Game.exe", names) + + def test_collect_skips_empty_entries(self): + import vocalinux.auto_pause_monitor as apm + + mock_psutil = self._psutil_mock() + proc = MagicMock() + proc.info = {"name": "", "exe": None} + mock_psutil.process_iter.return_value = [proc] + with patch.dict("sys.modules", {"psutil": mock_psutil}): + names = apm.collect_running_process_names() + self.assertEqual(names, set()) + + def test_collect_swallows_basename_errors(self): + import vocalinux.auto_pause_monitor as apm + + mock_psutil = self._psutil_mock() + proc = MagicMock() + # os.path.basename on non-path-like may raise; helper must continue + bad_exe = MagicMock() + bad_exe.__fspath__ = MagicMock(side_effect=TypeError("nope")) + proc.info = {"name": "ok", "exe": bad_exe} + mock_psutil.process_iter.return_value = [proc] + with patch.dict("sys.modules", {"psutil": mock_psutil}): + with patch("os.path.basename", side_effect=TypeError("bad")): + names = apm.collect_running_process_names() + self.assertIn("ok", names) + + +class TestUnloadModelAndAutoPauseBlock(unittest.TestCase): + """SpeechRecognitionManager.unload_model and start_recognition guard.""" + + def _make_manager(self): + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + with patch.object(SpeechRecognitionManager, "__init__", lambda self: None): + mgr = SpeechRecognitionManager() + + mgr.engine = "vosk" + mgr.state = RecognitionState.IDLE + mgr.model = MagicMock(name="model") + mgr.recognizer = MagicMock(name="recognizer") + mgr._model_initialized = True + mgr._auto_paused = False + mgr._idle_unloaded = False + mgr._model_lock = MagicMock() + mgr._model_lock.__enter__ = MagicMock(return_value=None) + mgr._model_lock.__exit__ = MagicMock(return_value=False) + mgr._http_session = None + mgr._reconnection_attempts = 0 + mgr._update_state = MagicMock() + mgr._init_vosk = MagicMock() + mgr._init_whisper = MagicMock() + mgr._init_whispercpp = MagicMock() + mgr._init_remote_api = MagicMock() + mgr.stop_recognition = MagicMock() + return mgr + + def test_unload_model_clears_resources_and_sets_flag(self): + mgr = self._make_manager() + + mgr.unload_model() + + self.assertIsNone(mgr.model) + self.assertIsNone(mgr.recognizer) + self.assertFalse(mgr._model_initialized) + self.assertTrue(mgr.is_auto_paused) + mgr.stop_recognition.assert_not_called() # already IDLE + + def test_unload_stops_active_recognition_first(self): + mgr = self._make_manager() + mgr.state = RecognitionState.LISTENING + + mgr.unload_model() + + mgr.stop_recognition.assert_called_once() + self.assertTrue(mgr.is_auto_paused) + + def test_unload_closes_http_session(self): + mgr = self._make_manager() + session = MagicMock() + mgr._http_session = session + + mgr.unload_model() + + session.close.assert_called_once() + self.assertIsNone(mgr._http_session) + + def test_unload_http_session_close_error_is_swallowed(self): + mgr = self._make_manager() + session = MagicMock() + session.close.side_effect = RuntimeError("already closed") + mgr._http_session = session + + mgr.unload_model() + + self.assertIsNone(mgr._http_session) + self.assertTrue(mgr.is_auto_paused) + + def test_unload_continues_if_gc_raises(self): + mgr = self._make_manager() + with patch("gc.collect", side_effect=RuntimeError("gc failed")): + mgr.unload_model() + self.assertTrue(mgr.is_auto_paused) + self.assertIsNone(mgr.model) + + def test_start_recognition_blocked_when_auto_paused(self): + mgr = self._make_manager() + mgr._auto_paused = True + # model not ready either; auto-pause should win first + with ( + patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"), + patch("vocalinux.speech_recognition.recognition_manager._show_notification"), + ): + mgr.start_recognition() + + self.assertEqual(mgr.state, RecognitionState.IDLE) + # Should not have transitioned to listening + mgr._update_state.assert_not_called() + + def test_unload_idle_keepalive_sets_idle_flag_not_auto_pause(self): + mgr = self._make_manager() + mgr.unload_model(reason="idle_keepalive") + self.assertFalse(mgr.is_auto_paused) + self.assertTrue(mgr.is_idle_unloaded) + self.assertIsNone(mgr.model) + + def test_ensure_model_loaded_reloads_after_idle_unload(self): + mgr = self._make_manager() + mgr.unload_model(reason="idle_keepalive") + + def init_vosk(): + mgr.model = MagicMock() + mgr._model_initialized = True + + mgr._init_vosk.side_effect = init_vosk + self.assertTrue(mgr.ensure_model_loaded()) + self.assertFalse(mgr.is_idle_unloaded) + self.assertTrue(mgr.model_ready) + + def test_ensure_model_loaded_blocked_when_auto_paused(self): + mgr = self._make_manager() + mgr.unload_model(reason="auto_pause") + self.assertFalse(mgr.ensure_model_loaded()) + mgr._init_vosk.assert_not_called() + + def test_start_recognition_lazy_reloads_after_idle_unload(self): + mgr = self._make_manager() + mgr.unload_model(reason="idle_keepalive") + + def init_vosk(): + mgr.model = MagicMock() + mgr._model_initialized = True + + mgr._init_vosk.side_effect = init_vosk + mgr.should_record = False + mgr._recognition_mode = "toggle" + mgr.audio_buffer = [] + mgr._segment_queue = None + mgr.audio_thread = None + mgr.recognition_thread = None + + with ( + patch("vocalinux.speech_recognition.recognition_manager.play_start_sound"), + patch( + "vocalinux.speech_recognition.recognition_manager.threading.Thread" + ) as mock_thread, + ): + mock_thread.return_value = MagicMock() + mgr.start_recognition() + + self.assertFalse(mgr.is_idle_unloaded) + mgr._update_state.assert_called() + + def test_reinitialize_clears_auto_pause_and_reloads(self): + mgr = self._make_manager() + mgr._auto_paused = True + mgr.model = None + mgr._model_initialized = False + + # Simulate successful reinit setting model ready + def init_vosk(): + mgr.model = MagicMock() + mgr._model_initialized = True + + mgr._init_vosk.side_effect = init_vosk + + mgr.reinitialize_after_resume() + + mgr._init_vosk.assert_called_once() + self.assertFalse(mgr.is_auto_paused) + self.assertIsNotNone(mgr.model) + + def test_pause_then_resume_via_monitor_and_engine(self): + """End-to-end of the shipped path: match → unload → clear → reinit.""" + mgr = self._make_manager() + state = {"running": {"overwatch"}} + + def on_pause(): + mgr.unload_model() + + def on_resume(): + def init_vosk(): + mgr.model = MagicMock() + mgr._model_initialized = True + + mgr._init_vosk.side_effect = init_vosk + mgr.reinitialize_after_resume() + + monitor = AutoPauseMonitor( + get_config=lambda: (True, ["overwatch"], 5.0), + on_pause=on_pause, + on_resume=on_resume, + process_snapshot=lambda: state["running"], + use_glib=False, + ) + + monitor.check_once() + self.assertTrue(mgr.is_auto_paused) + self.assertIsNone(mgr.model) + + state["running"] = set() + monitor.check_once() + self.assertFalse(mgr.is_auto_paused) + self.assertIsNotNone(mgr.model) + self.assertTrue(mgr._model_initialized) + + +class TestAutoPauseConfigDefaults(unittest.TestCase): + """DEFAULT_CONFIG schema for auto_pause section.""" + + def test_default_config_has_auto_pause_section(self): + self.assertIn("auto_pause", DEFAULT_CONFIG) + section = DEFAULT_CONFIG["auto_pause"] + self.assertFalse(section["enabled"]) + self.assertEqual(section["apps"], []) + self.assertIn("poll_interval_seconds", section) + + def test_default_config_has_model_keepalive_section(self): + self.assertIn("model_keepalive", DEFAULT_CONFIG) + section = DEFAULT_CONFIG["model_keepalive"] + self.assertFalse(section["enabled"]) + self.assertEqual(section["idle_timeout_seconds"], 300) + + def test_config_manager_round_trip_auto_pause(self): + import json + import tempfile + from unittest.mock import patch + + from vocalinux.ui.config_manager import ConfigManager + + with tempfile.TemporaryDirectory() as tmp: + config_dir = os.path.join(tmp, "vocalinux") + config_file = os.path.join(config_dir, "config.json") + os.makedirs(config_dir, exist_ok=True) + + with ( + patch("vocalinux.ui.config_manager.CONFIG_DIR", config_dir), + patch("vocalinux.ui.config_manager.CONFIG_FILE", config_file), + ): + cm = ConfigManager() + self.assertFalse(cm.get_bool("auto_pause", "enabled", True)) + self.assertEqual(cm.get("auto_pause", "apps", None), []) + + cm.set("auto_pause", "enabled", True) + cm.set("auto_pause", "apps", ["Overwatch", "steam"]) + self.assertTrue(cm.save_config()) + + cm2 = ConfigManager() + self.assertTrue(cm2.get_bool("auto_pause", "enabled", False)) + self.assertEqual(cm2.get("auto_pause", "apps"), ["Overwatch", "steam"]) + + with open(config_file) as f: + raw = json.load(f) + self.assertIn("auto_pause", raw) + self.assertEqual(raw["auto_pause"]["apps"], ["Overwatch", "steam"]) + + +# Tray wiring tests live in test_suspend_handler.py (TestTrayAutoPauseWiring) +# so this module never imports tray_indicator and cannot pin a stale gi/GLib +# mock for later suite files. + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_model_keepalive.py b/tests/test_model_keepalive.py new file mode 100644 index 0000000..e035a00 --- /dev/null +++ b/tests/test_model_keepalive.py @@ -0,0 +1,156 @@ +"""Tests for model keep-alive idle unload helper.""" + +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from vocalinux.model_keepalive import ( + DEFAULT_IDLE_TIMEOUT_SECONDS, + ModelKeepAlive, + clamp_idle_timeout, +) +from vocalinux.ui.config_manager import DEFAULT_CONFIG + + +class TestClampIdleTimeout(unittest.TestCase): + def test_clamps_low_and_high(self): + self.assertEqual(clamp_idle_timeout(10), 60) + self.assertEqual(clamp_idle_timeout(99999), 3600) + self.assertEqual(clamp_idle_timeout(300), 300) + + def test_invalid_falls_back_to_default(self): + self.assertEqual(clamp_idle_timeout("nope"), float(DEFAULT_IDLE_TIMEOUT_SECONDS)) + self.assertEqual(clamp_idle_timeout(None), float(DEFAULT_IDLE_TIMEOUT_SECONDS)) + + +class TestModelKeepAlive(unittest.TestCase): + def test_bump_when_disabled_cancels(self): + fired = [] + ka = ModelKeepAlive( + get_config=lambda: (False, 300), + on_idle_unload=lambda: fired.append(1), + use_glib=False, + ) + ka.start() + ka.bump() + self.assertFalse(ka.armed) + self.assertFalse(ka.fire_if_due()) + self.assertEqual(fired, []) + + def test_bump_arms_and_fire_unloads(self): + fired = [] + ka = ModelKeepAlive( + get_config=lambda: (True, 300), + on_idle_unload=lambda: fired.append(1), + is_safe_to_unload=lambda: True, + use_glib=False, + ) + ka.start() + ka.bump() + self.assertTrue(ka.armed) + self.assertTrue(ka.fire_if_due()) + self.assertEqual(fired, [1]) + self.assertFalse(ka.armed) + # Second fire without re-arm is a no-op + self.assertFalse(ka.fire_if_due()) + + def test_cancel_prevents_fire(self): + fired = [] + ka = ModelKeepAlive( + get_config=lambda: (True, 120), + on_idle_unload=lambda: fired.append(1), + use_glib=False, + ) + ka.start() + ka.bump() + ka.cancel() + self.assertFalse(ka.armed) + self.assertFalse(ka.fire_if_due()) + self.assertEqual(fired, []) + + def test_unsafe_predicate_skips_unload(self): + fired = [] + ka = ModelKeepAlive( + get_config=lambda: (True, 120), + on_idle_unload=lambda: fired.append(1), + is_safe_to_unload=lambda: False, + use_glib=False, + ) + ka.start() + ka.bump() + self.assertFalse(ka.fire_if_due()) + self.assertEqual(fired, []) + + def test_callback_error_is_swallowed(self): + def boom(): + raise RuntimeError("unload failed") + + ka = ModelKeepAlive( + get_config=lambda: (True, 120), + on_idle_unload=boom, + is_safe_to_unload=lambda: True, + use_glib=False, + ) + ka.start() + ka.bump() + self.assertFalse(ka.fire_if_due()) + + def test_config_error_disables(self): + def bad_config(): + raise RuntimeError("config boom") + + ka = ModelKeepAlive(get_config=bad_config, use_glib=False) + ka.start() + ka.bump() + self.assertFalse(ka.armed) + + def test_stop_and_shutdown(self): + ka = ModelKeepAlive(get_config=lambda: (True, 120), use_glib=False) + ka.start() + ka.bump() + ka.shutdown() + self.assertFalse(ka.active) + self.assertFalse(ka.armed) + + def test_start_idempotent(self): + ka = ModelKeepAlive(get_config=lambda: (True, 120), use_glib=False) + ka.start() + ka.start() + self.assertTrue(ka.active) + + def test_glib_schedule_and_timeout(self): + """Drive GLib scheduling via mocked gi.repository (same pattern as auto-pause).""" + mock_glib = MagicMock() + mock_glib.timeout_add_seconds.return_value = 42 + fired = [] + + with patch.dict( + "sys.modules", + {"gi": MagicMock(), "gi.repository": MagicMock(GLib=mock_glib)}, + ): + ka = ModelKeepAlive( + get_config=lambda: (True, 90), + on_idle_unload=lambda: fired.append(1), + is_safe_to_unload=lambda: True, + use_glib=True, + ) + ka.start() + ka.bump() + mock_glib.timeout_add_seconds.assert_called() + # Invoke the scheduled callback + cb = mock_glib.timeout_add_seconds.call_args[0][1] + self.assertFalse(cb()) + self.assertEqual(fired, [1]) + # Callback clears _timeout_id; shutdown should be a no-op cancel + ka.shutdown() + self.assertFalse(ka.active) + self.assertFalse(ka.armed) + + def test_default_config_section(self): + self.assertIn("model_keepalive", DEFAULT_CONFIG) + self.assertFalse(DEFAULT_CONFIG["model_keepalive"]["enabled"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_suspend_handler.py b/tests/test_suspend_handler.py index 1101850..5eb7cf5 100644 --- a/tests/test_suspend_handler.py +++ b/tests/test_suspend_handler.py @@ -457,5 +457,191 @@ class TestTrayIndicatorResumeFlow(unittest.TestCase): mock_glib.source_remove.assert_not_called() +class TestTrayAutoPauseWiring(unittest.TestCase): + """TrayIndicator auto-pause config + pause/resume handlers. + + Kept here (not in test_auto_pause_monitor) so tray_indicator is only + imported from this module's existing tray tests, avoiding early import + under a collection-time gi MagicMock that would pin a stale GLib on the + module and break SOURCE_REMOVE comparisons later in the suite. + """ + + def _make_tray_indicator(self): + from vocalinux.ui.tray_indicator import TrayIndicator + + with patch.object(TrayIndicator, "__init__", lambda self: None): + indicator = TrayIndicator.__new__(TrayIndicator) + + indicator.speech_engine = MagicMock() + indicator.speech_engine.state = RecognitionState.IDLE + indicator.speech_engine.unload_model = MagicMock() + indicator.speech_engine.reinitialize_after_resume = MagicMock() + indicator.speech_engine.is_auto_paused = False + indicator.config_manager = MagicMock() + indicator._setup_keyboard_shortcuts = MagicMock() + indicator._input_monitor = None + indicator._settle_timer_id = None + indicator._monitor_timeout_id = None + indicator._model_keepalive = MagicMock() + indicator._auto_pause_monitor = MagicMock() + indicator._auto_pause_monitor.paused = False + return indicator + + def test_get_auto_pause_config(self): + indicator = self._make_tray_indicator() + indicator.config_manager.get_bool.return_value = True + indicator.config_manager.get.return_value = ["steam"] + indicator.config_manager.get_float.return_value = 7.0 + + enabled, apps, interval = indicator._get_auto_pause_config() + + assert enabled is True + assert apps == ["steam"] + assert interval == 7.0 + indicator.config_manager.get_bool.assert_called_with("auto_pause", "enabled", False) + indicator.config_manager.get.assert_called_with("auto_pause", "apps", []) + + def test_on_auto_pause_calls_unload(self): + indicator = self._make_tray_indicator() + indicator._on_auto_pause() + indicator.speech_engine.unload_model.assert_called_once_with(reason="auto_pause") + indicator._model_keepalive.cancel.assert_called_once() + + def test_on_auto_resume_calls_reinit(self): + indicator = self._make_tray_indicator() + indicator._on_auto_resume() + indicator.speech_engine.reinitialize_after_resume.assert_called_once() + indicator._model_keepalive.bump.assert_called_once() + + def test_get_model_keepalive_config(self): + indicator = self._make_tray_indicator() + indicator.config_manager.get_bool.return_value = True + indicator.config_manager.get_float.return_value = 600.0 + + enabled, timeout = indicator._get_model_keepalive_config() + + assert enabled is True + assert timeout == 600.0 + indicator.config_manager.get_bool.assert_called_with("model_keepalive", "enabled", False) + + def test_on_keepalive_idle_unload(self): + indicator = self._make_tray_indicator() + indicator._on_keepalive_idle_unload() + indicator.speech_engine.unload_model.assert_called_once_with(reason="idle_keepalive") + + def test_is_safe_for_keepalive_unload(self): + indicator = self._make_tray_indicator() + indicator.speech_engine.state = RecognitionState.IDLE + indicator.speech_engine.is_auto_paused = False + assert indicator._is_safe_for_keepalive_unload() is True + + indicator.speech_engine.state = RecognitionState.LISTENING + assert indicator._is_safe_for_keepalive_unload() is False + + indicator.speech_engine.state = RecognitionState.IDLE + indicator.speech_engine.is_auto_paused = True + assert indicator._is_safe_for_keepalive_unload() is False + + @patch("vocalinux.ui.tray_indicator.GLib") + def test_system_resume_skips_reinit_when_auto_paused(self, mock_glib): + indicator = self._make_tray_indicator() + indicator.speech_engine.is_auto_paused = True + indicator._start_input_device_monitor = MagicMock() + + indicator._on_system_resume() + + assert mock_glib.timeout_add_seconds.call_count == 1 + assert ( + mock_glib.timeout_add_seconds.call_args[0][1] == indicator._start_input_device_monitor + ) + + @patch("vocalinux.ui.tray_indicator.GLib") + def test_system_resume_reinits_when_not_auto_paused(self, mock_glib): + indicator = self._make_tray_indicator() + indicator.speech_engine.is_auto_paused = False + indicator._reinit_speech_after_resume = MagicMock() + indicator._start_input_device_monitor = MagicMock() + + indicator._on_system_resume() + + assert mock_glib.timeout_add_seconds.call_count == 2 + + def test_get_auto_pause_config_coerces_non_list_apps(self): + indicator = self._make_tray_indicator() + indicator.config_manager.get_bool.return_value = True + indicator.config_manager.get.return_value = "steam" # not a list + indicator.config_manager.get_float.return_value = 5.0 + + enabled, apps, interval = indicator._get_auto_pause_config() + + assert enabled is True + assert apps == [] + assert interval == 5.0 + + def test_get_auto_pause_config_treats_none_apps_as_empty(self): + indicator = self._make_tray_indicator() + indicator.config_manager.get_bool.return_value = False + indicator.config_manager.get.return_value = None + indicator.config_manager.get_float.return_value = 5.0 + + enabled, apps, _interval = indicator._get_auto_pause_config() + + assert enabled is False + assert apps == [] + + def test_on_auto_pause_fallback_stops_when_no_unload(self): + indicator = self._make_tray_indicator() + # Non-callable attribute so getattr finds it but callable() is False + indicator.speech_engine.unload_model = None + indicator.speech_engine.state = RecognitionState.LISTENING + indicator.speech_engine.stop_recognition = MagicMock() + + indicator._on_auto_pause() + + indicator.speech_engine.stop_recognition.assert_called_once() + + def test_on_auto_pause_fallback_idle_is_noop(self): + indicator = self._make_tray_indicator() + indicator.speech_engine.unload_model = None + indicator.speech_engine.state = RecognitionState.IDLE + indicator.speech_engine.stop_recognition = MagicMock() + + indicator._on_auto_pause() + + indicator.speech_engine.stop_recognition.assert_not_called() + + def test_on_auto_resume_swallows_reinit_errors(self): + indicator = self._make_tray_indicator() + indicator.speech_engine.reinitialize_after_resume.side_effect = RuntimeError("boom") + + indicator._on_auto_resume() # must not raise + + indicator.speech_engine.reinitialize_after_resume.assert_called_once() + + def test_on_auto_resume_without_reinit_method(self): + indicator = self._make_tray_indicator() + indicator.speech_engine.reinitialize_after_resume = None + + indicator._on_auto_resume() # no-op path + indicator._model_keepalive.bump.assert_called_once() + + def test_quit_shuts_down_keepalive(self): + indicator = self._make_tray_indicator() + monitor = MagicMock() + indicator._auto_pause_monitor = monitor + indicator._suspend_handler = MagicMock() + indicator._cleanup_input_monitor = MagicMock() + indicator.shortcut_manager = MagicMock() + indicator.text_injector = None + + with patch("vocalinux.ui.tray_indicator.Gtk") as mock_gtk: + indicator._quit() + + monitor.shutdown.assert_called_once() + indicator._model_keepalive.shutdown.assert_called_once() + indicator._suspend_handler.shutdown.assert_called_once() + mock_gtk.main_quit.assert_called_once() + + if __name__ == "__main__": unittest.main()