From dbe0215aadef473f000a3a30ed339aab00caf114 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 30 Sep 2026 01:05:18 -0700 Subject: [PATCH] feat(shortcuts): add language switching shortcuts, fix PTT races, and add dictation history (#880) * feat(shortcuts): per-language dictation shortcuts (#805) Bind extra keys that start dictation in a specific catalog language for one utterance, then restore the configured language. Bindings live in shortcuts.language_shortcuts as [{shortcut, language}], normalized against the language catalog and the backend's shortcut parser. The tray spawns one KeyboardShortcutManager per binding in the active mode; toggle presses route through start_recognition_with_language, and push-to-talk arms press/release the same way. The manager keeps the override one-shot: it applies before the utterance starts (outranking follow-keyboard-layout) and unwinds on the next IDLE/ERROR transition or when a mid-dictation reconfigure picks a new language. Engines that cannot re-language without a reload (VOSK, Parakeet, English-only whisper.cpp) refuse with the error cue and a notification instead of dictating in the wrong language. Settings gains a Language Shortcuts group under Shortcuts: each row is a language picker, an editable key entry, a Record button that retargets the shared key-capture, and a remove button; edits persist immediately and rebuild the live listeners. A non-dict shortcuts section now rebuilds instead of crashing config load. * fix(shortcuts): class-level default for language shortcut listeners TrayIndicator stubs built with __new__ (e.g. test_suspend_handler) skip __init__ but still run _quit(); give _language_shortcut_managers a class-level default so teardown iterates an empty list. * Fix Greptile review findings on per-language shortcuts (#805) - Bind each dictation session's language for transcription workers so the one-shot restore cannot rewrite a queued segment's language - Judge a shortcut's capability against the model the layout refresh will serve (whisper.cpp sibling swap) and refuse requests the engine drops - Add _oneshot_language_restore to the reconfigure snapshot and settle a revived pending restore on failure - Dedupe language bindings by gesture (modifier reordering, enter/return alias, pure-modifier side sets) instead of by string - Stop per-language listeners under external activation and never re-arm - Gate push-to-talk release on the binding that started the session, and stop a held hold before rebuilding listeners - Keep a row's last valid binding during partial edits, report rejected duplicates, cancel a capture aimed at a removed row, disarm the previous Record target, persist a recorded key once, and add the missing type hints * fix(shortcuts): resolve remaining Greptile race findings on PR #880 - Stamp each queued audio segment with the session language it was recorded under, so a worker that outlives stop_recognition's joins transcribes each utterance in its own language rather than whatever a newer session stored in _session_language. - Serialize push-to-talk press/release behind _ptt_lock: a release landing while start_recognition runs now waits for the owner claim instead of being dropped, so a quick key tap cannot leave the microphone running. - Add the missing return type hint on _stop_language_shortcut_managers. * test(shortcuts): stub _segment_started_at on bare manager stubs * fix(shortcuts): carry segment language as attribute, default language on bare managers * fix(shortcuts): keep (segment, language) queue tuples, stamp segment.language too * fix(settings): apply history toggle and limit to the live store --------- Co-authored-by: Jatin K Malik --- src/vocalinux/common_types.py | 4 + .../engines/faster_whisper_engine.py | 17 +- .../speech_recognition/recognition_manager.py | 273 +++++- src/vocalinux/ui/config_manager.py | 73 +- .../ui/keyboard_backends/__init__.py | 2 + src/vocalinux/ui/keyboard_backends/base.py | 62 ++ src/vocalinux/ui/settings_dialog.py | 322 ++++++- src/vocalinux/ui/tray_indicator.py | 238 ++++- tests/test_language_shortcuts.py | 911 ++++++++++++++++++ .../test_recognition_manager_device_config.py | 29 +- tests/test_record_during_reload.py | 2 +- 11 files changed, 1857 insertions(+), 76 deletions(-) create mode 100644 tests/test_language_shortcuts.py diff --git a/src/vocalinux/common_types.py b/src/vocalinux/common_types.py index f438662..0c792f9 100644 --- a/src/vocalinux/common_types.py +++ b/src/vocalinux/common_types.py @@ -60,6 +60,10 @@ class SpeechRecognitionManagerProtocol(Protocol): """Start the speech recognition process. Returns True if listening began.""" ... + def start_recognition_with_language(self, language: str, mode: str = "toggle") -> bool: + """Start recognition in ``language`` for this utterance only (#805).""" + ... + def stop_recognition(self) -> None: """Stop the speech recognition process.""" ... diff --git a/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py b/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py index e6e7f40..478a5c7 100644 --- a/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py +++ b/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py @@ -121,19 +121,24 @@ class FasterWhisperEngine: self._model = None self._model_initialized = False - def _normalize_language(self) -> Optional[str]: + def _normalize_language(self, language: Optional[str] = None) -> Optional[str]: """Map Vocalinux language codes to faster-whisper language codes.""" - if self.language == "auto": + language = self.language if language is None else language + if language == "auto": return None - if self.language == "en-us" or self.language == "en-in": + if language == "en-us" or language == "en-in": return "en" - return self.language + return language - def transcribe(self, audio_buffer: list[bytes]) -> str: + def transcribe(self, audio_buffer: list[bytes], language: Optional[str] = None) -> str: """Transcribe the provided audio buffer. Args: audio_buffer: List of audio data chunks (16-bit PCM at 16kHz). + language: Per-call override of the configured engine language + (#805): a one-shot dictation binds its session language here + so a later restore cannot rewrite it mid-flight. None uses + ``self.language``. Returns: Recognized text. @@ -152,7 +157,7 @@ class FasterWhisperEngine: segments, _info = model.transcribe( audio_float, - language=self._normalize_language(), + language=self._normalize_language(language), task="transcribe", beam_size=5, best_of=5, diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 516d271..41ce068 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -22,6 +22,7 @@ from typing import TYPE_CHECKING, Callable, Optional if TYPE_CHECKING: import numpy as np + import requests from ..audio.capture import ( # noqa: F401 (re-exported for compatibility) _STICKY_LOCK_MIN_MEAN_SQUARE, @@ -90,9 +91,18 @@ class _AudioSegment(list): from speech captured after it, however late the decode finishes. """ - def __init__(self, chunks: list[bytes], started_at: Optional[float]) -> None: + def __init__( + self, + chunks: list[bytes], + started_at: Optional[float], + language: Optional[str] = None, + ) -> None: super().__init__(chunks) self.started_at = started_at + # The language this utterance was dictated under (#805): a worker that + # outlives its session must not transcribe a stale segment in a newer + # session's language. + self.language = language def _pywhispercpp_distribution_version() -> Optional[tuple[int, ...]]: @@ -380,6 +390,14 @@ class SpeechRecognitionManager: # start of every dictation while the sentinel is in force (#821). self.language_preference = language self.language = normalize_language_for_engine(engine, resolve_language_preference(language)) + # One-shot language override set by a per-language shortcut (#805): + # consumed by the next start attempt, restored when dictation ends. + self._pending_language_override: Optional[str] = None + self._oneshot_language_restore: Optional[str] = None + # The language the in-flight dictation session resolves to. Workers + # read this so a one-shot restore during a lingering transcribe cannot + # change a queued segment's language (#805). + self._session_language: Optional[str] = None self.stop_sound_guard_ms = kwargs.get("stop_sound_guard_ms", 200) self.state = RecognitionState.IDLE self.audio_thread = None @@ -574,9 +592,17 @@ class SpeechRecognitionManager: #: "not following the layout" instead of raising on the dictation path. language_preference: str = "auto" + #: Same for the one-shot override state (#805): __new__-built stubs in tests + #: must not raise on the dictation path either. + language: str = "auto" + _pending_language_override: Optional[str] = None + _oneshot_language_restore: Optional[str] = None + _session_language: Optional[str] = None + #: Logged once rather than per dictation, so an unsupported pairing does not #: spam the log on every hotkey press. _warned_follow_layout_unsupported: bool = False + _warned_language_override_unsupported: bool = False # Every live field ``reconfigure()`` can write. A failed engine switch # restores this set so dictation matches the persisted config, not the @@ -586,6 +612,7 @@ class SpeechRecognitionManager: "model_size", "language", "language_preference", + "_oneshot_language_restore", "vad_sensitivity", "silence_timeout", "audio_device_index", @@ -704,6 +731,107 @@ class SpeechRecognitionManager: "multilingual model in Settings to use this mode." ) + def _dictation_language(self, segment_language: Optional[str] = None) -> str: + """The language a queued audio segment was recorded under (#805). + + A transcription worker can outlive ``stop_recognition``'s timed joins, + so ``self.language`` may already be restored to the configured value + while a queued segment still needs the one-shot override — and a newer + dictation may already have overwritten ``_session_language`` by the + time a stale worker reads it. Queued segments therefore carry a + ``segment_language`` snapshot stamped at enqueue time; calls without + one (tests and direct transcribe calls) fall back to the session + binding, then the configured language. + """ + if segment_language is not None: + return segment_language + return self._session_language or self.language + + def _refuse_language_override(self, language: str) -> None: + """Play the refusal cue for a per-language shortcut the model can't serve.""" + if not self._warned_language_override_unsupported: + self._warned_language_override_unsupported = True + logger.warning( + f"Cannot dictate in {language!r}: the loaded " + f"{self.engine}/{self.model_size!r} model cannot transcribe " + "it without a reload. Pick a multilingual model in Settings " + "or switch the dictation language instead." + ) + play_error_sound() + _show_notification( + "Language Shortcut Unavailable", + f"Vocalinux cannot dictate in {language} with the loaded model. " + "Choose a multilingual model in Settings to use this shortcut.", + "dialog-warning", + ) + + def _serving_model_supports_language_switch(self) -> bool: + """Whether the model that will serve this utterance can switch language. + + ``start_recognition`` refreshes the layout before a shortcut's override + applies, and on follow-keyboard-layout that refresh can swap an + English-only whisper.cpp model for a downloaded multilingual sibling. + Judge the model the refresh leaves loaded, not the one in memory now. + """ + if self._can_relanguage_without_reload(): + return True + if self.engine == "whisper_cpp" and self.language_preference == LANGUAGE_FOLLOWS_LAYOUT: + sibling = _multilingual_sibling(self.model_size) + if sibling != self.model_size and is_model_downloaded(sibling): + return True + return False + + def _apply_pending_language_override(self) -> bool: + """Point this utterance at a per-language shortcut's language (#805). + + Consumed once, right after the layout refresh, so the explicit shortcut + wins over the follow-mode resolution *and* the capability check runs + against the model that will actually transcribe. The previous effective + language is kept in ``_oneshot_language_restore`` and put back by + ``_restore_language_after_oneshot`` once dictation ends. + + Returns False — refusing to start — when the engine drops the request + (Parakeet normalizes every catalog language to ``auto``, so a labeled + shortcut could never select it) or the loaded model cannot transcribe + it without a reload; dictating silently in the wrong language would be + worse than playing the refusal cue. + """ + requested = self._pending_language_override + if requested is None: + return True + target = normalize_language_for_engine(self.engine, requested) + if target != requested: + self._refuse_language_override(requested) + return False + if target == self.language: + return True + if not self._can_relanguage_without_reload(): + self._refuse_language_override(requested) + return False + self._oneshot_language_restore = self.language + self.language = target + self.command_processor.set_language(target) + if self._faster_whisper_engine is not None: + self._faster_whisper_engine.language = target + logger.debug(f"Dictating in {target} via per-language shortcut") + return True + + def _restore_language_after_oneshot(self) -> None: + """Undo a per-language shortcut's one-shot override (#805). + + Runs when dictation drops to IDLE or ERROR, so the next plain hotkey + dictation resolves the configured language again. + """ + previous = self._oneshot_language_restore + if previous is None: + return + self._oneshot_language_restore = None + self.language = previous + self.command_processor.set_language(previous) + if self._faster_whisper_engine is not None: + self._faster_whisper_engine.language = previous + logger.debug(f"Restored dictation language {previous} after one-shot") + def _init_vosk(self): """Initialize the VOSK speech recognition engine.""" # VOSK doesn't support auto-detect, so fall back to en-us for "auto" @@ -879,12 +1007,16 @@ class SpeechRecognitionManager: self.state = RecognitionState.ERROR raise - def _transcribe_with_whisper(self, audio_buffer: list[bytes]) -> str: + def _transcribe_with_whisper( + self, audio_buffer: list[bytes], language: Optional[str] = None + ) -> str: """ Transcribe audio buffer using Whisper. Args: audio_buffer: List of audio data chunks (16-bit PCM at 16kHz) + language: Per-utterance snapshot stamped on the queued segment + (#805); None resolves the live session language. Returns: Transcribed text @@ -919,7 +1051,7 @@ class SpeechRecognitionManager: import torch use_fp16 = self.model.device != torch.device("cpu") - lang = resolve_whisper_language(self.language) + lang = resolve_whisper_language(self._dictation_language(language)) # Transcribe with Whisper (handles variable length audio automatically) result = self.model.transcribe( @@ -1131,12 +1263,16 @@ class SpeechRecognitionManager: self.state = RecognitionState.ERROR raise - def _transcribe_with_faster_whisper(self, audio_buffer: list[bytes]) -> str: + def _transcribe_with_faster_whisper( + self, audio_buffer: list[bytes], language: Optional[str] = None + ) -> str: """ Transcribe audio buffer using faster-whisper. Args: audio_buffer: List of audio data chunks (16-bit PCM at 16kHz) + language: Per-utterance snapshot stamped on the queued segment + (#805); None resolves the live session language. Returns: Transcribed text @@ -1146,7 +1282,9 @@ class SpeechRecognitionManager: logger.warning("faster-whisper engine not ready during transcription") return "" - return self._faster_whisper_engine.transcribe(audio_buffer) + return self._faster_whisper_engine.transcribe( + audio_buffer, language=self._dictation_language(language) + ) except (RuntimeError, OSError, ValueError) as e: logger.error(f"Error in faster-whisper transcription: {e}", exc_info=True) return "" @@ -1552,7 +1690,9 @@ class SpeechRecognitionManager: logger.info("Successfully loaded model with CPU backend") return cpu_backend - def _transcribe_with_whispercpp(self, audio_buffer: list[bytes]) -> str: + def _transcribe_with_whispercpp( + self, audio_buffer: list[bytes], language: Optional[str] = None + ) -> str: """ Transcribe audio buffer using whisper.cpp. @@ -1583,7 +1723,7 @@ class SpeechRecognitionManager: ) # Prepare language parameter - lang = resolve_whisper_language(self.language) + lang = resolve_whisper_language(self._dictation_language(language)) logger.debug(f"whisper.cpp using language: {lang or 'auto-detect'}") @@ -1953,7 +2093,12 @@ class SpeechRecognitionManager: self._model_initialized = True logger.info("Remote API engine setup complete.") - def _transcribe_with_remote_api(self, audio_buffer: list[bytes], session) -> str: + def _transcribe_with_remote_api( + self, + audio_buffer: list[bytes], + session: Optional["requests.Session"], + language: Optional[str] = None, + ) -> str: """Transcribe audio via remote API. Package audio buffer into WAV format and send to remote server via HTTP POST. @@ -1963,6 +2108,8 @@ class SpeechRecognitionManager: Args: audio_buffer: Audio data chunk list (16-bit PCM at 16kHz) session: A requests.Session snapshot (obtained under _model_lock) + language: Per-utterance snapshot stamped on the queued segment + (#805); None resolves the live session language. Returns: Transcribed text @@ -1998,7 +2145,7 @@ class SpeechRecognitionManager: ) # Prepare language parameters - lang = resolve_whisper_language(self.language) + lang = resolve_whisper_language(self._dictation_language(language)) # Prepare HTTP request headers headers = {} @@ -2749,6 +2896,11 @@ class SpeechRecognitionManager: new_state: The new recognition state """ self.state = new_state + if new_state in (RecognitionState.IDLE, RecognitionState.ERROR): + # A per-language shortcut's override lasts for exactly one + # dictation (#805); restore the configured language here so every + # exit path — normal stop, error, quit — unwinds it. + self._restore_language_after_oneshot() for callback in self.state_callbacks: callback(new_state) @@ -2919,6 +3071,17 @@ class SpeechRecognitionManager: # user is typing in right now rather than the one they had at startup. self._refresh_language_from_layout() + # A per-language shortcut's override runs after the layout refresh so + # the explicit key press still wins over follow-mode resolution, and + # so its capability check sees the model the refresh will serve (#805). + if not self._apply_pending_language_override(): + return False + + # Bind this session's language for the transcription workers: they can + # outlive stop_recognition's joins, and the one-shot restore must not + # rewrite the language of a segment still in the queue (#805). + self._session_language = self.language + logger.info("Starting speech recognition") self._update_state(RecognitionState.LISTENING) @@ -3078,6 +3241,43 @@ class SpeechRecognitionManager: if getattr(self, "_buffered_reload_session", False) and self._buffered_capture_failed: self._capture_finished.set() + def start_recognition_with_language(self, language: str, mode: str = "toggle") -> bool: + """Start dictation in ``language`` for this utterance only (#805). + + The configured preference is untouched: when this dictation ends the + engine's language is restored, so the next plain hotkey behaves as + before. When the loaded engine/model cannot transcribe ``language`` + without a reload (VOSK, Parakeet, an English-only whisper.cpp model), + starting is refused with the usual error cue rather than dictating in + the wrong language. + + Args: + language: A catalog id (e.g. "de", "auto") as stored by + ``shortcuts.language_shortcuts``. + mode: "toggle" or "push_to_talk", same as start_recognition. + + Returns: + True if recognition actually started, False otherwise. + """ + target = normalize_language_for_engine(self.engine, language) + if target != language: + # The engine drops the requested language entirely: Parakeet maps + # every catalog id to auto, so a labeled shortcut could never + # select it. + self._refuse_language_override(language) + return False + if target != self.language and not self._serving_model_supports_language_switch(): + self._refuse_language_override(language) + return False + + self._pending_language_override = language + try: + return self.start_recognition(mode=mode) + finally: + # The override only ever feeds the next start attempt, whether it + # ran or was refused by the guards inside start_recognition. + self._pending_language_override = None + def _record_audio(self): """Record audio from the microphone with reconnection logic.""" # Lazy import to avoid circular dependency @@ -3370,10 +3570,20 @@ class SpeechRecognitionManager: self._process_audio_buffer(audio_buffer) - def _process_audio_buffer(self, audio_buffer: list[bytes]): - """Process an immutable audio segment for transcription and commands.""" + def _process_audio_buffer( + self, audio_buffer: list[bytes], language: Optional[str] = None + ) -> None: + """Process an immutable audio segment for transcription and commands. + + ``language`` is the per-utterance snapshot stamped on the segment when + it was queued (#805); None falls back to the segment's own stamp and + then to the live session language. + """ if not audio_buffer: return + if language is None: + language = getattr(audio_buffer, "language", None) + dictation_language = self._dictation_language(language) if self.engine == "vosk": # Lock recognizer access to prevent race condition with reconfigure @@ -3389,16 +3599,16 @@ class SpeechRecognitionManager: text = result.get("text", "") elif self.engine == "whisper": - text = self._transcribe_with_whisper(audio_buffer) + text = self._transcribe_with_whisper(audio_buffer, dictation_language) elif self.engine == "whisper_cpp": - text = self._transcribe_with_whispercpp(audio_buffer) + text = self._transcribe_with_whispercpp(audio_buffer, dictation_language) elif self.engine == "parakeet": text = self._transcribe_with_parakeet(audio_buffer) elif self.engine == "faster_whisper": - text = self._transcribe_with_faster_whisper(audio_buffer) + text = self._transcribe_with_faster_whisper(audio_buffer, dictation_language) elif self.engine == "remote_api": # Snapshot the HTTP session under lock to prevent race with @@ -3412,7 +3622,7 @@ class SpeechRecognitionManager: if session is None: logger.error("Remote API HTTP session not initialized") return - text = self._transcribe_with_remote_api(audio_buffer, session) + text = self._transcribe_with_remote_api(audio_buffer, session, dictation_language) else: logger.error(f"Unknown engine: {self.engine}") @@ -3459,7 +3669,7 @@ class SpeechRecognitionManager: f"Recognition loop - should_record={self.should_record}, queue_empty={self._segment_queue.empty()}" ) try: - segment = self._segment_queue.get(timeout=0.1) + queued = self._segment_queue.get(timeout=0.1) except queue.Empty: # Only exit if we're not recording AND queue is empty if not self.should_record and self._segment_queue.empty(): @@ -3468,7 +3678,7 @@ class SpeechRecognitionManager: ) # Give a brief moment for any final items to be enqueued try: - segment = self._segment_queue.get(timeout=0.5) + queued = self._segment_queue.get(timeout=0.5) except queue.Empty: logger.debug("Recognition loop - no more items, exiting") break @@ -3476,28 +3686,30 @@ class SpeechRecognitionManager: logger.debug("Recognition loop - queue timeout, continuing") continue - if segment is None: + if queued is None: logger.debug("Recognition loop - got None signal, draining remaining items...") # Drain any remaining items before exiting while not self._segment_queue.empty(): try: remaining = self._segment_queue.get_nowait() if remaining is not None: + remaining_segment, remaining_language = remaining logger.debug( - f"Recognition loop - processing remaining segment with {len(remaining)} chunks" + f"Recognition loop - processing remaining segment with {len(remaining_segment)} chunks" ) if self.should_record: self._update_state(RecognitionState.PROCESSING) - self._process_audio_buffer(remaining) + self._process_audio_buffer(remaining_segment, remaining_language) except queue.Empty: break logger.debug("Recognition loop - exiting after None signal") break + segment, segment_language = queued logger.debug(f"Recognition loop - processing segment with {len(segment)} chunks") if self.should_record: self._update_state(RecognitionState.PROCESSING) - self._process_audio_buffer(segment) + self._process_audio_buffer(segment, segment_language) if self.should_record: self._update_state(RecognitionState.LISTENING) finally: @@ -3517,14 +3729,20 @@ class SpeechRecognitionManager: logger.debug(f"_enqueue_audio_segment called with {len(segment)} chunks") + # Stamp this utterance's language on the segment: a worker that + # outlives its dictation (or drains another session's queue) must + # transcribe each segment in the language it was recorded under, not + # whatever a newer session stored in _session_language (#805). + segment.language = self._dictation_language() + stamped = (segment, segment.language) try: - self._segment_queue.put_nowait(segment) + self._segment_queue.put_nowait(stamped) logger.debug("Enqueued segment successfully") except queue.Full: logger.warning("Transcription queue is full, dropping oldest pending segment") try: self._segment_queue.get_nowait() - self._segment_queue.put_nowait(segment) + self._segment_queue.put_nowait(stamped) except queue.Empty: logger.warning("Could not recover queue space for transcription segment") @@ -3596,6 +3814,10 @@ class SpeechRecognitionManager: # VOSK needs to load a different model for the new language language_changed = False if language is not None: + # A new configured language replaces any pending one-shot restore: + # if this reconfigure stops an in-flight per-language dictation, + # the restore must not undo what was just set here (#805). + self._oneshot_language_restore = None # Settings hands over the preference, which may be the sentinel. Keep # it so the follow mode survives, and resolve what the engine gets. self.language_preference = language @@ -3708,6 +3930,11 @@ class SpeechRecognitionManager: # with unsaved VAD/device/API knobs while the UI shows the # previous configuration. self._restore_reconfigure_state(previous) + # The snapshot can revive a one-shot language restore + # whose dictation was already stopped above; settle it + # now so the shortcut language does not leak into the + # next ordinary dictation (#805). + self._restore_language_after_oneshot() self._defer_download = True try: self._init_selected_engine() diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index 0e7ad17..b3379c1 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -157,6 +157,9 @@ DEFAULT_CONFIG = { # left_/right_ variants) — double-tap (toggle) or hold (push_to_talk). # Modifier+key combos are also supported, e.g. "alt+r", "ctrl+alt+r", # "super+space" — press (toggle) or hold (push_to_talk). + # Per-language shortcuts (#805): each {"shortcut", "language"} entry + # starts dictation in that catalog language for one utterance. + "language_shortcuts": [], }, "ui": { "start_minimized": False, @@ -230,6 +233,52 @@ DEFAULT_CONFIG = { } +def _is_valid_language_shortcut(shortcut: Any) -> bool: + """Return whether ``shortcut`` parses as a bindable shortcut string. + + Imported lazily: config_manager loads before the GTK stack in several + entry points, and the keyboard package must not become a hard dependency + of configuration access. + """ + from .keyboard_backends import is_valid_shortcut + + return isinstance(shortcut, str) and is_valid_shortcut(shortcut) + + +def normalize_language_shortcuts(raw: Any) -> list[dict[str, str]]: + """Normalize a ``shortcuts.language_shortcuts`` value (#805). + + Returns ``[{"shortcut": ..., "language": ...}]`` in stored order. Entries + that are not objects, lack a valid shortcut, or name a language outside the + catalog are dropped; the first binding wins for a duplicated shortcut. + ``"auto"`` stays valid (dictate with per-utterance detection); ``"layout"`` + is not a catalog id and cannot be bound. + """ + if not isinstance(raw, list): + return [] + + entries: list[dict[str, str]] = [] + seen_shortcuts: set[str] = set() + for item in raw: + if not isinstance(item, dict): + continue + shortcut = item.get("shortcut") + language = item.get("language") + if not isinstance(shortcut, str) or not isinstance(language, str): + continue + shortcut = shortcut.strip().lower() + language = language.strip() + if language not in SUPPORTED_LANGUAGES: + continue + if not _is_valid_language_shortcut(shortcut): + continue + if shortcut in seen_shortcuts: + continue + seen_shortcuts.add(shortcut) + entries.append({"shortcut": shortcut, "language": language}) + return entries + + def _multilingual_sibling(model_name: str) -> str: """Drop the ``.en`` infix so medium.en / medium.en-q5_0 become multilingual.""" if ".en" not in model_name: @@ -454,7 +503,12 @@ class ConfigManager: def _migrate_shortcuts_config(self, user_config: Optional[dict] = None): """Migrate deprecated shortcuts and preserve legacy defaults when omitted.""" - shortcuts_config = self.config.get("shortcuts", {}) + shortcuts_config = self.config.get("shortcuts") + if not isinstance(shortcuts_config, dict): + # A hand-edited config can hold a scalar here; rebuild the section + # so the migrations below and later readers find a dict. + shortcuts_config = {} + self.config["shortcuts"] = shortcuts_config shortcut = shortcuts_config.get("toggle_recognition") changed = False @@ -802,6 +856,23 @@ class ConfigManager: self.config["text_injection"] = {} self.config["text_injection"]["paste_shortcut"] = normalize_paste_shortcut(shortcut) + def get_language_shortcuts(self) -> list[dict[str, str]]: + """Return validated ``[{"shortcut", "language"}]`` bindings (#805). + + Malformed or stale entries (unknown language, unparsable shortcut) + never reach the listeners: they are filtered out here. + """ + shortcuts = self.config.get("shortcuts") + if not isinstance(shortcuts, dict): + return [] + return normalize_language_shortcuts(shortcuts.get("language_shortcuts")) + + def set_language_shortcuts(self, entries: Any) -> None: + """Store per-language shortcut bindings after normalization (#805).""" + if not isinstance(self.config.get("shortcuts"), dict): + self.config["shortcuts"] = {} + self.config["shortcuts"]["language_shortcuts"] = normalize_language_shortcuts(entries) + def is_overlay_enabled(self) -> bool: """Check if the floating dictation overlay is enabled (default True).""" return self.get_bool("ui", "show_overlay", True) diff --git a/src/vocalinux/ui/keyboard_backends/__init__.py b/src/vocalinux/ui/keyboard_backends/__init__.py index 287a0a6..9e42c57 100644 --- a/src/vocalinux/ui/keyboard_backends/__init__.py +++ b/src/vocalinux/ui/keyboard_backends/__init__.py @@ -24,6 +24,8 @@ from .base import ( is_valid_shortcut, parse_shortcut, parse_shortcut_spec, + shortcut_gesture, + shortcut_gestures_collide, ) logger = logging.getLogger(__name__) diff --git a/src/vocalinux/ui/keyboard_backends/base.py b/src/vocalinux/ui/keyboard_backends/base.py index e6b0028..8f3ecec 100644 --- a/src/vocalinux/ui/keyboard_backends/base.py +++ b/src/vocalinux/ui/keyboard_backends/base.py @@ -126,6 +126,29 @@ MODIFIER_NAMES = { "right_shift", } +# The physical modifier keys each token can satisfy. The side-agnostic family +# name covers both hands ("ctrl" fires on either Ctrl key), while left_/right_ +# tokens cover one side only. Two gestures are the same to a backend whenever +# the key sets they can fire on overlap (#880). +MODIFIER_KEY_SIDES: dict[str, frozenset] = { + "ctrl": frozenset({"left_ctrl", "right_ctrl"}), + "alt": frozenset({"left_alt", "right_alt"}), + "shift": frozenset({"left_shift", "right_shift"}), + "super": frozenset({"left_super", "right_super"}), + "left_ctrl": frozenset({"left_ctrl"}), + "right_ctrl": frozenset({"right_ctrl"}), + "left_alt": frozenset({"left_alt"}), + "right_alt": frozenset({"right_alt"}), + "left_shift": frozenset({"left_shift"}), + "right_shift": frozenset({"right_shift"}), +} + +# Shortcut tokens that resolve to the same physical main key. +_MAIN_KEY_ALIASES = { + "return": "enter", + "escape": "esc", +} + # Named (non-alphanumeric) main keys accepted in a combo. Single letters/digits # and function keys (f1-f24) are accepted by rule and need not be listed here. _NAMED_MAIN_KEYS = { @@ -315,6 +338,45 @@ def is_valid_shortcut(shortcut_string: str) -> bool: return False +def shortcut_gesture(shortcut_string: str) -> Optional[Tuple[str, frozenset]]: + """The physical gesture a shortcut string describes, as backends match it. + + A pure-modifier hold normalizes to ``("hold", sides)``: the set of + physical keys whose press fires it, so "ctrl+ctrl" and + "left_ctrl+left_ctrl" overlap on the left Ctrl key. A combo normalizes to + ``("combo", {main key})`` because backends fire on the main key while + every required modifier is held — extra held keys do not block a combo, + so any two combos on the same key can both fire. Tokens that alias to the + same physical key ("return"/"enter", "escape"/"esc") normalize together. + + Returns None when the string does not parse. + """ + try: + spec = parse_shortcut_spec(shortcut_string) + except (ValueError, AttributeError): + return None + if spec.key is None: + return ("hold", MODIFIER_KEY_SIDES[spec.modifiers[0]]) + return ("combo", frozenset({_MAIN_KEY_ALIASES.get(spec.key, spec.key)})) + + +def shortcut_gestures_collide(first: str, second: str) -> bool: + """Whether two shortcut strings can be triggered by the same gesture (#805). + + String comparison alone misses gestures the backends treat as identical, + like "ctrl+alt+r" vs "alt+ctrl+r" or "ctrl+ctrl" vs + "left_ctrl+left_ctrl". + """ + first_gesture = shortcut_gesture(first) + second_gesture = shortcut_gesture(second) + if first_gesture is None or second_gesture is None: + return False + if first_gesture[0] != second_gesture[0]: + # A modifier press and a main-key press are different trigger events. + return False + return bool(first_gesture[1] & second_gesture[1]) + + def _main_key_label(token: str) -> str: """Human-readable label for a main-key token (e.g. 'r' -> 'R').""" if token in _NAMED_KEY_LABELS: diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index ed5a231..6d12f90 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -24,7 +24,8 @@ import re import threading import time from contextlib import contextmanager -from typing import TYPE_CHECKING, Any, Callable, Iterator, Literal, NamedTuple, Optional +from functools import partial +from typing import TYPE_CHECKING, Any, Callable, Iterator, Literal, NamedTuple, Optional, cast import gi @@ -2568,6 +2569,8 @@ class SettingsDialog(Gtk.Dialog): update_status_callback: callable = None, overlay_enabled_callback: Optional[Callable[[bool], None]] = None, hotkey_listener_update_callback: Optional[Callable[[], None]] = None, + language_shortcuts_update_callback: Optional[Callable[[], None]] = None, + history_update_callback: Optional[Callable[[], None]] = None, ): super().__init__(title="Vocalinux Settings", transient_for=parent, flags=0) # Force window decorations (title-bar close) on all WMs. An in-window @@ -2582,6 +2585,13 @@ class SettingsDialog(Gtk.Dialog): self.update_status_callback = update_status_callback self.overlay_enabled_callback = overlay_enabled_callback self.hotkey_listener_update_callback = hotkey_listener_update_callback + self.language_shortcuts_update_callback = language_shortcuts_update_callback + self.history_update_callback = history_update_callback + # Per-language shortcut rows (#805): a dict of row widgets per binding, + # populated by _build_language_shortcuts_section. + self._language_shortcut_rows: list[dict] = [] + self._language_shortcut_add_row: Optional[PreferenceRow] = None + self._recording_shortcut_target = None self._test_active = False self._test_result = "" self._initializing = True # Flag to prevent auto-apply during initialization @@ -2738,6 +2748,7 @@ class SettingsDialog(Gtk.Dialog): # Build UI sections into their topic pages self._build_shortcuts_section() + self._build_language_shortcuts_section() self._build_recognition_section() self._build_simple_model_section() self._build_engine_section() @@ -3639,6 +3650,8 @@ class SettingsDialog(Gtk.Dialog): logger.info(f"Transcription history toggled: {enabled}") self.config_manager.set("history", "enabled", enabled) self.config_manager.save_settings() + if self.history_update_callback: + self.history_update_callback() return False def _on_history_max_items_changed(self, widget: Gtk.SpinButton) -> None: @@ -3650,6 +3663,8 @@ class SettingsDialog(Gtk.Dialog): logger.info(f"Transcription history max items: {max_items}") self.config_manager.set("history", "max_items", max_items) self.config_manager.save_settings() + if self.history_update_callback: + self.history_update_callback() def _on_autostart_toggled(self, widget, state): """Handle toggle of the autostart switch.""" @@ -4457,8 +4472,11 @@ class SettingsDialog(Gtk.Dialog): self.custom_shortcut_row.set_no_show_all(True) group.add_row(self.custom_shortcut_row) - # Key-capture state for the Record button. + # Key-capture state for the Record button. ``_recording_shortcut_target`` + # names (entry, apply_fn, button, hint_label) so one recorder serves the + # main shortcut and every per-language row (#805). self._recording_shortcut = False + self._recording_shortcut_target = None self._evdev_shortcut_recorder = None self.connect("key-press-event", self._on_shortcut_key_press) self.connect("destroy", self._on_shortcut_recorder_destroy) @@ -4532,6 +4550,254 @@ class SettingsDialog(Gtk.Dialog): return False + def _build_language_shortcuts_section(self) -> None: + """Build the Language Shortcuts section (#805). + + One row per shortcut → language binding, an Add row at the bottom, and + an info label the shared key-recorder writes its hints into. + """ + group = PreferencesGroup( + title="Language Shortcuts", + description=( + "Start dictation in a specific language with its own key. The " + "binding lasts for that one dictation; your main language is " + "restored after. Needs a multilingual model." + ), + keywords=("language", "shortcut", "multilingual"), + ) + self.language_shortcuts_group = group + + for entry in self.config_manager.get_language_shortcuts(): + self._add_language_shortcut_row(entry["language"], entry["shortcut"]) + + add_button = Gtk.Button(label="Add Language Shortcut") + add_button.set_tooltip_text("Bind another language to its own key") + add_button.connect("clicked", self._on_add_language_shortcut_clicked) + self._language_shortcut_add_row = PreferenceRow( + title="Add a binding", + subtitle="Pick a language, then record a key for it", + widget=add_button, + keywords=("add", "language", "shortcut"), + ) + group.add_row(self._language_shortcut_add_row) + + self.shortcuts_tab.pack_start(group, False, False, 0) + + info_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + info_box.get_style_context().add_class("info-box") + info_box.set_margin_start(4) + info_box.set_margin_end(4) + info_box.set_margin_top(4) + + info_icon = Gtk.Image.new_from_icon_name("dialog-information-symbolic", Gtk.IconSize.MENU) + info_box.pack_start(info_icon, False, False, 0) + + self.language_shortcuts_info_label = Gtk.Label( + label="Language shortcuts take effect immediately and follow the " + "shortcut mode above.", + xalign=0, + wrap=True, + ) + self.language_shortcuts_info_label.get_style_context().add_class("tip-label") + info_box.pack_start(self.language_shortcuts_info_label, True, True, 0) + + self.shortcuts_tab.pack_start(info_box, False, False, 0) + + def _add_language_shortcut_row(self, language: str, shortcut: str) -> None: + """Append one language → shortcut binding row (#805).""" + hbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + + language_picker = SearchablePicker() + _style_combo(language_picker, _CONTROL_WIDTH - 40) + _prevent_scroll_on_hover(language_picker) + for lang_code, lang_info in SUPPORTED_LANGUAGES.items(): + language_picker.append(lang_code, str(cast(dict, lang_info)["name"])) + hbox.pack_start(language_picker, False, False, 0) + + shortcut_entry = Gtk.Entry() + shortcut_entry.set_placeholder_text("e.g. ctrl+alt+d") + shortcut_entry.set_width_chars(14) + shortcut_entry.set_tooltip_text( + "A modifier plus a key (ctrl+alt+d) or a function key (f10)" + ) + shortcut_entry.set_text(shortcut) + shortcut_entry.connect("changed", self._on_language_shortcut_row_changed) + hbox.pack_start(shortcut_entry, True, True, 0) + + record_button = Gtk.Button(label="Record") + record_button.set_tooltip_text("Click, then press your desired key combo") + hbox.pack_start(record_button, False, False, 0) + + remove_button = Gtk.Button.new_from_icon_name("list-remove-symbolic", Gtk.IconSize.BUTTON) + remove_button.set_tooltip_text("Remove this binding") + hbox.pack_start(remove_button, False, False, 0) + + # ``refs["row"]`` fills in below: handlers are wired before the row + # exists so the picker's initial set_active_id cannot emit stale titles. + refs = { + "row": None, + "language_picker": language_picker, + "shortcut_entry": shortcut_entry, + "record_button": record_button, + # The last binding this row actually persisted; a mid-edit entry + # (empty or half-typed) falls back to it rather than erasing the + # binding. + "last_valid_shortcut": shortcut.strip().lower(), + } + language_picker.connect("changed", partial(self._on_language_shortcut_picker_changed, refs)) + record_button.connect("clicked", partial(self._on_record_language_shortcut_clicked, refs)) + remove_button.connect("clicked", partial(self._on_remove_language_shortcut_clicked, refs)) + if not language_picker.set_active_id(language): + language_picker.set_active_id("auto") + + lang_info = cast(Optional[dict], SUPPORTED_LANGUAGES.get(language)) + lang_name = str(lang_info["name"]) if lang_info else language + row = PreferenceRow( + title=lang_name, + subtitle="", + widget=hbox, + keywords=("language", "shortcut", lang_name), + ) + refs["row"] = row + + # Newest binding above the Add row so the group reads top-down. + if self._language_shortcut_add_row is not None: + index = self.language_shortcuts_group.rows.index(self._language_shortcut_add_row) + self.language_shortcuts_group.listbox.insert(row, index) + self.language_shortcuts_group.rows.insert(index, row) + else: + self.language_shortcuts_group.add_row(row) + self._language_shortcut_rows.append(refs) + + def _on_language_shortcut_picker_changed( + self, refs: dict[str, Any], picker: SearchablePicker + ) -> None: + """Apply a row's language change: retitle it and persist (#805).""" + row = refs.get("row") + if row is not None: + lang_id = picker.get_active_id() or "" + lang_info = cast(Optional[dict], SUPPORTED_LANGUAGES.get(lang_id)) + lang_name = str(lang_info["name"]) if lang_info else lang_id + row.set_title(lang_name) + self._on_language_shortcut_row_changed(picker) + + def _on_language_shortcut_row_changed(self, *args: object) -> None: + """Persist the bindings whenever a row's language or key changes (#805).""" + self._persist_language_shortcuts() + + def _on_record_language_shortcut_clicked( + self, refs: dict[str, Any], button: Gtk.Button + ) -> None: + """Arm key-capture aimed at a language row's entry (#805).""" + + def _apply(shortcut: str) -> None: + # _commit_recorded_shortcut already wrote the captured key to the + # entry, whose changed signal persisted it; setting the same text + # again would write config a second time and rebuild the + # listeners for one recorded key. + if refs["shortcut_entry"].get_text() != shortcut: + refs["shortcut_entry"].set_text(shortcut) + + self._begin_shortcut_recording( + refs["shortcut_entry"], + _apply, + refs["record_button"], + self.language_shortcuts_info_label, + ) + + def _on_remove_language_shortcut_clicked( + self, refs: dict[str, Any], button: Gtk.Button + ) -> None: + """Remove a language binding row and persist (#805).""" + # An armed Record capture aimed at this row's entry must be cancelled + # first: the commit would otherwise write into a destroyed widget. + target = getattr(self, "_recording_shortcut_target", None) + if ( + getattr(self, "_recording_shortcut", False) + and target is not None + and target[0] is refs["shortcut_entry"] + ): + self._stop_recording_shortcut() + row = refs["row"] + self.language_shortcuts_group.listbox.remove(row) + if row in self.language_shortcuts_group.rows: + self.language_shortcuts_group.rows.remove(row) + self._language_shortcut_rows.remove(refs) + row.destroy() + self._persist_language_shortcuts() + + def _on_add_language_shortcut_clicked(self, button: Gtk.Button) -> None: + """Append a fresh binding row (defaults to auto-detect) (#805).""" + self._add_language_shortcut_row("auto", "") + self.language_shortcuts_group.show_all() + self._persist_language_shortcuts() + + def _collect_language_shortcuts(self) -> list[dict[str, str]]: + """Read the current rows as [{shortcut, language}] for config (#805).""" + entries = [] + for refs in self._language_shortcut_rows: + entries.append( + { + "shortcut": refs["shortcut_entry"].get_text(), + "language": refs["language_picker"].get_active_id() or "auto", + } + ) + return entries + + def _persist_language_shortcuts(self) -> None: + """Write the rows to config and refresh the live listeners (#805). + + A row mid-edit — empty, half-typed, or claiming a key another row + already holds — would be dropped by config normalization, erasing its + binding when the dialog closes. Such rows keep their last persisted + binding instead, and a rejected duplicate is called out on the info + label. + """ + if self._initializing: + return + entries = [] + rejected = [] + claimed = set() + for refs in self._language_shortcut_rows: + language = refs["language_picker"].get_active_id() or "auto" + shortcut = refs["shortcut_entry"].get_text().strip().lower() + last_valid = refs.get("last_valid_shortcut", "") + if shortcut and is_valid_shortcut(shortcut) and shortcut not in claimed: + refs["last_valid_shortcut"] = shortcut + claimed.add(shortcut) + elif last_valid and last_valid not in claimed: + if shortcut and is_valid_shortcut(shortcut): + rejected.append(shortcut) + shortcut = last_valid + claimed.add(shortcut) + else: + if shortcut and is_valid_shortcut(shortcut): + rejected.append(shortcut) + continue + entries.append({"shortcut": shortcut, "language": language}) + self.config_manager.set_language_shortcuts(entries) + self.config_manager.save_settings() + if self.language_shortcuts_update_callback: + self.language_shortcuts_update_callback() + self._report_language_shortcut_rejections(rejected) + + def _report_language_shortcut_rejections(self, rejected: list[str]) -> None: + """Surface dropped duplicate keys on the section's info label (#805).""" + label = getattr(self, "language_shortcuts_info_label", None) + if label is None: + return + if rejected: + keys = ", ".join(sorted(set(rejected))) + label.set_markup( + f"Shortcut " + f"{GLib.markup_escape_text(keys)} is already bound on " + "another row." + ) + elif not getattr(self, "_recording_shortcut", False): + label.set_text( + "Language shortcuts take effect immediately and follow the " "shortcut mode above." + ) + def _is_preset_shortcut(self, shortcut: str) -> bool: """Return True if shortcut is one of the built-in double-tap presets.""" return shortcut in SUPPORTED_SHORTCUTS @@ -4636,20 +4902,46 @@ class SettingsDialog(Gtk.Dialog): def _on_record_shortcut_clicked(self, widget): """Begin capturing the next key combo pressed in the dialog.""" + self._begin_shortcut_recording( + self.custom_shortcut_entry, + self._apply_custom_shortcut, + self.record_shortcut_button, + self.shortcut_info_label, + ) + + def _begin_shortcut_recording( + self, + entry: Gtk.Entry, + apply_fn: Callable[[str], None], + button: Gtk.Button, + hint_label: Gtk.Label, + ) -> None: + """Arm key-capture for one shortcut target. + + ``apply_fn`` receives the captured shortcut string; ``hint_label`` gets + the press-keys hint and any error/cancel messages for this capture. + """ + if self._recording_shortcut: + # Only one capture can be armed: reset the previously armed + # button instead of leaving two targets captioned "Press keys…". + self._stop_recording_shortcut() self._recording_shortcut = True - self.record_shortcut_button.set_label("Press keys…") - self.shortcut_info_label.set_markup( + self._recording_shortcut_target = (entry, apply_fn, button, hint_label) + button.set_label("Press keys…") + hint_label.set_markup( "Press a modifier + key (e.g. Alt+R), or an F-key. Press Esc to cancel." ) self._start_evdev_shortcut_recorder() - def _stop_recording_shortcut(self): - """Exit key-capture mode and restore the Record button.""" + def _stop_recording_shortcut(self) -> None: + """Exit key-capture mode and restore the armed Record button.""" self._recording_shortcut = False self._stop_evdev_shortcut_recorder() - if getattr(self, "record_shortcut_button", None) is not None: + target = getattr(self, "_recording_shortcut_target", None) + self._recording_shortcut_target = None + if target is not None: try: - self.record_shortcut_button.set_label("Record") + target[2].set_label("Record") except Exception: pass @@ -4677,9 +4969,13 @@ class SettingsDialog(Gtk.Dialog): return False if not shortcut or not is_valid_shortcut(shortcut): return False - self.custom_shortcut_entry.set_text(shortcut) + target = getattr(self, "_recording_shortcut_target", None) + if target is None: + return False + entry, apply_fn, _button, _hint_label = target + entry.set_text(shortcut) self._stop_recording_shortcut() - self._apply_custom_shortcut(shortcut) + apply_fn(shortcut) return True def _on_evdev_recorded_shortcut(self, shortcut: str) -> None: @@ -4715,12 +5011,14 @@ class SettingsDialog(Gtk.Dialog): return True if not getattr(self, "_recording_shortcut", False): return True + target = getattr(self, "_recording_shortcut_target", None) + hint_label = target[3] if target is not None else self.shortcut_info_label if keyname == "Escape" and not shortcut: self._stop_recording_shortcut() - self.shortcut_info_label.set_text("Recording cancelled.") + hint_label.set_text("Recording cancelled.") return True - self.shortcut_info_label.set_markup( + hint_label.set_markup( "Need a modifier + key, or an F1–F24 " "function key alone. Try again or press Esc to cancel." ) diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index b5ca42e..9e367ee 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -10,6 +10,7 @@ import os import signal import threading import time +from functools import partial from typing import Any, Callable, Optional, Sequence, cast import gi @@ -53,10 +54,15 @@ from ..utils.update_monitor import UpdateMonitor from ..utils.whispercpp_model_info import TDRZ_MODEL, WHISPERCPP_MODEL_INFO, is_model_downloaded from . import notifications from .config_manager import get_shared_config_manager -from .keyboard_backends import DEFAULT_SHORTCUT, DEFAULT_SHORTCUT_MODE +from .keyboard_backends import ( + DEFAULT_SHORTCUT, + DEFAULT_SHORTCUT_MODE, + SHORTCUT_MODES, + shortcut_gestures_collide, +) from .keyboard_shortcuts import KeyboardShortcutManager from .settings_dialog import ModelDownloadDialog, SettingsDialog, recommended_model_for_engine -from .transcription_history import TranscriptionHistory +from .transcription_history import DEFAULT_MAX_ITEMS, TranscriptionHistory logger = logging.getLogger(__name__) @@ -125,6 +131,19 @@ class TrayIndicator: the speech recognition process. """ + # Tests build TrayIndicator stubs with __new__ that skip __init__; keep a + # class-level default so teardown paths like _quit() still find the list. + _language_shortcut_managers: list = [] + + #: Same for the push-to-talk owner slot: a __new__-built stub must answer + #: "no binding owns the live session" instead of raising on release paths. + _ptt_owner: Optional["KeyboardShortcutManager"] = None + + #: Serializes push-to-talk press/release handling. Press and release + #: callbacks run on separate backend threads, so a release landing while + #: start_recognition is still running must not be lost (#805). + _ptt_lock = threading.RLock() + def __init__( self, speech_engine: SpeechRecognitionManagerProtocol, @@ -168,6 +187,15 @@ class TrayIndicator: # Initialize keyboard shortcut manager with configured shortcut and mode self.shortcut_manager = KeyboardShortcutManager(shortcut=shortcut, mode=mode) + # One listener per bound language shortcut (#805); rebuilt by + # _setup_language_shortcuts alongside the main one. + self._language_shortcut_managers: list[KeyboardShortcutManager] = [] + # The binding whose press started the live push-to-talk session, so an + # unrelated key's release cannot end it (#805). + self._ptt_owner: Optional[KeyboardShortcutManager] = None + # Serializes the press/release pair across backend threads, so a + # release landing while start_recognition runs cannot be lost. + self._ptt_lock = threading.RLock() # Ensure icon directory exists os.makedirs(ICON_DIR, exist_ok=True) @@ -328,6 +356,9 @@ class TrayIndicator: # in over D-Bus (see VocalinuxDBusService). if self._external_activation_active(): logger.info("Internal hotkey listener disabled (external activation via D-Bus)") + # The whole internal listener family is off; per-language + # listeners must not keep firing either. + self._stop_language_shortcut_managers() return # Get configured mode from config @@ -340,11 +371,109 @@ class TrayIndicator: elif mode == "push_to_talk": # Register press/release callbacks for push-to-talk mode self.shortcut_manager.register_press_callback(self._start_recognition) - self.shortcut_manager.register_release_callback(self._stop_recognition) + self.shortcut_manager.register_release_callback(self._release_main_push_to_talk) # Start the keyboard shortcut manager self.shortcut_manager.start() + self._setup_language_shortcuts() + + def _setup_language_shortcuts(self) -> None: + """(Re)build a listener per configured language shortcut (#805). + + Each entry gets its own KeyboardShortcutManager because the backends + open the input devices read-only and never grab them, so parallel + listeners do not interfere. A binding whose gesture collides with the + main shortcut or an earlier language binding is skipped: the same + gesture cannot fire both. + """ + # Same stop-first protection as _setup_keyboard_shortcuts: rebuilding + # below removes the release callback a held push-to-talk key needs to + # end its session, so the hold must be ended before it can be lost. + if self.speech_engine.state != RecognitionState.IDLE: + logger.info("Stopping active recognition before rebuilding language shortcuts") + self._stop_recognition() + + self._stop_language_shortcut_managers() + + mode = self.config_manager.get_str("shortcuts", "mode", DEFAULT_SHORTCUT_MODE) + if mode not in SHORTCUT_MODES: + # A hand-edited mode must not take down the whole shortcut setup + # via a backend constructor's ValueError. + logger.warning(f"Unknown shortcut mode {mode!r} for language shortcuts") + mode = DEFAULT_SHORTCUT_MODE + main_shortcut = ( + self.config_manager.get_str("shortcuts", "toggle_recognition", DEFAULT_SHORTCUT) + .strip() + .lower() + ) + + # Collisions are judged on the gestures the backends match, not the + # strings: "ctrl+alt+r" and "alt+ctrl+r" name the same key press, and + # "left_ctrl+left_ctrl" fires inside "ctrl+ctrl"'s key set. + bound = [main_shortcut] + for entry in self.config_manager.get_language_shortcuts(): + shortcut = entry["shortcut"] + language = entry["language"] + if any(shortcut_gestures_collide(shortcut, other) for other in bound): + logger.warning( + f"Skipping language shortcut {shortcut!r} for {language}: " + "it shares its gesture with another binding" + ) + continue + bound.append(shortcut) + + try: + manager = KeyboardShortcutManager(shortcut=shortcut, mode=mode) + except ValueError as exc: + logger.warning(f"Skipping language shortcut {shortcut!r} for {language}: {exc}") + continue + if mode == "toggle": + manager.register_toggle_callback( + partial(self._toggle_recognition_in_language, language) + ) + else: + manager.register_press_callback( + partial(self._start_recognition_in_language, language, manager) + ) + manager.register_release_callback( + partial(self._release_language_push_to_talk, manager) + ) + manager.start() + self._language_shortcut_managers.append(manager) + logger.info(f"Language shortcut {shortcut} -> {language} armed ({mode} mode)") + + def _apply_history_settings(self) -> None: + """Push the saved history preferences onto the live store (#805). + + Toggling history off must stop retention immediately — not after a + restart — and the snippets-keep limit applies to what's already held. + """ + if self.transcription_history is None: + return + enabled = self.config_manager.get_bool("history", "enabled", True) + max_items = self.config_manager.get_int("history", "max_items", DEFAULT_MAX_ITEMS) + self.transcription_history.set_max_items(max_items) + self.transcription_history.set_enabled(enabled) + + def _stop_language_shortcut_managers(self) -> None: + """Stop every per-language listener and drop the managers.""" + for manager in self._language_shortcut_managers: + try: + manager.stop() + except Exception: + logger.exception("Failed to stop a language shortcut manager") + self._language_shortcut_managers = [] + + def refresh_language_shortcuts(self) -> None: + """Rebuild the per-language listeners after a settings change (#805).""" + if self._external_activation_active(): + # The internal listener family is off; a binding edit must not + # arm language listeners behind the external-activation setting. + self._stop_language_shortcut_managers() + return + self._setup_language_shortcuts() + def _init_icons(self): """Initialize the icon files for the tray indicator.""" # Ensure icon directory exists @@ -558,17 +687,79 @@ class TrayIndicator: dialog.show() return False - def _toggle_recognition(self): + def _toggle_recognition(self) -> None: """Toggle the recognition state between IDLE and LISTENING.""" - if self.speech_engine.state == RecognitionState.IDLE: - self.speech_engine.start_recognition() - else: - self.speech_engine.stop_recognition() + with self._ptt_lock: + self._ptt_owner = None + if self.speech_engine.state == RecognitionState.IDLE: + self.speech_engine.start_recognition() + else: + self.speech_engine.stop_recognition() - def _start_recognition(self): + def _start_recognition(self) -> None: """Start voice recognition (for push-to-talk mode).""" - if self.speech_engine.state == RecognitionState.IDLE: - self.speech_engine.start_recognition(mode="push_to_talk") + self._push_to_talk_press( + self.shortcut_manager, + lambda: self.speech_engine.start_recognition(mode="push_to_talk"), + ) + + def _push_to_talk_press( + self, manager: Optional[KeyboardShortcutManager], start: Callable[[], bool] + ) -> None: + """Push-to-talk press shared by the main and per-language bindings (#805). + + The release runs on another backend thread and used to be dropped when + it landed between LISTENING and the owner assignment, leaving + dictation running with the key already up. Holding the lock through + the whole start makes the release wait instead: it then finds the + owner and ends the session normally. + """ + with self._ptt_lock: + if self.speech_engine.state == RecognitionState.IDLE and start(): + self._ptt_owner = manager + + def _push_to_talk_release(self, manager: KeyboardShortcutManager) -> None: + """Push-to-talk release: end the session only when this binding owns it.""" + with self._ptt_lock: + if self._ptt_owner is manager: + self._stop_recognition() + + def _release_main_push_to_talk(self) -> None: + """End a push-to-talk session only when the main binding started it (#805).""" + self._push_to_talk_release(self.shortcut_manager) + + def _release_language_push_to_talk(self, manager: KeyboardShortcutManager) -> None: + """End a push-to-talk session only when this binding started it (#805). + + Sharing one stop callback across listeners let an unrelated binding's + release end a session it never started. + """ + self._push_to_talk_release(manager) + + def _toggle_recognition_in_language(self, language: str) -> None: + """Toggle-style trigger for a per-language shortcut (#805). + + While idle it starts a one-shot dictation in ``language``; while + dictating it stops, matching the main toggle's behaviour regardless + of which language the current utterance is in. + """ + with self._ptt_lock: + self._ptt_owner = None + if self.speech_engine.state == RecognitionState.IDLE: + self.speech_engine.start_recognition_with_language(language) + else: + self.speech_engine.stop_recognition() + + def _start_recognition_in_language( + self, language: str, manager: Optional[KeyboardShortcutManager] = None + ) -> None: + """Push-to-talk start for a per-language shortcut (#805).""" + self._push_to_talk_press( + manager, + lambda: self.speech_engine.start_recognition_with_language( + language, mode="push_to_talk" + ), + ) def _external_start(self) -> None: """Start recognition for an external (D-Bus) trigger. @@ -577,18 +768,24 @@ class TrayIndicator: `vocalinux --start` transcribes immediately/with silence detection rather than deferring until a Stop. """ - if self.speech_engine.state == RecognitionState.IDLE: - self.speech_engine.start_recognition() + with self._ptt_lock: + self._ptt_owner = None + if self.speech_engine.state == RecognitionState.IDLE: + self.speech_engine.start_recognition() def _external_stop(self) -> None: """Stop recognition for an external (D-Bus) trigger.""" - if self.speech_engine.state != RecognitionState.IDLE: - self.speech_engine.stop_recognition() + with self._ptt_lock: + self._ptt_owner = None + if self.speech_engine.state != RecognitionState.IDLE: + self.speech_engine.stop_recognition() - def _stop_recognition(self): + def _stop_recognition(self) -> None: """Stop voice recognition (for push-to-talk mode).""" - if self.speech_engine.state != RecognitionState.IDLE: - self.speech_engine.stop_recognition() + with self._ptt_lock: + self._ptt_owner = None + if self.speech_engine.state != RecognitionState.IDLE: + self.speech_engine.stop_recognition() def _add_menu_item(self, label: str, callback: Callable): """ @@ -1159,6 +1356,8 @@ class TrayIndicator: ), overlay_enabled_callback=self.set_overlay_enabled, hotkey_listener_update_callback=self._setup_keyboard_shortcuts, + language_shortcuts_update_callback=self.refresh_language_shortcuts, + history_update_callback=self._apply_history_settings, ) dialog.connect("response", self._on_settings_dialog_response) dialog.connect("destroy", self._on_settings_dialog_destroyed) @@ -1552,8 +1751,9 @@ class TrayIndicator: self._cleanup_input_monitor() - # Stop the keyboard shortcut manager + # Stop the keyboard shortcut managers self.shortcut_manager.stop() + self._stop_language_shortcut_managers() if getattr(self, "overlay", None) is not None: self.overlay.destroy() diff --git a/tests/test_language_shortcuts.py b/tests/test_language_shortcuts.py new file mode 100644 index 0000000..b210682 --- /dev/null +++ b/tests/test_language_shortcuts.py @@ -0,0 +1,911 @@ +"""Per-language dictation shortcuts: config schema, one-shot engine override, +tray listeners, and settings-row plumbing (#805).""" + +from __future__ import annotations + +import importlib +import os +import queue +import sys +import tempfile +import threading +from collections.abc import Iterator +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +import vocalinux.ui +from vocalinux.common_types import RecognitionState +from vocalinux.ui.config_manager import ConfigManager, normalize_language_shortcuts +from vocalinux.utils.vosk_model_info import SUPPORTED_LANGUAGES + + +@pytest.fixture(scope="module") +def settings_dialog() -> Iterator[Any]: + """Import settings_dialog with real bases for its GTK subclasses. + + Same approach as tests/test_follow_keyboard_layout.py; both sys.modules + and the package attribute are restored so later tests patching the module + by name do not reach a second module object (#686). + """ + repository = sys.modules["gi.repository"] + bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")} + saved_module = sys.modules.pop("vocalinux.ui.settings_dialog", None) + saved_attribute = getattr(vocalinux.ui, "settings_dialog", None) + try: + with patch.object(repository, "Gtk", MagicMock(**bases)): + module = importlib.import_module("vocalinux.ui.settings_dialog") + yield module + finally: + sys.modules.pop("vocalinux.ui.settings_dialog", None) + if saved_module is not None: + sys.modules["vocalinux.ui.settings_dialog"] = saved_module + if saved_attribute is not None: + vocalinux.ui.settings_dialog = saved_attribute + elif hasattr(vocalinux.ui, "settings_dialog"): + del vocalinux.ui.settings_dialog + + +@pytest.fixture +def dialog_class(settings_dialog: Any) -> type[Any]: + return settings_dialog.SettingsDialog + + +# --- config normalization --------------------------------------------------- + + +def test_normalize_language_shortcuts_keeps_valid_bindings() -> None: + entries = normalize_language_shortcuts( + [ + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "ctrl+alt+g", "language": "hu"}, + {"shortcut": "f5", "language": "auto"}, + ] + ) + assert entries == [ + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "ctrl+alt+g", "language": "hu"}, + {"shortcut": "f5", "language": "auto"}, + ] + + +@pytest.mark.parametrize("raw", [None, "alt+d", 42, {"shortcut": "alt+d"}]) +def test_normalize_language_shortcuts_rejects_non_lists(raw: Any) -> None: + assert normalize_language_shortcuts(raw) == [] + + +@pytest.mark.parametrize( + "item", + [ + "alt+d", # not a mapping + {"shortcut": "alt+d"}, # missing language + {"language": "de"}, # missing shortcut + {"shortcut": 5, "language": "de"}, # wrong types + {"shortcut": "alt+d", "language": "klingon"}, # not a catalog id + {"shortcut": "alt", "language": "de"}, # bare modifier, unbindable + {"shortcut": "", "language": "de"}, # empty shortcut + {"shortcut": "alt+d", "language": "layout"}, # sentinel, not a language + ], +) +def test_normalize_language_shortcuts_drops_bad_entries(item: Any) -> None: + assert normalize_language_shortcuts([item]) == [] + + +def test_normalize_language_shortcuts_dedupes_by_shortcut() -> None: + """The first binding wins when two rows claim the same key.""" + entries = normalize_language_shortcuts( + [ + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "alt+d", "language": "fr"}, + ] + ) + assert entries == [{"shortcut": "alt+d", "language": "de"}] + + +def test_language_shortcuts_roundtrip_through_config(tmp_path: Any) -> None: + config_dir = tmp_path / "vocalinux" + config_dir.mkdir() + config_file = config_dir / "config.json" + with ( + patch("vocalinux.ui.config_manager.CONFIG_DIR", str(config_dir)), + patch("vocalinux.ui.config_manager.CONFIG_FILE", str(config_file)), + patch("vocalinux.utils.system_language.detect_system_language", return_value=None), + ): + manager = ConfigManager() + assert manager.get_language_shortcuts() == [] + + manager.set_language_shortcuts( + [ + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "broken", "language": "fr"}, # dropped by normalize + ] + ) + assert manager.config["shortcuts"]["language_shortcuts"] == [ + {"shortcut": "alt+d", "language": "de"} + ] + assert manager.get_language_shortcuts() == [{"shortcut": "alt+d", "language": "de"}] + + # Persisted to disk. + manager.save_config() + reloaded = ConfigManager() + assert reloaded.get_language_shortcuts() == [{"shortcut": "alt+d", "language": "de"}] + + +def test_get_language_shortcuts_survives_a_broken_shortcuts_section() -> None: + """A hand-edited config replacing the section with a scalar reads as empty.""" + with tempfile.TemporaryDirectory() as temp_dir: + config_dir = os.path.join(temp_dir, "vocalinux") + config_file = os.path.join(config_dir, "config.json") + os.makedirs(config_dir, exist_ok=True) + with open(config_file, "w") as handle: + handle.write('{"shortcuts": "oops"}') + with ( + patch("vocalinux.ui.config_manager.CONFIG_DIR", config_dir), + patch("vocalinux.ui.config_manager.CONFIG_FILE", config_file), + patch( + "vocalinux.utils.system_language.detect_system_language", + return_value=None, + ), + ): + manager = ConfigManager() + assert manager.get_language_shortcuts() == [] + manager.set_language_shortcuts([{"shortcut": "alt+d", "language": "de"}]) + assert manager.get_language_shortcuts() == [{"shortcut": "alt+d", "language": "de"}] + + +# --- one-shot language override on the manager ------------------------------ + + +def _manager_stub( + engine: str = "whisper_cpp", + model_size: str = "small", + language: str = "en-us", + language_preference: str = "en-us", +) -> Any: + """A manager carrying only the fields the one-shot path touches.""" + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + manager = SpeechRecognitionManager.__new__(SpeechRecognitionManager) + manager.engine = engine + manager.model_size = model_size + manager.language = language + manager.language_preference = language_preference + manager.state = RecognitionState.IDLE + manager.state_callbacks = [] + manager._faster_whisper_engine = None + manager.command_processor = MagicMock() + manager._pending_language_override = None + manager._oneshot_language_restore = None + manager._segment_started_at = None + return manager + + +def test_language_shortcut_starts_dictation_in_that_language() -> None: + import vocalinux.speech_recognition.recognition_manager as rm + + manager = _manager_stub() + manager.start_recognition = MagicMock( + side_effect=lambda mode="toggle": (manager._apply_pending_language_override(), True)[1] + ) + + assert manager.start_recognition_with_language("de") is True + manager.start_recognition.assert_called_once_with(mode="toggle") + assert manager.language == "de" + assert manager.command_processor.set_language.call_args.args == ("de",) + # Consumed: a later plain start must not see the override. + assert manager._pending_language_override is None + assert manager._oneshot_language_restore == "en-us" + + +def test_language_shortcut_wins_over_follow_layout() -> None: + """An explicit key press outranks the follow-mode resolution (#821).""" + import vocalinux.speech_recognition.recognition_manager as rm + + manager = _manager_stub(language_preference="layout") + + def fake_start(mode: str = "toggle") -> bool: + manager._refresh_language_from_layout() + manager._apply_pending_language_override() + return True + + manager.start_recognition = MagicMock(side_effect=fake_start) + with patch.object(rm, "language_for_active_layout", return_value="fr"): + assert manager.start_recognition_with_language("de") is True + + assert manager.language == "de" + assert manager._oneshot_language_restore == "fr" + manager._update_state(RecognitionState.IDLE) + assert manager.language == "fr" + + +def test_language_is_restored_when_dictation_ends() -> None: + """The configured language comes back on the IDLE transition.""" + manager = _manager_stub() + manager._oneshot_language_restore = "en-us" + manager.language = "de" + + manager._update_state(RecognitionState.IDLE) + + assert manager.language == "en-us" + assert manager._oneshot_language_restore is None + manager.command_processor.set_language.assert_called_with("en-us") + + +def test_language_is_restored_on_error_too() -> None: + manager = _manager_stub() + manager._oneshot_language_restore = "en-us" + manager.language = "de" + + manager._update_state(RecognitionState.ERROR) + + assert manager.language == "en-us" + assert manager._oneshot_language_restore is None + + +def test_shortcut_language_matching_current_language_leaves_no_restore() -> None: + """No-op overrides never register a restore.""" + manager = _manager_stub(language="de") + manager._pending_language_override = "de" + + manager._apply_pending_language_override() + + assert manager.language == "de" + assert manager._oneshot_language_restore is None + manager.command_processor.set_language.assert_not_called() + + +@pytest.mark.parametrize("engine,model_size", [("vosk", "small"), ("parakeet", "turbo")]) +def test_unsupported_engine_refuses_instead_of_dictating_wrong( + engine: str, model_size: str +) -> None: + """VOSK/Parakeet cannot honour a per-utterance language; refuse loudly.""" + import vocalinux.speech_recognition.recognition_manager as rm + + manager = _manager_stub(engine=engine, model_size=model_size, language="en-us") + manager.start_recognition = MagicMock(return_value=True) + + with ( + patch.object(rm, "play_error_sound") as mock_error, + patch.object(rm, "_show_notification") as mock_notify, + ): + assert manager.start_recognition_with_language("de") is False + + manager.start_recognition.assert_not_called() + mock_error.assert_called_once() + mock_notify.assert_called_once() + assert manager.language == "en-us" + assert manager._pending_language_override is None + + +def test_english_only_whispercpp_model_refuses_the_shortcut() -> None: + """small.en has no non-English weights to aim the shortcut at.""" + import vocalinux.speech_recognition.recognition_manager as rm + + manager = _manager_stub(model_size="small.en") + manager.start_recognition = MagicMock(return_value=True) + + with ( + patch.object(rm, "play_error_sound") as mock_error, + patch.object(rm, "_show_notification") as mock_notify, + ): + assert manager.start_recognition_with_language("de") is False + + manager.start_recognition.assert_not_called() + mock_error.assert_called_once() + mock_notify.assert_called_once() + + +def test_same_language_shortcut_runs_on_english_only_model() -> None: + """No language switch is needed to dictate in the model's own language.""" + import vocalinux.speech_recognition.recognition_manager as rm + + manager = _manager_stub(model_size="small.en", language="en-us") + manager.start_recognition = MagicMock(return_value=True) + + with patch.object(rm, "_show_notification") as mock_notify: + assert manager.start_recognition_with_language("en-us") is True + + mock_notify.assert_not_called() + + +def test_pending_override_is_consumed_even_when_start_is_refused() -> None: + """A refused start (auto-pause, model missing) must not leak the override.""" + manager = _manager_stub() + manager.start_recognition = MagicMock(return_value=False) + + assert manager.start_recognition_with_language("de") is False + assert manager._pending_language_override is None + assert manager._oneshot_language_restore is None + + +def test_reconfiguring_the_language_clears_a_pending_restore() -> None: + """A mid-dictation language change must not be undone by the restore.""" + manager = _manager_stub() + manager._oneshot_language_restore = "en-us" + manager.language = "de" + manager._voice_commands_preference = None + manager._snapshot_reconfigure_state = MagicMock(return_value={}) + + manager.reconfigure(language="fr") + + assert manager._oneshot_language_restore is None + assert manager.language == "fr" + + +# --- tray wiring ------------------------------------------------------------ + + +def _tray_stub(shortcut_mode: str = "toggle", entries: Any = None) -> Any: + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = TrayIndicator.__new__(TrayIndicator) + tray.config_manager = MagicMock() + config_values = { + "mode": shortcut_mode, + "toggle_recognition": "right_alt+right_alt", + } + tray.config_manager.get_str.side_effect = lambda section, key, default="": config_values.get( + key, default + ) + tray.config_manager.get_language_shortcuts.return_value = entries or [] + tray.speech_engine = MagicMock() + tray.speech_engine.state = RecognitionState.IDLE + tray._language_shortcut_managers = [] + # D-Bus activation unavailable -> the internal listener family is on, + # which is what these tests exercise. + tray._external_activation_unavailable = True + return tray + + +def test_tray_builds_one_listener_per_binding() -> None: + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub( + entries=[ + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "ctrl+alt+g", "language": "hu"}, + ] + ) + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + created = [MagicMock(), MagicMock()] + manager_class.side_effect = created + + TrayIndicator._setup_language_shortcuts(tray) + + assert manager_class.call_count == 2 + manager_class.assert_any_call(shortcut="alt+d", mode="toggle") + manager_class.assert_any_call(shortcut="ctrl+alt+g", mode="toggle") + assert tray._language_shortcut_managers == created + for manager in created: + manager.register_toggle_callback.assert_called_once() + manager.start.assert_called_once() + + +def test_tray_skips_bindings_that_collide() -> None: + """A language shortcut equal to the main one cannot fire both.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub( + entries=[ + {"shortcut": "right_alt+right_alt", "language": "de"}, # main + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "alt+d", "language": "fr"}, # duplicate of previous + ] + ) + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + TrayIndicator._setup_language_shortcuts(tray) + + manager_class.assert_called_once_with(shortcut="alt+d", mode="toggle") + + +def test_tray_language_managers_follow_push_to_talk() -> None: + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub( + shortcut_mode="push_to_talk", + entries=[{"shortcut": "alt+d", "language": "de"}], + ) + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + manager = MagicMock() + manager_class.return_value = manager + + TrayIndicator._setup_language_shortcuts(tray) + + manager_class.assert_called_once_with(shortcut="alt+d", mode="push_to_talk") + manager.register_press_callback.assert_called_once() + # The release is routed through the owner-gated helper for this binding, + # not a stop callback shared by every listener. + release_callback = manager.register_release_callback.call_args.args[0] + assert release_callback.func == tray._release_language_push_to_talk + assert release_callback.args == (manager,) + + +def test_tray_toggle_in_language_starts_a_one_shot() -> None: + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub() + tray.speech_engine.state = RecognitionState.IDLE + + TrayIndicator._toggle_recognition_in_language(tray, "de") + + tray.speech_engine.start_recognition_with_language.assert_called_once_with("de") + + +def test_tray_toggle_in_language_stops_while_dictating() -> None: + """Same as the main toggle: a press while dictating ends it.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub() + tray.speech_engine.state = RecognitionState.LISTENING + + TrayIndicator._toggle_recognition_in_language(tray, "de") + + tray.speech_engine.stop_recognition.assert_called_once() + tray.speech_engine.start_recognition_with_language.assert_not_called() + + +def test_tray_push_to_talk_in_language_starts_a_one_shot() -> None: + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub() + TrayIndicator._start_recognition_in_language(tray, "de") + + tray.speech_engine.start_recognition_with_language.assert_called_once_with( + "de", mode="push_to_talk" + ) + + +def test_tray_language_release_stops_only_its_own_session() -> None: + """Another binding's release must not end a hold it did not start.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub(shortcut_mode="push_to_talk") + german, french = MagicMock(), MagicMock() + tray.speech_engine.state = RecognitionState.LISTENING + tray._ptt_owner = german + + TrayIndicator._release_language_push_to_talk(tray, french) + tray.speech_engine.stop_recognition.assert_not_called() + + TrayIndicator._release_language_push_to_talk(tray, german) + tray.speech_engine.stop_recognition.assert_called_once() + assert tray._ptt_owner is None + + +def test_tray_ptt_release_during_start_still_ends_the_session() -> None: + """A release landing mid-start waits for the owner claim, then stops (#805).""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub(shortcut_mode="push_to_talk") + german = MagicMock() + tray.speech_engine.state = RecognitionState.IDLE + + def fake_start(*_args: Any, **_kwargs: Any) -> bool: + tray.speech_engine.state = RecognitionState.LISTENING + releaser.start() + return True + + releaser = threading.Thread( + target=TrayIndicator._release_language_push_to_talk, args=(tray, german) + ) + tray.speech_engine.start_recognition_with_language.side_effect = fake_start + + TrayIndicator._start_recognition_in_language(tray, "de", german) + releaser.join(timeout=5) + + tray.speech_engine.stop_recognition.assert_called_once() + assert tray._ptt_owner is None + + +def test_tray_main_ptt_release_during_start_still_ends_the_session() -> None: + """The main binding has the same deferred-release guarantee (#805).""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub(shortcut_mode="push_to_talk") + tray.shortcut_manager = MagicMock() + tray.speech_engine.state = RecognitionState.IDLE + + def fake_start(*_args: Any, **_kwargs: Any) -> bool: + tray.speech_engine.state = RecognitionState.LISTENING + releaser.start() + return True + + releaser = threading.Thread(target=TrayIndicator._release_main_push_to_talk, args=(tray,)) + tray.speech_engine.start_recognition.side_effect = fake_start + + TrayIndicator._start_recognition(tray) + releaser.join(timeout=5) + + tray.speech_engine.stop_recognition.assert_called_once() + assert tray._ptt_owner is None + + +def test_tray_skips_gesture_equivalent_bindings() -> None: + """Different spellings of the same gesture still collide.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub( + entries=[ + {"shortcut": "ctrl+alt+r", "language": "de"}, + {"shortcut": "alt+ctrl+r", "language": "fr"}, # reordered, same gesture + {"shortcut": "ctrl+enter", "language": "es"}, + {"shortcut": "alt+return", "language": "hu"}, # enter/return alias + ] + ) + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + TrayIndicator._setup_language_shortcuts(tray) + + armed = [call.kwargs["shortcut"] for call in manager_class.call_args_list] + assert armed == ["ctrl+alt+r", "ctrl+enter"] + + +def test_tray_skips_pure_modifier_gesture_overlap() -> None: + """A one-sided hold fires inside the both-sides binding's key set.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub( + entries=[ + {"shortcut": "left_ctrl+left_ctrl", "language": "de"}, + {"shortcut": "ctrl+ctrl", "language": "fr"}, # left ctrl fires both + {"shortcut": "right_ctrl+right_ctrl", "language": "es"}, # other side + ] + ) + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + TrayIndicator._setup_language_shortcuts(tray) + + armed = [call.kwargs["shortcut"] for call in manager_class.call_args_list] + assert armed == ["left_ctrl+left_ctrl", "right_ctrl+right_ctrl"] + + +def test_refresh_does_not_rearm_under_external_activation() -> None: + """A binding edit must not arm listeners while internal hotkeys are off.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub(entries=[{"shortcut": "alt+d", "language": "de"}]) + tray._external_activation_unavailable = False + tray.config_manager.get_bool.return_value = True # disable_internal_hotkey + + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + TrayIndicator.refresh_language_shortcuts(tray) + + manager_class.assert_not_called() + assert tray._language_shortcut_managers == [] + + +def test_refresh_stops_a_held_push_to_talk_first() -> None: + """Rebuilding listeners cannot strand a session waiting on its release.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub( + shortcut_mode="push_to_talk", + entries=[{"shortcut": "alt+d", "language": "de"}], + ) + tray.speech_engine.state = RecognitionState.LISTENING + + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager"): + TrayIndicator.refresh_language_shortcuts(tray) + + tray.speech_engine.stop_recognition.assert_called_once() + + +def test_refresh_rebuilds_the_language_listeners() -> None: + """Settings changes must reach the live listeners immediately.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub(entries=[{"shortcut": "alt+d", "language": "de"}]) + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + old = MagicMock() + manager_class.return_value = old + TrayIndicator._setup_language_shortcuts(tray) + + tray.config_manager.get_language_shortcuts.return_value = [ + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "ctrl+alt+f", "language": "fr"}, + ] + manager_class.side_effect = [MagicMock(), MagicMock()] + + TrayIndicator.refresh_language_shortcuts(tray) + + old.stop.assert_called_once() + assert len(tray._language_shortcut_managers) == 2 + + +def test_stop_language_shortcut_managers_survives_a_bad_stop() -> None: + """One failing listener must not orphan the rest during teardown.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub() + bad, good = MagicMock(), MagicMock() + bad.stop.side_effect = OSError("device vanished") + tray._language_shortcut_managers = [bad, good] + + TrayIndicator._stop_language_shortcut_managers(tray) + + bad.stop.assert_called_once() + good.stop.assert_called_once() + assert tray._language_shortcut_managers == [] + + +def test_oneshot_restore_is_part_of_the_reconfigure_snapshot() -> None: + """Rollback must be able to revive the pending one-shot restore (#805).""" + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + assert "_oneshot_language_restore" in SpeechRecognitionManager._RECONFIGURE_STATE_ATTRS + + +def test_failed_reconfigure_settles_the_pending_oneshot_restore() -> None: + """Rollback revives the pending restore, then settles it before re-init.""" + manager = _manager_stub(language="de") + manager._oneshot_language_restore = "en-us" + manager._voice_commands_preference = None + manager._defer_download = True + manager._model_lock = threading.Lock() + manager._snapshot_reconfigure_state = MagicMock( + return_value={"language": "de", "_oneshot_language_restore": "en-us"} + ) + manager._init_selected_engine = MagicMock(side_effect=[RuntimeError("boom"), None]) + + with pytest.raises(RuntimeError): + manager.reconfigure(language="fr", model_size="tiny") + + assert manager.language == "en-us" + assert manager._oneshot_language_restore is None + + +def test_language_shortcut_allowed_when_layout_will_swap_model() -> None: + """Follow-layout can swap in the multilingual sibling the shortcut needs.""" + import vocalinux.speech_recognition.recognition_manager as rm + + manager = _manager_stub(model_size="small.en", language_preference="layout") + manager.start_recognition = MagicMock(return_value=True) + + with ( + patch.object(rm, "is_model_downloaded", return_value=True), + patch.object(rm, "_multilingual_sibling", return_value="small"), + ): + assert manager.start_recognition_with_language("de") is True + + manager.start_recognition.assert_called_once() + + +def test_parakeet_auto_shortcut_is_still_allowed() -> None: + """Normalizing to auto is only a refusal when it drops the request.""" + manager = _manager_stub(engine="parakeet", model_size="turbo", language="auto") + manager.start_recognition = MagicMock(return_value=True) + + assert manager.start_recognition_with_language("auto") is True + + +def test_dictation_language_tracks_the_session_snapshot() -> None: + """Workers read the session binding, not the live configured language.""" + manager = _manager_stub(language="en-us") + manager._session_language = "de" + assert manager._dictation_language() == "de" + manager._session_language = None + assert manager._dictation_language() == "en-us" + + +def test_dictation_language_prefers_the_segment_snapshot() -> None: + """The stamped snapshot beats the session binding and the preference.""" + manager = _manager_stub(language="en-us") + manager._session_language = "fr" + assert manager._dictation_language("de") == "de" + + +def test_stale_worker_keeps_the_segments_own_language() -> None: + """A worker draining after a newer start must not adopt its language.""" + manager = _manager_stub(language="en-us") + manager._segment_queue = queue.Queue(maxsize=32) + manager.should_record = False + manager._session_language = "de" + manager._process_audio_buffer = MagicMock() + + manager._enqueue_audio_segment([b"audio"]) + # A newer dictation binds its language before the stale worker drains. + manager._session_language = "fr" + manager._signal_recognition_stop() + manager._perform_recognition() + + manager._process_audio_buffer.assert_called_once() + args = manager._process_audio_buffer.call_args.args + assert args[0] == [b"audio"] + assert args[0].language == "de" + assert args[1] == "de" + + +# --- settings dialog plumbing ------------------------------------------------ + + +def _dialog_stub() -> Any: + """A dialog carrying only what the language-shortcut plumbing reads.""" + dialog = MagicMock() + dialog._initializing = False + dialog.config_manager = MagicMock() + dialog.language_shortcuts_update_callback = MagicMock() + dialog._language_shortcut_rows = [] + return dialog + + +def test_collect_language_shortcuts_reads_every_row(dialog_class: type[Any]) -> None: + dialog = _dialog_stub() + picker, entry = MagicMock(), MagicMock() + picker.get_active_id.return_value = "de" + entry.get_text.return_value = "alt+d" + dialog._language_shortcut_rows = [{"language_picker": picker, "shortcut_entry": entry}] + + assert dialog_class._collect_language_shortcuts(dialog) == [ + {"shortcut": "alt+d", "language": "de"} + ] + + +def test_persist_writes_config_and_refreshes_listeners( + dialog_class: type[Any], +) -> None: + dialog = _dialog_stub() + picker, entry = MagicMock(), MagicMock() + picker.get_active_id.return_value = "de" + entry.get_text.return_value = "alt+d" + dialog._language_shortcut_rows = [{"language_picker": picker, "shortcut_entry": entry}] + collected = dialog_class._collect_language_shortcuts(dialog) + assert collected == [{"shortcut": "alt+d", "language": "de"}] + dialog._collect_language_shortcuts = lambda: collected + + dialog_class._persist_language_shortcuts(dialog) + + dialog.config_manager.set_language_shortcuts.assert_called_once_with( + [{"shortcut": "alt+d", "language": "de"}] + ) + dialog.config_manager.save_settings.assert_called_once() + dialog.language_shortcuts_update_callback.assert_called_once() + + +def test_persist_is_a_no_op_during_dialog_build( + dialog_class: type[Any], +) -> None: + """Rows being populated must not write config or rebuild listeners.""" + dialog = _dialog_stub() + dialog._initializing = True + + dialog_class._persist_language_shortcuts(dialog) + + dialog.config_manager.set_language_shortcuts.assert_not_called() + dialog.language_shortcuts_update_callback.assert_not_called() + + +def test_recorded_key_is_written_to_the_row_entry( + dialog_class: type[Any], +) -> None: + """The shared recorder must serve a language row, not just the main entry.""" + dialog = _dialog_stub() + entry, apply_fn = MagicMock(), MagicMock() + dialog._recording_shortcut = True + dialog._recording_shortcut_target = (entry, apply_fn, MagicMock(), MagicMock()) + + assert dialog_class._commit_recorded_shortcut(dialog, "alt+d") is True + + entry.set_text.assert_called_once_with("alt+d") + apply_fn.assert_called_once_with("alt+d") + + +def test_recording_stays_target_free_when_not_recording( + dialog_class: type[Any], +) -> None: + dialog = _dialog_stub() + dialog._recording_shortcut = False + + assert dialog_class._commit_recorded_shortcut(dialog, "alt+d") is False + + +def test_picker_change_retitles_the_row(dialog_class: type[Any]) -> None: + """The row title follows the picked language so search stays honest.""" + dialog = _dialog_stub() + row = MagicMock() + picker = MagicMock() + picker.get_active_id.return_value = "de" + refs = {"row": row} + dialog._persist_language_shortcuts = lambda: dialog_class._persist_language_shortcuts(dialog) + dialog._on_language_shortcut_row_changed = ( + lambda *a: dialog_class._on_language_shortcut_row_changed(dialog, *a) + ) + + dialog_class._on_language_shortcut_picker_changed(dialog, refs, picker) + + row.set_title.assert_called_once_with(SUPPORTED_LANGUAGES["de"]["name"]) + dialog.language_shortcuts_update_callback.assert_called_once() + + +def test_persist_keeps_last_valid_binding_during_partial_edit( + dialog_class: type[Any], +) -> None: + """A half-typed key cannot erase the row's persisted binding.""" + dialog = _dialog_stub() + picker, entry = MagicMock(), MagicMock() + picker.get_active_id.return_value = "de" + entry.get_text.return_value = "alt+" # mid-typing + dialog._language_shortcut_rows = [ + { + "language_picker": picker, + "shortcut_entry": entry, + "last_valid_shortcut": "alt+d", + } + ] + + dialog_class._persist_language_shortcuts(dialog) + + dialog.config_manager.set_language_shortcuts.assert_called_once_with( + [{"shortcut": "alt+d", "language": "de"}] + ) + + +def test_persist_reports_a_rejected_duplicate(dialog_class: type[Any]) -> None: + """A row claiming a taken key keeps its last binding and says why.""" + dialog = _dialog_stub() + first_picker, second_picker = MagicMock(), MagicMock() + first_entry, second_entry = MagicMock(), MagicMock() + first_picker.get_active_id.return_value = "de" + second_picker.get_active_id.return_value = "fr" + first_entry.get_text.return_value = "alt+d" + second_entry.get_text.return_value = "alt+d" + dialog._language_shortcut_rows = [ + { + "language_picker": first_picker, + "shortcut_entry": first_entry, + "last_valid_shortcut": "alt+d", + }, + { + "language_picker": second_picker, + "shortcut_entry": second_entry, + "last_valid_shortcut": "alt+f", + }, + ] + dialog._report_language_shortcut_rejections = ( + lambda rejected: dialog_class._report_language_shortcut_rejections(dialog, rejected) + ) + + dialog_class._persist_language_shortcuts(dialog) + + dialog.config_manager.set_language_shortcuts.assert_called_once_with( + [ + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "alt+f", "language": "fr"}, + ] + ) + dialog.language_shortcuts_info_label.set_markup.assert_called_once() + + +def test_removing_an_armed_row_cancels_recording(dialog_class: type[Any]) -> None: + """A capture aimed at a removed row is cancelled before it is destroyed.""" + dialog = _dialog_stub() + entry, button = MagicMock(), MagicMock() + refs = { + "row": MagicMock(), + "language_picker": MagicMock(), + "shortcut_entry": entry, + "record_button": button, + } + dialog._language_shortcut_rows = [refs] + dialog._recording_shortcut = True + dialog._recording_shortcut_target = (entry, MagicMock(), button, MagicMock()) + + dialog_class._on_remove_language_shortcut_clicked(dialog, refs, button) + + dialog._stop_recording_shortcut.assert_called_once() + + +def test_begin_recording_disarms_the_previous_target(dialog_class: type[Any]) -> None: + """Arming a second row resets the first row's 'Press keys…' button.""" + dialog = _dialog_stub() + dialog._recording_shortcut = True + dialog._recording_shortcut_target = ( + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + ) + + dialog_class._begin_shortcut_recording( + dialog, MagicMock(), MagicMock(), MagicMock(), MagicMock() + ) + + dialog._stop_recording_shortcut.assert_called_once() diff --git a/tests/test_recognition_manager_device_config.py b/tests/test_recognition_manager_device_config.py index f22577f..6385e95 100644 --- a/tests/test_recognition_manager_device_config.py +++ b/tests/test_recognition_manager_device_config.py @@ -2175,15 +2175,16 @@ class TestEnqueueAudioSegment(unittest.TestCase): buf = [b"\x00\x01", b"\x02\x03"] manager._enqueue_audio_segment(buf) assert not manager._segment_queue.empty() - assert manager._segment_queue.get_nowait() == buf + # Segments carry the session language they were recorded under (#805). + assert manager._segment_queue.get_nowait() == (buf, "en-us") def test_enqueue_drops_oldest_when_queue_full(self): """When queue is full, oldest item is dropped and new one inserted.""" manager = _make_manager() manager._segment_queue = queue.Queue(maxsize=1) - manager._segment_queue.put_nowait([b"old"]) + manager._segment_queue.put_nowait(([b"old"], "en-us")) manager._enqueue_audio_segment([b"new"]) - assert manager._segment_queue.get_nowait() == [b"new"] + assert manager._segment_queue.get_nowait() == ([b"new"], "en-us") class TestPerformRecognition(unittest.TestCase): @@ -2206,7 +2207,7 @@ class TestPerformRecognition(unittest.TestCase): with patch.object(manager, "_process_audio_buffer") as mock_proc: with patch.object(manager, "_update_state"): t = self._run_recognition(manager) - manager._segment_queue.put([b"\x00\x01"]) + manager._segment_queue.put(([b"\x00\x01"], "en-us")) manager.should_record = False manager._segment_queue.put(None) t.join(timeout=2) @@ -2222,8 +2223,8 @@ class TestPerformRecognition(unittest.TestCase): with patch.object(manager, "_process_audio_buffer") as mock_proc: with patch.object(manager, "_update_state"): t = self._run_recognition(manager) - manager._segment_queue.put([b"first"]) - manager._segment_queue.put([b"second"]) + manager._segment_queue.put(([b"first"], "en-us")) + manager._segment_queue.put(([b"second"], "en-us")) manager.should_record = False manager._segment_queue.put(None) t.join(timeout=2) @@ -2285,13 +2286,13 @@ class TestPerformRecognition(unittest.TestCase): states = self._track_states(manager) with patch.object(manager, "_process_audio_buffer") as mock_proc: - manager._segment_queue.put([b"leftover"]) + manager._segment_queue.put(([b"leftover"], "en-us")) manager._segment_queue.put(None) t = threading.Thread(target=manager._perform_recognition) t.start() t.join(timeout=2) assert not t.is_alive() - mock_proc.assert_called_once_with([b"leftover"]) + mock_proc.assert_called_once_with([b"leftover"], "en-us") assert RecognitionState.PROCESSING not in states assert RecognitionState.LISTENING not in states @@ -2308,12 +2309,12 @@ class TestPerformRecognition(unittest.TestCase): with patch.object(manager, "_process_audio_buffer") as mock_proc: # None is consumed first; leftover is drained from the remaining queue. manager._segment_queue.put(None) - manager._segment_queue.put([b"leftover"]) + manager._segment_queue.put(([b"leftover"], "en-us")) t = threading.Thread(target=manager._perform_recognition) t.start() t.join(timeout=2) assert not t.is_alive() - mock_proc.assert_called_once_with([b"leftover"]) + mock_proc.assert_called_once_with([b"leftover"], "en-us") assert RecognitionState.PROCESSING not in states assert manager.state == RecognitionState.IDLE @@ -2327,13 +2328,13 @@ class TestPerformRecognition(unittest.TestCase): states = self._track_states(manager) processed = threading.Event() - def _after_process(_segment): + def _after_process(_segment, _language): processed.set() with patch.object(manager, "_process_audio_buffer", side_effect=_after_process): t = threading.Thread(target=manager._perform_recognition) t.start() - manager._segment_queue.put([b"live"]) + manager._segment_queue.put(([b"live"], "en-us")) assert processed.wait(timeout=2) deadline = time.time() + 1 while RecognitionState.LISTENING not in states and time.time() < deadline: @@ -2361,11 +2362,11 @@ class TestPerformRecognition(unittest.TestCase): manager.state = RecognitionState.IDLE states = self._track_states(manager) - def _start_new_session(_segment): + def _start_new_session(_segment, _language): manager.state = RecognitionState.LISTENING with patch.object(manager, "_process_audio_buffer", side_effect=_start_new_session): - manager._segment_queue.put([b"leftover"]) + manager._segment_queue.put(([b"leftover"], "en-us")) manager._segment_queue.put(None) t = threading.Thread(target=manager._perform_recognition) t.start() diff --git a/tests/test_record_during_reload.py b/tests/test_record_during_reload.py index 3d2cac3..ab59325 100644 --- a/tests/test_record_during_reload.py +++ b/tests/test_record_during_reload.py @@ -53,7 +53,7 @@ def session(monkeypatch: pytest.MonkeyPatch) -> Generator[Session, None, None]: manager.model = object() manager._model_initialized = True - def collect_transcription(audio: list[bytes]) -> None: + def collect_transcription(audio: list[bytes], language: str | None = None) -> None: transcribed.append(audio) monkeypatch.setattr(manager, "_record_audio", record)