mrleo1nidandClaude Opus 5
20a666e0d4
- Окно настроек: модель и провайдер (со списком моделей с сервера), озвучка,
голосовой ввод, слово активации, автозапуск с Windows
- Сохранение в config.yaml через ruamel.yaml с комментариями и ссылками ${VAR},
проверка значений до записи; переключатели трея тоже сохраняются
- Настройки LLM и переключатели применяются на лету, для остального — перезапуск
- Один экземпляр приложения, лог в файл, аргументы --config и --wait-pid, русские диалоги Qt
- Сборка PyInstaller (onedir) с библиотеками CUDA, иконка приложения
- Тесты сохранения настроек и окна настроек
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
121 lines
3.4 KiB
Python
121 lines
3.4 KiB
Python
import threading
|
|
import time
|
|
from collections.abc import Callable
|
|
|
|
import numpy as np
|
|
from PySide6.QtCore import QCoreApplication
|
|
|
|
from agr_assistent.core.speech import Speaker
|
|
|
|
|
|
class FakeEngine:
|
|
sample_rate = 1000
|
|
|
|
def __init__(self, fail_on: str | None = None) -> None:
|
|
self.fail_on = fail_on
|
|
|
|
def load(self) -> None:
|
|
pass
|
|
|
|
def synthesize(self, text: str) -> np.ndarray:
|
|
if text == self.fail_on:
|
|
raise RuntimeError("boom")
|
|
return np.zeros(10, dtype=np.float32)
|
|
|
|
|
|
class FakePlayer:
|
|
def __init__(self, play_seconds: float = 0.0) -> None:
|
|
self.play_seconds = play_seconds
|
|
self.played = 0
|
|
self.finished = 0
|
|
self.aborted = 0
|
|
self.lock = threading.Lock()
|
|
|
|
def play(self, audio: np.ndarray, sample_rate: int, should_continue: Callable[[], bool]) -> None:
|
|
deadline = time.monotonic() + self.play_seconds
|
|
while time.monotonic() < deadline:
|
|
if not should_continue():
|
|
self.abort()
|
|
return
|
|
time.sleep(0.005)
|
|
with self.lock:
|
|
self.played += 1
|
|
|
|
def finish(self) -> None:
|
|
self.finished += 1
|
|
|
|
def abort(self) -> None:
|
|
self.aborted += 1
|
|
|
|
|
|
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 test_reply_is_spoken_sentence_by_sentence(qapp: QCoreApplication) -> None:
|
|
player = FakePlayer()
|
|
speaker = Speaker(FakeEngine(), player, enabled=True)
|
|
events: list[str] = []
|
|
speaker.playback_started.connect(lambda: events.append("started"))
|
|
speaker.finished.connect(lambda: events.append("finished"))
|
|
|
|
speaker.begin()
|
|
speaker.feed("Раз. Два! ")
|
|
speaker.feed("Три")
|
|
speaker.end()
|
|
|
|
_wait_until(lambda: "finished" in events)
|
|
assert events == ["started", "finished"]
|
|
assert player.played == 3
|
|
assert player.finished == 1
|
|
assert not speaker.is_active
|
|
|
|
|
|
def test_stop_interrupts_playback(qapp: QCoreApplication) -> None:
|
|
player = FakePlayer(play_seconds=0.5)
|
|
speaker = Speaker(FakeEngine(), player, enabled=True)
|
|
finished: list[bool] = []
|
|
speaker.finished.connect(lambda: finished.append(True))
|
|
|
|
speaker.begin()
|
|
speaker.feed("Первая фраза. Вторая фраза. ")
|
|
_wait_until(lambda: speaker.is_playing)
|
|
speaker.stop()
|
|
|
|
assert finished == [True]
|
|
_wait_until(lambda: player.aborted >= 1)
|
|
time.sleep(0.1)
|
|
assert player.played == 0
|
|
|
|
|
|
def test_disabled_speaker_ignores_replies(qapp: QCoreApplication) -> None:
|
|
player = FakePlayer()
|
|
speaker = Speaker(FakeEngine(), player, enabled=False)
|
|
|
|
speaker.begin()
|
|
speaker.feed("Привет. ")
|
|
speaker.end()
|
|
|
|
assert not speaker.is_active
|
|
time.sleep(0.1)
|
|
assert player.played == 0
|
|
|
|
|
|
def test_synthesis_error_stops_reply_and_reports(qapp: QCoreApplication) -> None:
|
|
speaker = Speaker(FakeEngine(fail_on="Плохо."), FakePlayer(), enabled=True)
|
|
errors: list[str] = []
|
|
speaker.error_occurred.connect(errors.append)
|
|
|
|
speaker.begin()
|
|
speaker.feed("Плохо. Хорошо. Ещё. ")
|
|
speaker.end()
|
|
|
|
_wait_until(lambda: bool(errors))
|
|
assert len(errors) == 1
|
|
assert "boom" in errors[0]
|
|
assert not speaker.is_active
|