e66bd218d9
* fix(audio): bound capture reads and support cancellation * fix(recognition): bound and serialize capture shutdown * fix(recognition): cancel retries and report stalled microphones * fix(audio): prefer pulse when the ALSA default is unavailable * fix(recognition): guard late callbacks and finalize capture tests * docs(audio): explain input selection and stall recovery * fix(speech): enforce auto-pause across buffered reload timeouts * fix(speech): recover auto-resume after reload timeout * test(speech): fix auto-pause follow-up test regressions * fix(speech): retain auto-resume after reconfigure timeout * fix(tray): preserve auto-pause recovery across suspend * test(capture-shutdown): give the bare tray the suspend path's shortcut managers _on_system_suspend now releases the main and per-language hotkey listeners before sleep (#1025); the fixture builds TrayIndicator without __init__, so it needs the two attributes present. * fix(tray): release hotkey listeners on suspend even while auto-paused Merging #1025's suspend cleanup with this branch's auto-pause handoff left an early return that skipped shortcut_manager.stop() whenever a configured game was active, keeping EVIOCGRABs held through sleep. The paused branch now only owns the model-lifecycle work; listener release and stopping active recognition run for every suspend. * fix(tray): hold the apply lock only around the download's verify and save Wrapping the whole recommended-model download in _apply_settings_lock blocked every GTK-thread settings apply (e.g. Test Dictation) for the length of the transfer. The engine already serializes reconfigures, so the lock only needs to cover verifying the download still owns the receipt and saving it; an apply that lands mid-download now proceeds instead of freezing the app. --------- Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
1297 lines
49 KiB
Python
1297 lines
49 KiB
Python
"""
|
|
Deep coverage tests for recognition_manager.py.
|
|
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 ----
|
|
_MOCK_KEYS = [
|
|
"vosk",
|
|
"whisper",
|
|
"torch",
|
|
"pyaudio",
|
|
"pywhispercpp",
|
|
"pywhispercpp.model",
|
|
]
|
|
_ORIG_MODULES = {}
|
|
for _k in _MOCK_KEYS:
|
|
_ORIG_MODULES[_k] = sys.modules.get(_k)
|
|
if _k not in sys.modules:
|
|
sys.modules[_k] = MagicMock()
|
|
|
|
if "gi" not in sys.modules:
|
|
sys.modules["gi"] = MagicMock()
|
|
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 ( # noqa: E402
|
|
SpeechRecognitionManager,
|
|
_AudioSegment,
|
|
_get_supported_channels,
|
|
_get_supported_sample_rate,
|
|
_ReleaseGate,
|
|
_setup_alsa_error_handler,
|
|
get_audio_input_devices,
|
|
)
|
|
|
|
# Restore stdlib modules immediately after import
|
|
for _k, _v in _ORIG_MODULES.items():
|
|
if _v is not None:
|
|
sys.modules[_k] = _v
|
|
elif _k in sys.modules and isinstance(sys.modules[_k], MagicMock):
|
|
del sys.modules[_k]
|
|
|
|
|
|
def _make_manager(engine="whisper_cpp", **kw):
|
|
"""Create a manager with deferred init to avoid actually loading models."""
|
|
with (
|
|
patch.object(SpeechRecognitionManager, "_init_vosk"),
|
|
patch.object(SpeechRecognitionManager, "_init_whisper"),
|
|
patch.object(SpeechRecognitionManager, "_init_whispercpp"),
|
|
patch.object(SpeechRecognitionManager, "_init_parakeet"),
|
|
patch.object(SpeechRecognitionManager, "_init_faster_whisper"),
|
|
):
|
|
mgr = SpeechRecognitionManager(
|
|
engine=engine, model_size="small", language="en-us", defer_download=True, **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:
|
|
mock_cdll.return_value = MagicMock()
|
|
_setup_alsa_error_handler()
|
|
mock_cdll.assert_called_once_with("libasound.so.2")
|
|
|
|
def test_setup_oserror(self):
|
|
with patch("ctypes.CDLL", side_effect=OSError("not found")):
|
|
_setup_alsa_error_handler() # Should not raise
|
|
|
|
|
|
class TestGetAudioInputDevices(unittest.TestCase):
|
|
def test_returns_devices(self):
|
|
mock_pa_inst = MagicMock()
|
|
mock_pa_inst.get_device_count.return_value = 3
|
|
mock_pa_inst.get_device_info_by_index.side_effect = [
|
|
{"name": "Mic 1", "maxInputChannels": 2, "index": 0},
|
|
{"name": "Speaker", "maxInputChannels": 0, "index": 1},
|
|
{"name": "Mic 2", "maxInputChannels": 1, "index": 2},
|
|
]
|
|
mock_pyaudio = MagicMock()
|
|
mock_pyaudio.PyAudio.return_value = mock_pa_inst
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
devices = get_audio_input_devices()
|
|
self.assertEqual(len(devices), 2)
|
|
|
|
def test_handles_error(self):
|
|
mock_pyaudio = MagicMock()
|
|
mock_pyaudio.PyAudio.side_effect = OSError("no audio")
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
devices = get_audio_input_devices()
|
|
self.assertEqual(devices, [])
|
|
|
|
|
|
class TestSupportedChannels(unittest.TestCase):
|
|
def _run_with_pyaudio(self, func):
|
|
mock_pyaudio = MagicMock()
|
|
mock_pyaudio.paInt16 = 8
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
return func()
|
|
|
|
def test_mono_supported(self):
|
|
mock_pa = MagicMock()
|
|
mock_pa.is_format_supported.return_value = True
|
|
result = self._run_with_pyaudio(lambda: _get_supported_channels(mock_pa, None))
|
|
self.assertEqual(result, 1)
|
|
|
|
def test_mono_fails_stereo_ok(self):
|
|
mock_pa = MagicMock()
|
|
call_count = [0]
|
|
mock_stream = MagicMock()
|
|
|
|
def open_side_effect(**kw):
|
|
call_count[0] += 1
|
|
if kw.get("channels") == 1:
|
|
raise IOError("Invalid number of channels -9998")
|
|
return mock_stream
|
|
|
|
mock_pa.open.side_effect = open_side_effect
|
|
result = self._run_with_pyaudio(lambda: _get_supported_channels(mock_pa, None))
|
|
self.assertEqual(result, 2)
|
|
|
|
def test_both_fail(self):
|
|
mock_pa = MagicMock()
|
|
mock_pa.is_format_supported.side_effect = ValueError("nope")
|
|
result = self._run_with_pyaudio(lambda: _get_supported_channels(mock_pa, None))
|
|
self.assertEqual(result, 1)
|
|
|
|
|
|
class TestSupportedSampleRate(unittest.TestCase):
|
|
def _run_with_pyaudio(self, func):
|
|
mock_pyaudio = MagicMock()
|
|
mock_pyaudio.paInt16 = 8
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
return func()
|
|
|
|
def test_default_rate_used(self):
|
|
mock_pa = MagicMock()
|
|
mock_pa.get_default_input_device_info.return_value = {"defaultSampleRate": 48000}
|
|
mock_stream = MagicMock()
|
|
mock_pa.open.return_value = mock_stream
|
|
result = self._run_with_pyaudio(lambda: _get_supported_sample_rate(mock_pa, None, 1))
|
|
self.assertEqual(result, 48000)
|
|
|
|
def test_all_open_fail_returns_16000(self):
|
|
mock_pa = MagicMock()
|
|
mock_pa.get_default_input_device_info.return_value = {"defaultSampleRate": 0}
|
|
mock_pa.open.side_effect = IOError("no rate works")
|
|
result = self._run_with_pyaudio(lambda: _get_supported_sample_rate(mock_pa, None, 1))
|
|
self.assertEqual(result, 16000)
|
|
|
|
|
|
class TestManagerConstruction(unittest.TestCase):
|
|
def test_vosk_engine(self):
|
|
mgr = _make_manager(engine="vosk")
|
|
self.assertEqual(mgr.engine, "vosk")
|
|
|
|
def test_whisper_engine(self):
|
|
mgr = _make_manager(engine="whisper")
|
|
self.assertEqual(mgr.engine, "whisper")
|
|
|
|
def test_whispercpp_engine(self):
|
|
mgr = _make_manager(engine="whisper_cpp")
|
|
self.assertEqual(mgr.engine, "whisper_cpp")
|
|
|
|
def test_default_state(self):
|
|
mgr = _make_manager()
|
|
self.assertEqual(mgr.state, RecognitionState.IDLE)
|
|
self.assertIsNone(mgr.model)
|
|
|
|
|
|
class TestCallbacks(unittest.TestCase):
|
|
def test_register_text_callback(self):
|
|
mgr = _make_manager()
|
|
cb = MagicMock()
|
|
mgr.register_text_callback(cb)
|
|
self.assertIn(cb, mgr.text_callbacks)
|
|
|
|
def test_register_state_callback(self):
|
|
mgr = _make_manager()
|
|
cb = MagicMock()
|
|
mgr.register_state_callback(cb)
|
|
self.assertIn(cb, mgr.state_callbacks)
|
|
|
|
def test_register_action_callback(self):
|
|
mgr = _make_manager()
|
|
cb = MagicMock()
|
|
mgr.register_action_callback(cb)
|
|
self.assertIn(cb, mgr.action_callbacks)
|
|
|
|
def test_register_audio_level_callback(self):
|
|
mgr = _make_manager()
|
|
cb = MagicMock()
|
|
mgr.register_audio_level_callback(cb)
|
|
self.assertIn(cb, mgr._audio_level_callbacks)
|
|
|
|
def test_update_state(self):
|
|
mgr = _make_manager()
|
|
cb = MagicMock()
|
|
mgr.register_state_callback(cb)
|
|
mgr._update_state(RecognitionState.LISTENING)
|
|
self.assertEqual(mgr.state, RecognitionState.LISTENING)
|
|
cb.assert_called_with(RecognitionState.LISTENING)
|
|
|
|
|
|
class TestCancelDownload(unittest.TestCase):
|
|
def test_cancel_download(self):
|
|
mgr = _make_manager()
|
|
mgr._download_cancelled = False
|
|
mgr.cancel_download()
|
|
self.assertTrue(mgr._download_cancelled)
|
|
|
|
|
|
class TestStartStopRecognition(unittest.TestCase):
|
|
|
|
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()
|
|
mgr.state = RecognitionState.LISTENING
|
|
mgr.should_record = True
|
|
mgr.audio_thread = MagicMock()
|
|
mgr.audio_thread.is_alive.return_value = True
|
|
events = []
|
|
|
|
def capture_exited(timeout: float | None = None) -> None:
|
|
events.append("join")
|
|
mgr.audio_thread.is_alive.return_value = False
|
|
|
|
mgr.audio_thread.join.side_effect = capture_exited
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.play_stop_sound",
|
|
side_effect=lambda: events.append("sound"),
|
|
):
|
|
mgr.stop_recognition()
|
|
|
|
# sound must play after the audio thread has joined so the cue cannot be captured
|
|
self.assertEqual(events[:2], ["join", "sound"])
|
|
|
|
def test_stop_sound_guard_chunk_calculation(self):
|
|
mgr = _make_manager()
|
|
|
|
self.assertEqual(mgr._get_stop_sound_guard_chunks(), 3)
|
|
|
|
mgr.stop_sound_guard_ms = 0
|
|
self.assertEqual(mgr._get_stop_sound_guard_chunks(), 0)
|
|
|
|
|
|
class TestReconfigure(unittest.TestCase):
|
|
def test_reconfigure_language(self):
|
|
mgr = _make_manager()
|
|
mgr.state = RecognitionState.IDLE
|
|
with patch.object(mgr, "_init_whispercpp"):
|
|
mgr.reconfigure(language="fr")
|
|
self.assertEqual(mgr.language, "fr")
|
|
|
|
def test_reconfigure_model_size(self):
|
|
mgr = _make_manager()
|
|
mgr.state = RecognitionState.IDLE
|
|
with patch.object(mgr, "_init_whispercpp"):
|
|
mgr.reconfigure(model_size="medium", force_download=False)
|
|
self.assertEqual(mgr.model_size, "medium")
|
|
|
|
def test_reconfigure_engine(self):
|
|
mgr = _make_manager()
|
|
mgr.state = RecognitionState.IDLE
|
|
with patch.object(mgr, "_init_vosk"):
|
|
mgr.reconfigure(engine="vosk", force_download=False)
|
|
self.assertEqual(mgr.engine, "vosk")
|
|
|
|
def test_reconfigure_language_candidates(self) -> None:
|
|
mgr = _make_manager()
|
|
mgr.state = RecognitionState.IDLE
|
|
with patch.object(mgr, "_init_whispercpp"):
|
|
mgr.reconfigure(whispercpp_language_candidates="en,es", force_download=False)
|
|
self.assertEqual(mgr.whispercpp_language_candidates, ["en", "es"])
|
|
|
|
|
|
class TestProcessFinalBuffer(unittest.TestCase):
|
|
def test_process_final_buffer_whispercpp(self):
|
|
mgr = _make_manager(engine="whisper_cpp")
|
|
mgr.audio_buffer = [b"\x00\x00" * 512] * 5
|
|
mgr._capture_sample_rate = 16000
|
|
cb = MagicMock()
|
|
mgr.register_text_callback(cb)
|
|
mgr.command_processor = MagicMock()
|
|
mgr.command_processor.process_text.return_value = "hello"
|
|
|
|
with patch.object(mgr, "_transcribe_with_whispercpp", return_value="hello world"):
|
|
mgr._process_final_buffer()
|
|
|
|
def test_process_final_buffer_whisper(self):
|
|
mgr = _make_manager(engine="whisper")
|
|
mgr.audio_buffer = [b"\x00\x00" * 512] * 5
|
|
mgr._capture_sample_rate = 16000
|
|
mgr.command_processor = MagicMock()
|
|
mgr.command_processor.process_text.return_value = "hello"
|
|
|
|
with patch.object(mgr, "_transcribe_with_whisper", return_value="hello world"):
|
|
mgr._process_final_buffer()
|
|
|
|
def test_process_final_buffer_empty(self):
|
|
mgr = _make_manager(engine="whisper_cpp")
|
|
mgr.audio_buffer = []
|
|
mgr._process_final_buffer() # Should handle empty gracefully
|
|
|
|
|
|
class TestTranscribeWhispercpp(unittest.TestCase):
|
|
def test_normalize_language_candidates(self) -> None:
|
|
self.assertEqual(
|
|
SpeechRecognitionManager._normalize_language_candidates(" en-US, es ; fr "),
|
|
["en", "es", "fr"],
|
|
)
|
|
self.assertEqual(
|
|
SpeechRecognitionManager._normalize_language_candidates(["EN", "es", "en"]),
|
|
["en", "es"],
|
|
)
|
|
self.assertEqual(SpeechRecognitionManager._normalize_language_candidates(""), [])
|
|
self.assertEqual(SpeechRecognitionManager._normalize_language_candidates(None), [])
|
|
|
|
def test_transcribe_success(self):
|
|
mgr = _make_manager()
|
|
mgr.language = "en-us"
|
|
mock_segment = MagicMock()
|
|
mock_segment.text = "hello world"
|
|
mgr.model = MagicMock()
|
|
mgr.model.transcribe.return_value = [mock_segment]
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512])
|
|
# Should return transcribed text
|
|
|
|
def test_transcribe_model_none(self):
|
|
mgr = _make_manager()
|
|
mgr.model = None
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512])
|
|
self.assertEqual(result, "")
|
|
|
|
def test_transcribe_empty_buffer(self):
|
|
mgr = _make_manager()
|
|
mock_np = MagicMock()
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whispercpp([])
|
|
self.assertEqual(result, "")
|
|
|
|
def test_transcribe_exception(self):
|
|
mgr = _make_manager()
|
|
mgr.model = MagicMock()
|
|
mgr.model.transcribe.side_effect = RuntimeError("crash")
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512])
|
|
self.assertEqual(result, "")
|
|
|
|
def test_transcribe_auto_language(self):
|
|
mgr = _make_manager()
|
|
mgr.language = "auto"
|
|
mock_segment = MagicMock()
|
|
mock_segment.text = "bonjour"
|
|
mgr.model = MagicMock()
|
|
mgr.model.transcribe.return_value = [mock_segment]
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512])
|
|
|
|
self.assertEqual(result, "bonjour")
|
|
mgr.model.transcribe.assert_called_once()
|
|
self.assertIsNone(mgr.model.transcribe.call_args.kwargs["language"])
|
|
|
|
def test_transcribe_specific_language(self) -> None:
|
|
mgr = _make_manager()
|
|
mgr.language = "es"
|
|
mock_segment = MagicMock()
|
|
mock_segment.text = "hola"
|
|
mgr.model = MagicMock()
|
|
mgr.model.transcribe.return_value = [mock_segment]
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512])
|
|
|
|
self.assertEqual(result, "hola")
|
|
mgr.model.transcribe.assert_called_once()
|
|
self.assertEqual(mgr.model.transcribe.call_args.kwargs["language"], "es")
|
|
|
|
def test_transcribe_candidate_languages_restricts_to_english(self) -> None:
|
|
mgr = _make_manager(whispercpp_language_candidates="en,es")
|
|
mgr.language = "auto"
|
|
mock_segment = MagicMock()
|
|
mock_segment.text = "hello"
|
|
mgr.model = MagicMock()
|
|
mgr.model.auto_detect_language.return_value = (("en", 0.8), {"en": 0.7, "es": 0.3})
|
|
mgr.model.transcribe.return_value = [mock_segment]
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512])
|
|
|
|
self.assertEqual(result, "hello")
|
|
mgr.model.auto_detect_language.assert_called_once()
|
|
mgr.model.transcribe.assert_called_once()
|
|
self.assertEqual(mgr.model.transcribe.call_args.kwargs["language"], "en")
|
|
|
|
def test_transcribe_candidate_languages_restricts_to_spanish(self) -> None:
|
|
mgr = _make_manager(whispercpp_language_candidates=["en", "es", "fr"])
|
|
mgr.language = "auto"
|
|
mock_segment = MagicMock()
|
|
mock_segment.text = "hola"
|
|
mgr.model = MagicMock()
|
|
mgr.model.auto_detect_language.return_value = (
|
|
("de", 0.6),
|
|
{"en": 0.2, "es": 0.5, "fr": 0.1},
|
|
)
|
|
mgr.model.transcribe.return_value = [mock_segment]
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512])
|
|
|
|
self.assertEqual(result, "hola")
|
|
mgr.model.auto_detect_language.assert_called_once()
|
|
mgr.model.transcribe.assert_called_once()
|
|
self.assertEqual(mgr.model.transcribe.call_args.kwargs["language"], "es")
|
|
|
|
|
|
class TestTranscribeWhisper(unittest.TestCase):
|
|
def test_transcribe_success(self):
|
|
mgr = _make_manager(engine="whisper")
|
|
mgr.language = "en-us"
|
|
mgr.model = MagicMock()
|
|
mgr.model.transcribe.return_value = {"text": "hello world"}
|
|
mgr.model.device = MagicMock()
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
|
|
mock_torch = MagicMock()
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np, "torch": mock_torch}):
|
|
result = mgr._transcribe_with_whisper([b"\x00\x00" * 512])
|
|
|
|
def test_transcribe_empty(self):
|
|
mgr = _make_manager(engine="whisper")
|
|
mock_np = MagicMock()
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_whisper([])
|
|
self.assertEqual(result, "")
|
|
|
|
def test_transcribe_model_none(self):
|
|
mgr = _make_manager(engine="whisper")
|
|
mgr.model = None
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000)
|
|
mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
mock_np.float32 = "float32"
|
|
with patch.dict("sys.modules", {"numpy": mock_np, "torch": MagicMock()}):
|
|
result = mgr._transcribe_with_whisper([b"\x00\x00" * 512])
|
|
self.assertEqual(result, "")
|
|
|
|
|
|
class TestTranscribeParakeet(unittest.TestCase):
|
|
def test_transcribe_create_stream_exception(self):
|
|
mgr = _make_manager()
|
|
mgr.model = MagicMock()
|
|
mgr.model.create_stream.side_effect = RuntimeError("decode boom")
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock()
|
|
mock_np.frombuffer.return_value.astype.return_value = MagicMock()
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np}):
|
|
result = mgr._transcribe_with_parakeet([b"\x00\x00" * 512])
|
|
self.assertEqual(result, "")
|
|
mgr.model.create_stream.assert_called_once()
|
|
|
|
|
|
class TestInitVosk(unittest.TestCase):
|
|
def test_init_vosk_model_exists(self):
|
|
mgr = _make_manager(engine="vosk")
|
|
with patch("os.path.exists", return_value=True):
|
|
with patch("os.path.isdir", return_value=True):
|
|
mock_vosk = MagicMock()
|
|
mock_model = MagicMock()
|
|
mock_recognizer = MagicMock()
|
|
mock_vosk.Model.return_value = mock_model
|
|
mock_vosk.KaldiRecognizer.return_value = mock_recognizer
|
|
with patch.dict("sys.modules", {"vosk": mock_vosk}):
|
|
mgr._init_vosk()
|
|
self.assertTrue(mgr._model_initialized)
|
|
self.assertEqual(mgr.model, mock_model)
|
|
self.assertEqual(mgr.recognizer, mock_recognizer)
|
|
|
|
def test_init_vosk_no_model(self):
|
|
mgr = _make_manager(engine="vosk")
|
|
mock_vosk = MagicMock()
|
|
with patch("os.path.exists", return_value=False):
|
|
with patch.dict("sys.modules", {"vosk": mock_vosk}):
|
|
mgr._init_vosk()
|
|
# When model doesn't exist and defer_download is True, should not initialize
|
|
self.assertFalse(mgr._model_initialized)
|
|
|
|
|
|
class TestInitWhisper(unittest.TestCase):
|
|
def test_init_whisper(self):
|
|
mgr = _make_manager(engine="whisper")
|
|
mock_model = MagicMock()
|
|
mock_whisper = MagicMock()
|
|
mock_whisper.load_model.return_value = mock_model
|
|
mock_torch = MagicMock()
|
|
mock_torch.cuda.is_available.return_value = False
|
|
with patch.dict("sys.modules", {"whisper": mock_whisper, "torch": mock_torch}):
|
|
with patch("os.path.exists", return_value=True):
|
|
mgr._init_whisper()
|
|
self.assertTrue(mgr._model_initialized)
|
|
self.assertEqual(mgr.model, mock_model)
|
|
|
|
|
|
class TestInitWhispercpp(unittest.TestCase):
|
|
|
|
def test_init_whispercpp_vulkan_fallback(self):
|
|
mgr = _make_manager(engine="whisper_cpp")
|
|
with patch("os.path.exists", return_value=True):
|
|
mock_pywhispercpp = MagicMock()
|
|
call_count = [0]
|
|
|
|
def model_side_effect(*a, **kw):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
raise RuntimeError("16-bit storage not supported")
|
|
return MagicMock()
|
|
|
|
mock_pywhispercpp.Model.side_effect = model_side_effect
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": MagicMock(),
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
},
|
|
):
|
|
with patch.object(
|
|
SpeechRecognitionManager,
|
|
"_whispercpp_model_is_verified",
|
|
return_value=True,
|
|
):
|
|
try:
|
|
mgr._init_whispercpp()
|
|
except Exception:
|
|
pass
|
|
|
|
def test_cpu_fallback_filters_unsupported_whispercpp_params(self) -> None:
|
|
for override, original_threads, expected_threads in (
|
|
(None, 2, 6),
|
|
(3, 2, 3),
|
|
):
|
|
with self.subTest(override=override):
|
|
mgr = _make_manager(engine="whisper_cpp", whispercpp_n_threads=override)
|
|
model = MagicMock()
|
|
mock_pywhispercpp = MagicMock()
|
|
mock_pywhispercpp.Model.return_value = model
|
|
model_kwargs = {"n_threads": original_threads, "no_context": True}
|
|
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": MagicMock(),
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
},
|
|
):
|
|
with patch.object(
|
|
mgr,
|
|
"_get_supported_whispercpp_params",
|
|
return_value={"n_threads"},
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.multiprocessing.cpu_count",
|
|
return_value=8,
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager._show_notification"
|
|
):
|
|
with patch.dict("os.environ", {}, clear=False):
|
|
loaded_backend = mgr._handle_gpu_fallback(
|
|
RuntimeError("16-bit storage not supported"),
|
|
"/tmp/model.bin",
|
|
model_kwargs,
|
|
"cpu",
|
|
)
|
|
|
|
self.assertEqual(loaded_backend, "cpu")
|
|
self.assertIs(mgr.model, model)
|
|
self.assertEqual(mock_pywhispercpp.Model.call_count, 1)
|
|
self.assertEqual(
|
|
mock_pywhispercpp.Model.call_args.kwargs,
|
|
{"n_threads": expected_threads},
|
|
)
|
|
self.assertEqual(model_kwargs["n_threads"], original_threads)
|
|
|
|
|
|
class TestDownloadVoskModel(unittest.TestCase):
|
|
pass
|
|
|
|
|
|
class TestBufferManagement(unittest.TestCase):
|
|
def test_set_buffer_limit(self):
|
|
mgr = _make_manager()
|
|
mgr.set_buffer_limit(100)
|
|
self.assertEqual(mgr._max_buffer_size, 100)
|
|
|
|
def test_get_buffer_stats(self):
|
|
mgr = _make_manager()
|
|
stats = mgr.get_buffer_stats()
|
|
self.assertIsInstance(stats, dict)
|
|
|
|
|
|
class TestWhispercppModelKwargs(unittest.TestCase):
|
|
def test_default_model_kwargs(self) -> None:
|
|
mgr = _make_manager(engine="whisper_cpp")
|
|
kwargs = mgr._build_whispercpp_model_kwargs(n_threads=4)
|
|
self.assertTrue(kwargs.get("no_timestamps"))
|
|
self.assertTrue(kwargs.get("no_context"))
|
|
self.assertFalse(kwargs["print_progress"])
|
|
self.assertNotIn("initial_prompt", kwargs)
|
|
self.assertEqual(kwargs.get("temperature"), 0.0)
|
|
self.assertEqual(kwargs.get("n_threads"), 4)
|
|
|
|
def test_model_kwargs_all_switches_off(self):
|
|
mgr = _make_manager(
|
|
engine="whisper_cpp",
|
|
whispercpp_no_timestamps=False,
|
|
whispercpp_no_context=False,
|
|
)
|
|
kwargs = mgr._build_whispercpp_model_kwargs(n_threads=4)
|
|
self.assertNotIn("no_timestamps", kwargs)
|
|
self.assertNotIn("no_context", kwargs)
|
|
|
|
def test_model_kwargs_with_initial_prompt(self):
|
|
mgr = _make_manager(
|
|
engine="whisper_cpp",
|
|
whispercpp_initial_prompt="test prompt",
|
|
)
|
|
kwargs = mgr._build_whispercpp_model_kwargs(n_threads=4)
|
|
self.assertEqual(kwargs.get("initial_prompt"), "test prompt")
|
|
|
|
def test_model_kwargs_custom_numerics(self):
|
|
mgr = _make_manager(
|
|
engine="whisper_cpp",
|
|
whispercpp_temperature=0.7,
|
|
whispercpp_temperature_inc=0.2,
|
|
whispercpp_entropy_thold=3.0,
|
|
whispercpp_logprob_thold=-2.5,
|
|
whispercpp_no_speech_thold=0.8,
|
|
)
|
|
kwargs = mgr._build_whispercpp_model_kwargs(n_threads=8)
|
|
self.assertEqual(kwargs.get("temperature"), 0.7)
|
|
self.assertEqual(kwargs.get("temperature_inc"), 0.2)
|
|
self.assertEqual(kwargs.get("entropy_thold"), 3.0)
|
|
self.assertEqual(kwargs.get("logprob_thold"), -2.5)
|
|
self.assertEqual(kwargs.get("no_speech_thold"), 0.8)
|
|
self.assertEqual(kwargs.get("n_threads"), 8)
|
|
|
|
def test_model_kwargs_filter_unsupported_native_params(self) -> None:
|
|
mgr = _make_manager(engine="whisper_cpp")
|
|
kwargs = mgr._build_whispercpp_model_kwargs(n_threads=4)
|
|
|
|
filtered = mgr._filter_whispercpp_model_kwargs(
|
|
kwargs,
|
|
supported_params={
|
|
"n_threads",
|
|
"suppress_blank",
|
|
"no_speech_thold",
|
|
"entropy_thold",
|
|
"logprob_thold",
|
|
"temperature",
|
|
"temperature_inc",
|
|
"no_context",
|
|
"initial_prompt",
|
|
},
|
|
)
|
|
|
|
self.assertNotIn("no_timestamps", filtered)
|
|
self.assertNotIn("print_progress", filtered)
|
|
self.assertEqual(filtered["n_threads"], 4)
|
|
self.assertTrue(filtered["no_context"])
|
|
|
|
def test_reconfigure_whispercpp_params(self):
|
|
mgr = _make_manager(engine="whisper_cpp")
|
|
original_temp_inc = mgr.whispercpp_temperature_inc
|
|
with patch.object(SpeechRecognitionManager, "_init_whispercpp"):
|
|
mgr.reconfigure(
|
|
whispercpp_temperature=0.5,
|
|
whispercpp_no_timestamps=False,
|
|
whispercpp_initial_prompt="hello",
|
|
force_download=False,
|
|
)
|
|
self.assertEqual(mgr.whispercpp_temperature, 0.5)
|
|
self.assertFalse(mgr.whispercpp_no_timestamps)
|
|
self.assertEqual(mgr.whispercpp_initial_prompt, "hello")
|
|
self.assertEqual(mgr.whispercpp_temperature_inc, original_temp_inc)
|
|
|
|
|
|
class TestVoiceCommandsProperty(unittest.TestCase):
|
|
def test_voice_commands_vosk_default(self):
|
|
mgr = _make_manager(engine="vosk")
|
|
# For vosk engine, voice commands should be enabled by default
|
|
self.assertTrue(mgr._resolve_voice_commands_enabled())
|
|
|
|
def test_voice_commands_whisper_default(self):
|
|
mgr = _make_manager(engine="whisper")
|
|
self.assertFalse(mgr._resolve_voice_commands_enabled())
|
|
|
|
def test_voice_commands_explicit_on(self):
|
|
mgr = _make_manager(engine="whisper", voice_commands_enabled=True)
|
|
self.assertTrue(mgr._resolve_voice_commands_enabled())
|
|
|
|
def test_voice_commands_explicit_off(self):
|
|
mgr = _make_manager(engine="vosk", voice_commands_enabled=False)
|
|
self.assertFalse(mgr._resolve_voice_commands_enabled())
|
|
|
|
|
|
class TestPendingTranscription(unittest.TestCase):
|
|
def test_live_non_current_worker_is_pending(self) -> None:
|
|
manager = _make_manager()
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def worker() -> None:
|
|
started.set()
|
|
release.wait(timeout=2.0)
|
|
|
|
thread = threading.Thread(target=worker, daemon=True)
|
|
thread.start()
|
|
try:
|
|
self.assertTrue(started.wait(timeout=1.0))
|
|
manager.recognition_thread = thread
|
|
self.assertTrue(manager.has_pending_transcription())
|
|
finally:
|
|
release.set()
|
|
thread.join(timeout=1.0)
|
|
|
|
def test_superseded_worker_is_still_pending(self) -> None:
|
|
manager = _make_manager()
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def worker_a() -> None:
|
|
started.set()
|
|
release.wait(timeout=2.0)
|
|
|
|
def worker_b() -> None:
|
|
return
|
|
|
|
first = threading.Thread(target=worker_a, daemon=True)
|
|
manager._track_recognition_worker(first)
|
|
first.start()
|
|
self.assertTrue(started.wait(timeout=1.0))
|
|
|
|
second = threading.Thread(target=worker_b, daemon=True)
|
|
second.start()
|
|
second.join(timeout=1.0)
|
|
manager.recognition_thread = second
|
|
|
|
try:
|
|
self.assertTrue(manager.has_pending_transcription())
|
|
finally:
|
|
release.set()
|
|
first.join(timeout=1.0)
|
|
|
|
self.assertFalse(manager.has_pending_transcription())
|
|
|
|
def test_track_recognition_worker_prunes_finished_workers(self) -> None:
|
|
manager = _make_manager()
|
|
|
|
def finished_worker() -> None:
|
|
return
|
|
|
|
finished = threading.Thread(target=finished_worker, daemon=True)
|
|
finished.start()
|
|
finished.join(timeout=1.0)
|
|
manager._track_recognition_worker(finished)
|
|
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def active_worker() -> None:
|
|
started.set()
|
|
release.wait(timeout=2.0)
|
|
|
|
active = threading.Thread(target=active_worker, daemon=True)
|
|
manager._track_recognition_worker(active)
|
|
active.start()
|
|
|
|
try:
|
|
self.assertTrue(started.wait(timeout=1.0))
|
|
self.assertEqual(manager._recognition_workers, [active])
|
|
finally:
|
|
release.set()
|
|
active.join(timeout=1.0)
|
|
|
|
def test_start_recognition_tracks_worker(self) -> None:
|
|
manager = _make_manager()
|
|
manager._model_initialized = True
|
|
manager.model = MagicMock()
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def record_audio() -> None:
|
|
return
|
|
|
|
def perform_recognition() -> None:
|
|
started.set()
|
|
release.wait(timeout=2.0)
|
|
|
|
manager._record_audio = record_audio
|
|
manager._perform_recognition = perform_recognition
|
|
|
|
try:
|
|
with (
|
|
patch("vocalinux.speech_recognition.recognition_manager.play_start_sound"),
|
|
patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"),
|
|
patch.object(manager, "_arm_playback_duck"),
|
|
):
|
|
self.assertTrue(manager.start_recognition())
|
|
worker = manager.recognition_thread
|
|
self.assertIsNotNone(worker)
|
|
self.assertIn(worker, manager._recognition_workers)
|
|
self.assertTrue(started.wait(timeout=1.0))
|
|
release.set()
|
|
manager.stop_recognition()
|
|
finally:
|
|
release.set()
|
|
if manager.recognition_thread is not None:
|
|
manager.recognition_thread.join(timeout=1.0)
|
|
|
|
def test_queued_audio_without_live_worker_is_not_pending(self) -> None:
|
|
manager = _make_manager()
|
|
manager._segment_queue.put(([], "en"))
|
|
|
|
self.assertFalse(manager.has_pending_transcription())
|
|
|
|
def test_idle_manager_without_queued_audio_is_not_pending(self) -> None:
|
|
manager = _make_manager()
|
|
|
|
self.assertFalse(manager.has_pending_transcription())
|
|
|
|
def test_stop_recognition_signal_is_not_pending(self) -> None:
|
|
manager = _make_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.should_record = True
|
|
started = threading.Event()
|
|
|
|
def worker() -> None:
|
|
started.set()
|
|
manager._perform_recognition()
|
|
|
|
thread = threading.Thread(target=worker, daemon=True)
|
|
manager.recognition_thread = thread
|
|
thread.start()
|
|
self.assertTrue(started.wait(timeout=1.0))
|
|
|
|
with patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"):
|
|
manager.stop_recognition()
|
|
|
|
self.assertFalse(manager.has_pending_transcription())
|
|
|
|
def test_final_segment_is_handed_off_before_worker_exits(self) -> None:
|
|
manager = _make_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.should_record = True
|
|
manager._process_audio_buffer = MagicMock()
|
|
started = threading.Event()
|
|
audio_release = threading.Event()
|
|
|
|
def worker() -> None:
|
|
started.set()
|
|
manager._perform_recognition()
|
|
|
|
def audio_worker() -> None:
|
|
audio_release.wait(timeout=1.0)
|
|
|
|
recognition_thread = threading.Thread(target=worker, daemon=True)
|
|
manager.recognition_thread = recognition_thread
|
|
recognition_thread.start()
|
|
self.assertTrue(started.wait(timeout=1.0))
|
|
|
|
audio_thread = threading.Thread(target=audio_worker, daemon=True)
|
|
manager.audio_thread = audio_thread
|
|
audio_thread.start()
|
|
manager.audio_buffer = [b"\x00\x00"] * 8
|
|
manager._recording_segment_has_speech = True
|
|
|
|
try:
|
|
with patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"):
|
|
manager.stop_recognition()
|
|
finally:
|
|
audio_release.set()
|
|
audio_thread.join(timeout=1.0)
|
|
recognition_thread.join(timeout=1.0)
|
|
|
|
with manager._segment_queue.mutex:
|
|
self.assertFalse(any(item is not None for item in manager._segment_queue.queue))
|
|
manager._process_audio_buffer.assert_called_once()
|
|
self.assertFalse(manager.has_pending_transcription())
|
|
|
|
def test_worker_does_not_count_itself_as_pending(self) -> None:
|
|
manager = _make_manager()
|
|
manager.recognition_thread = threading.current_thread()
|
|
|
|
self.assertFalse(manager.has_pending_transcription())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|