mrleo1nidandClaude Opus 5
513d010c19
- Глобальная горячая клавиша через RegisterHotKey (по умолчанию Win+Alt+Space) - Запись с микрофона, конец фразы по паузе через Silero VAD, звуковые сигналы - Распознавание faster-whisper large-v3-turbo на GPU, CUDA-библиотеки из pip-пакетов nvidia-* - Состояния «слушаю» и «распознаю», кнопка «Говорить» в чате и пункт в трее - Голосовая команда прерывает текущий ответ и озвучку - Тесты горячей клавиши, детектора конца фразы и сеанса голосового ввода Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
128 lines
3.8 KiB
Python
128 lines
3.8 KiB
Python
import threading
|
|
import time
|
|
from collections.abc import Callable
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from PySide6.QtCore import QCoreApplication
|
|
|
|
from agr_assistent.core.voice import VoiceInput
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def qapp() -> QCoreApplication:
|
|
return QCoreApplication.instance() or QCoreApplication([])
|
|
|
|
|
|
class FakeRecorder:
|
|
def __init__(self, *, wait_for_stop: bool = False, speech: bool = True) -> None:
|
|
self.wait_for_stop = wait_for_stop
|
|
self.speech = speech
|
|
self.recording = threading.Event()
|
|
|
|
def load(self) -> None:
|
|
pass
|
|
|
|
def record(self, should_stop: Callable[[], bool]) -> np.ndarray | None:
|
|
self.recording.set()
|
|
while self.wait_for_stop and not should_stop():
|
|
time.sleep(0.005)
|
|
return np.zeros(160, dtype=np.float32) if self.speech else None
|
|
|
|
|
|
class FakeRecognizer:
|
|
def __init__(self, text: str = "привет", error: Exception | None = None) -> None:
|
|
self.text = text
|
|
self.error = error
|
|
|
|
def load(self) -> None:
|
|
pass
|
|
|
|
def transcribe(self, audio: np.ndarray) -> str:
|
|
if self.error:
|
|
raise self.error
|
|
return self.text
|
|
|
|
|
|
def _wait_until(condition: Callable[[], bool], timeout: float = 3.0) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while not condition():
|
|
assert time.monotonic() < deadline, "условие не выполнилось вовремя"
|
|
QCoreApplication.processEvents()
|
|
time.sleep(0.005)
|
|
|
|
|
|
def _voice(recorder: FakeRecorder, recognizer: FakeRecognizer) -> VoiceInput:
|
|
return VoiceInput(recorder, recognizer, sound_cues=False)
|
|
|
|
|
|
def test_recognized_text_is_emitted(qapp: QCoreApplication) -> None:
|
|
voice = _voice(FakeRecorder(), FakeRecognizer("какая погода"))
|
|
events: list[str] = []
|
|
voice.recognizing_started.connect(lambda: events.append("recognizing"))
|
|
voice.recognized.connect(lambda text: events.append(text))
|
|
|
|
voice.start()
|
|
assert voice.is_listening
|
|
|
|
_wait_until(lambda: "какая погода" in events)
|
|
assert events == ["recognizing", "какая погода"]
|
|
assert not voice.is_active
|
|
|
|
|
|
def test_stop_finishes_recording_early(qapp: QCoreApplication) -> None:
|
|
recorder = FakeRecorder(wait_for_stop=True)
|
|
voice = _voice(recorder, FakeRecognizer())
|
|
recognized: list[str] = []
|
|
voice.recognized.connect(recognized.append)
|
|
|
|
voice.start()
|
|
assert recorder.recording.wait(1)
|
|
voice.stop()
|
|
|
|
_wait_until(lambda: recognized == ["привет"])
|
|
|
|
|
|
def test_cancel_discards_result(qapp: QCoreApplication) -> None:
|
|
recorder = FakeRecorder(wait_for_stop=True)
|
|
voice = _voice(recorder, FakeRecognizer())
|
|
recognized: list[str] = []
|
|
finished: list[bool] = []
|
|
voice.recognized.connect(recognized.append)
|
|
voice.finished.connect(lambda: finished.append(True))
|
|
|
|
voice.start()
|
|
assert recorder.recording.wait(1)
|
|
voice.cancel()
|
|
|
|
assert finished == [True]
|
|
time.sleep(0.1)
|
|
QCoreApplication.processEvents()
|
|
assert recognized == []
|
|
assert finished == [True]
|
|
|
|
|
|
def test_no_speech_finishes_silently(qapp: QCoreApplication) -> None:
|
|
voice = _voice(FakeRecorder(speech=False), FakeRecognizer())
|
|
events: list[str] = []
|
|
voice.recognized.connect(lambda text: events.append("recognized"))
|
|
voice.error_occurred.connect(lambda message: events.append("error"))
|
|
voice.finished.connect(lambda: events.append("finished"))
|
|
|
|
voice.start()
|
|
|
|
_wait_until(lambda: "finished" in events)
|
|
assert events == ["finished"]
|
|
|
|
|
|
def test_recognition_error_is_reported(qapp: QCoreApplication) -> None:
|
|
voice = _voice(FakeRecorder(), FakeRecognizer(error=RuntimeError("нет CUDA")))
|
|
errors: list[str] = []
|
|
voice.error_occurred.connect(errors.append)
|
|
|
|
voice.start()
|
|
|
|
_wait_until(lambda: bool(errors))
|
|
assert "нет CUDA" in errors[0]
|
|
assert not voice.is_active
|