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

feat(recognition): cancel an in-progress transcription with the recording hotkey (#1004)

Browse Source 
* feat(recognition): cancel in-progress transcription with hotkey

Adds per-session cancellation across the manager, tray, and Test Dictation.

Fixes #991

* fix(settings): keep Test Dictation cancellation responsive

Fixes #991

* fix(recognition): keep late workers cancellable (#991)

* test(tray): avoid early tray import in cancellation tests

* fix(recognition): keep a cancelled capture thread out of the next session

A cancel returns to IDLE without joining the capture thread, so a thread
still in its reconnection backoff could wake inside a newer recording,
reopen that session's device and write into its buffer. Tie the capture
loop and the reconnection to the session's cancel token: a cancelled or
replaced session stops reading, skips the reopen and exits without
raising an error.
This commit is contained in:
Jatin K Malik
2026-10-21 01:33:58 -07:00
committed by GitHub
parent 9b5cdf686f
commit 138a233892
15 changed files with 1485 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vocalinux/common_types.py
+8
View File
@@ -94,6 +94,14 @@ class SpeechRecognitionManagerProtocol(Protocol):
"""Start recognition in ``language`` for this utterance only (#805)."""
...
def is_transcribing(self) -> bool:
"""Whether a stopped dictation session is still being transcribed."""
...
def cancel_transcription(self) -> bool:
"""Cancel the active dictation session and discard its result."""
...
def stop_recognition(self) -> None:
"""Stop the speech recognition process."""
...
src/vocalinux/speech_recognition/recognition_manager.py
+288 -62
View File
@@ -121,6 +121,7 @@ class _AudioSegment(list):
chunks: list[bytes],
started_at: Optional[float],
language: Optional[str] = None,
cancel_event: Optional[threading.Event] = None,
release_gate: Optional[_ReleaseGate] = None,
) -> None:
super().__init__(chunks)
@@ -129,6 +130,7 @@ class _AudioSegment(list):
# outlives its session must not transcribe a stale segment in a newer
# session's language.
self.language = language
self.cancel_event = cancel_event
self.release_gate = release_gate
@@ -239,6 +241,20 @@ def resolve_language_preference(language: str) -> str:
logger = logging.getLogger(__name__)
_WHISPERCPP_ABORT_CALLBACK_SUPPORT: dict[type, bool] = {}
def _supports_whispercpp_abort_callback(model_class: type) -> bool:
"""Return whether a whisper.cpp model class accepts abort_callback."""
if model_class not in _WHISPERCPP_ABORT_CALLBACK_SUPPORT:
try:
_WHISPERCPP_ABORT_CALLBACK_SUPPORT[model_class] = (
"abort_callback" in inspect.signature(getattr(model_class, "transcribe")).parameters
)
except (TypeError, ValueError):
_WHISPERCPP_ABORT_CALLBACK_SUPPORT[model_class] = False
return _WHISPERCPP_ABORT_CALLBACK_SUPPORT[model_class]
def _filter_non_speech(text: str) -> str:
"""
@@ -414,6 +430,10 @@ class SpeechRecognitionManager:
speech recognition engines (VOSK and Whisper).
"""
_STOP_WORKER_JOIN_TIMEOUT_S: float = 5.0
_STOP_WORKER_FINAL_JOIN_TIMEOUT_S: float = 1.0
_idle_on_worker_exit: Optional[threading.Event] = None
def __init__(
self,
engine: str = "vosk",
@@ -454,6 +474,8 @@ class SpeechRecognitionManager:
self.state = RecognitionState.IDLE
self.audio_thread: Optional[threading.Thread] = None
self.recognition_thread: Optional[threading.Thread] = None
self._worker_exited: Optional[threading.Event] = None
self._idle_on_worker_exit: Optional[threading.Event] = None
self._recognition_workers: list[threading.Thread] = []
self._recognition_workers_lock = threading.Lock()
self.model: Any = None
@@ -551,6 +573,7 @@ class SpeechRecognitionManager:
# Recording control flags
self.should_record = False
self._session_cancel = threading.Event()
self._recognition_mode = "toggle" # "toggle" or "push_to_talk"
self.audio_buffer: list[bytes] = []
# Monotonic time the current buffer's first chunk was captured,
@@ -1940,6 +1963,11 @@ class SpeechRecognitionManager:
"""
import time
cancel_event = getattr(audio_buffer, "cancel_event", None)
if cancel_event is not None and cancel_event.is_set():
logger.debug("Skipping whisper.cpp transcription for a cancelled session")
return ""
try:
import numpy as np
@@ -1972,6 +2000,10 @@ class SpeechRecognitionManager:
logger.warning("Model is None during transcription, returning empty result")
return ""
if cancel_event is not None and cancel_event.is_set():
logger.debug("Skipping whisper.cpp language detection for a cancelled session")
return ""
if lang is None and self.whispercpp_language_candidates:
detected, lang_probs = self.model.auto_detect_language(
audio_float,
@@ -2021,10 +2053,14 @@ class SpeechRecognitionManager:
# Transcribe with whisper.cpp
# pywhispercpp expects audio as numpy array
transcribe_start = time.time()
transcribe_kwargs = {"language": lang}
transcribe_kwargs: dict[str, Any] = {"language": lang}
# pywhispercpp reuses native parameter state. Passing an empty
# value explicitly clears a prompt that was active last segment.
transcribe_kwargs["initial_prompt"] = self._get_whispercpp_prompt() or ""
if cancel_event is not None and _supports_whispercpp_abort_callback(
type(self.model)
):
transcribe_kwargs["abort_callback"] = cancel_event.is_set
segments = self.model.transcribe(audio_float, **transcribe_kwargs)
transcribe_duration = time.time() - transcribe_start
@@ -3339,6 +3375,10 @@ class SpeechRecognitionManager:
self._session_language = self.language
logger.info("Starting speech recognition")
with self._buffer_lock:
self._session_cancel = threading.Event()
self._worker_exited = threading.Event()
self._idle_on_worker_exit = None
self._update_state(RecognitionState.LISTENING)
# Play the start sound
@@ -3367,13 +3407,60 @@ class SpeechRecognitionManager:
self._arm_playback_duck()
return True
def stop_recognition(self):
def is_transcribing(self) -> bool:
"""True while a stopped session's audio is still being transcribed."""
return self.state == RecognitionState.PROCESSING and not self.should_record
def cancel_transcription(self) -> bool:
"""Abort the current dictation session and discard its text.
Returns True if something was cancelled.
"""
if self.state not in (RecognitionState.LISTENING, RecognitionState.PROCESSING):
return False
cancel = self._session_cancel
with self._buffer_lock:
if (
self._session_cancel is not cancel
or self.state not in (RecognitionState.LISTENING, RecognitionState.PROCESSING)
or cancel.is_set()
):
return False
logger.info("Cancelling transcription")
cancel.set()
self._idle_on_worker_exit = None
segment_queue = self._segment_queue
buffered = getattr(self, "_buffered_reload_session", False)
if buffered:
self._cancel_buffered_session.set()
while True:
try:
segment_queue.get_nowait()
except queue.Empty:
break
self._signal_recognition_stop()
if self.should_record:
self.stop_recognition()
elif not buffered:
self._recognition_mode = "toggle"
self._update_state(RecognitionState.IDLE)
return True
def stop_recognition(self) -> None:
"""Stop the speech recognition process."""
if self.state == RecognitionState.IDLE:
return
if getattr(self, "_buffered_reload_session", False) and not self.should_record:
return
session_cancel = self._session_cancel
audio_thread = self.audio_thread
recognition_thread = self.recognition_thread
worker_exited = getattr(self, "_worker_exited", None)
logger.info("Stopping speech recognition")
# Mark recording over before touching the duck so a concurrent arm
@@ -3387,8 +3474,8 @@ class SpeechRecognitionManager:
# Wait for audio thread to finish recording and enqueue any pending audio
# This is critical to prevent race condition where recognition thread exits
# before the final audio segment is enqueued
if self.audio_thread and self.audio_thread.is_alive():
self.audio_thread.join(timeout=2.0)
if audio_thread and audio_thread.is_alive():
audio_thread.join(timeout=2.0)
self.release_playback_duck()
@@ -3396,34 +3483,53 @@ class SpeechRecognitionManager:
# Kept before buffer processing so the cue still feels immediate.
play_stop_sound()
# Trim only a small tail to avoid the stop sound without clipping the user's final word.
with self._buffer_lock:
stop_sound_guard_chunks = self._get_stop_sound_guard_chunks()
if stop_sound_guard_chunks > 0 and len(self.audio_buffer) > stop_sound_guard_chunks:
discarded_chunks = self.audio_buffer[-stop_sound_guard_chunks:]
self.audio_buffer = self.audio_buffer[:-stop_sound_guard_chunks]
logger.debug(
"Discarded %s audio chunks (~%sms) to avoid transcribing feedback sound",
len(discarded_chunks),
self.stop_sound_guard_ms,
)
if self._session_cancel is session_cancel:
# Trim only a small tail to avoid the stop sound without clipping the user's final word.
with self._buffer_lock:
if self._session_cancel is not session_cancel:
return
if session_cancel.is_set():
if self.audio_buffer:
logger.debug("Dropping final audio buffer from a cancelled session")
self.audio_buffer = []
self._recording_segment_has_speech = False
else:
stop_sound_guard_chunks = self._get_stop_sound_guard_chunks()
if (
stop_sound_guard_chunks > 0
and len(self.audio_buffer) > stop_sound_guard_chunks
):
discarded_chunks = self.audio_buffer[-stop_sound_guard_chunks:]
self.audio_buffer = self.audio_buffer[:-stop_sound_guard_chunks]
logger.debug(
"Discarded %s audio chunks (~%sms) to avoid transcribing feedback sound",
len(discarded_chunks),
self.stop_sound_guard_ms,
)
if self.audio_buffer and self._recording_segment_has_speech:
logger.debug(f"Enqueuing final speech buffer with {len(self.audio_buffer)} chunks")
self._enqueue_audio_segment(
self.audio_buffer,
release_gate=(
self._key_release_gate if self._recognition_mode == "push_to_talk" else None
),
)
self.audio_buffer = []
elif self.audio_buffer:
logger.debug(
"Dropping final audio buffer with no detected speech "
f"({len(self.audio_buffer)} chunks)"
)
self.audio_buffer = []
self._recording_segment_has_speech = False
if self.audio_buffer and self._recording_segment_has_speech:
logger.debug(
f"Enqueuing final speech buffer with {len(self.audio_buffer)} chunks"
)
self._enqueue_audio_segment(
self.audio_buffer,
release_gate=(
self._key_release_gate
if self._recognition_mode == "push_to_talk"
else None
),
)
self.audio_buffer = []
elif self.audio_buffer:
logger.debug(
"Dropping final audio buffer with no detected speech "
f"({len(self.audio_buffer)} chunks)"
)
self.audio_buffer = []
self._recording_segment_has_speech = False
if self._session_cancel is not session_cancel:
return
if getattr(self, "_buffered_reload_session", False):
# Keep the session busy until its queued audio has been transcribed.
@@ -3434,16 +3540,38 @@ class SpeechRecognitionManager:
return
# Wake up recognition thread so it can drain queued segments and stop
self._signal_recognition_stop()
if not session_cancel.is_set():
self._signal_recognition_stop()
if self.recognition_thread and self.recognition_thread.is_alive():
self.recognition_thread.join(timeout=5.0) # Increased timeout for transcription
self._signal_recognition_stop()
if recognition_thread and recognition_thread.is_alive():
recognition_thread.join(timeout=self._STOP_WORKER_JOIN_TIMEOUT_S)
if not session_cancel.is_set():
self._signal_recognition_stop()
if self.recognition_thread and self.recognition_thread.is_alive():
self.recognition_thread.join(timeout=1.0)
if recognition_thread and recognition_thread.is_alive():
recognition_thread.join(timeout=self._STOP_WORKER_FINAL_JOIN_TIMEOUT_S)
self._recognition_mode = "toggle"
with self._buffer_lock:
if self._session_cancel is not session_cancel:
return
worker_still_running = (
recognition_thread is not None
and worker_exited is not None
and not worker_exited.is_set()
and not session_cancel.is_set()
)
if worker_still_running:
self._idle_on_worker_exit = session_cancel
else:
self._recognition_mode = "toggle"
if worker_still_running:
logger.info(
"Transcription still running after stop; staying in PROCESSING until it finishes "
"or is cancelled"
)
return
if self._session_cancel is not session_cancel:
return
self._update_state(RecognitionState.IDLE)
def _track_recognition_worker(self, worker: threading.Thread) -> None:
@@ -3580,6 +3708,7 @@ class SpeechRecognitionManager:
def _record_audio(self) -> None:
"""Record audio from the microphone with reconnection logic."""
gate = self._key_release_gate
session_cancel = self._session_cancel
# Lazy import to avoid circular dependency
from ..ui.audio_feedback import play_error_sound # noqa: F401
@@ -3641,7 +3770,12 @@ class SpeechRecognitionManager:
logger.error("This may indicate a problem with the audio device or permissions.")
# Attempt reconnection
if not self._attempt_audio_reconnection(audio):
if not self._attempt_audio_reconnection(audio, session_cancel):
if not self._capture_session_current(session_cancel):
logger.info("Audio capture stopped: its session was cancelled or replaced")
if audio is not None:
audio.terminate()
return
self.should_record = False
self.release_playback_duck()
play_error_sound()
@@ -3676,10 +3810,13 @@ class SpeechRecognitionManager:
if self._silero_vad is not None:
self._silero_vad.reset()
while self.should_record:
while self._capture_session_current(session_cancel):
try:
# Check buffer size and enforce limits (with lock for thread safety)
with self._buffer_lock:
# A newer session owns the shared buffer once this one is cancelled.
if session_cancel.is_set() or self._session_cancel is not session_cancel:
break
if len(self.audio_buffer) >= self._max_buffer_size:
if getattr(self, "_reload_pending", False):
raise RuntimeError("Audio buffer filled while reloading the model")
@@ -3692,6 +3829,8 @@ class SpeechRecognitionManager:
logger.info(f"Buffer trimmed by {remove_count} chunks")
data = source.read_chunk()
if session_cancel.is_set() or self._session_cancel is not session_cancel:
break
if not self.audio_buffer:
# First chunk of a new segment: remember when its
@@ -3838,6 +3977,9 @@ class SpeechRecognitionManager:
),
)
except (IOError, OSError) as e:
if not self._capture_session_current(session_cancel):
logger.info("Audio capture stopped: its session was cancelled or replaced")
break
current_time = time.time()
logger.error(f"Audio device error: {e}")
@@ -3847,13 +3989,18 @@ class SpeechRecognitionManager:
): # Prevent rapid reconnection attempts
self._last_audio_error_time = current_time
if self._attempt_audio_reconnection(audio):
if self._attempt_audio_reconnection(audio, session_cancel):
logger.info("Audio reconnection successful, continuing recording")
# The session source may have been rebuilt (e.g.
# the configured device switched between a mic and
# a PipeWire sink) — keep reading the new source.
source = self._capture_source
continue # Continue recording with the reopened source
elif not self._capture_session_current(session_cancel):
logger.info(
"Audio capture stopped: its session was cancelled or replaced"
)
break
else:
logger.error("Audio reconnection failed, stopping recording")
capture_failed = True
@@ -3976,12 +4123,26 @@ class SpeechRecognitionManager:
self,
gate: Optional[_ReleaseGate],
force: bool = False,
cancel: Optional[threading.Event] = None,
) -> None:
"""Deliver queued transcripts once their push-to-talk gate opens."""
if gate is None or (not gate.is_set() and not force):
"""Deliver queued transcripts once their push-to-talk gate opens.
Held text from a cancelled session is dropped instead of delivered.
"""
if gate is None:
return
if cancel is not None and cancel.is_set():
if gate.held:
logger.info("Discarding held push-to-talk text from a cancelled session")
gate.held.clear()
return
if not gate.is_set() and not force:
return
while gate.held:
if cancel is not None and cancel.is_set():
gate.held.clear()
return
transcript = gate.held.pop(0)
self._deliver_transcript(
transcript.text,
@@ -3998,6 +4159,10 @@ class SpeechRecognitionManager:
it was queued (#805); None falls back to the segment's own stamp and
then to the live session language.
"""
cancel_event = getattr(audio_buffer, "cancel_event", None)
if cancel_event is not None and cancel_event.is_set():
logger.debug("Skipping audio processing for a cancelled session")
return
if not audio_buffer:
return
if language is None:
@@ -4023,10 +4188,13 @@ class SpeechRecognitionManager:
logger.error(f"Unknown engine: {self.engine}")
return
text = transcriber()
if cancel_event is not None and cancel_event.is_set():
logger.info("Discarding transcription from a cancelled session")
return
if text is None:
gate = getattr(audio_buffer, "release_gate", None)
if gate is not None and gate.is_set():
self._release_held_transcripts(gate)
self._release_held_transcripts(gate, cancel=cancel_event)
return
# Process text - either with voice commands or pass through directly
@@ -4045,6 +4213,10 @@ class SpeechRecognitionManager:
processed_text = text.strip()
actions = []
if cancel_event is not None and cancel_event.is_set():
logger.info("Discarding transcription from a cancelled session")
return
gate = getattr(audio_buffer, "release_gate", None)
if gate is not None and not gate.is_set():
if processed_text or actions:
@@ -4058,7 +4230,7 @@ class SpeechRecognitionManager:
logger.info("Holding transcribed text until the push-to-talk key is released")
return
elif gate is not None:
self._release_held_transcripts(gate)
self._release_held_transcripts(gate, cancel=cancel_event)
self._deliver_transcript(
processed_text,
@@ -4070,23 +4242,37 @@ class SpeechRecognitionManager:
"""Perform speech recognition in real-time."""
gate = self._key_release_gate
logger.debug("_perform_recognition thread started")
cancel = self._session_cancel
segment_queue = self._segment_queue
worker_exited = getattr(self, "_worker_exited", None)
try:
while True:
if cancel.is_set():
logger.debug("Recognition loop exiting for a cancelled session")
break
logger.debug(
f"Recognition loop - should_record={self.should_record}, queue_empty={self._segment_queue.empty()}"
f"Recognition loop - should_record={self.should_record}, queue_empty={segment_queue.empty()}"
)
try:
queued = self._segment_queue.get(timeout=0.1)
queued = segment_queue.get(timeout=0.1)
if cancel.is_set():
logger.debug("Recognition loop discarded an item for a cancelled session")
break
except queue.Empty:
self._release_held_transcripts(gate)
self._release_held_transcripts(gate, cancel=cancel)
# Only exit if we're not recording AND queue is empty
if not self.should_record and self._segment_queue.empty():
if not self.should_record and segment_queue.empty():
logger.debug(
"Recognition loop - not recording and queue empty, checking for final items..."
)
# Give a brief moment for any final items to be enqueued
try:
queued = self._segment_queue.get(timeout=0.5)
queued = segment_queue.get(timeout=0.5)
if cancel.is_set():
logger.debug(
"Recognition loop discarded an item for a cancelled session"
)
break
except queue.Empty:
logger.debug("Recognition loop - no more items, exiting")
break
@@ -4097,9 +4283,17 @@ class SpeechRecognitionManager:
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():
while not segment_queue.empty():
if cancel.is_set():
logger.debug("Recognition loop drain stopped for a cancelled session")
break
try:
remaining = self._segment_queue.get_nowait()
remaining = segment_queue.get_nowait()
if cancel.is_set():
logger.debug(
"Recognition loop discarded a drained item for a cancelled session"
)
break
if remaining is not None:
remaining_segment, remaining_language = remaining
logger.debug(
@@ -4107,7 +4301,7 @@ class SpeechRecognitionManager:
)
if self.should_record:
self._update_state(RecognitionState.PROCESSING)
self._release_held_transcripts(gate)
self._release_held_transcripts(gate, cancel=cancel)
self._process_audio_buffer(remaining_segment, remaining_language)
except queue.Empty:
break
@@ -4118,18 +4312,28 @@ class SpeechRecognitionManager:
logger.debug(f"Recognition loop - processing segment with {len(segment)} chunks")
if self.should_record:
self._update_state(RecognitionState.PROCESSING)
self._release_held_transcripts(gate)
self._release_held_transcripts(gate, cancel=cancel)
self._process_audio_buffer(segment, segment_language)
if self.should_record:
if self.should_record and self._session_cancel is cancel:
self._update_state(RecognitionState.LISTENING)
finally:
logger.debug("_perform_recognition thread exiting")
self._release_held_transcripts(gate, force=True)
# Do not force IDLE here. stop_recognition() already does that,
# and a leftover worker can finish after the user has started
# again (LISTENING, should_record still false during the start
# sound). Forcing IDLE would paint the tray idle and make the
# next Alt+F12 start a second session (#739).
self._release_held_transcripts(gate, force=True, cancel=cancel)
hand_off = False
if worker_exited is not None:
with self._buffer_lock:
worker_exited.set()
hand_off = self._idle_on_worker_exit is cancel
if hand_off:
self._idle_on_worker_exit = None
if (
hand_off
and self._session_cancel is cancel
and not cancel.is_set()
and not self.should_record
):
self._recognition_mode = "toggle"
self._update_state(RecognitionState.IDLE)
def _enqueue_audio_segment(
self,
@@ -4140,6 +4344,7 @@ class SpeechRecognitionManager:
segment = _AudioSegment(
audio_buffer.copy(),
self._segment_started_at,
cancel_event=self._session_cancel,
release_gate=release_gate,
)
if not segment:
@@ -4467,12 +4672,29 @@ class SpeechRecognitionManager:
if source.audio is not None:
self._pyaudio_instance = source.audio
def _attempt_audio_reconnection(self, audio_instance) -> bool:
def _capture_session_current(self, session_cancel: threading.Event) -> bool:
"""Whether a capture thread's session is still recording and not cancelled.
A cancel returns to IDLE without joining the capture thread, so a
thread still sleeping in reconnection must not touch a newer session.
"""
return (
self.should_record
and self._session_cancel is session_cancel
and not session_cancel.is_set()
)
def _attempt_audio_reconnection(
self, audio_instance: Any, session_cancel: Optional[threading.Event] = None
) -> bool:
"""
Attempt to reconnect to the audio device.
Args:
audio_instance: The PyAudio instance to use for reconnection
session_cancel: The capture thread's session token. When given, a
session cancelled or replaced during the backoff gives up
without reopening the device.
Returns:
bool: True if reconnection was successful, False otherwise
@@ -4494,6 +4716,10 @@ class SpeechRecognitionManager:
# Wait before attempting reconnection
time.sleep(delay)
if session_cancel is not None and not self._capture_session_current(session_cancel):
logger.info("Skipping audio reconnection for a cancelled or replaced session")
return False
source = self._capture_source_for_session()
# The source may not know about the stream it is asked to replace
# (e.g. when tests drive this method directly).
src/vocalinux/ui/settings_dialog.py
+94 -27
View File
@@ -2523,6 +2523,8 @@ class SettingsDialog(AboutSectionMixin):
None
)
self._test_active = False
self._test_token = 0
self._test_cancelled = False
self._test_result = ""
self._initializing = True # Flag to prevent auto-apply during initialization
self._populating_models = False # Flag to prevent model change handler during population
@@ -5925,12 +5927,22 @@ class SettingsDialog(AboutSectionMixin):
self.recognition_audio_level = self.audio_level_bar
footer.pack_start(self.audio_level_bar, False, False, 0)
test_button_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
self.test_button = Gtk.Button(label="Test Dictation")
self.test_button.set_tooltip_text(
"Record 3 seconds of speech and show the transcription here"
)
self.test_button.connect("clicked", self._on_test_clicked)
footer.pack_start(self.test_button, False, False, 0)
test_button_row.pack_start(self.test_button, False, False, 0)
self.test_cancel_button = Gtk.Button(label="Cancel")
self.test_cancel_button.set_tooltip_text("Stop the test and discard its result")
self.test_cancel_button.connect("clicked", self._on_test_cancel_clicked)
self.test_cancel_button.set_no_show_all(True)
self.test_cancel_button.hide()
test_button_row.pack_start(self.test_cancel_button, False, False, 0)
footer.pack_start(test_button_row, False, False, 0)
# One shared status line (audio test results and recognition info)
self.progress_info_label = Gtk.Label(label="", use_markup=True, xalign=0)
@@ -9166,7 +9178,7 @@ class SettingsDialog(AboutSectionMixin):
return settings
def _on_test_clicked(self, widget):
def _on_test_clicked(self, widget: Any) -> None:
"""Handle click on the test button."""
if self._test_active:
logger.warning("Test already in progress.")
@@ -9249,9 +9261,13 @@ class SettingsDialog(AboutSectionMixin):
return
self._test_active = True
self._test_token += 1
self._test_cancelled = False
self._test_timeout_cancel_attempted = False
self.test_button.set_sensitive(False)
self.test_button.set_label("Testing… Speak Now!")
self.test_cancel_button.set_sensitive(True)
self.test_cancel_button.show()
self.test_output_revealer.set_reveal_child(True)
self.test_buffer.set_text("")
self._test_result = ""
@@ -9259,13 +9275,54 @@ class SettingsDialog(AboutSectionMixin):
# Enough room for one utterance plus the configured silence window.
delay = float(selected_settings.get("silence_timeout", 1.2)) + 2.0
threading.Thread(target=self._stop_test_after_delay, args=(delay,)).start()
threading.Thread(target=self._stop_test_after_delay, args=(delay, self._test_token)).start()
def _test_text_callback(self, text: str):
def _on_test_cancel_clicked(self, widget: Any) -> None:
"""Cancel the test and discard any text already captured."""
if not self._test_active or self._test_cancelled:
return
self._test_cancelled = True
self.test_button.set_sensitive(False)
self.test_cancel_button.set_sensitive(False)
self.test_button.set_label("Cancelling…")
def cancel_test() -> None:
try:
self.speech_engine.cancel_transcription()
except (AttributeError, OSError, RuntimeError, TypeError):
logger.exception("Test Dictation cancellation failed")
GLib.idle_add(self._finish_test_after_user_cancel)
threading.Thread(
target=cancel_test,
daemon=True,
name="vocalinux-test-dictation-cancel",
).start()
def _finish_test_after_user_cancel(self) -> bool:
"""Restore test UI and callbacks without checking for captured speech."""
self._test_active = False
self._test_timeout_cancel_attempted = False
self.test_button.set_sensitive(True)
self.test_button.set_label("Test Dictation")
self.test_cancel_button.hide()
self.test_cancel_button.set_sensitive(True)
self.update_recognition_progress("Idle")
if hasattr(self, "_saved_text_callbacks"):
self.speech_engine.set_text_callbacks(self._saved_text_callbacks)
del self._saved_text_callbacks
self.speech_engine.test_capture_ceiling = time.monotonic()
self.test_buffer.set_text("(Test cancelled)")
return False
def _test_text_callback(self, text: str) -> None:
"""Callback specifically for the test recognition."""
GLib.idle_add(self._append_test_result, text)
def _append_test_result(self, text: str):
def _append_test_result(self, text: str) -> bool:
if getattr(self, "_test_cancelled", False) is True:
return False
current_text = self.test_buffer.get_text(
self.test_buffer.get_start_iter(), self.test_buffer.get_end_iter(), False
)
@@ -9275,10 +9332,10 @@ class SettingsDialog(AboutSectionMixin):
self.test_textview.scroll_to_mark(mark, 0.0, True, 0.0, 1.0)
return False
def _stop_test_after_delay(self, delay: float):
def _stop_test_after_delay(self, delay: float, token: Optional[int] = None) -> None:
"""Stops the recognition test after a specified delay."""
time.sleep(delay)
GLib.idle_add(self._finalize_test)
GLib.idle_add(self._finalize_test, token)
def _test_engine_busy(self) -> bool:
"""Return whether recognition can still deliver text to the test callback."""
@@ -9290,32 +9347,36 @@ class SettingsDialog(AboutSectionMixin):
or (callable(has_pending_transcription) and has_pending_transcription() is True)
)
def _finalize_test(self) -> bool:
def _finalize_test(self, token: Optional[int] = None) -> bool:
"""Finalize the test state and UI updates."""
if not self._test_active:
if (token is not None and token != getattr(self, "_test_token", 0)) or not getattr(
self, "_test_active", False
):
return False
if getattr(self, "_test_cancelled", False):
return False
self.speech_engine.stop_recognition()
self.test_button.set_sensitive(False)
self.test_button.set_label("Processing…")
self.test_cancel_button.set_sensitive(True)
self.test_cancel_button.show()
self.update_recognition_progress("Processing")
# Keep test callbacks installed until no worker can deliver the test
# utterance. Use GLib polling rather than joining on the GTK thread.
if self._test_engine_busy():
self.test_button.set_sensitive(False)
self.test_button.set_label("Processing…")
self.update_recognition_progress("Processing")
self._test_idle_wait_ticks = 0
self._test_timeout_cancel_attempted = False
GLib.timeout_add(100, self._wait_for_idle_then_restore_callbacks)
return False
def stop_test() -> None:
try:
self.speech_engine.stop_recognition()
except (AttributeError, OSError, RuntimeError, TypeError):
logger.exception("Test Dictation stop failed")
# Briefly settle for any GLib.idle_add text delivery already in flight.
GLib.timeout_add(500, self._restore_callbacks_and_check_result)
threading.Thread(
target=stop_test,
daemon=True,
name="vocalinux-test-dictation-stop",
).start()
self._test_active = False
self.test_button.set_sensitive(True)
self.test_button.set_label("Test Dictation")
self.update_recognition_progress("Idle")
self._test_idle_wait_ticks = 0
self._test_timeout_cancel_attempted = False
GLib.timeout_add(100, self._wait_for_idle_then_restore_callbacks)
return False
@@ -9333,10 +9394,14 @@ class SettingsDialog(AboutSectionMixin):
test callbacks and ``_test_active`` so live injectors are not restored
early and later dictation cannot be hijacked into the test buffer.
"""
if getattr(self, "_test_cancelled", False) is True:
return False
if not hasattr(self, "_saved_text_callbacks"):
self._test_active = False
self.test_button.set_sensitive(True)
self.test_button.set_label("Test Dictation")
self.test_cancel_button.hide()
return False
if self._test_engine_busy():
@@ -9430,6 +9495,8 @@ class SettingsDialog(AboutSectionMixin):
self._test_timeout_cancel_attempted = False
self.test_button.set_sensitive(True)
self.test_button.set_label("Test Dictation")
self.test_cancel_button.hide()
self.test_cancel_button.set_sensitive(True)
self.update_recognition_progress("Idle")
return self._restore_callbacks_and_check_result()
src/vocalinux/ui/tray_indicator.py
+63 -2
View File
@@ -134,6 +134,7 @@ class TrayIndicator:
# 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 = []
_indicator_title: str = "Vocalinux"
#: 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.
@@ -235,6 +236,7 @@ class TrayIndicator:
# IconThemePath so gray→red→gray→red actually turns red again.
self._last_icon_key: Optional[str] = None
self._icon_theme_nudge = False
self._indicator_title = "Vocalinux"
# Register for speech recognition state changes
self.speech_engine.register_state_callback(self._on_recognition_state_changed)
@@ -557,6 +559,7 @@ class TrayIndicator:
self.indicator.set_icon_theme_path(ICON_DIR)
self.indicator.set_status(AppIndicator3.IndicatorStatus.ACTIVE)
self.indicator.set_title("Vocalinux")
self._indicator_title = "Vocalinux"
except Exception as e:
logger.error(f"Failed to create AppIndicator: {e}")
GLib.idle_add(self._show_appindicator_error_dialog, str(e))
@@ -576,6 +579,11 @@ class TrayIndicator:
# Add menu items
self._add_menu_item("Start Voice Typing", self._on_start_clicked)
self._add_menu_item("Stop Voice Typing", self._on_stop_clicked)
self._cancel_menu_item = self._add_menu_item(
"Cancel Transcription", self._on_cancel_transcription_clicked
)
self._cancel_menu_item.set_no_show_all(True)
self._cancel_menu_item.hide()
self._add_menu_item("Transcribe Audio File\u2026", self._on_transcribe_file_clicked)
self._add_menu_separator()
@@ -706,6 +714,8 @@ class TrayIndicator:
def _toggle_recognition(self) -> None:
"""Toggle the recognition state between IDLE and LISTENING."""
if self._cancel_if_transcribing():
return
with self._ptt_lock:
self._ptt_owner = None
if self.speech_engine.state == RecognitionState.IDLE:
@@ -715,11 +725,23 @@ class TrayIndicator:
def _start_recognition(self) -> None:
"""Start voice recognition (for push-to-talk mode)."""
if self._cancel_if_transcribing():
return
self._push_to_talk_press(
self.shortcut_manager,
lambda: self.speech_engine.start_recognition(mode="push_to_talk"),
)
def _cancel_if_transcribing(self) -> bool:
"""Cancel a running transcription; the recording hotkey doubles as cancel (#991)."""
if self.speech_engine.state != RecognitionState.PROCESSING:
return False
is_transcribing = getattr(self.speech_engine, "is_transcribing", None)
if not callable(is_transcribing) or is_transcribing() is not True:
return False
cancel = getattr(self.speech_engine, "cancel_transcription", None)
return callable(cancel) and cancel() is True
def _push_to_talk_press(
self, manager: Optional[KeyboardShortcutManager], start: Callable[[], bool]
) -> None:
@@ -760,6 +782,8 @@ class TrayIndicator:
dictating it stops, matching the main toggle's behaviour regardless
of which language the current utterance is in.
"""
if self._cancel_if_transcribing():
return
with self._ptt_lock:
self._ptt_owner = None
if self.speech_engine.state == RecognitionState.IDLE:
@@ -771,6 +795,8 @@ class TrayIndicator:
self, language: str, manager: Optional[KeyboardShortcutManager] = None
) -> None:
"""Push-to-talk start for a per-language shortcut (#805)."""
if self._cancel_if_transcribing():
return
self._push_to_talk_press(
manager,
lambda: self.speech_engine.start_recognition_with_language(
@@ -1158,7 +1184,13 @@ class TrayIndicator:
self.indicator.set_icon_full(icon_key, description)
self._last_icon_key = icon_key
def _update_ui(self, state: RecognitionState):
def _set_indicator_title(self, title: str) -> None:
if getattr(self, "_indicator_title", "Vocalinux") == title:
return
self.indicator.set_title(title)
self._indicator_title = title
def _update_ui(self, state: RecognitionState) -> bool:
"""
Update the UI based on the recognition state.
@@ -1181,19 +1213,36 @@ class TrayIndicator:
if not hasattr(self, "indicator"):
return False
is_transcribing = getattr(self.speech_engine, "is_transcribing", None)
transcribing = callable(is_transcribing) and is_transcribing() is True
show_cancel = state == RecognitionState.PROCESSING and transcribing
self._set_cancel_menu_item_visible(show_cancel)
if state == RecognitionState.IDLE:
self._set_indicator_title("Vocalinux")
self._set_indicator_icon(self._icon_keys["default"], "Microphone off")
self._set_menu_item_enabled("Start Voice Typing", True)
self._set_menu_item_enabled("Stop Voice Typing", False)
elif state == RecognitionState.LISTENING:
self._set_indicator_title("Vocalinux")
self._set_indicator_icon(self._icon_keys["active"], "Microphone on")
self._set_menu_item_enabled("Start Voice Typing", False)
self._set_menu_item_enabled("Stop Voice Typing", True)
elif state == RecognitionState.PROCESSING:
self._set_indicator_icon(self._icon_keys["processing"], "Processing speech")
if transcribing:
shortcut_manager = getattr(self, "shortcut_manager", None)
name = getattr(shortcut_manager, "shortcut_display_name", None)
if not isinstance(name, str) or not name:
name = "the recording shortcut"
description = f"Transcribing, press {name} again to cancel"
else:
description = "Processing speech"
self._set_indicator_title(description if transcribing else "Vocalinux")
self._set_indicator_icon(self._icon_keys["processing"], description)
self._set_menu_item_enabled("Start Voice Typing", False)
self._set_menu_item_enabled("Stop Voice Typing", True)
elif state == RecognitionState.ERROR:
self._set_indicator_title("Vocalinux")
self._set_indicator_icon(self._icon_keys["default"], "Error")
self._set_menu_item_enabled("Start Voice Typing", True)
self._set_menu_item_enabled("Stop Voice Typing", False)
@@ -1236,6 +1285,13 @@ class TrayIndicator:
item.set_sensitive(enabled)
break
def _set_cancel_menu_item_visible(self, visible: bool) -> None:
item = getattr(self, "_cancel_menu_item", None)
if item is None or getattr(self, "menu", None) is None:
return
item.set_sensitive(visible)
item.set_visible(visible)
def _on_start_clicked(self, widget):
"""Handle click on the Start Voice Typing menu item."""
logger.debug("Start Voice Typing clicked")
@@ -1246,6 +1302,11 @@ class TrayIndicator:
logger.debug("Stop Voice Typing clicked")
self.speech_engine.stop_recognition()
def _on_cancel_transcription_clicked(self, widget: Any) -> None:
"""Cancel an in-progress transcription from the tray menu."""
logger.debug("Cancel Transcription clicked")
self.speech_engine.cancel_transcription()
def _refresh_history_menu(self) -> bool:
"""Rebuild the Recent Snippets submenu from the current history."""
if self._history_menu_item is None or self.transcription_history is None:
tests/test_auto_pause_monitor.py
+2
View File
@@ -458,6 +458,8 @@ class TestUnloadModelAndAutoPauseBlock(unittest.TestCase):
mgr._model_lock = MagicMock()
mgr._model_lock.__enter__ = MagicMock(return_value=None)
mgr._model_lock.__exit__ = MagicMock(return_value=False)
mgr._buffer_lock = threading.Lock()
mgr._session_cancel = threading.Event()
mgr._http_session = None
mgr._faster_whisper_engine = None
mgr._reconnection_attempts = 0
tests/test_cancel_transcription.py
+685
View File
@@ -0,0 +1,685 @@
"""Tests for cancelling an active speech recognition session."""
import queue
import threading
import time
from typing import Callable, Optional
from unittest.mock import MagicMock, patch
import vocalinux.speech_recognition.recognition_manager as recognition_module
from vocalinux.common_types import RecognitionState
from vocalinux.speech_recognition.recognition_manager import (
SpeechRecognitionManager,
_AudioSegment,
_ReleaseGate,
)
def _make_manager() -> SpeechRecognitionManager:
duck = MagicMock()
duck.enabled.return_value = False
with patch.object(SpeechRecognitionManager, "_init_whispercpp"):
manager = SpeechRecognitionManager(
engine="whisper_cpp",
model_size="small",
language="en-us",
playback_duck=duck,
)
manager.model = MagicMock()
manager._model_initialized = True
return manager
def _start_session(manager: SpeechRecognitionManager, record_audio: Callable[[], None]) -> None:
manager._record_audio = record_audio
with patch.object(recognition_module, "play_start_sound"):
with patch.object(manager, "_arm_playback_duck"):
assert manager.start_recognition()
def _stop_session(manager: SpeechRecognitionManager) -> None:
with patch.object(recognition_module, "play_stop_sound"):
manager.stop_recognition()
def _no_audio_capture() -> None:
return None
def _stop_with_short_worker_joins(manager: SpeechRecognitionManager) -> None:
with patch.object(SpeechRecognitionManager, "_STOP_WORKER_JOIN_TIMEOUT_S", 0.05):
with patch.object(SpeechRecognitionManager, "_STOP_WORKER_FINAL_JOIN_TIMEOUT_S", 0.01):
_stop_session(manager)
def _wait_for_state(manager: SpeechRecognitionManager, state: RecognitionState) -> None:
deadline = time.monotonic() + 2.0
while manager.state is not state and time.monotonic() < deadline:
time.sleep(0.005)
assert manager.state is state
def _record_until_stopped(
manager: SpeechRecognitionManager, capture_started: threading.Event
) -> None:
manager._recording_segment_has_speech = False
capture_started.set()
while manager.should_record:
time.sleep(0.005)
def test_processing_cancel_is_immediate_and_new_session_remains_isolated() -> None:
manager = _make_manager()
capture_started = threading.Event()
transcriber_started = threading.Event()
release_old_transcriber = threading.Event()
new_transcription_finished = threading.Event()
delivered: list[str] = []
segments: list[str] = []
transcriptions: list[str] = []
def record_audio() -> None:
_record_until_stopped(manager, capture_started)
def transcribe(audio_buffer: list[bytes], language: Optional[str] = None) -> str:
del audio_buffer, language
if not transcriptions:
transcriptions.append("old")
transcriber_started.set()
release_old_transcriber.wait(timeout=5.0)
return "old session"
transcriptions.append("new")
new_transcription_finished.set()
return "new session"
manager._transcribe_with_whispercpp = transcribe
manager._voice_commands_enabled = False
manager.register_text_callback(delivered.append)
manager.register_segment_callback(lambda text, _started_at: segments.append(text))
capture_started.clear()
_start_session(manager, record_audio)
assert capture_started.wait(timeout=1.0)
old_cancel = manager._session_cancel
with manager._buffer_lock:
manager.audio_buffer = [b"old audio"]
manager._recording_segment_has_speech = True
stop_thread = threading.Thread(target=_stop_session, args=(manager,))
stop_thread.start()
assert transcriber_started.wait(timeout=2.0)
assert manager.is_transcribing()
old_queue = manager._segment_queue
cancel_started = time.monotonic()
assert manager.cancel_transcription()
assert time.monotonic() - cancel_started < 0.5
assert manager.state is RecognitionState.IDLE
assert old_queue.get_nowait() is None
assert old_queue.empty()
capture_started.clear()
_start_session(manager, record_audio)
assert capture_started.wait(timeout=1.0)
new_cancel = manager._session_cancel
assert new_cancel is not old_cancel
assert not new_cancel.is_set()
manager._enqueue_audio_segment([b"new audio"])
assert new_transcription_finished.wait(timeout=2.0)
assert delivered == ["new session"]
assert segments == ["new session"]
release_old_transcriber.set()
stop_thread.join(timeout=2.0)
assert not stop_thread.is_alive()
assert manager.state is RecognitionState.LISTENING
assert delivered == ["new session"]
_stop_session(manager)
assert manager.state is RecognitionState.IDLE
def test_cancel_while_listening_discards_final_buffer() -> None:
manager = _make_manager()
capture_started = threading.Event()
manager._voice_commands_enabled = False
def record_audio() -> None:
_record_until_stopped(manager, capture_started)
_start_session(manager, record_audio)
assert capture_started.wait(timeout=1.0)
with manager._buffer_lock:
manager.audio_buffer = [b"final audio"]
manager._recording_segment_has_speech = True
with patch.object(manager, "_enqueue_audio_segment") as enqueue:
assert manager.cancel_transcription()
assert manager.state is RecognitionState.IDLE
assert manager._session_cancel.is_set()
assert not manager.audio_buffer
enqueue.assert_not_called()
def test_cancel_when_idle_returns_false() -> None:
manager = _make_manager()
assert not manager.cancel_transcription()
assert manager.state is RecognitionState.IDLE
def test_is_transcribing_requires_stopped_processing_state() -> None:
manager = _make_manager()
manager.state = RecognitionState.PROCESSING
manager.should_record = False
assert manager.is_transcribing()
manager.should_record = True
assert not manager.is_transcribing()
manager.should_record = False
manager.state = RecognitionState.LISTENING
assert not manager.is_transcribing()
def test_cancel_rejects_an_already_cancelled_session() -> None:
manager = _make_manager()
manager.state = RecognitionState.PROCESSING
manager._session_cancel.set()
assert not manager.cancel_transcription()
assert manager.state is RecognitionState.PROCESSING
def test_cancel_buffered_reload_sets_flag_and_leaves_processing_state() -> None:
manager = _make_manager()
manager.state = RecognitionState.PROCESSING
manager._buffered_reload_session = True
assert manager.cancel_transcription()
assert manager._session_cancel.is_set()
assert manager._cancel_buffered_session.is_set()
assert manager.state is RecognitionState.PROCESSING
assert manager._segment_queue.get_nowait() is None
def test_enqueue_stamps_segment_with_current_session_cancel_event() -> None:
manager = _make_manager()
manager._enqueue_audio_segment([b"audio"])
segment, _language = manager._segment_queue.get_nowait()
assert isinstance(segment, _AudioSegment)
assert segment.cancel_event is manager._session_cancel
def test_cancelled_transcriber_result_is_discarded_before_callbacks() -> None:
manager = _make_manager()
manager._voice_commands_enabled = True
cancel_event = threading.Event()
segment = _AudioSegment([b"audio"], None, cancel_event=cancel_event)
transcriber_started = threading.Event()
release_transcriber = threading.Event()
callbacks: list[str] = []
actions: list[str] = []
manager.command_processor.process_text = MagicMock(return_value=("discard me", ["action"]))
manager.action_callbacks.append(actions.append)
def transcribe(audio_buffer: list[bytes], language: Optional[str] = None) -> str:
del audio_buffer, language
transcriber_started.set()
release_transcriber.wait(timeout=2.0)
return "discard me"
manager._transcribe_with_whispercpp = transcribe
manager.register_text_callback(callbacks.append)
manager.register_segment_callback(lambda text, _started_at: callbacks.append(text))
worker = threading.Thread(target=manager._process_audio_buffer, args=(segment,))
worker.start()
assert transcriber_started.wait(timeout=1.0)
cancel_event.set()
release_transcriber.set()
worker.join(timeout=2.0)
assert not worker.is_alive()
assert not callbacks
assert not actions
manager.command_processor.process_text.assert_not_called()
def test_cancelled_buffer_is_not_transcribed() -> None:
manager = _make_manager()
cancel_event = threading.Event()
cancel_event.set()
segment = _AudioSegment([b"audio"], None, cancel_event=cancel_event)
manager._transcribe_with_whispercpp = MagicMock(return_value="must not run")
manager._process_audio_buffer(segment)
manager._transcribe_with_whispercpp.assert_not_called()
def test_perform_recognition_worker_keeps_its_original_queue() -> None:
manager = _make_manager()
manager.should_record = True
old_cancel = manager._session_cancel
old_queue: queue.Queue[tuple[_AudioSegment, str]] = queue.Queue()
new_queue: queue.Queue[tuple[_AudioSegment, str]] = queue.Queue()
manager._segment_queue = old_queue
process_started = threading.Event()
release_process = threading.Event()
processed: list[str] = []
def process(segment: list[bytes], _language: Optional[str] = None) -> None:
processed.append(segment[0].decode())
process_started.set()
release_process.wait(timeout=2.0)
manager._process_audio_buffer = process
old_queue.put((_AudioSegment([b"old"], None), "en-us"))
worker = threading.Thread(target=manager._perform_recognition)
worker.start()
assert process_started.wait(timeout=1.0)
old_cancel.set()
manager._session_cancel = threading.Event()
manager._segment_queue = new_queue
new_queue.put((_AudioSegment([b"new"], None), "en-us"))
release_process.set()
worker.join(timeout=2.0)
assert not worker.is_alive()
assert processed == ["old"]
assert new_queue.qsize() == 1
def test_whispercpp_abort_callback_is_passed_and_tracks_cancellation() -> None:
class FakeAudioArray:
def astype(self, _dtype: object) -> "FakeAudioArray":
return self
def __truediv__(self, _divisor: float) -> "FakeAudioArray":
return self
def __len__(self) -> int:
return 16000
class FakeNumpy:
int16 = object()
float32 = object()
@staticmethod
def frombuffer(_audio: bytes, dtype: object) -> FakeAudioArray:
del dtype
return FakeAudioArray()
class SupportedModel:
def __init__(self) -> None:
self.abort_callback: Optional[Callable[[], bool]] = None
def transcribe(
self,
_audio: FakeAudioArray,
language: Optional[str] = None,
initial_prompt: str = "",
abort_callback: Optional[Callable[[], bool]] = None,
) -> list[object]:
del language, initial_prompt
self.abort_callback = abort_callback
return []
manager = _make_manager()
model = SupportedModel()
manager.model = model
cancel_event = threading.Event()
segment = _AudioSegment([b"\x00\x00"], None, cancel_event=cancel_event)
with patch.dict("sys.modules", {"numpy": FakeNumpy}):
assert manager._transcribe_with_whispercpp(segment) == ""
assert model.abort_callback is not None
assert not model.abort_callback()
cancel_event.set()
assert model.abort_callback()
def test_whispercpp_without_abort_callback_signature_still_transcribes() -> None:
class FakeAudioArray:
def astype(self, _dtype: object) -> "FakeAudioArray":
return self
def __truediv__(self, _divisor: float) -> "FakeAudioArray":
return self
def __len__(self) -> int:
return 16000
class FakeNumpy:
int16 = object()
float32 = object()
@staticmethod
def frombuffer(_audio: bytes, dtype: object) -> FakeAudioArray:
del dtype
return FakeAudioArray()
class UnsupportedModel:
def __init__(self) -> None:
self.called = False
def transcribe(
self,
_audio: FakeAudioArray,
language: Optional[str] = None,
initial_prompt: str = "",
) -> list[object]:
del language, initial_prompt
self.called = True
return []
manager = _make_manager()
model = UnsupportedModel()
manager.model = model
segment = _AudioSegment([b"\x00\x00"], None, cancel_event=threading.Event())
with patch.dict("sys.modules", {"numpy": FakeNumpy}):
assert manager._transcribe_with_whispercpp(segment) == ""
assert model.called
def test_whispercpp_skips_language_detection_when_cancelled() -> None:
manager = _make_manager()
manager.whispercpp_language_candidates = ["en"]
manager.model.auto_detect_language = MagicMock()
cancel_event = threading.Event()
cancel_event.set()
segment = _AudioSegment([b"\x00\x00"], None, cancel_event=cancel_event)
assert manager._transcribe_with_whispercpp(segment) == ""
manager.model.auto_detect_language.assert_not_called()
def test_late_worker_handoff_keeps_processing_until_transcription_finishes() -> None:
manager = _make_manager()
transcriber_started = threading.Event()
release_transcriber = threading.Event()
delivered: list[str] = []
def transcribe(_audio: list[bytes], _language: Optional[str] = None) -> str:
transcriber_started.set()
assert release_transcriber.wait(timeout=3.0)
return "late text"
manager._transcribe_with_whispercpp = transcribe
manager._voice_commands_enabled = False
manager.register_text_callback(delivered.append)
_start_session(manager, _no_audio_capture)
manager._enqueue_audio_segment([b"slow audio"])
assert transcriber_started.wait(timeout=1.0)
worker_thread = manager.recognition_thread
_stop_with_short_worker_joins(manager)
assert manager.state is RecognitionState.PROCESSING
assert manager.is_transcribing()
release_transcriber.set()
_wait_for_state(manager, RecognitionState.IDLE)
assert delivered == ["late text"]
assert manager._worker_exited is not None
assert manager._worker_exited.is_set()
assert worker_thread is not None
worker_thread.join(timeout=1.0)
assert not worker_thread.is_alive()
def test_cancel_late_worker_discards_text_and_next_session_succeeds() -> None:
manager = _make_manager()
transcriber_started = threading.Event()
release_transcriber = threading.Event()
next_transcription_finished = threading.Event()
next_text_delivered = threading.Event()
delivered: list[str] = []
transcribe_calls = 0
def transcribe(_audio: list[bytes], _language: Optional[str] = None) -> str:
nonlocal transcribe_calls
transcribe_calls += 1
if transcribe_calls == 1:
transcriber_started.set()
assert release_transcriber.wait(timeout=3.0)
return "cancelled text"
next_transcription_finished.set()
return "next session"
def on_text(text: str) -> None:
delivered.append(text)
if text == "next session":
next_text_delivered.set()
manager._transcribe_with_whispercpp = transcribe
manager._voice_commands_enabled = False
manager.register_text_callback(on_text)
_start_session(manager, _no_audio_capture)
manager._enqueue_audio_segment([b"slow audio"])
assert transcriber_started.wait(timeout=1.0)
worker_exited = manager._worker_exited
assert worker_exited is not None
worker_thread = manager.recognition_thread
_stop_with_short_worker_joins(manager)
assert manager.is_transcribing()
assert manager.cancel_transcription()
assert manager.state is RecognitionState.IDLE
release_transcriber.set()
assert worker_exited.wait(timeout=2.0)
assert worker_thread is not None
worker_thread.join(timeout=1.0)
assert not worker_thread.is_alive()
assert manager.state is RecognitionState.IDLE
assert delivered == []
_start_session(manager, _no_audio_capture)
manager._enqueue_audio_segment([b"next audio"])
assert next_transcription_finished.wait(timeout=1.0)
assert next_text_delivered.wait(timeout=1.0)
assert delivered == ["next session"]
_stop_session(manager)
assert manager.state is RecognitionState.IDLE
def test_cancelled_old_worker_does_not_change_new_session_state() -> None:
manager = _make_manager()
old_transcriber_started = threading.Event()
release_old_transcriber = threading.Event()
new_transcription_finished = threading.Event()
new_text_delivered = threading.Event()
delivered: list[str] = []
transcribe_calls = 0
def transcribe(_audio: list[bytes], _language: Optional[str] = None) -> str:
nonlocal transcribe_calls
transcribe_calls += 1
if transcribe_calls == 1:
old_transcriber_started.set()
assert release_old_transcriber.wait(timeout=3.0)
return "old text"
new_transcription_finished.set()
return "new text"
def on_text(text: str) -> None:
delivered.append(text)
if text == "new text":
new_text_delivered.set()
manager._transcribe_with_whispercpp = transcribe
manager._voice_commands_enabled = False
manager.register_text_callback(on_text)
_start_session(manager, _no_audio_capture)
manager._enqueue_audio_segment([b"old audio"])
assert old_transcriber_started.wait(timeout=1.0)
old_worker_exited = manager._worker_exited
assert old_worker_exited is not None
old_worker_thread = manager.recognition_thread
_stop_with_short_worker_joins(manager)
old_cancel = manager._session_cancel
assert manager._idle_on_worker_exit is old_cancel
assert manager.cancel_transcription()
assert manager._idle_on_worker_exit is None
_start_session(manager, _no_audio_capture)
new_cancel = manager._session_cancel
assert new_cancel is not old_cancel
manager._enqueue_audio_segment([b"new audio"])
assert new_transcription_finished.wait(timeout=1.0)
assert new_text_delivered.wait(timeout=1.0)
assert delivered == ["new text"]
release_old_transcriber.set()
assert old_worker_thread is not None
old_worker_thread.join(timeout=2.0)
assert not old_worker_thread.is_alive()
assert old_worker_exited.is_set()
_wait_for_state(manager, RecognitionState.LISTENING)
assert manager._session_cancel is new_cancel
assert not new_cancel.is_set()
assert delivered == ["new text"]
_stop_session(manager)
assert manager.state is RecognitionState.IDLE
def test_fast_worker_still_returns_to_idle_from_stop() -> None:
manager = _make_manager()
transcriber_finished = threading.Event()
def transcribe(_audio: list[bytes], _language: Optional[str] = None) -> str:
transcriber_finished.set()
return "quick text"
manager._transcribe_with_whispercpp = transcribe
manager._voice_commands_enabled = False
_start_session(manager, _no_audio_capture)
manager._enqueue_audio_segment([b"quick audio"])
assert transcriber_finished.wait(timeout=1.0)
_stop_session(manager)
assert manager.state is RecognitionState.IDLE
assert manager._worker_exited is not None
assert manager._worker_exited.is_set()
def test_cancel_drops_push_to_talk_text_held_until_release() -> None:
manager = _make_manager()
manager._voice_commands_enabled = False
delivered: list[str] = []
manager.register_text_callback(delivered.append)
cancel_event = threading.Event()
gate = _ReleaseGate()
segment = _AudioSegment([b"audio"], None, cancel_event=cancel_event, release_gate=gate)
manager._transcribe_with_whispercpp = MagicMock(return_value="held words")
manager._process_audio_buffer(segment)
assert len(gate.held) == 1
cancel_event.set()
gate.set()
manager._release_held_transcripts(gate, force=True, cancel=cancel_event)
assert not gate.held
assert not delivered
def test_cancelled_worker_exit_does_not_force_release_held_text() -> None:
manager = _make_manager()
delivered: list[str] = []
manager.register_text_callback(delivered.append)
manager.should_record = False
gate = _ReleaseGate()
manager._key_release_gate = gate
manager._session_cancel.set()
gate.held.append(recognition_module._HeldTranscript("held words", [], None))
manager._perform_recognition()
assert not gate.held
assert not delivered
def test_uncancelled_release_still_delivers_held_text() -> None:
manager = _make_manager()
delivered: list[str] = []
manager.register_text_callback(delivered.append)
gate = _ReleaseGate()
gate.held.append(recognition_module._HeldTranscript("held words", [], None))
gate.set()
manager._release_held_transcripts(gate, cancel=threading.Event())
assert delivered == ["held words"]
def test_capture_session_current_tracks_cancel_and_replacement() -> None:
manager = _make_manager()
manager.should_record = True
session = manager._session_cancel
assert manager._capture_session_current(session)
manager._session_cancel = threading.Event()
assert not manager._capture_session_current(session)
manager._session_cancel = session
session.set()
assert not manager._capture_session_current(session)
def test_stale_capture_reconnection_does_not_touch_new_session_device() -> None:
manager = _make_manager()
manager.should_record = True
old_session = manager._session_cancel
new_source = MagicMock()
manager._capture_source = new_source
manager._capture_source_for_session = MagicMock(return_value=new_source)
def new_session_starts(_delay: float) -> None:
old_session.set()
manager._session_cancel = threading.Event()
with patch.object(recognition_module.time, "sleep", side_effect=new_session_starts):
assert manager._attempt_audio_reconnection(MagicMock(), old_session) is False
manager._capture_source_for_session.assert_not_called()
new_source.reopen.assert_not_called()
def test_cancelled_capture_thread_does_not_write_into_new_session_buffer() -> None:
manager = _make_manager()
manager._silero_vad = None
manager.should_record = True
old_session = manager._session_cancel
reads = 0
source = MagicMock()
source.requires_pyaudio = False
source.sample_rate = 16000
def read_chunk() -> bytes:
nonlocal reads
reads += 1
if reads == 1:
old_session.set()
manager._session_cancel = threading.Event()
manager.audio_buffer = [b"new-session"]
return b"\x00\x00" * 1024
source.read_chunk.side_effect = read_chunk
manager._new_capture_source = MagicMock(return_value=source)
with patch.object(recognition_module, "play_error_sound", create=True):
manager._record_audio()
assert reads == 1
assert manager.audio_buffer == [b"new-session"]
assert manager.state is not RecognitionState.ERROR
source.close.assert_called_once()
tests/test_common_types_enums_protocols.py
+3 -1
View File
@@ -299,7 +299,7 @@ class TestEnumImports(unittest.TestCase):
class TestProtocolImports(unittest.TestCase):
"""Test that all protocols are importable."""
def test_protocol_imports(self):
def test_protocol_imports(self) -> None:
"""Test that protocols can be imported."""
from vocalinux.common_types import (
SpeechRecognitionManagerProtocol,
@@ -309,6 +309,8 @@ class TestProtocolImports(unittest.TestCase):
# Verify protocols are types with expected methods
self.assertTrue(hasattr(SpeechRecognitionManagerProtocol, "start_recognition"))
self.assertTrue(hasattr(SpeechRecognitionManagerProtocol, "stop_recognition"))
self.assertTrue(hasattr(SpeechRecognitionManagerProtocol, "is_transcribing"))
self.assertTrue(hasattr(SpeechRecognitionManagerProtocol, "cancel_transcription"))
self.assertTrue(hasattr(SpeechRecognitionManagerProtocol, "register_state_callback"))
self.assertTrue(hasattr(TextInjectorProtocol, "inject_text"))
tests/test_language_shortcuts.py
+1
View File
@@ -183,6 +183,7 @@ def _manager_stub(
manager._pending_language_override = None
manager._oneshot_language_restore = None
manager._segment_started_at = None
manager._session_cancel = threading.Event()
manager._key_release_gate = _ReleaseGate()
manager._key_release_gate.set()
return manager
tests/test_playback_ducker.py
+2
View File
@@ -729,6 +729,8 @@ def test_recognition_hook_schedules_after_start_and_restores_before_stop_cue(
):
thread_cls.return_value = MagicMock()
assert manager.start_recognition() is True
assert manager._worker_exited is not None
manager._worker_exited.set()
manager.stop_recognition()
start_at = timeline.index("start_sound")
tests/test_recognition_manager.py
+1
View File
@@ -214,6 +214,7 @@ class TestSpeechRecognition(unittest.TestCase):
# Stop recognition
manager.audio_thread = self.threadInstance
manager.recognition_thread = self.threadInstance
manager._worker_exited.set()
manager.stop_recognition()
# Verify stopped state
tests/test_recognition_manager_core.py
+2
View File
@@ -504,6 +504,7 @@ class TestStartStopRecognition(unittest.TestCase):
manager.start_recognition()
manager.audio_thread = self.threadInstance
manager.recognition_thread = self.threadInstance
manager._worker_exited.set()
manager.stop_recognition()
@@ -587,6 +588,7 @@ class TestPushToTalkMode(unittest.TestCase):
manager.audio_thread = self.threadInstance
manager.recognition_thread = self.threadInstance
manager._worker_exited.set()
manager.stop_recognition()
self.assertEqual(manager._recognition_mode, "toggle")
tests/test_settings_dialog.py
+14 -1
View File
@@ -1254,9 +1254,22 @@ class TestSettingsNavigation(unittest.TestCase):
footer_body = self.source_code.split("def _build_sidebar_footer")[1].split("\n def ")[0]
box_pack = footer_body.index("footer.pack_start(self.test_output_revealer")
status_pack = footer_body.index("footer.pack_start(status_row")
button_pack = footer_body.index("footer.pack_start(self.test_button")
button_pack = footer_body.index("footer.pack_start(test_button_row")
test_button = footer_body.index("test_button_row.pack_start(self.test_button")
cancel_button = footer_body.index("test_button_row.pack_start(self.test_cancel_button")
self.assertLess(box_pack, status_pack)
self.assertLess(status_pack, button_pack)
self.assertLess(test_button, cancel_button)
def test_test_dictation_stop_runs_off_gtk_thread(self) -> None:
"""Finalization keeps the Cancel button usable while recognition stops."""
finalize_body = self.source_code.split("def _finalize_test")[1].split("\n def ")[0]
self.assertIn("def stop_test", finalize_body)
self.assertIn('name="vocalinux-test-dictation-stop"', finalize_body)
self.assertIn(
"GLib.timeout_add(100, self._wait_for_idle_then_restore_callbacks)",
finalize_body,
)
def test_test_dictation_scrolled_window_has_no_shadow(self) -> None:
"""GTK3 ScrolledWindow default shadow is a left stripe on GNOME/Wayland (#847)."""
tests/test_settings_test_dictation.py
+128 -13
View File
@@ -68,8 +68,11 @@ def _dialog_for_test(
) -> Mock:
dialog = Mock()
dialog._test_active = False
dialog._test_token = 0
dialog._test_cancelled = False
dialog._applying_settings = False
dialog.test_button = Mock()
dialog.test_cancel_button = Mock()
dialog.test_output_revealer = Mock()
dialog.test_buffer = _text_buffer()
dialog.test_textview = Mock()
@@ -184,12 +187,60 @@ def test_test_dictation_started_arms_listen_timer() -> None:
SettingsDialog._on_test_clicked(dialog, None)
thread_cls.assert_called_once()
assert thread_cls.call_args.kwargs["args"] == (4.0,)
assert thread_cls.call_args.kwargs["args"] == (4.0, 1)
assert dialog._test_active is True
assert dialog._test_cancelled is False
dialog.test_button.set_label.assert_called_with("Testing… Speak Now!")
dialog.test_cancel_button.show.assert_called_once()
dialog.apply_settings.assert_not_called()
def test_test_dictation_cancel_runs_off_thread_and_restores_callbacks() -> None:
dialog = _dialog_for_test(start_return=True)
with (
patch.object(settings_dialog.threading, "Thread"),
patch.object(settings_dialog, "_undownloaded_model_info", return_value=None),
):
SettingsDialog._on_test_clicked(dialog, None)
live_callbacks = dialog.speech_engine.get_text_callbacks.return_value
dialog.speech_engine.cancel_transcription.side_effect = RuntimeError("cancel failed")
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
SettingsDialog._on_test_cancel_clicked(dialog, None)
assert dialog._test_cancelled is True
dialog.test_button.set_sensitive.assert_called_with(False)
dialog.test_cancel_button.set_sensitive.assert_called_with(False)
dialog.test_button.set_label.assert_called_with("Cancelling…")
thread_cls.assert_called_once()
assert thread_cls.call_args.kwargs["daemon"] is True
assert thread_cls.call_args.kwargs["name"] == "vocalinux-test-dictation-cancel"
with patch.object(settings_dialog, "GLib") as glib:
thread_cls.call_args.kwargs["target"]()
glib.idle_add.assert_called_once_with(dialog._finish_test_after_user_cancel)
result = SettingsDialog._finish_test_after_user_cancel(dialog)
assert result is False
glib.timeout_add.assert_not_called()
dialog.speech_engine.cancel_transcription.assert_called_once()
dialog.speech_engine.set_text_callbacks.assert_called_with(live_callbacks)
assert dialog.speech_engine.test_capture_ceiling is not None
assert dialog.test_buffer._text == "(Test cancelled)"
assert dialog._test_active is False
dialog.test_cancel_button.hide.assert_called()
dialog.update_recognition_progress.assert_called_with("Idle")
def test_cancelled_test_ignores_a_queued_text_callback() -> None:
dialog = Mock()
dialog._test_cancelled = True
dialog.test_buffer = _text_buffer()
assert SettingsDialog._append_test_result(dialog, "late result") is False
assert dialog.test_buffer._text == ""
def test_test_dictation_reconfigures_when_live_engine_differs_from_ui() -> None:
"""UI matches the file but the live manager is still on another engine."""
dialog = _dialog_for_test(start_return=True, model_ready=True)
@@ -257,9 +308,12 @@ def _dialog_for_finalize(
"""Minimal dialog wired for _finalize_test / idle-wait helpers."""
dialog = Mock()
dialog._test_active = True
dialog._test_token = 2
dialog._test_cancelled = False
dialog._test_idle_wait_ticks = 0
dialog._test_timeout_cancel_attempted = False
dialog.test_button = Mock()
dialog.test_cancel_button = Mock()
dialog.test_buffer = _text_buffer()
dialog.update_recognition_progress = Mock()
dialog._saved_text_callbacks = (
@@ -282,6 +336,9 @@ def _dialog_for_finalize(
)
dialog._keep_waiting_after_timeout = SettingsDialog._keep_waiting_after_timeout.__get__(dialog)
dialog._finish_test_restore_ui = SettingsDialog._finish_test_restore_ui.__get__(dialog)
dialog._finish_test_after_user_cancel = SettingsDialog._finish_test_after_user_cancel.__get__(
dialog
)
dialog._test_engine_busy = SettingsDialog._test_engine_busy.__get__(dialog)
dialog._check_test_result = Mock(return_value=False)
@@ -308,6 +365,15 @@ def _dialog_for_finalize(
return dialog
def test_stale_finalize_token_does_not_stop_new_test() -> None:
dialog = _dialog_for_finalize(state=RecognitionState.LISTENING)
assert SettingsDialog._finalize_test(dialog, token=1) is False
dialog.speech_engine.stop_recognition.assert_not_called()
assert dialog._test_active is True
def test_open_dictation_pad_row_invokes_callback_on_click() -> None:
callback = Mock()
dialog = Mock()
@@ -363,19 +429,29 @@ def test_dictation_pad_switch_persists_routing_preference() -> None:
def test_finalize_test_buffered_reload_waits_for_idle_before_restore() -> None:
"""stop_recognition may return while PROCESSING; keep test callbacks."""
"""Stop recognition runs off the GTK thread while test callbacks remain installed."""
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=True)
live = dialog._saved_text_callbacks[0]
with patch.object(settings_dialog, "GLib") as glib:
with (
patch.object(settings_dialog, "GLib") as glib,
patch.object(settings_dialog.threading, "Thread") as thread_cls,
):
SettingsDialog._finalize_test(dialog)
dialog.speech_engine.stop_recognition.assert_not_called()
thread_cls.assert_called_once()
assert thread_cls.call_args.kwargs["daemon"] is True
assert thread_cls.call_args.kwargs["name"] == "vocalinux-test-dictation-stop"
thread_cls.call_args.kwargs["target"]()
dialog.speech_engine.stop_recognition.assert_called_once()
# Must not restore live injector yet — worker still holds the utterance.
dialog.speech_engine.set_text_callbacks.assert_not_called()
assert dialog._saved_text_callbacks == [live]
assert dialog._test_active is True
dialog.test_button.set_label.assert_called_with("Processing…")
dialog.test_cancel_button.set_sensitive.assert_called_with(True)
dialog.test_cancel_button.show.assert_called_once()
glib.timeout_add.assert_called_once_with(100, dialog._wait_for_idle_then_restore_callbacks)
@@ -411,21 +487,57 @@ def test_wait_for_idle_treats_buffered_flag_as_busy_even_if_state_idle() -> None
dialog.speech_engine.set_text_callbacks.assert_not_called()
def test_finalize_test_idle_path_uses_settle_delay() -> None:
"""When no work remains, keep the legacy 500ms settle."""
def test_finalize_test_idle_path_still_polls_after_threaded_stop() -> None:
"""Even an already-idle session uses the off-thread stop and idle poll."""
dialog = _dialog_for_finalize(state=RecognitionState.IDLE, buffered_reload=False)
with patch.object(settings_dialog, "GLib") as glib:
with (
patch.object(settings_dialog, "GLib") as glib,
patch.object(settings_dialog.threading, "Thread") as thread_cls,
):
SettingsDialog._finalize_test(dialog)
dialog.speech_engine.stop_recognition.assert_called_once()
assert dialog._test_active is False
dialog.test_button.set_label.assert_called_with("Test Dictation")
glib.timeout_add.assert_called_once_with(500, dialog._restore_callbacks_and_check_result)
# Live callbacks still saved until the settle timeout fires.
dialog.speech_engine.stop_recognition.assert_not_called()
assert dialog._test_active is True
dialog.test_button.set_label.assert_called_with("Processing…")
dialog.test_cancel_button.show.assert_called_once()
thread_cls.assert_called_once()
glib.timeout_add.assert_called_once_with(100, dialog._wait_for_idle_then_restore_callbacks)
assert hasattr(dialog, "_saved_text_callbacks")
def test_cancel_during_processing_restores_once_without_result_check() -> None:
"""A queued poll after cancellation must not restore callbacks a second time."""
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=False)
live = dialog._saved_text_callbacks[0]
with (
patch.object(settings_dialog, "GLib") as glib,
patch.object(settings_dialog.threading, "Thread") as thread_cls,
):
SettingsDialog._finalize_test(dialog)
assert thread_cls.call_args.kwargs["name"] == "vocalinux-test-dictation-stop"
dialog._test_cancelled = True
assert SettingsDialog._finish_test_after_user_cancel(dialog) is False
assert dialog.test_buffer._text == "(Test cancelled)"
dialog.speech_engine.set_text_callbacks.assert_called_once_with([live])
assert SettingsDialog._wait_for_idle_then_restore_callbacks(dialog) is False
dialog.speech_engine.set_text_callbacks.assert_called_once_with([live])
dialog._check_test_result.assert_not_called()
assert glib.timeout_add.call_count == 1
def test_idle_poll_does_not_restore_after_cancel_flag() -> None:
dialog = _dialog_for_finalize(state=RecognitionState.IDLE, buffered_reload=False)
dialog._test_cancelled = True
assert SettingsDialog._wait_for_idle_then_restore_callbacks(dialog) is False
dialog.speech_engine.set_text_callbacks.assert_not_called()
dialog._check_test_result.assert_not_called()
def test_finalize_waits_for_pending_transcription_and_keeps_test_callback() -> None:
"""Late recognition text stays on the test path after state becomes IDLE."""
live_text: list[str] = []
@@ -435,11 +547,14 @@ def test_finalize_waits_for_pending_transcription_and_keeps_test_callback() -> N
dialog = _dialog_for_finalize(state=RecognitionState.IDLE)
dialog._saved_text_callbacks = [live_callback]
dialog.speech_engine.has_pending_transcription = Mock(side_effect=[True, True, False])
dialog.speech_engine.has_pending_transcription = Mock(side_effect=[True, False])
dialog._test_text_callback = SettingsDialog._test_text_callback.__get__(dialog)
dialog._append_test_result = SettingsDialog._append_test_result.__get__(dialog)
with patch.object(settings_dialog, "GLib") as glib:
with (
patch.object(settings_dialog, "GLib") as glib,
patch.object(settings_dialog.threading, "Thread"),
):
glib.idle_add.side_effect = lambda callback, *args: callback(*args)
SettingsDialog._finalize_test(dialog)
tests/test_suspend_handler.py
+70
View File
@@ -6,6 +6,7 @@ SpeechRecognitionManager.
"""
import sys
import threading
import unittest
from unittest.mock import MagicMock, patch
@@ -491,6 +492,75 @@ class TestTrayIndicatorResumeFlow(unittest.TestCase):
mock_glib.source_remove.assert_not_called()
def test_tray_hotkey_cancels_late_worker_after_stop_join_timeout(self) -> None:
import vocalinux.speech_recognition.recognition_manager as recognition_module
from vocalinux.speech_recognition.recognition_manager import (
SpeechRecognitionManager,
)
duck = MagicMock()
duck.enabled.return_value = False
with patch.object(SpeechRecognitionManager, "_init_whispercpp"):
manager = SpeechRecognitionManager(
engine="whisper_cpp",
model_size="small",
language="en-us",
playback_duck=duck,
)
manager.model = MagicMock()
manager._model_initialized = True
manager._voice_commands_enabled = False
transcriber_started = threading.Event()
release_transcriber = threading.Event()
delivered: list[str] = []
def record_audio() -> None:
return None
def transcribe(_audio: list[bytes], _language: str | None = None) -> str:
transcriber_started.set()
release_transcriber.wait(timeout=3.0)
return "cancelled text"
manager._record_audio = record_audio
manager._transcribe_with_whispercpp = transcribe
manager.register_text_callback(delivered.append)
with patch.object(recognition_module, "play_start_sound"):
with patch.object(manager, "_arm_playback_duck"):
assert manager.start_recognition()
manager._enqueue_audio_segment([b"slow audio"])
assert transcriber_started.wait(timeout=1.0)
worker_exited = manager._worker_exited
assert worker_exited is not None
worker_thread = manager.recognition_thread
assert worker_thread is not None
try:
with patch.object(SpeechRecognitionManager, "_STOP_WORKER_JOIN_TIMEOUT_S", 0.05):
with patch.object(
SpeechRecognitionManager, "_STOP_WORKER_FINAL_JOIN_TIMEOUT_S", 0.01
):
with patch.object(recognition_module, "play_stop_sound"):
manager.stop_recognition()
assert manager.state is RecognitionState.PROCESSING
assert manager.is_transcribing()
tray = self._make_tray_indicator()
tray.speech_engine = manager
tray._ptt_lock = threading.Lock()
tray._ptt_owner = None
tray._toggle_recognition()
assert manager.state is RecognitionState.IDLE
finally:
release_transcriber.set()
worker_thread.join(timeout=2.0)
assert not worker_thread.is_alive()
assert worker_exited.is_set()
assert manager.state is RecognitionState.IDLE
assert delivered == []
class TestTrayAutoPauseWiring(unittest.TestCase):
"""TrayIndicator auto-pause config + pause/resume handlers.
tests/test_tray_indicator.py
+124 -1
View File
@@ -7,7 +7,9 @@ The tests focus on the business logic of the TrayIndicator class.
import os
import sys
import threading
import unittest
from threading import Thread
from types import SimpleNamespace
from unittest.mock import MagicMock, PropertyMock, patch
@@ -190,12 +192,133 @@ class TestTrayIndicator(unittest.TestCase):
self.mock_speech_engine.stop_recognition.assert_called_once()
self.mock_speech_engine.start_recognition.assert_not_called()
def test_toggle_recognition_from_processing(self):
def test_toggle_recognition_from_processing(self) -> None:
"""Test toggling recognition state from PROCESSING."""
self.mock_speech_engine.state = self.RecognitionState.PROCESSING
self.tray_indicator._toggle_recognition()
self.mock_speech_engine.stop_recognition.assert_called_once()
self.mock_speech_engine.start_recognition.assert_not_called()
self.mock_speech_engine.cancel_transcription.assert_not_called()
def test_toggle_recognition_cancels_transcription_before_lock(self) -> None:
self.mock_speech_engine.state = self.RecognitionState.PROCESSING
self.mock_speech_engine.is_transcribing.return_value = True
self.mock_speech_engine.cancel_transcription.return_value = True
self.tray_indicator._toggle_recognition()
self.mock_speech_engine.cancel_transcription.assert_called_once()
self.mock_speech_engine.stop_recognition.assert_not_called()
self.mock_speech_engine.start_recognition.assert_not_called()
def test_push_to_talk_cancel_clears_owner_and_release_is_noop(self) -> None:
self.mock_speech_engine.state = self.RecognitionState.PROCESSING
self.mock_speech_engine.is_transcribing.return_value = True
self.mock_speech_engine.cancel_transcription.return_value = True
self.tray_indicator._start_recognition()
self.tray_indicator._release_main_push_to_talk()
self.mock_speech_engine.cancel_transcription.assert_called_once()
self.mock_speech_engine.start_recognition.assert_not_called()
self.mock_speech_engine.stop_recognition.assert_not_called()
def test_language_hotkeys_cancel_transcription(self) -> None:
self.mock_speech_engine.state = self.RecognitionState.PROCESSING
self.mock_speech_engine.is_transcribing.return_value = True
self.mock_speech_engine.cancel_transcription.return_value = True
self.tray_indicator._toggle_recognition_in_language("fr")
self.tray_indicator._start_recognition_in_language("fr", self.mock_ksm)
self.assertEqual(self.mock_speech_engine.cancel_transcription.call_count, 2)
self.mock_speech_engine.start_recognition_with_language.assert_not_called()
self.mock_speech_engine.stop_recognition.assert_not_called()
def test_cancel_hotkey_does_not_wait_for_ptt_lock(self) -> None:
self.mock_speech_engine.state = self.RecognitionState.PROCESSING
self.mock_speech_engine.is_transcribing.return_value = True
self.mock_speech_engine.cancel_transcription.return_value = True
lock_held = threading.Event()
release_lock = threading.Event()
hotkey_finished = threading.Event()
def hold_lock() -> None:
with self.tray_indicator._ptt_lock:
lock_held.set()
release_lock.wait(timeout=2.0)
def press_hotkey() -> None:
self.tray_indicator._toggle_recognition()
hotkey_finished.set()
holder = Thread(target=hold_lock)
holder.start()
self.assertTrue(lock_held.wait(timeout=1.0))
hotkey = Thread(target=press_hotkey)
hotkey.start()
try:
self.assertTrue(hotkey_finished.wait(timeout=0.5))
finally:
release_lock.set()
holder.join(timeout=1.0)
hotkey.join(timeout=1.0)
self.mock_speech_engine.cancel_transcription.assert_called_once()
def test_cancel_menu_visibility_and_transcription_tooltip(self) -> None:
self.tray_indicator.indicator = MagicMock()
self.tray_indicator.menu = MagicMock()
self.tray_indicator._cancel_menu_item = MagicMock()
self.tray_indicator._icon_keys = {
"default": "off",
"active": "on",
"processing": "processing",
}
self.tray_indicator.shortcut_manager.shortcut_display_name = "Ctrl+Alt+V"
self.mock_speech_engine.state = self.RecognitionState.PROCESSING
self.mock_speech_engine.is_transcribing.return_value = True
with patch.object(self.tray_indicator, "_set_indicator_icon"):
with patch.object(self.tray_indicator, "_set_menu_item_enabled"):
with patch.object(self.tray_indicator, "_update_overlay"):
self.tray_indicator._update_ui(self.RecognitionState.PROCESSING)
self.assertEqual(
self.tray_indicator.indicator.set_title.call_args.args[0],
"Transcribing, press Ctrl+Alt+V again to cancel",
)
self.tray_indicator._cancel_menu_item.set_sensitive.assert_called_with(True)
self.tray_indicator._cancel_menu_item.set_visible.assert_called_with(True)
self.mock_speech_engine.state = self.RecognitionState.LISTENING
self.tray_indicator._update_ui(self.RecognitionState.LISTENING)
self.tray_indicator._cancel_menu_item.set_sensitive.assert_called_with(False)
self.tray_indicator._cancel_menu_item.set_visible.assert_called_with(False)
self.tray_indicator.indicator.set_title.assert_called_with("Vocalinux")
def test_cancel_tooltip_uses_fallback_without_shortcut_name(self) -> None:
self.tray_indicator.indicator = MagicMock()
self.tray_indicator.menu = MagicMock()
self.tray_indicator._cancel_menu_item = MagicMock()
self.tray_indicator._icon_keys = {"default": "off", "active": "on", "processing": "busy"}
self.tray_indicator.shortcut_manager = MagicMock()
self.mock_speech_engine.state = self.RecognitionState.PROCESSING
self.mock_speech_engine.is_transcribing.return_value = True
with patch.object(self.tray_indicator, "_set_indicator_icon") as set_icon:
with patch.object(self.tray_indicator, "_set_menu_item_enabled"):
with patch.object(self.tray_indicator, "_update_overlay"):
self.tray_indicator._update_ui(self.RecognitionState.PROCESSING)
set_icon.assert_called_once_with(
"busy", "Transcribing, press the recording shortcut again to cancel"
)
def test_cancel_menu_callback_calls_manager_directly(self) -> None:
self.tray_indicator._on_cancel_transcription_clicked(None)
self.mock_speech_engine.cancel_transcription.assert_called_once()
def test_on_start_clicked(self):
"""Test start button click handler."""