diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 19c5ef0..8f3a2a3 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -19,7 +19,7 @@ import sys import threading import time from pathlib import Path -from typing import TYPE_CHECKING, Any, Callable, Optional, Union, cast +from typing import TYPE_CHECKING, Any, Callable, NamedTuple, Optional, Union, cast if TYPE_CHECKING: import numpy as np @@ -88,6 +88,22 @@ if TYPE_CHECKING: from ..custom_dictionary import CustomDictionaryManager +class _HeldTranscript(NamedTuple): + """A transcript held until its push-to-talk session is released.""" + + text: str + actions: list[str] + started_at: Optional[float] + + +class _ReleaseGate(threading.Event): + """Session event that gates delivery while allowing transcription to continue.""" + + def __init__(self) -> None: + super().__init__() + self.held: list[_HeldTranscript] = [] + + class _AudioSegment(list): """A queued audio segment that knows when its capture began. @@ -96,6 +112,7 @@ class _AudioSegment(list): ``time.monotonic()`` timestamp of its first captured chunk. That stamp is what lets a listener tell speech captured before a point in time from speech captured after it, however late the decode finishes. + A push-to-talk ``release_gate`` holds transcribed text until the key is released. """ def __init__( @@ -103,6 +120,7 @@ class _AudioSegment(list): chunks: list[bytes], started_at: Optional[float], language: Optional[str] = None, + release_gate: Optional[_ReleaseGate] = None, ) -> None: super().__init__(chunks) self.started_at = started_at @@ -110,6 +128,7 @@ class _AudioSegment(list): # outlives its session must not transcribe a stale segment in a newer # session's language. self.language = language + self.release_gate = release_gate def _pywhispercpp_distribution_version() -> Optional[tuple[int, ...]]: @@ -537,13 +556,23 @@ class SpeechRecognitionManager: self._buffer_lock = threading.Lock() # Thread safety for audio_buffer self._model_lock = threading.Lock() # Thread safety for model/recognizer access self._segment_queue: queue.Queue[tuple[_AudioSegment, str]] = queue.Queue(maxsize=32) + self._key_release_gate = _ReleaseGate() + self._key_release_gate.set() # Reliability improvements - Issue #92 self._max_buffer_size = 5000 # Maximum number of audio chunks in buffer + # Longest stretch of continuous speech kept in one buffer before it is + # queued for transcription. A push-to-talk hold (or a toggle session + # that never goes quiet) otherwise grows the buffer until the safety + # cap above discards its oldest audio. The effective flush point is + # also capped at 75% of that chunk budget, so a device with short + # chunks (high capture rate) can never reach the destructive trim + # before the segment is flushed. + self._max_segment_seconds = 90.0 self._reconnection_attempts = 0 self._max_reconnection_attempts = 5 self._reconnection_delay = 1.0 # Initial delay in seconds - self._last_audio_error_time = 0 + self._last_audio_error_time = 0.0 self._audio_stream = None self._pyaudio_instance = None self._capture_sample_rate = 16000 # Default, updated when device is opened @@ -3280,6 +3309,7 @@ class SpeechRecognitionManager: self.audio_buffer = [] self._segment_started_at = None self._segment_queue = queue.Queue(maxsize=32) + self._key_release_gate = _ReleaseGate() # Start the audio recording thread self.audio_thread = threading.Thread(target=self._record_audio) @@ -3308,6 +3338,7 @@ class SpeechRecognitionManager: # sees the dictation as ended. The volume goes back only once the # microphone thread has left the device, and before the stop cue. self.should_record = False + self._key_release_gate.set() self._update_state(RecognitionState.PROCESSING) self._cancel_pending_playback_duck() @@ -3337,7 +3368,12 @@ class SpeechRecognitionManager: 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) + 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( @@ -3477,25 +3513,26 @@ class SpeechRecognitionManager: # ran or was refused by the guards inside start_recognition. self._pending_language_override = None - def _record_audio(self): + def _record_audio(self) -> None: """Record audio from the microphone with reconnection logic.""" + gate = self._key_release_gate # Lazy import to avoid circular dependency from ..ui.audio_feedback import play_error_sound # noqa: F401 try: - import numpy as np - except ImportError as e: - logger.error(f"Failed to import required audio libraries: {e}") - logger.error("Please install required dependencies: pip install pyaudio numpy") - self.should_record = False - self.release_playback_duck() - play_error_sound() - self._buffered_capture_failed = True - self._update_state(RecognitionState.ERROR) - self._signal_buffered_capture_done() - return + try: + import numpy as np + except ImportError as e: + logger.error(f"Failed to import required audio libraries: {e}") + logger.error("Please install required dependencies: pip install pyaudio numpy") + self.should_record = False + self.release_playback_duck() + play_error_sound() + self._buffered_capture_failed = True + self._update_state(RecognitionState.ERROR) + self._signal_buffered_capture_done() + return - try: # PyAudio configuration CHUNK = 1024 @@ -3557,7 +3594,7 @@ class SpeechRecognitionManager: logger.info("Audio recording started") # Record audio while should_record is True - silence_counter = 0 + silence_counter = 0.0 speech_detected_in_session = False self._recording_segment_has_speech = False log_level_interval = 0 # Counter for periodic level logging @@ -3699,6 +3736,43 @@ class SpeechRecognitionManager: ) speech_detected_in_session = True silence_counter = 0 + + # Continuous speech with no pause (a held push-to-talk + # key, or a toggle session that never goes quiet) keeps + # one segment growing until the safety cap silently drops + # its oldest audio. Queue the segment long before that so + # every word is transcribed in order while recording keeps + # going. Capture chunks are 16 kHz mono, so their duration + # follows from the byte count at any negotiated rate. + # Push-to-talk text is held until the key is released. + chunk_seconds = len(data) / (2 * 16000) + if chunk_seconds > 0 and self._recording_segment_has_speech: + buffered_seconds = len(self.audio_buffer) * chunk_seconds + flush_seconds = min( + self._max_segment_seconds, + self._max_buffer_size * chunk_seconds * 0.75, + ) + if buffered_seconds >= flush_seconds and not getattr( + self, "_reload_pending", False + ): + with self._buffer_lock: + segment = self.audio_buffer.copy() + self.audio_buffer = [] + self._recording_segment_has_speech = False + silence_counter = 0 + logger.info( + "Continuing dictation: queued %.0fs segment " + "(%d chunks) for transcription; push-to-talk text " + "will be held until release", + buffered_seconds, + len(segment), + ) + self._enqueue_audio_segment( + segment, + release_gate=( + gate if self._recognition_mode == "push_to_talk" else None + ), + ) except (IOError, OSError) as e: current_time = time.time() logger.error(f"Audio device error: {e}") @@ -3772,6 +3846,7 @@ class SpeechRecognitionManager: self._buffered_capture_failed = True self._update_state(RecognitionState.ERROR) finally: + gate.set() self._signal_buffered_capture_done() def _process_final_buffer(self): @@ -3808,6 +3883,48 @@ class SpeechRecognitionManager: return None return self._transcribe_with_remote_api(audio_buffer, session, language) + def _deliver_transcript( + self, + text: str, + actions: list[str], + started_at: Optional[float], + ) -> None: + """Deliver a transcribed segment to registered callbacks.""" + logger.debug( + f"processed_text='{text[:50] if text else '(empty)'}...', callbacks={len(self.text_callbacks)}" + ) + if text: + if started_at is None: + started_at = time.monotonic() + for segment_callback in self.segment_callbacks: + segment_callback(text, started_at) + for callback in self.text_callbacks: + logger.debug( + f"invoking text callback: {callback.__name__ if hasattr(callback, '__name__') else callback}" + ) + callback(text) + + for action in actions: + for callback in self.action_callbacks: + callback(action) + + def _release_held_transcripts( + self, + gate: Optional[_ReleaseGate], + force: bool = False, + ) -> None: + """Deliver queued transcripts once their push-to-talk gate opens.""" + if gate is None or (not gate.is_set() and not force): + return + + while gate.held: + transcript = gate.held.pop(0) + self._deliver_transcript( + transcript.text, + transcript.actions, + transcript.started_at, + ) + def _process_audio_buffer( self, audio_buffer: list[bytes], language: Optional[str] = None ) -> None: @@ -3843,10 +3960,15 @@ class SpeechRecognitionManager: return text = transcriber() 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) return # Process text - either with voice commands or pass through directly logger.debug(f"_process_audio_buffer got text='{text[:50] if text else '(empty)'}...'") + processed_text = "" + actions: list[str] = [] if text: # Corrections must precede command parsing: a safe replacement can # stop a model mishearing from triggering a destructive action. @@ -3859,29 +3981,30 @@ class SpeechRecognitionManager: processed_text = text.strip() actions = [] - # Call text callbacks with processed text - logger.debug( - f"processed_text='{processed_text[:50] if processed_text else '(empty)'}...', callbacks={len(self.text_callbacks)}" - ) - if processed_text: - started_at = getattr(audio_buffer, "started_at", None) - if started_at is None: - started_at = time.monotonic() - for segment_callback in self.segment_callbacks: - segment_callback(processed_text, started_at) - for callback in self.text_callbacks: - logger.debug( - f"invoking text callback: {callback.__name__ if hasattr(callback, '__name__') else callback}" + gate = getattr(audio_buffer, "release_gate", None) + if gate is not None and not gate.is_set(): + if processed_text or actions: + gate.held.append( + _HeldTranscript( + processed_text, + actions, + getattr(audio_buffer, "started_at", None), ) - callback(processed_text) + ) + logger.info("Holding transcribed text until the push-to-talk key is released") + return + elif gate is not None: + self._release_held_transcripts(gate) - # Call action callbacks for each action - for action in actions: - for callback in self.action_callbacks: - callback(action) + self._deliver_transcript( + processed_text, + actions, + getattr(audio_buffer, "started_at", None), + ) - def _perform_recognition(self): + def _perform_recognition(self) -> None: """Perform speech recognition in real-time.""" + gate = self._key_release_gate logger.debug("_perform_recognition thread started") try: while True: @@ -3891,6 +4014,7 @@ class SpeechRecognitionManager: try: queued = self._segment_queue.get(timeout=0.1) except queue.Empty: + self._release_held_transcripts(gate) # Only exit if we're not recording AND queue is empty if not self.should_record and self._segment_queue.empty(): logger.debug( @@ -3919,6 +4043,7 @@ class SpeechRecognitionManager: ) if self.should_record: self._update_state(RecognitionState.PROCESSING) + self._release_held_transcripts(gate) self._process_audio_buffer(remaining_segment, remaining_language) except queue.Empty: break @@ -3929,20 +4054,30 @@ 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._process_audio_buffer(segment, segment_language) if self.should_record: 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). - def _enqueue_audio_segment(self, audio_buffer: list[bytes]): + def _enqueue_audio_segment( + self, + audio_buffer: list[bytes], + release_gate: Optional[_ReleaseGate] = None, + ) -> None: """Queue an audio segment for asynchronous transcription.""" - segment = _AudioSegment(audio_buffer.copy(), self._segment_started_at) + segment = _AudioSegment( + audio_buffer.copy(), + self._segment_started_at, + release_gate=release_gate, + ) if not segment: logger.warning("_enqueue_audio_segment called with empty buffer") return diff --git a/tests/test_language_shortcuts.py b/tests/test_language_shortcuts.py index 4c751ce..ab2d8d6 100644 --- a/tests/test_language_shortcuts.py +++ b/tests/test_language_shortcuts.py @@ -166,7 +166,10 @@ def _manager_stub( 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 + from vocalinux.speech_recognition.recognition_manager import ( + SpeechRecognitionManager, + _ReleaseGate, + ) manager = SpeechRecognitionManager.__new__(SpeechRecognitionManager) manager.engine = engine @@ -180,6 +183,8 @@ def _manager_stub( manager._pending_language_override = None manager._oneshot_language_restore = None manager._segment_started_at = None + manager._key_release_gate = _ReleaseGate() + manager._key_release_gate.set() return manager diff --git a/tests/test_recognition_manager.py b/tests/test_recognition_manager.py index f1b1e3a..162eb66 100644 --- a/tests/test_recognition_manager.py +++ b/tests/test_recognition_manager.py @@ -185,12 +185,16 @@ class TestSpeechRecognition(unittest.TestCase): text_callback.assert_called_once_with("processed text") action_callback.assert_called_once_with("action1") - def test_start_stop_recognition(self): + def test_start_stop_recognition(self) -> None: """Test starting and stopping recognition.""" manager = SpeechRecognitionManager(engine="vosk") + initial_gate = manager._key_release_gate # Start recognition manager.start_recognition() + session_gate = manager._key_release_gate + self.assertIsNot(session_gate, initial_gate) + self.assertFalse(session_gate.is_set()) # Verify state and calls self.assertEqual(manager.state, RecognitionState.LISTENING) @@ -215,9 +219,15 @@ class TestSpeechRecognition(unittest.TestCase): # Verify stopped state self.assertEqual(manager.state, RecognitionState.IDLE) self.assertFalse(manager.should_record) + self.assertTrue(session_gate.is_set()) mock_audio_feedback.play_stop_sound.assert_called_once() self.threadInstance.join.assert_called() + manager.start_recognition() + self.assertIsNot(manager._key_release_gate, session_gate) + self.assertFalse(manager._key_release_gate.is_set()) + manager.stop_recognition() + def test_whisper_engine(self): """Test initialization and usage with Whisper engine.""" # Setup Whisper and torch mocks diff --git a/tests/test_recognition_manager_advanced.py b/tests/test_recognition_manager_advanced.py index 23cb0a3..ba338f5 100644 --- a/tests/test_recognition_manager_advanced.py +++ b/tests/test_recognition_manager_advanced.py @@ -6,10 +6,12 @@ Properly isolated - saves/restores sys.modules to avoid polluting other tests. import importlib import json import os +import queue import sys import threading import time import unittest +from typing import Optional from unittest.mock import MagicMock, call, patch # ---- Safely mock external deps, preserving stdlib ---- @@ -35,8 +37,10 @@ if "gi.repository" not in sys.modules: from vocalinux.common_types import RecognitionState # noqa: E402 from vocalinux.speech_recognition.recognition_manager import ( # noqa: E402 SpeechRecognitionManager, + _AudioSegment, _get_supported_channels, _get_supported_sample_rate, + _ReleaseGate, _setup_alsa_error_handler, get_audio_input_devices, ) @@ -64,6 +68,301 @@ def _make_manager(engine="whisper_cpp", **kw): return mgr +def _identity(text: str) -> str: + return text + + +class TestPushToTalkReleaseGate(unittest.TestCase): + def test_gated_segment_is_held_until_release(self) -> None: + """A gated segment is transcribed but delivers callbacks only on release.""" + manager = _make_manager(engine="vosk") + manager._voice_commands_enabled = True + gate = _ReleaseGate() + segment = _AudioSegment([b"audio"], None, release_gate=gate) + delivered: list[tuple[str, str]] = [] + + def on_segment(text: str, _started_at: float) -> None: + delivered.append(("segment", text)) + + def on_text(text: str) -> None: + delivered.append(("text", text)) + + def on_action(action: str) -> None: + delivered.append(("action", action)) + + manager.segment_callbacks.append(on_segment) + manager.text_callbacks.append(on_text) + manager.action_callbacks.append(on_action) + + with ( + patch.object(manager, "_transcribe_with_vosk", return_value="raw"), + patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), + patch.object( + manager.command_processor, + "process_text", + return_value=("hello", ["action"]), + ), + ): + manager._process_audio_buffer(segment) + self.assertEqual(delivered, []) + self.assertEqual(len(gate.held), 1) + self.assertEqual( + (gate.held[0].text, gate.held[0].actions, gate.held[0].started_at), + ("hello", ["action"], None), + ) + gate.set() + manager._release_held_transcripts(gate) + + self.assertEqual( + delivered, + [("segment", "hello"), ("text", "hello"), ("action", "action")], + ) + self.assertEqual(gate.held, []) + + def test_old_session_gate_does_not_release_new_session_text(self) -> None: + """A slow worker's gate must not release a newer session's held text.""" + manager = _make_manager(engine="vosk") + manager._voice_commands_enabled = False + delivered: list[str] = [] + manager.text_callbacks.append(delivered.append) + old_gate = _ReleaseGate() + new_gate = _ReleaseGate() + + def transcribe(audio_buffer: list[bytes]) -> str: + return audio_buffer[0].decode() + + with ( + patch.object(manager, "_transcribe_with_vosk", side_effect=transcribe), + patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), + ): + manager._process_audio_buffer(_AudioSegment([b"old"], None, release_gate=old_gate)) + manager._process_audio_buffer(_AudioSegment([b"new"], None, release_gate=new_gate)) + self.assertEqual(delivered, []) + + # The old session's key goes up; the new session is still held. + old_gate.set() + manager._release_held_transcripts(old_gate) + self.assertEqual(delivered, ["old"]) + self.assertEqual(len(new_gate.held), 1) + + new_gate.set() + manager._release_held_transcripts(new_gate) + + self.assertEqual(delivered, ["old", "new"]) + + def test_empty_transcription_while_held_does_not_flush(self) -> None: + """A blank mid-hold result must not release text before the key is up.""" + manager = _make_manager(engine="vosk") + manager._voice_commands_enabled = False + delivered: list[str] = [] + manager.text_callbacks.append(delivered.append) + gate = _ReleaseGate() + + with ( + patch.object(manager, "_transcribe_with_vosk", return_value="hello"), + patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), + ): + manager._process_audio_buffer(_AudioSegment([b"audio"], None, release_gate=gate)) + self.assertEqual(delivered, []) + self.assertEqual(len(gate.held), 1) + + with patch.object(manager, "_transcribe_with_vosk", return_value=""): + manager._process_audio_buffer(_AudioSegment([b"blank"], None, release_gate=gate)) + self.assertEqual(delivered, []) + self.assertEqual(len(gate.held), 1) + + gate.set() + manager._release_held_transcripts(gate) + + self.assertEqual(delivered, ["hello"]) + + def test_ungated_and_released_segments_deliver_immediately(self) -> None: + """Segments without a pending gate never block callback delivery.""" + manager = _make_manager(engine="vosk") + manager._voice_commands_enabled = False + delivered: list[str] = [] + manager.text_callbacks.append(delivered.append) + opened_gate = _ReleaseGate() + opened_gate.set() + + with ( + patch.object(manager, "_transcribe_with_vosk", return_value="hello"), + patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), + ): + for gate in (None, opened_gate): + manager._process_audio_buffer(_AudioSegment([b"audio"], None, release_gate=gate)) + + self.assertEqual(delivered, ["hello", "hello"]) + + def test_worker_transcribes_ahead_and_delivers_after_release(self) -> None: + """Queued segments transcribe during a hold and deliver FIFO after release.""" + manager = _make_manager(engine="vosk") + manager._voice_commands_enabled = False + manager.should_record = True + manager._segment_queue = queue.Queue() + gate = _ReleaseGate() + manager._key_release_gate = gate + transcribed = {name: threading.Event() for name in ("first", "second", "final")} + processed_both = threading.Event() + delivered: list[str] = [] + manager.text_callbacks.append(delivered.append) + + first = _AudioSegment([b"first"], None, release_gate=gate) + second = _AudioSegment([b"second"], None, release_gate=gate) + final = _AudioSegment([b"final"], None, release_gate=gate) + for segment in (first, second): + manager._segment_queue.put((segment, "en-us")) + + def transcribe(audio_buffer: list[bytes]) -> str: + text = audio_buffer[0].decode() + transcribed[text].set() + return text + + original_process = manager._process_audio_buffer + processed_count = 0 + + def process_and_signal( + audio_buffer: list[bytes], + language: Optional[str] = None, + ) -> None: + nonlocal processed_count + original_process(audio_buffer, language) + processed_count += 1 + if processed_count == 2: + processed_both.set() + + with ( + patch.object(manager, "_transcribe_with_vosk", side_effect=transcribe), + patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), + patch.object(manager, "_process_audio_buffer", side_effect=process_and_signal), + ): + worker = threading.Thread(target=manager._perform_recognition) + worker.start() + try: + self.assertTrue(transcribed["first"].wait(timeout=2)) + self.assertTrue(transcribed["second"].wait(timeout=2)) + self.assertTrue(processed_both.wait(timeout=2)) + self.assertEqual(len(gate.held), 2) + self.assertEqual(delivered, []) + self.assertTrue(worker.is_alive()) + gate.set() + manager._segment_queue.put((final, "en-us")) + finally: + gate.set() + manager.should_record = False + manager._segment_queue.put(None) + worker.join(timeout=2) + + self.assertFalse(worker.is_alive()) + self.assertEqual(delivered, ["first", "second", "final"]) + + def test_gate_release_without_final_segment_drains_on_timeout(self) -> None: + """Held transcripts drain when the open gate is noticed on queue timeout.""" + manager = _make_manager(engine="vosk") + manager._voice_commands_enabled = False + manager.should_record = True + manager._segment_queue = queue.Queue() + gate = _ReleaseGate() + manager._key_release_gate = gate + transcribed = {name: threading.Event() for name in ("first", "second")} + processed_both = threading.Event() + delivered_both = threading.Event() + delivered: list[str] = [] + + def on_text(text: str) -> None: + delivered.append(text) + if len(delivered) == 2: + delivered_both.set() + + manager.text_callbacks.append(on_text) + for name in ("first", "second"): + manager._segment_queue.put( + (_AudioSegment([name.encode()], None, release_gate=gate), "en-us") + ) + + def transcribe(audio_buffer: list[bytes]) -> str: + text = audio_buffer[0].decode() + transcribed[text].set() + return text + + original_process = manager._process_audio_buffer + processed_count = 0 + + def process_and_signal( + audio_buffer: list[bytes], + language: Optional[str] = None, + ) -> None: + nonlocal processed_count + original_process(audio_buffer, language) + processed_count += 1 + if processed_count == 2: + processed_both.set() + + with ( + patch.object(manager, "_transcribe_with_vosk", side_effect=transcribe), + patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), + patch.object(manager, "_process_audio_buffer", side_effect=process_and_signal), + ): + worker = threading.Thread(target=manager._perform_recognition) + worker.start() + try: + self.assertTrue(transcribed["first"].wait(timeout=2)) + self.assertTrue(transcribed["second"].wait(timeout=2)) + self.assertTrue(processed_both.wait(timeout=2)) + self.assertEqual(delivered, []) + gate.set() + self.assertTrue(delivered_both.wait(timeout=2)) + self.assertEqual(delivered, ["first", "second"]) + finally: + gate.set() + manager.should_record = False + manager._segment_queue.put(None) + worker.join(timeout=2) + + self.assertFalse(worker.is_alive()) + + def test_worker_exit_force_releases_held_transcripts(self) -> None: + """The worker's finally block delivers held text even if the gate stays closed.""" + manager = _make_manager(engine="vosk") + manager._voice_commands_enabled = False + manager.should_record = False + manager._segment_queue = queue.Queue() + gate = _ReleaseGate() + manager._key_release_gate = gate + delivered: list[str] = [] + manager.text_callbacks.append(delivered.append) + for name in ("first", "second"): + manager._segment_queue.put( + (_AudioSegment([name.encode()], None, release_gate=gate), "en-us") + ) + manager._segment_queue.put(None) + + def transcribe(audio_buffer: list[bytes]) -> str: + return audio_buffer[0].decode() + + with ( + patch.object( + manager, + "_transcribe_with_vosk", + side_effect=transcribe, + ), + patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), + ): + worker = threading.Thread(target=manager._perform_recognition) + worker.start() + try: + worker.join(timeout=2) + self.assertFalse(worker.is_alive()) + self.assertFalse(gate.is_set()) + finally: + if worker.is_alive(): + gate.set() + worker.join(timeout=2) + + self.assertEqual(delivered, ["first", "second"]) + self.assertEqual(gate.held, []) + + class TestALSAErrorHandler(unittest.TestCase): def test_setup_success(self): with patch("ctypes.CDLL") as mock_cdll: @@ -220,15 +519,39 @@ class TestCancelDownload(unittest.TestCase): class TestStartStopRecognition(unittest.TestCase): - def test_stop_recognition(self): + def test_stop_recognition(self) -> None: mgr = _make_manager() mgr.state = RecognitionState.LISTENING mgr.should_record = True + gate = _ReleaseGate() + mgr._key_release_gate = gate mgr.audio_thread = MagicMock() mgr.audio_thread.is_alive.return_value = True mgr.audio_thread.join = MagicMock() mgr.stop_recognition() self.assertFalse(mgr.should_record) + self.assertTrue(gate.is_set()) + + def test_stop_enqueues_final_buffer_with_gate_only_for_push_to_talk(self) -> None: + for mode in ("push_to_talk", "toggle"): + with self.subTest(mode=mode): + mgr = _make_manager() + mgr.state = RecognitionState.LISTENING + mgr.should_record = True + mgr._recognition_mode = mode + mgr._key_release_gate = _ReleaseGate() + mgr.audio_buffer = [b"speech"] + mgr._recording_segment_has_speech = True + mgr._enqueue_audio_segment = MagicMock() + + with patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"): + mgr.stop_recognition() + + expected_gate = mgr._key_release_gate if mode == "push_to_talk" else None + mgr._enqueue_audio_segment.assert_called_once_with( + [b"speech"], + release_gate=expected_gate, + ) def test_stop_sound_plays_after_audio_thread_join(self): mgr = _make_manager() diff --git a/tests/test_recognition_manager_download_buffer.py b/tests/test_recognition_manager_download_buffer.py index 6c235a2..2836cf4 100644 --- a/tests/test_recognition_manager_download_buffer.py +++ b/tests/test_recognition_manager_download_buffer.py @@ -320,7 +320,7 @@ class TestBufferManagement(unittest.TestCase): with patch("vocalinux.ui.audio_feedback.play_stop_sound"): manager.stop_recognition() - enqueue_mock.assert_called_once_with([b"small"]) + enqueue_mock.assert_called_once_with([b"small"], release_gate=None) assert manager.audio_buffer == [] def test_stop_recognition_large_buffer_trim(self): diff --git a/tests/test_recognition_manager_silero.py b/tests/test_recognition_manager_silero.py index 952055f..1679232 100644 --- a/tests/test_recognition_manager_silero.py +++ b/tests/test_recognition_manager_silero.py @@ -12,8 +12,10 @@ returning scripted speech probabilities. Verifies: """ import sys +import threading import time import unittest # noqa: E402 +from typing import Optional from unittest.mock import MagicMock, patch # noqa: E402 # Earlier test modules install `sys.modules["numpy"] = MagicMock()` at module @@ -40,7 +42,10 @@ if "gi.repository" not in sys.modules: sys.modules["gi.repository"] = MagicMock() from vocalinux.common_types import RecognitionState # noqa: E402 -from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager # noqa: E402 +from vocalinux.speech_recognition.recognition_manager import ( # noqa: E402 + SpeechRecognitionManager, + _ReleaseGate, +) for _k, _v in _ORIG_MODULES.items(): if _v is not None: @@ -171,7 +176,16 @@ class TestRecordAudioSileroPath(unittest.TestCase): # Capture flushes triggered by the silence-detection branch self.enqueued = [] - self.mgr._enqueue_audio_segment = lambda buf: self.enqueued.append(list(buf)) + self.enqueued_gates: list[Optional[_ReleaseGate]] = [] + + def enqueue_segment( + audio_buffer: list[bytes], + release_gate: Optional[_ReleaseGate] = None, + ) -> None: + self.enqueued.append(list(audio_buffer)) + self.enqueued_gates.append(release_gate) + + self.mgr._enqueue_audio_segment = enqueue_segment def _drive( self, @@ -234,6 +248,46 @@ class TestRecordAudioSileroPath(unittest.TestCase): self._drive(probs=[0.95] * 20, vad_sensitivity=3) self.assertEqual(len(self.enqueued), 0, "speech should keep silence counter at 0, no flush") + def test_continuous_speech_flushes_before_buffer_cap(self) -> None: + """A pause-free hold must be queued, not trimmed, when it grows long.""" + # 20 chunks * 0.064s = 1.28s. With a 0.5s segment limit the loop + # queues ~0.5s of audio and starts a fresh segment, twice. + self.mgr._max_segment_seconds = 0.5 + self._drive(probs=[0.95] * 20, vad_sensitivity=3) + self.assertEqual(len(self.enqueued), 2) + self.assertEqual(len(self.mgr.audio_buffer), 4) + self.assertLess(len(self.mgr.audio_buffer), self.mgr._max_buffer_size) + + def test_chunk_budget_flushes_before_cap_on_short_chunks(self) -> None: + """The 75% chunk-budget limit must win when chunks are short. + + A high capture rate makes each chunk short, so a 90 s time limit would + be reached long after the destructive cap. The flush point has to + derive from the chunk budget in that case, or the trim sees speech. + """ + # 10 chunks * 0.064s = 0.64s of buffer, so 75% is 7.5 chunks -> flush + # at chunk 8, two chunks before the cap could trim. + self.mgr._max_buffer_size = 10 + self.mgr._max_segment_seconds = 90.0 + self._drive(probs=[0.95] * 10, vad_sensitivity=3) + self.assertEqual(len(self.enqueued), 1) + self.assertEqual(len(self.enqueued[0]), 8) + self.assertEqual(len(self.mgr.audio_buffer), 2) + self.assertLess(len(self.mgr.audio_buffer), self.mgr._max_buffer_size) + # Every captured chunk is either queued or still buffered: no trim. + self.assertEqual( + sum(len(segment) for segment in self.enqueued) + len(self.mgr.audio_buffer), + 10, + ) + + def test_speech_during_reload_is_not_flushed(self) -> None: + """A model reload keeps the utterance intact rather than enqueuing it.""" + self.mgr._max_segment_seconds = 0.5 + self.mgr._reload_pending = True + self._drive(probs=[0.95] * 20, vad_sensitivity=3) + self.assertEqual(self.enqueued, []) + self.assertEqual(len(self.mgr.audio_buffer), 20) + def test_reload_keeps_speech_across_silence(self) -> None: """Speech during reload stays together despite intervening silence.""" self.mgr._reload_pending = True @@ -301,6 +355,30 @@ class TestRecordAudioSileroPath(unittest.TestCase): self._drive(probs=[0.6] * 20, vad_sensitivity=1) self.assertEqual(len(self.enqueued), 0) + def test_push_to_talk_flushes_carry_the_session_release_gate(self) -> None: + """Mid-hold segments carry the session gate, which capture exit opens.""" + self.mgr._recognition_mode = "push_to_talk" + self.mgr._max_segment_seconds = 0.5 + self.mgr._key_release_gate = _ReleaseGate() + gate = self.mgr._key_release_gate + + self._drive(probs=[0.95] * 20, vad_sensitivity=3) + + self.assertGreater(len(self.enqueued_gates), 0) + self.assertTrue(all(enqueued_gate is gate for enqueued_gate in self.enqueued_gates)) + self.assertTrue(gate.is_set()) + + def test_toggle_flushes_do_not_carry_a_release_gate(self) -> None: + """Toggle-mode segments are not gated even though capture exit opens its event.""" + self.mgr._max_segment_seconds = 0.5 + self.mgr._key_release_gate = _ReleaseGate() + gate = self.mgr._key_release_gate + + self._drive(probs=[0.95] * 20, vad_sensitivity=3) + + self.assertEqual(self.enqueued_gates, [None, None]) + self.assertTrue(gate.is_set()) + def test_sensitivity_highest_accepts_borderline_speech(self): """At sensitivity=5 (threshold 0.3), prob=0.6 is speech, no flush.""" self._drive(probs=[0.6] * 20, vad_sensitivity=5) @@ -350,7 +428,16 @@ class TestRecordAudioAmplitudeFallback(unittest.TestCase): self.mgr._silero_vad = None self.enqueued = [] - self.mgr._enqueue_audio_segment = lambda buf: self.enqueued.append(list(buf)) + self.enqueued_gates: list[Optional[_ReleaseGate]] = [] + + def enqueue_segment( + audio_buffer: list[bytes], + release_gate: Optional[_ReleaseGate] = None, + ) -> None: + self.enqueued.append(list(audio_buffer)) + self.enqueued_gates.append(release_gate) + + self.mgr._enqueue_audio_segment = enqueue_segment def _drive(self, payload, max_iters): stream = MagicMock() @@ -408,7 +495,7 @@ class TestAudioLevelCallback(unittest.TestCase): mgr.silence_timeout = 0.5 mgr.vad_sensitivity = 3 mgr._silero_vad = None - mgr._enqueue_audio_segment = lambda buf: None + mgr._enqueue_audio_segment = MagicMock() levels = [] mgr._audio_level_callbacks = [levels.append]