Этап 3: голосовой ввод по горячей клавише
- Глобальная горячая клавиша через RegisterHotKey (по умолчанию Win+Alt+Space) - Запись с микрофона, конец фразы по паузе через Silero VAD, звуковые сигналы - Распознавание faster-whisper large-v3-turbo на GPU, CUDA-библиотеки из pip-пакетов nvidia-* - Состояния «слушаю» и «распознаю», кнопка «Говорить» в чате и пункт в трее - Голосовая команда прерывает текущий ответ и озвучку - Тесты горячей клавиши, детектора конца фразы и сеанса голосового ввода Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
fec52dd1c2
commit
513d010c19
@@ -0,0 +1,127 @@
|
||||
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
|
||||
Reference in New Issue
Block a user