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)