xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat(shortcuts): add language switching shortcuts, fix PTT races, and add dictation history (#880)

Browse Source 
* feat(shortcuts): per-language dictation shortcuts (#805)

Bind extra keys that start dictation in a specific catalog language for
one utterance, then restore the configured language. Bindings live in
shortcuts.language_shortcuts as [{shortcut, language}], normalized
against the language catalog and the backend's shortcut parser.

The tray spawns one KeyboardShortcutManager per binding in the active
mode; toggle presses route through start_recognition_with_language, and
push-to-talk arms press/release the same way. The manager keeps the
override one-shot: it applies before the utterance starts (outranking
follow-keyboard-layout) and unwinds on the next IDLE/ERROR transition
or when a mid-dictation reconfigure picks a new language. Engines that
cannot re-language without a reload (VOSK, Parakeet, English-only
whisper.cpp) refuse with the error cue and a notification instead of
dictating in the wrong language.

Settings gains a Language Shortcuts group under Shortcuts: each row is
a language picker, an editable key entry, a Record button that retargets
the shared key-capture, and a remove button; edits persist immediately
and rebuild the live listeners. A non-dict shortcuts section now rebuilds
instead of crashing config load.

* fix(shortcuts): class-level default for language shortcut listeners

TrayIndicator stubs built with __new__ (e.g. test_suspend_handler)
skip __init__ but still run _quit(); give _language_shortcut_managers
a class-level default so teardown iterates an empty list.

* Fix Greptile review findings on per-language shortcuts (#805)

- Bind each dictation session's language for transcription workers so the
  one-shot restore cannot rewrite a queued segment's language
- Judge a shortcut's capability against the model the layout refresh will
  serve (whisper.cpp sibling swap) and refuse requests the engine drops
- Add _oneshot_language_restore to the reconfigure snapshot and settle a
  revived pending restore on failure
- Dedupe language bindings by gesture (modifier reordering, enter/return
  alias, pure-modifier side sets) instead of by string
- Stop per-language listeners under external activation and never re-arm
- Gate push-to-talk release on the binding that started the session, and
  stop a held hold before rebuilding listeners
- Keep a row's last valid binding during partial edits, report rejected
  duplicates, cancel a capture aimed at a removed row, disarm the previous
  Record target, persist a recorded key once, and add the missing type
  hints

* fix(shortcuts): resolve remaining Greptile race findings on PR #880

- Stamp each queued audio segment with the session language it was
  recorded under, so a worker that outlives stop_recognition's joins
  transcribes each utterance in its own language rather than whatever a
  newer session stored in _session_language.
- Serialize push-to-talk press/release behind _ptt_lock: a release
  landing while start_recognition runs now waits for the owner claim
  instead of being dropped, so a quick key tap cannot leave the
  microphone running.
- Add the missing return type hint on _stop_language_shortcut_managers.

* test(shortcuts): stub _segment_started_at on bare manager stubs

* fix(shortcuts): carry segment language as attribute, default language on bare managers

* fix(shortcuts): keep (segment, language) queue tuples, stamp segment.language too

* fix(settings): apply history toggle and limit to the live store

---------

Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
This commit is contained in:
devin-ai-integration[bot]
2026-09-30 01:05:18 -07:00
committed by GitHub
co-authored by Jatin K Malik
parent ca40f7d8b0
commit dbe0215aad
11 changed files with 1857 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vocalinux/common_types.py
+4
View File
@@ -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."""
...
src/vocalinux/speech_recognition/engines/faster_whisper_engine.py
+11 -6
View File
@@ -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,
src/vocalinux/speech_recognition/recognition_manager.py
+250 -23
View File
@@ -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()
src/vocalinux/ui/config_manager.py
+72 -1
View File
@@ -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)
src/vocalinux/ui/keyboard_backends/__init__.py
+2
View File
@@ -24,6 +24,8 @@ from .base import (
is_valid_shortcut,
parse_shortcut,
parse_shortcut_spec,
shortcut_gesture,
shortcut_gestures_collide,
)
logger = logging.getLogger(__name__)
src/vocalinux/ui/keyboard_backends/base.py
+62
View File
@@ -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:
src/vocalinux/ui/settings_dialog.py
+310 -12
View File
@@ -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"<span foreground='#e01b24'>Shortcut "
f"<b>{GLib.markup_escape_text(keys)}</b> is already bound on "
"another row.</span>"
)
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(
"<i>Press a modifier + key (e.g. Alt+R), or an F-key. Press Esc to cancel.</i>"
)
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(
"<span foreground='#e01b24'>Need a modifier + key, or an F1–F24 "
"function key alone. Try again or press Esc to cancel.</span>"
)
src/vocalinux/ui/tray_indicator.py
+219 -19
View File
@@ -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()
tests/test_language_shortcuts.py
+911
View File
@@ -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()
tests/test_recognition_manager_device_config.py
+15 -14
View File
@@ -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()
tests/test_record_during_reload.py
+1 -1
View File
@@ -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)