fix(ptt): prevent premature transcription during push-to-talk silence (#359)

* fix(ptt): prevent premature transcription during push-to-talk silence

In push-to-talk mode, a brief silence while holding the hotkey would
trigger transcription mid-hold, injecting text before key release.

Add recognition_mode parameter to start_recognition() so the silence
timeout in _record_audio() defers enqueuing audio in PTT mode. Audio
accumulates in the buffer until key release calls stop_recognition(),
which already handles final buffer enqueue. Toggle mode is unchanged.

* fix(clipboard): run clipboard copy in background thread to prevent blocking

The _copy_to_clipboard calls (wl-copy, xclip, xsel) have 5s timeouts
each and run synchronously on the recognition thread. When they hang,
they block stop_recognition() from completing, delaying the state
transition to IDLE and preventing the next push-to-talk press.

Move clipboard copy to a daemon thread on the two success paths so it
never blocks transcription or state transitions. The failure-fallback
path stays synchronous since it needs the return value.
This commit is contained in:
Jatin K Malik
2026-03-28 00:53:45 -07:00
committed by GitHub
parent 5cef9bfeac
commit 113c846d52
5 changed files with 92 additions and 10 deletions
+1 -1
View File
@@ -21,7 +21,7 @@ class SpeechRecognitionManagerProtocol(Protocol):
state: RecognitionState
def start_recognition(self) -> None:
def start_recognition(self, mode: str = "toggle") -> None:
"""Start the speech recognition process."""
...
@@ -561,6 +561,7 @@ class SpeechRecognitionManager:
# Recording control flags
self.should_record = False
self._recognition_mode = "toggle" # "toggle" or "push_to_talk"
self.audio_buffer = []
self._buffer_lock = threading.Lock() # Thread safety for audio_buffer
self._model_lock = threading.Lock() # Thread safety for model/recognizer access
@@ -1499,7 +1500,7 @@ class SpeechRecognitionManager:
"""Check if the model is initialized and ready for recognition."""
return self._model_initialized and self.model is not None
def start_recognition(self):
def start_recognition(self, mode: str = "toggle"):
"""Start the speech recognition process."""
if self.state != RecognitionState.IDLE:
logger.warning(f"Cannot start recognition in current state: {self.state}")
@@ -1526,6 +1527,7 @@ class SpeechRecognitionManager:
# Set recording flag
self.should_record = True
self._recognition_mode = mode
self.audio_buffer = []
self._segment_queue = queue.Queue(maxsize=32)
@@ -1588,6 +1590,7 @@ class SpeechRecognitionManager:
if self.recognition_thread and self.recognition_thread.is_alive():
self.recognition_thread.join(timeout=1.0)
self._recognition_mode = "toggle"
self._update_state(RecognitionState.IDLE)
def _record_audio(self):
@@ -1765,9 +1768,15 @@ class SpeechRecognitionManager:
silence_counter += CHUNK / RATE # Convert chunks to seconds
if silence_counter > self.silence_timeout: # Use self.silence_timeout
if len(self.audio_buffer) > 0:
logger.debug("Silence detected, queueing audio segment")
self._enqueue_audio_segment(self.audio_buffer)
self.audio_buffer = []
if self._recognition_mode == "push_to_talk":
logger.debug(
"Silence detected in push-to-talk mode, "
"deferring transcription until key release"
)
else:
logger.debug("Silence detected, queueing audio segment")
self._enqueue_audio_segment(self.audio_buffer)
self.audio_buffer = []
silence_counter = 0
else: # Speech
if not speech_detected_in_session:
+11 -4
View File
@@ -9,6 +9,7 @@ import logging
import os
import shutil
import subprocess
import threading
import time
from enum import Enum
from typing import Optional # noqa: F401
@@ -446,8 +447,11 @@ class TextInjector:
if result:
logger.info("Text injection completed successfully")
if self._should_copy_to_clipboard():
self._copy_to_clipboard(text)
logger.debug("Text also copied to clipboard (setting enabled)")
threading.Thread(
target=self._copy_to_clipboard,
args=(text,),
daemon=True,
).start()
return result
else:
logger.error("IBus injector not initialized")
@@ -478,8 +482,11 @@ class TextInjector:
logger.info("Text injection completed successfully")
if self._should_copy_to_clipboard():
self._copy_to_clipboard(text)
logger.debug("Text also copied to clipboard (setting enabled)")
threading.Thread(
target=self._copy_to_clipboard,
args=(text,),
daemon=True,
).start()
return True
except Exception as e:
+1 -1
View File
@@ -237,7 +237,7 @@ class TrayIndicator:
def _start_recognition(self):
"""Start voice recognition (for push-to-talk mode)."""
if self.speech_engine.state == RecognitionState.IDLE:
self.speech_engine.start_recognition()
self.speech_engine.start_recognition(mode="push_to_talk")
def _stop_recognition(self):
"""Stop voice recognition (for push-to-talk mode)."""
+66
View File
@@ -525,6 +525,72 @@ class TestStartStopRecognition(unittest.TestCase):
mock_audio_feedback.play_stop_sound.assert_not_called()
class TestPushToTalkMode(unittest.TestCase):
"""Test push-to-talk recognition mode prevents premature transcription on silence."""
def setUp(self):
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.downloadMock = self.mockDownload.start()
self.pathMock.return_value = "/mock/path/vosk-model"
self.recognizerMock = MagicMock()
self.kaldiMock.return_value = self.recognizerMock
self.recognizerMock.FinalResult.return_value = '{"text": ""}'
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
self.patcher_exists = patch("os.path.exists", return_value=True)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.mockDownload.stop()
self.patcher_exists.stop()
def test_default_recognition_mode_is_toggle(self):
manager = SpeechRecognitionManager(engine="vosk")
self.assertEqual(manager._recognition_mode, "toggle")
def test_start_recognition_sets_toggle_mode_by_default(self):
manager = SpeechRecognitionManager(engine="vosk")
manager.start_recognition()
self.assertEqual(manager._recognition_mode, "toggle")
def test_start_recognition_sets_push_to_talk_mode(self):
manager = SpeechRecognitionManager(engine="vosk")
manager.start_recognition(mode="push_to_talk")
self.assertEqual(manager._recognition_mode, "push_to_talk")
def test_stop_recognition_resets_mode_to_toggle(self):
manager = SpeechRecognitionManager(engine="vosk")
manager.start_recognition(mode="push_to_talk")
self.assertEqual(manager._recognition_mode, "push_to_talk")
manager.audio_thread = self.threadInstance
manager.recognition_thread = self.threadInstance
manager.stop_recognition()
self.assertEqual(manager._recognition_mode, "toggle")
class TestWhisperInitialization(unittest.TestCase):
"""Test cases for Whisper engine initialization."""