feat(offload): auto-pause apps + idle keep-alive model unload (#592)

* 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 <jatinkrmalik@users.noreply.github.com>

* 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 <jatinkrmalik@users.noreply.github.com>

* 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 <jatinkrmalik@users.noreply.github.com>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Jatin K Malik <jatinkrmalik@users.noreply.github.com>
This commit is contained in:
Jatin K Malik
2026-07-26 22:29:00 -07:00
committed by GitHub
co-authored by Jatin K Malik Cursor Agent
parent b945137d60
commit 4c070ded46
9 changed files with 2107 additions and 10 deletions
+255
View File
@@ -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
+185
View File
@@ -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
@@ -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)
+13
View File
@@ -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
},
+385 -3
View File
@@ -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
+112 -1
View File
@@ -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
+696
View File
@@ -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()
+156
View File
@@ -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()
+186
View File
@@ -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()