first commit

This commit is contained in:
workD12
2026-09-11 19:47:15 +03:00
commit 32754a1fe2
56 changed files with 3836 additions and 0 deletions
+1
View File
@@ -0,0 +1 @@
"""Модули проекта Blind — каждый модуль законченная проверяемая единица."""
+76
View File
@@ -0,0 +1,76 @@
# Модуль 5: горячие клавиши + машина состояний + память
Сердце ассистента: сшивает STT (М1), brain (М4), TTS (М3), аудио (М2)
в цельную программу с тремя способами управления.
## Управление (три режима)
| Режим | Как управлять | Где использовать |
|---|---|---|
| **Свободный** (`test_free.py`) | просто говори; пауза 2 с = фраза окончена (Silero-VAD) | целевой для отца |
| **Клавиши-консоль** (`test_assistant.py`) | Enter/s/r/q (имитация) | разработка в Linux |
| **Глобальные клавиши** (`--backend keys`) | space/esc/enter из любого окна | Windows-сборка |
## Машина состояний
```
idle ─▶ listening ─▶ thinking ─▶ speaking ─▶ idle
▲ │ │ │
└─────────┴────────────┴───────────┘ («Замолчи» из любого)
```
- Говорит, а отец начал говорить → речь мгновенно стихает, новая фраза обрабатывается.
- «Замолчи» гасит речь/отменяет запись/отбрасывает «думание» (счётчик поколений:
устаревшие ответы никогда не прозвучат).
- «Повтори» — последняя фраза (в idle).
## Долговременная память (data/memory.md)
- Слово прощания («до свидания», «прощай», «пока»…) → мгновенный прощальный ответ +
фоновым запросом модель формулирует конспект разговора → дописывается в файл.
- При старте конспект вшивается в системный промпт — ассистент «помнит» прошлые разговоры.
- Файл — обычный текст с датами, можно править руками; при переполнении (>8000 симв.)
старые записи сжимаются автоматически.
- Память не критична к сбоям: ошибка сети при конспекте просто логируется.
## Файлы
| Файл | Назначение |
|---|---|
| `modules/hotkeys/` | бэкенды клавиш: `console.py` (Linux), `keyboard_hook.py` (Windows), общий контракт `KeyEvents` |
| `modules/assistant/assistant.py` | машина состояний, конвейер STT→LLM→TTS, перебивание |
| `modules/assistant/memory.py` | файл-конспект между сессиями |
| `modules/assistant/farewell.py` | детект прощания + саммари диалога |
| `modules/audio_io/handsfree.py` | свободный режим: Silero-VAD в потоке, конец фразы по паузе |
| `test_assistant.py` | клавиши-консоль / глобальные клавиши |
| `test_free.py` | свободный режим (говори без кнопок) |
## Запуск
```bash
# свободный режим (целевой):
.venv/bin/python modules/assistant/test_free.py
.venv/bin/python modules/assistant/test_free.py --silence 2.5 # подольше пауза
# клавишный режим (Linux-разработка):
.venv/bin/python modules/assistant/test_assistant.py
```
Клавиши Windows-сборки настраиваются в .env: `HOTKEY_TALK/HOTKEY_STOP/HOTKEY_REPEAT`.
## Критерии приёмки — статус
1. ✅ Обычный цикл (живой тест юзером, 2026-09-11).
2. ✅ Контекст диалога внутри сессии.
3. ✅ «Замолчи» мгновенно; перебивание речи новым вопросом.
4. ✅ «Повтори».
5. ✅ Память между сессиями (мок-тест); живой тест двух запусков — на юзере.
6. 🟡 Свободный режим VAD — код готов, живой тест после установки silero-vad (CPU-torch).
## Известные детали
- Silero-VAD требует torch; для разработки ставим CPU-вариант (~200 МБ):
`pip install --index-url https://download.pytorch.org/whl/cpu torch silero-vad`.
В Windows-сборке заменим на ONNX-детектор (2 МБ, без torch).
- suppress-механизм: на время ответа микрофон глушится, чтобы не слышать саму себя;
при перебивании отцом suppress снимается мгновенно.
+6
View File
@@ -0,0 +1,6 @@
"""Модуль 5: assistant — машина состояний, сшивающая все модули в ассистента."""
from __future__ import annotations
from .assistant import Assistant
__all__ = ["Assistant"]
+422
View File
@@ -0,0 +1,422 @@
"""Машина состояний голосового ассистента (сердце Модуля 5).
Состояния и переходы:
idle ──talk_down──▶ listening ──talk_up──▶ thinking ──готово──▶ speaking ──конец──▶ idle
▲ │ │ │
└────── stop ────────┴───────────────────────┴────────────────────┘
Правила по ТЗ:
- если ассистент ГОВОРИТ, а отец нажал «Слушай» — речь мгновенно стихает
и начинается запись (одним нажатием);
- «Замолчи» гасит речь / отменяет запись / отбрасывает «думание»;
- «Повтори» — последняя фраза ассистента (в состоянии idle);
- удержание клавиши «Слушай» короче min_seconds отбрасывается.
Тяжёлая работа (STT→LLM→TTS) идёт в фоновом потоке; устаревшие результаты
отбрасываются по счётчику поколений _generation (если отец прервал и спросил
заново — старый ответ не прозвучит).
"""
from __future__ import annotations
import threading
import time
from pathlib import Path
from typing import Callable, Optional, Tuple
import numpy as np
from .errors import GENERAL_CUE, UNCLEAR_CUE, UNCLEAR_LIMIT, classify_error, cue_path
from .farewell import FAREWELL_REPLY, is_farewell, summarize_and_save
from .memory import MemoryStore
from .wake_word import WakeWordWatcher, matches_wake_word
def resample_for_player(audio: np.ndarray, source_rate: int, target_rate: int) -> np.ndarray:
"""Чанк в другой частоте (для склейки); edge всегда 24k — на всякий случай."""
from ..audio_io.resample import resample_to_16k
return resample_to_16k(audio, source_rate, target_rate)
class Assistant:
IDLE = "idle" # ждёт нажатия
LISTENING = "listening" # запись голоса
THINKING = "thinking" # STT → LLM → TTS
SPEAKING = "speaking" # воспроизведение ответа
def __init__(
self,
stt, brain, tts, # провайдеры модулей 1/4/3
recorder, player, # модуль 2
memory: Optional[MemoryStore] = None, # долговременная память (М5.1)
history_limit: int = 10, # реплик диалога в контексте (2×N сообщений)
on_state_change: Optional[Callable[[str, str], None]] = None, # для earcons М7
on_log: Optional[Callable[[str], None]] = None,
) -> None:
self.stt = stt
self.brain = brain
self.tts = tts
self.recorder = recorder
self.player = player
self.memory = memory or MemoryStore()
self.history_limit = history_limit
self.session_dialog: list[dict] = [] # реплики текущего разговора (для конспекта)
self.on_state_change = on_state_change or (lambda state, note: None)
self.on_log = on_log or (lambda msg: print(f" {msg}"))
# Долговременная память → в системный промпт LLM при каждом вопросе
memory_block = self.memory.as_prompt_block()
if memory_block:
self.brain.context_block = memory_block
self._log(f"Память загружена ({len(self.memory.load())} симв.)")
self.history: list[dict] = []
self._state = self.IDLE
self._lock = threading.Lock()
self._generation = 0 # инкремент при прерываниях; устаревшие воркеры молчат
self._last_speech: Optional[Tuple[np.ndarray, int]] = None
# Свободный режим (HandsFreeRecorder): событие «фраза закончена» от VAD
self._vad_mode = hasattr(recorder, "suppress") # duck-typing: HandsFreeRecorder
if self._vad_mode:
recorder.on_phrase = self._on_vad_phrase # публичный атрибут — переписываем
# СЕАНС ДИАЛОГА (пробел-тумблер): вне сеанса микрофон глушится
self._in_dialog = False
self._cues_dir = Path(__file__).resolve().parents[2] / "assets" / "earcons"
self._unclear_streak = 0 # неразборчивых фраз подряд (после UNCLEAR_LIMIT — куи)
# Wake-word: вне диалога слушаем тихо и ждём «Злата» (или своё из .env)
self.wake_word = WakeWordWatcher(on_detected=self._open_dialog)
if self._vad_mode:
recorder.suppress(False) # слушаем сразу — ждём wake-word
# Прогрев TTS-соединения и аудио-выхода в фоне (первый ответ звучит быстрее)
threading.Thread(target=self._warmup, daemon=True).start()
def _warmup(self) -> None:
"""Разогреть TTS (DNS/TLS до сервиса синтеза) и аудио-выход — беззвучно."""
try:
warm = getattr(self.tts, "warmup", None)
if warm:
warm()
except Exception:
pass
try:
import numpy as _np
self.player.play(_np.zeros(1600, dtype=_np.float32), 24000) # 67 мс тишины
except Exception:
pass
# ------------------------------------------------------------------ state
@property
def state(self) -> str:
return self._state
def _set_state(self, new_state: str, note: str = "") -> None:
with self._lock:
self._state = new_state
self.on_state_change(new_state, note)
def _log(self, msg: str) -> None:
self.on_log(msg)
# ------------------------------------------------- сеанс диалога (пробел)
def on_dialog_toggle(self) -> None:
"""ПРОБЕЛ: начать диалог или завершить его (куи-файлы — мгновенно)."""
if self._in_dialog:
self._close_dialog(play_end_cue=True)
else:
self._open_dialog()
def _open_dialog(self) -> None:
if self._state == self.SPEAKING:
self.player.stop()
if self._state == self.THINKING:
self._generation += 1 # старое «думание» отменяется
self._in_dialog = True
self._set_state(self.IDLE, f"Диалог начат (wake-word «{self.wake_word.wake_word}»)")
if self._vad_mode:
# микрофон слушает, но на время куи глушим (чтобы не услышать саму себя)
self.recorder.suppress(True)
self._log("Диалог начат (wake-word)")
self._play_cue("dialog_start.wav", unpause_after=True)
def _close_dialog(self, play_end_cue: bool) -> None:
if not self._in_dialog:
return
self._generation += 1 # всё текущее (запись/думание/речь) отменяется
self.player.stop()
if self._vad_mode:
self.recorder.suppress(False) # сбрасываем недозапись и слушаем снова
self._in_dialog = False
self._set_state(self.IDLE, f"Диалог завершён — жду «{self.wake_word.wake_word}»")
self._log("Диалог завершён — в режиме ожидания wake-word")
if play_end_cue:
self._play_cue("dialog_end.wav")
def _play_cue(self, filename: str, unpause_after: bool = False) -> None:
"""Мгновенно проиграть готовый wav-куи (без синтеза)."""
path = self._cues_dir / filename
if not path.is_file():
self._log(f"(куи-файл не найден: {filename})")
if unpause_after and self._vad_mode:
self.recorder.suppress(False)
return
def _watch() -> None:
while self.player.is_playing:
time.sleep(0.05)
if unpause_after and self._in_dialog and self._vad_mode:
self.recorder.suppress(False) # куи прозвучал — слушаем отца
self.player.play_file(str(path))
threading.Thread(target=_watch, daemon=True).start()
# ----------------------------------------------------- свободный режим
def _on_vad_phrase(self, audio: np.ndarray, samplerate: int) -> None:
"""VAD прислал законченную фразу (пауза тишины после речи).
Вне диалога: STT → если фраза начинается с wake-word («Злата») —
открыть диалог. Иначе молча игнорируем (фоновый разговор не будит).
В диалоге — обрабатываем как вопрос (перебивание работает).
"""
if not self._in_dialog:
self._generation += 1
generation = self._generation
self._set_state(self.LISTENING, "жду wake-word")
threading.Thread(
target=self._wake_check, args=(audio, generation), daemon=True
).start()
return
state = self._state
if state == self.SPEAKING:
self.player.stop()
if state == self.THINKING:
self._generation += 1
self._log("Перебиваю размышление новой фразой")
self._generation += 1
generation = self._generation
self._set_state(self.THINKING)
threading.Thread(target=self._pipeline, args=(audio, generation), daemon=True).start()
def _wake_check(self, audio: np.ndarray, generation: int) -> None:
"""Проверка фразы на wake-word (вне диалога). Дешёво: только STT."""
try:
t0 = time.perf_counter()
res = self.stt.transcribe(audio)
t_stt = time.perf_counter() - t0
text = res.text.strip()
if generation != self._generation:
return
if matches_wake_word(text, self.wake_word.wake_word):
self._log(f"Wake-word услышан ({t_stt:.1f} c): {text!r}")
self._open_dialog()
else:
# чужая речь/телевизор — молча, без реакций
self._log(f"(вне диалога фраза мимо wake-word: {text[:40]!r})")
self._set_state(self.IDLE, "жду wake-word")
except Exception as exc:
self._log(f"wake-check: {exc}")
def _speak(self, audio: np.ndarray, samplerate: int) -> None:
"""Озвучить и вернуть состояние в idle по окончании (или после прерывания)."""
generation = self._generation
self._set_state(self.SPEAKING)
# пока говорим — микрофон не слушает (не слышим сами себя из динамиков)
if self._vad_mode:
self.recorder.suppress(True)
self.player.play(audio, samplerate)
def _watch() -> None:
while self.player.is_playing:
time.sleep(0.05)
# сюда попадаем и при player.stop() — тогда state уже IDLE
if generation == self._generation and self._state == self.SPEAKING:
self._set_state(self.IDLE)
if self._vad_mode and generation == self._generation:
self.recorder.suppress(False) # снова слушаем
threading.Thread(target=_watch, daemon=True).start()
# ------------------------------------------------------------- события
def on_talk_down(self) -> None:
"""Клавиша «Слушай» нажата."""
state = self._state
if state == self.SPEAKING:
self.player.stop() # мгновенно стихаем (по ТЗ)
if state == self.THINKING:
self._generation += 1 # текущий ответ станет устаревшим
self._log("Прервал размышление — начинаю запись")
if state in (self.IDLE, self.SPEAKING, self.THINKING):
try:
self.recorder.start()
except RuntimeError as exc:
self._set_state(self.IDLE, f"микрофон недоступен: {exc}")
return
self._set_state(self.LISTENING)
self._log("Слушаю…")
# в LISTENING повторный down игнорируем (автодубли клавиши)
def on_talk_up(self) -> None:
"""Клавиша «Слушай» отпущена — запись закончена, обрабатываем."""
if self._state != self.LISTENING:
return
audio = self.recorder.stop()
if len(audio) == 0:
self._set_state(self.IDLE, "Слишком коротко — ничего не записал")
return
self._generation += 1
generation = self._generation
self._set_state(self.THINKING)
threading.Thread(target=self._pipeline, args=(audio, generation), daemon=True).start()
def on_stop(self) -> None:
"""Клавиша «Замолчи» — мгновенно тишина."""
state = self._state
self._generation += 1 # отбрасываем все текущие работы
if state == self.SPEAKING:
self.player.stop()
self._log("Прервал речь")
elif state == self.LISTENING:
self.recorder.cancel()
self._log("Запись отменена")
elif state == self.THINKING:
self._log("Размышление отменено")
self._set_state(self.IDLE)
def on_repeat(self) -> None:
"""Клавиша «Повтори» — снова озвучить последний ответ."""
if self._state != self.IDLE:
self._log("«Повтори» работает только в режиме ожидания")
return
if not self._last_speech:
self._log("Пока нечего повторять")
return
audio, samplerate = self._last_speech
self._log("Повторяю последний ответ")
self._speak(audio, samplerate)
# ------------------------------------------------------------ конвейер
def _pipeline(self, audio: np.ndarray, generation: int) -> None:
"""STT → LLM → TTS → озвучка (фоновый поток). Тайминги этапов — в лог."""
try:
t0 = time.perf_counter()
stt_res = self.stt.transcribe(audio)
t_stt = time.perf_counter() - t0
question = stt_res.text.strip()
if generation != self._generation:
return # устарело (прервали)
if not question or len(question) < 2:
self._handle_unclear("Речь не распознана")
return
self._unclear_streak = 0 # фраза распознана — счётчик в ноль
self._log(f"Вы: {question}")
# Прощание: отвечаем сразу, пишем конспект и ЗАКРЫВАЕМ сеанс диалога
if is_farewell(question) and self.session_dialog:
self._log("Прощание — записываю разговор в память")
threading.Thread(
target=summarize_and_save,
args=(self.brain, self.session_dialog + [{"role": "user", "text": question}], self.memory),
kwargs={"on_error": lambda msg: self._log(msg)},
daemon=True,
).start()
answer = FAREWELL_REPLY
answer_for_history = answer
llm = None
t_llm = 0.0
else:
t0 = time.perf_counter()
llm = self.brain.ask(question, history=self.history)
t_llm = time.perf_counter() - t0
if generation != self._generation:
return
answer = llm.cleaned_text # для ушей (без служебных маркеров)
answer_for_history = llm.raw_text # маркер [ЧАСТЬ i ИЗ n] остаётся — модель помнит часть
if not answer:
answer = "Извините, я задумалась и забыла, что хотела сказать. Спросите ещё раз."
answer_for_history = answer
self._log(f"Ассистент ({t_llm:.1f} с): {answer}")
# История и конспект-диалог пополняются одинаково для обоих путей озвучки
turn = [
{"role": "user", "text": question},
{"role": "assistant", "text": answer_for_history if not is_farewell(question) else answer},
]
self.history += turn
self.history = self.history[-self.history_limit:]
self.session_dialog += turn
# «Отвечаю» — сразу (синтез идёт далее в фоне, звук стартует по готовности)
self._log("Отвечаю...")
# Потоковая озвучка: чанки-предложения по мере готовности.
# Если синтезатор не умеет stream() — обычный путь (целиком).
streamer = getattr(self.tts, "stream", None)
if streamer is not None and generation == self._generation:
self._set_state(self.SPEAKING)
if self._vad_mode:
self.recorder.suppress(True)
self._last_speech = None # заполним по чанкам
pause = getattr(self.tts, "_pause", 0.0)
sr_prev = None
for audio, sr in streamer(answer):
if generation != self._generation: # перебили — гасим поток
self.player.stop()
break
piece = audio
if sr_prev is not None and sr != sr_prev:
piece = resample_for_player(audio, sr, sr_prev) # Rare: edge всегда 24k
sr_prev = sr
if self._last_speech is None:
self._last_speech = (piece, sr)
else:
prev, prev_sr = self._last_speech
gap = np.zeros(int(pause * prev_sr), dtype=np.float32)
self._last_speech = (np.concatenate([prev, gap, piece]), sr)
self.player.play(piece, sr)
# ждём конца чанка (если перебили — generation сместится и выйдем)
while self.player.is_playing and generation == self._generation:
time.sleep(0.05)
if generation != self._generation:
break
if generation == self._generation:
self._set_state(self.IDLE)
# Прощание: после ответа закрываем сеанс (микрофон глушится)
if is_farewell(question) and self._in_dialog:
self._close_dialog(play_end_cue=False)
elif self._vad_mode:
self.recorder.suppress(False)
return
t0 = time.perf_counter()
speech = self.tts.synthesize(answer)
t_tts = time.perf_counter() - t0
if generation != self._generation:
return
self._last_speech = (speech.audio, speech.samplerate)
self._speak(speech.audio, speech.samplerate)
t_play = getattr(self.player, "last_start_latency", 0.0)
self._log(f"тайминг: распознавание {t_stt:.1f} | модель {t_llm:.1f} | "
f"синтез {t_tts:.1f} | старт звука {t_play:.1f}")
except Exception as exc: # сеть, API, TTS — что угодно
if generation == self._generation:
self._log(f"Ошибка: {exc}")
cue = classify_error(exc)
self._set_state(self.IDLE, f"Ошибка: {cue}")
self._play_cue(cue)
# --------------------------------------------------- неразборчивая речь
def _handle_unclear(self, note: str) -> None:
"""Фраза не распознана: счётчик подряд; после UNCLEAR_LIMIT — куи и сброс."""
self._unclear_streak += 1
self._log(f"{note} (неразборчивых подряд: {self._unclear_streak})")
self._set_state(self.IDLE, note)
if self._unclear_streak >= UNCLEAR_LIMIT:
self._log("Много неразборчивых подряд — напоминаю про микрофон")
self._unclear_streak = 0
self._play_cue(UNCLEAR_CUE)
+35
View File
@@ -0,0 +1,35 @@
"""Классификация ошибок пайплайна → голосовые куи (текст + wav-файл).
Отец должен СЛЫШАТЬ, что случилось: нет сети / лимит / что-то ещё.
Куи — готовые wav из assets/earcons (мгновенно, без синтеза в момент ошибки).
"""
from __future__ import annotations
import re
from pathlib import Path
CUES_DIR = Path(__file__).resolve().parents[2] / "assets" / "earcons"
# Паттерны текста ошибки → куи
_PATTERNS: list[tuple[str, str]] = [
(r"Connection|ConnectError|Network|Timeout|timed out|getaddrinfo|SSLError|RemoteProtocol", "error_network.wav"),
(r"429|Rate limit|rate_limit|quota|Quota|billing|insufficient", "error_limit.wav"),
(r"401|403|Unauthorized|PermissionDenied|forbidden", "error_limit.wav"),
]
UNCLEAR_CUE = "unclear_question.wav"
GENERAL_CUE = "error_general.wav"
UNCLEAR_LIMIT = 5 # после N неразборчивых подряд — голосовое сообщение и сброс
def classify_error(exc: BaseException | str) -> str:
"""Текст ошибки → имя wav-куи в assets/earcons."""
text = str(exc)
for pattern, cue in _PATTERNS:
if re.search(pattern, text):
return cue
return GENERAL_CUE
def cue_path(cue: str) -> Path:
return CUES_DIR / cue
+57
View File
@@ -0,0 +1,57 @@
"""Конец разговора: прощание + запись конспекта в долговременную память.
Триггер — слово прощания во фразе отца («до свидания», «прощай», ...).
Ассистент отвечает прощальной фразой, затем ОТДЕЛЬНЫМ запросом просит модель
сформулировать конспект разговора и дописывает его в data/memory.md.
Конспект самоформулируется моделью (как предложил юзер), мы только сохраняем.
"""
from __future__ import annotations
import re
import threading
from typing import List, Optional
from .memory import MemoryStore
# Фразы-триггеры прощания (регистронезависимо, как часть слова «пока» тоже сработает)
_FAREWELL = re.compile(r"\b(до свидания|прощай|всего доброго|бай|пока)\b", re.I)
FAREWELL_REPLY = "До свидания! Я всё запомнила и буду ждать нового разговора. Всего вам доброго!"
SUMMARY_PROMPT = (
"Кратко сформулируй (3–5 простых предложений) что важно запомнить из этого "
"разговора на будущее: о чём человек спрашивал, что для него важно, "
"какие были договорённости или просьбы. Только сам конспект, без вступлений."
)
def is_farewell(text: str) -> bool:
return bool(_FAREWELL.search(text or ""))
def make_summary_prompt(dialog: List[dict]) -> str:
"""Собрать текст диалога в промпт для конспекта."""
lines = [f"{'Человек' if m['role'] == 'user' else 'Ассистент'}: {m['text']}"
for m in dialog]
return "\n".join([SUMMARY_PROMPT, ""] + lines)
def summarize_and_save(brain, dialog: List[dict], memory: MemoryStore,
on_error: Optional[callable] = None,
done: Optional[callable] = None) -> None:
"""Сформулировать конспект диалога и дописать в память (фоновая работа)."""
def _work() -> None:
try:
result = brain.ask(make_summary_prompt(dialog), history=[]) # без системного контекста ассистента
if result.cleaned_text:
memory.append(result.cleaned_text)
except Exception as exc:
# Память не критична, но тихо терять ошибку плохо при отладке
if on_error:
on_error(f"Не смог записать память: {exc}")
if done is None:
threading.Thread(target=_work, daemon=True).start()
else:
_work()
done()
+92
View File
@@ -0,0 +1,92 @@
"""Долговременная память ассистента — файл-конспект между сессиями.
Принцип (идея юзера 2026-09-11):
- в конце разговора («до свидания») модель просится сформулировать конспект,
он ДОПОЛНЯЕТСЯ в текстовый файл data/memory.md;
- при старте конспект кладётся в системный промпт — ассистент «помнит» прошлые
разговоры (кто отец, что обсуждали, договорённости).
Файл — простой читаемый текст: можно посмотреть и поправить руками.
Хранить сам диалог не нужно — конспект компактнее и токены дешевле.
"""
from __future__ import annotations
import re
import time
from pathlib import Path
from typing import List, Optional
class MemoryStore:
def __init__(self, path: str | Path = "data/memory.md", max_chars: int = 8000) -> None:
self._path = Path(path)
self._max_chars = max_chars # ограничение конспекта (защита от бесконечного роста)
self._path.parent.mkdir(parents=True, exist_ok=True)
# ------------------------------------------------------------------ чтение
@property
def path(self) -> Path:
return self._path
def load(self) -> str:
"""Конспект целиком (пустая строка, если памяти ещё нет)."""
if not self._path.is_file():
return ""
return self._path.read_text(encoding="utf-8").strip()
def as_prompt_block(self) -> str:
"""Блок для системного промпта (или пустая строка)."""
memory = self.load()
if not memory:
return ""
return (
"Вот что ты помнишь из прошлых разговоров с этим человеком "
"(его долговременная память):\n"
f"{memory}\n"
"Используй это естественно, без упоминания «записей» и «памяти»."
)
# ------------------------------------------------------------------ запись
def append(self, summary: str) -> None:
"""Дописать новый конспект с датой, не превышая лимит объёма."""
summary = summary.strip()
if not summary:
return
header = f"\n\n## {time.strftime('%d.%m.%Y')}\n"
current = self.load()
new_total = len(current) + len(header) + len(summary)
if new_total > self._max_chars:
# Файл распух: сжимаем старые записи в одну итоговую строку
current = self._condense(current)
with self._path.open("a", encoding="utf-8") as f:
if current and not current.endswith("\n"):
f.write("\n")
f.write(header + summary)
def clear(self) -> None:
"""Стереть память (для отладки)."""
self._path.write_text("", encoding="utf-8")
# ------------------------------------------------------------------ внутреннее
def _condense(self, text: str) -> str:
"""Сжать конспект: оставить последние записи целиком, старые — краткой выжимкой.
Сжатие делаем механически (первые предложения старых записей); качество
semantique-сжатия критично только для очень длинной памяти, а это редкость.
"""
sections: List[str] = [s.strip() for s in re.split(r"\n## ", "\n" + text) if s.strip()]
if len(sections) <= 1:
return text[: self._max_chars // 2]
keep_recent = sections[-4:] # последние 4 записи — целиком
old = "\n".join(sections[:-4])
condensed = "\n".join(
"• " + " ".join(re.split(r"(?<=[.!?]) ", chunk)[:2])
for chunk in old.split("•") if chunk.strip()
)
merged = ("— Итоги прошлых записей: " + condensed + "\n\n"
+ "\n## ".join(keep_recent))
return merged[: self._max_chars]
+108
View File
@@ -0,0 +1,108 @@
"""Тест Модуля 5: горячие клавиши + машина состояний = ЖИВОЙ ассистент.
Запуск:
python modules/assistant/test_assistant.py # console-бэкенд (Linux, без root)
python modules/assistant/test_assistant.py --backend keys # глобальные клавиши (Windows; Linux — root)
Консольный режим (имитация трёх клавиш):
Enter — нажать «Слушай» (первый раз), отпустить (второй раз) — аналог удержания
s + Enter — «Замолчи»
r + Enter — «Повтори»
q + Enter — выход
Критерии приёмки Модуля 5:
1. Пустой вопрос: Enter → молчание → Enter → «Слишком коротко».
2. Обычный цикл: Enter → вопрос → Enter → ответ голосом (диалог помнит контекст).
3. «Замолчи» во время речи — звук стихает мгновенно (цель < 100 мс).
4. «Повтори» — последняя фраза звучит снова.
5. Нажать «Слушай» ВО ВРЕМЯ речи — речь стихла, запись началась.
"""
from __future__ import annotations
import argparse
import sys
import threading
from pathlib import Path
try:
from modules.assistant import Assistant
from modules.audio_io import Player, Recorder
from modules.brain import get_provider as get_brain
from modules.hotkeys import KeyEvents, get_backend
from modules.stt import get_provider as get_stt
from modules.tts import make_tts
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from modules.assistant import Assistant
from modules.audio_io import Player, Recorder
from modules.brain import get_provider as get_brain
from modules.hotkeys import KeyEvents, get_backend
from modules.stt import get_provider as get_stt
from modules.tts import make_tts
STATES_RU = {
"idle": "⏸ ожидание",
"listening": "🎙 слушаю",
"thinking": "🧠 думаю…",
"speaking": "🔊 говорю",
}
def main() -> int:
parser = argparse.ArgumentParser(description="Живой ассистент (Модуль 5)")
parser.add_argument("--backend", default="console", choices=["console", "keys"])
parser.add_argument("--stt-model", default="small")
args = parser.parse_args()
print("Загружаю модули…")
stt = get_stt("faster-whisper", model_size=args.stt_model)
brain = get_brain("openrouter") # модель из .env
tts = make_tts() # провайдер/скорость из .env (TTS_PROVIDER, TTS_RATE)
recorder = Recorder()
player = Player()
def on_state(state: str, note: str) -> None:
line = STATES_RU.get(state, state)
print(f"\n[{line}]" + (f" {note}" if note else ""))
assistant = Assistant(stt, brain, tts, recorder, player, on_state_change=on_state)
quit_event = threading.Event()
events = KeyEvents(
on_talk_down=assistant.on_talk_down,
on_talk_up=assistant.on_talk_up,
on_stop=assistant.on_stop,
on_repeat=assistant.on_repeat,
on_quit=lambda: quit_event.set(),
)
if args.backend == "keys":
print("\nГлобальные клавиши: «Слушай» = space (удерживать), "
"«Замолчи» = esc, «Повтори» = enter. Ctrl+C — выход.")
print("(На Linux этот режим требует root: sudo -E .venv/bin/python …)")
else:
print("\nКонсольные клавиши (имитация удержания):")
print(" Enter — нажать «Слушай», ещё Enter — отпустить (начать обработку)")
print(" s + Enter — «Замолчи» | r + Enter — «Повтори» | q + Enter — выход")
backend = get_backend(args.backend, events)
backend.start()
try:
while not quit_event.wait(timeout=0.2):
pass
except KeyboardInterrupt:
print("\n(Ctrl+C)")
finally:
backend.stop()
try:
player.stop() # тишина при выходе
except Exception:
pass
print("Выход.")
return 0
if __name__ == "__main__":
sys.exit(main())
+128
View File
@@ -0,0 +1,128 @@
"""Тест М5.3: свободный режим с ПРОБЕЛОМ-тумблером диалога.
Запуск:
python modules/assistant/test_free.py
python modules/assistant/test_free.py --silence 1.5 # короче пауза конца фразы
Схема (целевая для отца):
ПРОБЕЛ (в консоли: d + Enter) — НАЧАТЬ диалог: куи «Ассистент готов к диалогу»
→ просто говори, пауза 2 c = фраза закончена → ответ.
ПРОБЕЛ в диалоге — ЗАВЕРШИТЬ: куи «Спасибо. До новых встреч».
Или скажи «до свидания» — то же самое + запомнит разговор.
Вне диалога микрофон заглушен. Выход: q + Enter.
"""
from __future__ import annotations
import argparse
import sys
import threading
from pathlib import Path
try:
from modules.assistant import Assistant
from modules.audio_io import HandsFreeRecorder, Player
from modules.brain import get_provider as get_brain
from modules.hotkeys import KeyEvents, get_backend
from modules.stt import get_provider as get_stt
from modules.tts import make_tts
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from modules.assistant import Assistant
from modules.audio_io import HandsFreeRecorder, Player
from modules.brain import get_provider as get_brain
from modules.hotkeys import KeyEvents, get_backend
from modules.stt import get_provider as get_stt
from modules.tts import make_tts
VAD_STATE_RU = {
"listening": "👂 слушаю тебя",
"speech_start": "🎙 начало фразы",
"speech_end": "✅ фраза собрана, обрабатываю",
"too_short": "⚠ слишком коротко — пропускаю",
"paused": "⏸ не слушаю (сама говорю)",
}
ASSISTANT_STATE_RU = {
"idle": "⏸ ожидание",
"listening": "🎙 запись",
"thinking": "🧠 думаю…",
"speaking": "🔊 говорю",
}
def main() -> int:
p = argparse.ArgumentParser(description="Свободный режим ассистента (VAD)")
p.add_argument("--silence", type=float, default=2.0,
help="пауза тишины, считающаяся концом фразы (сек)")
p.add_argument("--threshold", type=float, default=0.5,
help="порог VAD: больше — строже, меньше — чувствительнее")
p.add_argument("--min-speech", type=float, default=0.3,
help="минимальная доля голоса во фразе, сек")
p.add_argument("--stt-model", default="small")
args = p.parse_args()
print("Загружаю модули…")
stt = get_stt("faster-whisper", model_size=args.stt_model)
brain = get_brain("openrouter")
tts = make_tts() # провайдер/скорость из .env (TTS_PROVIDER, TTS_RATE)
player = Player()
quit_event = threading.Event()
def on_vad_state(state: str) -> None:
print(f"\n [микрофон] {VAD_STATE_RU.get(state, state)}")
def on_assistant_state(state: str, note: str) -> None:
print(f"\n[{ASSISTANT_STATE_RU.get(state, state)}]" + (f" {note}" if note else ""))
def on_log(msg: str) -> None:
print(f" {msg}")
recorder = HandsFreeRecorder(
silence_sec=args.silence,
speech_threshold=args.threshold,
min_speech_sec=args.min_speech,
on_state=on_vad_state,
on_error=lambda m: print(f" [ОШИБКА] {m}"),
)
assistant = Assistant(stt, brain, tts, recorder, player,
on_state_change=on_assistant_state, on_log=on_log)
print(f"\nУПРАВЛЕНИЕ (голосом, без клавиатуры):")
print(f" «{assistant.wake_word.wake_word.capitalize()}» — начать диалог (куи «готова к диалогу»)")
print(" «До свидания» — завершить диалог (куи «до новых встреч»)")
print(" Вне диалога ассистент слушает тихо и отвечает только на wake-word.")
print(" Консоль: s — «Замолчи», r — «Повтори», ПРОБЕЛ — тоже тумблер диалога, q — выход.\n")
quit_event = threading.Event()
events = KeyEvents(
on_talk_down=lambda: None,
on_talk_up=lambda: None,
on_stop=assistant.on_stop,
on_repeat=assistant.on_repeat,
on_quit=lambda: quit_event.set(),
on_dialog_toggle=assistant.on_dialog_toggle,
)
backend = get_backend("console", events)
backend.start()
recorder.start() # микрофон открыт, но заглушен до начала диалога
try:
while not quit_event.wait(timeout=0.2):
pass
except KeyboardInterrupt:
print("\n(Ctrl+C)")
finally:
backend.stop()
recorder.stop()
try:
player.stop()
except Exception:
pass
print("Выход.")
return 0
if __name__ == "__main__":
sys.exit(main())
+82
View File
@@ -0,0 +1,82 @@
"""Wake-word — голосовое «будильное слово» вместо клавиши (М5.4).
Отец говорит wake-word (например «Злата») → куи «Ассистент готов к диалогу»
→ свободный диалог → «до свидания» → снова ждёт слова.
Выбор слова (требования юзера): уникальное, редкое в бытовой речи,
однозначно распознаётся STT, простое для пожилого человека. По умолчанию
«Злата» (whisper пишет его без вариантов); настраивается в .env WAKE_WORD.
Защита от ложных срабатываний (слово из чужого разговора/ТВ):
- матч только по НАЧАЛУ фразы (первое слово), не по вхождению внутрь;
- нормализация: без пунктуации, нижний регистр, ё→е;
- совпадение по префиксу не менее min_prefix_len символов — устойчиво к
«Злат/Злата/златочка»;
- опционально max_edit_distance — допуск опечатки распознавания.
"""
from __future__ import annotations
import re
from typing import List, Optional
DEFAULT_WAKE_WORD = "марта"
_MIN_PREFIX = 4 # «злат» уже считается (обрубок «зла…» от микрофона)
_PART_MARKER_RE = re.compile(r"[^\wа-яё]+", re.IGNORECASE)
def normalize(text: str) -> List[str]:
"""Фраза → список слов (нижний регистр, без пунктуации, ё→е).
Дефис внутри слова склеивается («Ага-то» → «агато») — whisper любит
разбивать имена слогами через дефис.
"""
text = text.lower().replace("ё", "е")
text = re.sub(r"(\w)-(\w)", r"\1\2", text) # ага-то → агато
return [w for w in re.split(r"[^\w]+", text) if w]
def matches_wake_word(transcript: str, wake_word: str = DEFAULT_WAKE_WORD) -> bool:
"""True, если фраза НАЧИНАЕТСЯ с wake-word (целое слово, с допуском).
Whisper любит ломать имена: «Ага-то», «А гата» — поэтому проверяем
первое слово И склейку первых двух. «Агата»/«агат» (обрубок) триггерят;
«агатовые украшения» — нет: после корня продолжение слова.
"""
words = normalize(transcript)
if not words:
return False
ww = normalize(wake_word)[0] if normalize(wake_word) else ""
if not ww:
return False
candidates = [words[0]]
if len(words) >= 2:
candidates.append(words[0] + words[1]) # «а гата» → «агата»
for first in candidates:
if first == ww:
return True
if len(ww) >= _MIN_PREFIX and ww.startswith(first) and len(first) >= _MIN_PREFIX:
return True
if abs(len(first) - len(ww)) <= 1 and len(ww) >= _MIN_PREFIX \
and first[: len(ww) - 1] == ww[: len(ww) - 1]:
return True
return False
class WakeWordWatcher:
"""Проверяет распознанные фразы на наличие wake-word в режиме ожидания."""
def __init__(self, wake_word: Optional[str] = None, on_detected: Optional[callable] = None) -> None:
from ..brain.config import get_secret
self.wake_word = wake_word or get_secret("WAKE_WORD") or DEFAULT_WAKE_WORD
self.on_detected = on_detected or (lambda: None)
def feed_transcript(self, transcript: str) -> bool:
"""Распознанная фраза в режиме ожидания. True → wake-word сработал."""
if matches_wake_word(transcript, self.wake_word):
self.on_detected()
return True
return False
@property
def hint(self) -> str:
return f"Скажите «{self.wake_word.capitalize()}», чтобы начать диалог."
+24
View File
@@ -0,0 +1,24 @@
"""Модуль 2: audio_io — запись и воспроизведение звука.
Кроссплатформенно (Linux для разработки, Windows — цель).
Состав: Recorder (push-to-talk буфер в RAM), Player (мгновенный stop = «Замолчи»),
ресемплинг в 16 кГц mono для STT.
"""
from __future__ import annotations
from .player import Player
from .recorder import Recorder
from .resample import resample_to_16k
from .handsfree import HandsFreeRecorder
__all__ = ["Player", "Recorder", "HandsFreeRecorder", "resample_to_16k", "list_devices"]
def list_devices() -> None:
"""Печать аудио-устройств (вход/выход) — для диагностики."""
import sounddevice as sd
print(sd.query_devices())
default_in = sd.default.device[0]
default_out = sd.default.device[1]
print(f"\nПо умолчанию: вход={default_in}, выход={default_out}")
+34
View File
@@ -0,0 +1,34 @@
"""Декодирование сжатого аудио (mp3/ogg/...) в float32.
Использует PyAV — ffmpeg-библиотеки вкомпилированы в пакет av
(он уже есть как зависимость faster-whisper), системный ffmpeg не нужен.
"""
from __future__ import annotations
import io
from typing import Tuple
import av
import numpy as np
def decode_audio_bytes(data: bytes) -> Tuple[np.ndarray, int]:
"""Байты аудиофайла → (float32 mono [-1..1], samplerate)."""
container = av.open(io.BytesIO(data))
try:
stream = container.streams.audio[0]
samplerate = int(stream.rate)
resampler = av.AudioResampler(format="flt", layout="mono")
pieces = []
for frame in container.decode(stream):
for out in resampler.resample(frame):
arr = out.to_ndarray()
if arr.ndim == 2: # (channels, samples) → mono
arr = arr[0] if arr.shape[0] == 1 else arr.mean(axis=0)
pieces.append(np.asarray(arr, dtype=np.float32))
finally:
container.close()
if not pieces:
return np.zeros(0, dtype=np.float32), samplerate
return np.concatenate(pieces), samplerate
+207
View File
@@ -0,0 +1,207 @@
"""HandsFreeRecorder — свободный голосовой режим БЕЗ клавиш (М5.2).
Постоянно слушает микрофон; решение «речь/тишина» принимает Silero-VAD:
- началась речь → копим аудио (с pre-buffer 0.4 c, чтобы не терять первые слова);
- тишина silence_sec (по умолчанию 2.0 — «пауза 2 секунды» юзера) → фраза
закончена → вызов self.on_phrase(audio, samplerate);
- короче min_speech_sec — шум, отбраковка (событие on_state("discard"));
- длиннее max_seconds — принудительный срез;
- на время ответа ассистента прослушивание ставится на паузу (suppress),
чтобы микрофон не слышал её из динамиков.
Колбэки — публичные атрибуты: их можно переприсвоить и извне
(Assistant присваивает recorder.on_phrase = ...).
Фильтр шума: голосом считается окно, у которого prob VAD выше порога И
RMS выше noise_gate (Silero на белом шуме даёт ~0.85 — одной вероятности мало).
"""
from __future__ import annotations
import time
import warnings
from typing import Callable, List, Optional
import numpy as np
import sounddevice as sd
from .resample import TARGET_RATE, resample_to_16k
_VAD_WINDOW = 512 # 32 мс при 16 кГц — родной размер окна Silero
_PRE_BUFFER_SEC = 0.4 # буфер ДО начала речи (не терять начало фразы)
class HandsFreeRecorder:
def __init__(
self,
samplerate: int = TARGET_RATE,
device: Optional[int] = None,
silence_sec: float = 2.0,
speech_threshold: float = 0.5,
min_speech_sec: float = 0.3,
max_seconds: float = 30.0,
noise_gate_rms: float = 0.006, # ниже RMS — окно считается тишиной, как бы VAD ни хотел
on_phrase: Optional[Callable[[np.ndarray, int], None]] = None,
on_state: Optional[Callable[[str], None]] = None, # listening|speech_start|speech_end|too_short|paused
on_error: Optional[Callable[[str], None]] = None,
debug: bool = False,
) -> None:
self._sr = samplerate
self._device = device
self._silence_sec = silence_sec
self._threshold = speech_threshold
self._min_speech_sec = min_speech_sec
self._max_seconds = max_seconds
self._noise_gate = noise_gate_rms
# Публичные колбэки: Assistant переприсваивает on_phrase — так и задумано
self.on_phrase: Callable[[np.ndarray, int], None] = on_phrase or (lambda a, s: None)
self.on_state: Callable[[str], None] = on_state or (lambda s: None)
self.on_error: Callable[[str], None] = on_error or (lambda m: print(f" [VAD] {m}"))
self._vad = None
self._stream = None
self._running = False
self._suppress = False
self._recording = False
self._chunks: List[np.ndarray] = []
self._pre_buffer: List[np.ndarray] = []
self._speech_frames = 0
self._silence_frames = 0
self._native_sr = samplerate
self._carry: np.ndarray = np.zeros(0, dtype=np.float32) # хвост между колбэками
# ---------------------------------------------------------------- lifecycle
def start(self) -> None:
"""Открыть микрофон и слушать. Ошибки уходят в on_error (не роняют программу)."""
try:
warnings.filterwarnings("ignore", category=Warning)
from silero_vad import load_silero_vad
self._vad = load_silero_vad()
stream = self._open(self._sr)
if stream is None:
self._native_sr = int(
sd.query_devices(self._device, "input")["default_samplerate"]
)
stream = self._open(self._native_sr, required=True)
else:
self._native_sr = self._sr
stream.start()
self._stream = stream
self._running = True
self.on_state("listening")
except Exception as exc:
self.on_error(f"Свободный режим не запустился: {exc}")
def stop(self) -> None:
self._running = False
stream, self._stream = self._stream, None
if stream is not None:
try:
stream.stop()
stream.close()
except Exception:
pass
# ---------------------------------------------------------------- режим
@property
def is_listening(self) -> bool:
return self._running and not self._suppress
def suppress(self, on: bool) -> None:
"""Пауза прослушивания (пока сами говорим) + сброс недозаписи."""
if self._suppress == on:
return
self._suppress = on
self.on_state("paused" if on else "listening")
if on:
self._recording = False
self._chunks = []
self._pre_buffer = []
self._carry = np.zeros(0, dtype=np.float32)
# ---------------------------------------------------------------- внутреннее
def _open(self, samplerate: int, required: bool = False):
try:
return sd.InputStream(
samplerate=samplerate,
device=self._device,
channels=1,
dtype="float32",
blocksize=512 if samplerate == 16000 else 0,
callback=self._on_audio,
)
except Exception as exc:
if required:
raise RuntimeError(f"Не удалось открыть микрофон: {exc}") from exc
return None
def _vad_prob(self, window: np.ndarray) -> float:
"""Уверенность VAD для окна (0..1). JIT-модель требует torch.Tensor."""
import torch
with torch.no_grad():
return float(self._vad(torch.from_numpy(window), 16000))
def _on_audio(self, indata, frames, time_info, status) -> None: # noqa: ANN001
if status:
self.on_error(str(status))
if self._suppress or self._vad is None:
return
mono = indata[:, 0].copy()
if self._native_sr != 16000:
mono16 = resample_to_16k(mono, self._native_sr)
else:
mono16 = mono
# накопитель между колбэками: окно 512 всегда полное, хвост не теряется
buf = np.concatenate([self._carry, mono16])
n = len(buf) // _VAD_WINDOW
for i in range(n):
self._process_window(buf[i * _VAD_WINDOW:(i + 1) * _VAD_WINDOW])
self._carry = buf[n * _VAD_WINDOW:]
def _process_window(self, window: np.ndarray) -> None:
rms = float(np.sqrt(np.mean(np.square(window))))
voice = self._vad_prob(window) > self._threshold and rms > self._noise_gate
if voice:
if not self._recording:
self._recording = True
self._chunks = list(self._pre_buffer) if self._pre_buffer else []
self._pre_buffer = []
self._speech_frames = 0
self.on_state("speech_start")
self._chunks.append(window)
self._speech_frames += len(window)
self._silence_frames = 0
if self._speech_frames >= int(self._max_seconds * 16000):
self._finalize()
return
if self._recording:
self._chunks.append(window)
self._silence_frames += len(window)
if self._silence_frames >= int(self._silence_sec * 16000):
self._finalize()
else:
self._pre_buffer.append(window)
if len(self._pre_buffer) > int(_PRE_BUFFER_SEC * 16000 / _VAD_WINDOW):
self._pre_buffer.pop(0)
def _finalize(self) -> None:
chunks, self._chunks = self._chunks, []
self._recording = False
self._pre_buffer = []
speech_samples = self._speech_frames
self._speech_frames = 0
self._silence_frames = 0
total = sum(len(c) for c in chunks)
# Валидна именно ДОЛЯ голоса: суммарный буфер всегда ≥ паузе тишины
if speech_samples < int(self._min_speech_sec * 16000):
self.on_state("too_short") # видимая отбраковка, не тихая
return
self.on_state("speech_end")
self.on_phrase(np.concatenate(chunks).astype(np.float32), 16000)
+89
View File
@@ -0,0 +1,89 @@
"""Player — воспроизведение с мгновенной остановкой.
Клавиша «Замолчи» = player.stop(): sounddevice.stop() прерывает вывод
в течение ~миллисекунд, очередь очищается.
Воспроизведение идёт в фоновом потоке; is_playing отражает состояние.
last_start_latency — сколько времени занял вызов sd.play() (открытие потока
вывода до первого блока): замеряется для диагностики латентности ответа.
"""
from __future__ import annotations
import threading
import time
from pathlib import Path
from typing import Optional, Union
import numpy as np
import sounddevice as sd
import soundfile as sf
class Player:
def __init__(self, device: Optional[int] = None, volume: float = 1.0) -> None:
self._device = device
self.volume = volume # 0.0 .. 2.0
self._lock = threading.Lock()
self._playing = False
self._thread: Optional[threading.Thread] = None
self.last_start_latency = 0.0 # сек: длительность вызова sd.play()
@property
def is_playing(self) -> bool:
return self._playing
def play(self, data: np.ndarray, samplerate: int, blocking: bool = False,
lead_silence_sec: float = 0.1) -> None:
"""Воспроизвести float32 mono/stereo. Не блокирует (если blocking=False).
lead_silence_sec — тишина в начале: защита от «съедания» первого слова
при открытии потока вывода (недозаполненный буфер на некоторых ALSA/Pulse).
"""
audio = np.asarray(data, dtype=np.float32)
if audio.ndim == 1:
audio = audio[:, np.newaxis] # (frames,) → (frames, 1)
if lead_silence_sec > 0:
pad = np.zeros((int(lead_silence_sec * samplerate), audio.shape[1]),
dtype=np.float32)
audio = np.concatenate([pad, audio])
gain = float(np.clip(self.volume, 0.0, 2.0))
if gain != 1.0:
audio = np.clip(audio * gain, -1.0, 1.0)
with self._lock:
# Новая команда play отменяет предыдущую (накладывать нельзя)
sd.stop()
def _run() -> None:
try:
t0 = time.perf_counter()
sd.play(audio, samplerate, device=self._device)
self.last_start_latency = time.perf_counter() - t0
sd.wait() # вернётся сразу после stop() или конца аудио
finally:
self._playing = False
self._playing = True
if blocking:
_run()
else:
self._thread = threading.Thread(target=_run, daemon=True)
self._thread.start()
def play_file(self, path: Union[str, Path], blocking: bool = False) -> None:
"""Воспроизвести wav/flac/ogg с диска."""
path = Path(path)
data, sr = sf.read(str(path), dtype="float32", always_2d=False)
self.play(data, sr, blocking=blocking)
def stop(self) -> None:
"""Мгновенно заглушить воспроизведение (клавиша «Замолчи»)."""
with self._lock:
sd.stop()
self._playing = False
def wait(self) -> None:
"""Дождаться конца текущего воспроизведения."""
t = self._thread
if t is not None and t.is_alive():
t.join()
+145
View File
@@ -0,0 +1,145 @@
"""Recorder — push-to-talk запись в RAM.
Схема работы в будущем ассистенте:
клавиша «Слушай» нажата → recorder.start()
клавиша отпущена → audio = recorder.stop() → STT
Особенности:
- пробует открыть вход на 16 кГц; если устройство не умеет — пишет в нативной
частоте и ресемплирует в 16 кГц при stop() (нужно STT);
- поток в фоновом потоке PortAudio, данные копируются в список блоков;
- слишком короткие записи (< min_seconds) возвращаются пустыми — защита от
случайных кликов;
- потоко-безопасность через threading.Lock.
"""
from __future__ import annotations
import threading
import time
from typing import Callable, List, Optional
import numpy as np
import sounddevice as sd
from .resample import TARGET_RATE, resample_to_16k
class Recorder:
def __init__(
self,
samplerate: int = TARGET_RATE,
device: Optional[int] = None,
channels: int = 1,
blocksize: int = 800, # 50 мс при 16 кГц
max_seconds: float = 60.0, # защита от «забытой» клавиши
min_seconds: float = 0.3, # короче — считаем пустой записью
on_overrun: Optional[Callable[[str], None]] = None, # колбэк о проблемах
) -> None:
self._target_sr = samplerate
self._device = device
self._channels = channels
self._blocksize = blocksize
self._max_seconds = max_seconds
self._min_seconds = min_seconds
self._on_overrun = on_overrun
self._frames: List[np.ndarray] = []
self._stream: Optional[sd.InputStream] = None
self._native_sr: int = samplerate
self._started_at: float = 0.0
self._lock = threading.Lock()
# --- состояние -------------------------------------------------------
@property
def is_recording(self) -> bool:
return self._stream is not None and self._stream.active
@property
def duration(self) -> float:
"""Длительность текущей записи в секундах (0 если не пишем)."""
if not self.is_recording:
return 0.0
return time.perf_counter() - self._started_at
@property
def native_samplerate(self) -> int:
"""Частота, на которой реально шло устройство (до ресемплинга)."""
return self._native_sr
# --- управление ------------------------------------------------------
def _callback(self, indata, frames, time_info, status) -> None: # noqa: ANN001
if status and self._on_overrun:
self._on_overrun(str(status))
self._frames.append(indata.copy())
def start(self) -> None:
"""Начать запись. Бросает RuntimeError, если запись уже идёт."""
if self.is_recording:
raise RuntimeError("Запись уже идёт")
with self._lock:
self._frames = []
stream = self._try_open(self._target_sr)
if stream is None:
# Устройство не поддерживает 16 кГц — пишем в нативной и ресемплируем при stop()
self._native_sr = int(
sd.query_devices(self._device, "input")["default_samplerate"]
)
stream = self._try_open(self._native_sr, required=True)
else:
self._native_sr = self._target_sr
stream.start()
self._stream = stream
self._started_at = time.perf_counter()
def stop(self) -> np.ndarray:
"""Остановить запись и вернуть float32 mono, 16 кГц, [-1..1]."""
stream, self._stream = self._stream, None
if stream is None:
return np.zeros(0, dtype=np.float32)
stream.stop()
stream.close()
with self._lock:
blocks = self._frames
self._frames = []
audio = np.concatenate(blocks, axis=0) if blocks else np.zeros((0, self._channels))
mono = audio.mean(axis=1) if audio.ndim > 1 else audio
mono = np.clip(mono, -1.0, 1.0).astype(np.float32)
if self._native_sr != self._target_sr:
mono = resample_to_16k(mono, self._native_sr, self._target_sr)
min_len = int(self._min_seconds * self._target_sr)
if len(mono) < min_len:
return np.zeros(0, dtype=np.float32) # слишком коротко — пусто
return mono
def cancel(self) -> None:
"""Остановить запись, выбросив данные (например, нажали «Замолчи»)."""
stream, self._stream = self._stream, None
if stream is not None:
stream.stop()
stream.close()
with self._lock:
self._frames = []
# --- внутреннее ------------------------------------------------------
def _try_open(self, samplerate: int, required: bool = False) -> Optional[sd.InputStream]:
try:
return sd.InputStream(
samplerate=samplerate,
device=self._device,
channels=self._channels,
dtype="float32",
blocksize=self._blocksize,
callback=self._callback,
)
except (sd.PortAudioError, OSError) as exc:
if required:
raise RuntimeError(f"Не удалось открыть входное устройство: {exc}") from exc
return None
+34
View File
@@ -0,0 +1,34 @@
"""Ресемплинг аудио (линейная интерполяция) — 16 кГц mono для STT.
Для речи линейной интерполяции достаточно; качество-loss заметно меньше,
чем выигрыш от простоты и отсутствия зависимостей.
"""
from __future__ import annotations
import numpy as np
TARGET_RATE = 16000
def resample_to_16k(data: np.ndarray, source_rate: int, target_rate: int = TARGET_RATE) -> np.ndarray:
"""float32 mono → float32 mono с частотой target_rate."""
if source_rate == target_rate:
return np.asarray(data, dtype=np.float32)
x = np.asarray(data, dtype=np.float32)
if x.ndim > 1:
x = x.mean(axis=1)
if len(x) == 0:
return x
n_out = int(round(len(x) * target_rate / source_rate))
if n_out <= 1:
return np.zeros(max(n_out, 0), dtype=np.float32)
# Позиции выходных сэмплов во входной шкале (линейная интерполяция)
pos = np.linspace(0.0, len(x) - 1.0, num=n_out, dtype=np.float64)
i0 = pos.astype(np.int64)
i1 = np.minimum(i0 + 1, len(x) - 1)
frac = (pos - i0).astype(np.float32)
out = x[i0] * (1.0 - frac) + x[i1] * frac
return np.clip(out, -1.0, 1.0)
+132
View File
@@ -0,0 +1,132 @@
"""Тест Модуля 2: audio_io (запись / воспроизведение / ресемплинг / мгновенный stop).
Запуск:
python modules/audio_io/test_audio_io.py devices # список аудио-устройств
python modules/audio_io/test_audio_io.py auto # автоматический смоук (без участия человека)
python modules/audio_io/test_audio_io.py talk # ИНТЕРАКТИВНЫЙ: push-to-talk через Enter
Режим talk:
Enter → начать запись (говори), Enter → закончить и услышать себя,
затем записанное уходит в STT (проверка склейки Модуль 2 + Модуль 1).
"""
from __future__ import annotations
import sys
import time
from pathlib import Path
import numpy as np
try:
from modules.audio_io import Player, Recorder, list_devices, resample_to_16k
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from modules.audio_io import Player, Recorder, list_devices, resample_to_16k
SAMPLES_DIR = Path(__file__).resolve().parent / "samples"
# --------------------------------------------------------------------------
def smoke_test() -> None:
"""Автоматический смоук: ресемплинг, тоны, мгновенная остановка, короткая запись."""
print("== 1. Ресемплинг ==")
t = np.linspace(0.0, 1.0, 44100, endpoint=False, dtype=np.float32)
sine44k = 0.5 * np.sin(2 * np.pi * 440.0 * t)
out16k = resample_to_16k(sine44k, 44100)
assert len(out16k) == 16000, f"длина {len(out16k)} != 16000"
print(f" 44100 Гц ({len(sine44k)} сэмплов) → 16000 Гц ({len(out16k)}) OK")
# оценка частоты тона после ресемплинга (через число пересечений нуля)
zero_cross = np.count_nonzero(np.diff(np.signbit(out16k)))
freq = zero_cross / 2.0
print(f" тон после ресемплинга: ~{freq:.0f} Гц (ожидалось 440) OK")
print("== 2. Воспроизведение и мгновенный stop ==")
player = Player()
long_tone = 0.3 * np.sin(2 * np.pi * 440.0 * np.linspace(0, 5, 5 * 48000, endpoint=False))
player.play(long_tone, 48000)
time.sleep(0.8)
assert player.is_playing, "тон должен играть"
t0 = time.perf_counter()
player.stop()
stop_ms = (time.perf_counter() - t0) * 1000
print(f" stop() сработал за {stop_ms:.0f} мс (цель < 100 мс)")
assert not player.is_playing
print("== 3. Запись реального устройства (2 с, без участия человека) ==")
rec = Recorder(min_seconds=0.1)
rec.start()
time.sleep(2.0)
audio = rec.stop()
rms = float(np.sqrt(np.mean(np.square(audio)))) if len(audio) else 0.0
print(f" записано {len(audio)/16000:.1f} c, RMS={rms:.4f} (0 = тишина/нет данных)")
if len(audio) == 0:
print(" ВНИМАНИЕ: устройство вернуло слишком мало данных")
print("== 4. Короткая запись отбрасывается (защита) ==")
rec2 = Recorder(min_seconds=0.5)
rec2.start()
time.sleep(0.05)
empty = rec2.stop()
assert len(empty) == 0, "короткая запись должна возвращать пустой массив"
print(" OK: 50 мс → пусто")
print("\nСМОУК ПРОЙДЕН ✅")
# --------------------------------------------------------------------------
def talk_test() -> None:
"""Интерактивный push-to-talk: Enter — говорить, Enter — стоп, потом плеер и STT."""
from modules.stt import get_provider
print("Загружаю STT (faster-whisper small)...")
stt = get_provider("faster-whisper", model_size="small")
player = Player()
rec = Recorder()
while True:
print("\n[Enter] удержи-режим: нажми Enter и ГОВОРИ, затем Enter — стоп. "
"Пустая строка после фразы — выход.")
cmd = input("> ")
if cmd.strip() == "":
print("Выход.")
return
rec.start()
input("…запись идёт, Enter = закончить ")
audio = rec.stop()
if len(audio) == 0:
print("Слишком коротко — отброшено. Ещё раз.")
continue
print(f"Записано {len(audio)/16000:.1f} c (частота устройства {rec.native_samplerate} Гц).")
wav = Path(__file__).resolve().parent / "samples" / "ptt_last.wav"
wav.parent.mkdir(parents=True, exist_ok=True)
import soundfile as sf
sf.write(str(wav), audio, 16000)
print(f"Сохранено: {wav}")
print("Воспроизвожу твой голос (проверь качество, Esc/Enter — прервать)...")
player.play(audio, 16000)
try:
input(" (Enter — если хочешь прервать звук) ")
player.stop()
except KeyboardInterrupt:
player.stop()
res = stt.transcribe(audio)
print(f"STT: {res}")
def main() -> int:
mode = sys.argv[1] if len(sys.argv) > 1 else "auto"
if mode == "devices":
list_devices()
elif mode == "talk":
talk_test()
else:
smoke_test()
return 0
if __name__ == "__main__":
sys.exit(main())
+41
View File
@@ -0,0 +1,41 @@
"""Модуль 4: brain — «мозг» ассистента (LLM через OpenRouter) + очистка текста.
Использование:
from modules.brain import get_provider
brain = get_provider("openrouter", model="openai/gpt-4o-mini")
result = brain.ask("Какая погода?")
print(result.cleaned_text) # чистый текст для озвучки
"""
from __future__ import annotations
from .base import DialogHistory, LlmProvider, LlmResult
from .clean_text import clean_for_speech
from .prompts import SYSTEM_PROMPT
from .provider_openrouter import OpenRouterProvider
__all__ = [
"DialogHistory",
"LlmProvider",
"LlmResult",
"OpenRouterProvider",
"SYSTEM_PROMPT",
"clean_for_speech",
"get_provider",
]
_PROVIDERS = {
OpenRouterProvider.name: OpenRouterProvider,
}
def get_provider(name: str, **kwargs) -> LlmProvider:
"""Фабрика провайдеров по имени (выбор из конфига)."""
try:
provider_cls = _PROVIDERS[name]
except KeyError:
raise ValueError(
f"Неизвестный LLM-провайдер: {name!r}. Доступны: {sorted(_PROVIDERS)}"
) from None
return provider_cls(**kwargs)
+41
View File
@@ -0,0 +1,41 @@
"""Базовый интерфейс «мозга» (LLM) + результат запроса.
История диалога — простой список {"role": "user"|"assistant", "text": str},
чтобы провайдеры и будущая машина состояний не зависели от формата API.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import List, Optional
DialogHistory = List[dict] # [{"role": "user"|"assistant", "text": str}, ...]
@dataclass
class LlmResult:
"""Ответ LLM: сырой текст + очищенный для озвучки."""
question: str
raw_text: str # как прислала модель (с разметкой, если была)
cleaned_text: str # после clean_for_speech() — именно это озвучиваем
generation_sec: float
model: str = ""
def __str__(self) -> str:
return (f"[{self.model}] {self.generation_sec:.2f} c, "
f"{len(self.cleaned_text)} симв.: {self.cleaned_text!r}")
class LlmProvider(ABC):
"""Интерфейс провайдера LLM (сейчас OpenRouter, при надобности — локальный ollama)."""
name: str = "base"
context_block: str = "" # доп. контекст (долговременная память); подставляется в system
@abstractmethod
def ask(self, question: str, history: Optional[DialogHistory] = None) -> LlmResult:
"""Задать вопрос, получить ответ. history — предыдущие реплики диалога."""
def close(self) -> None:
"""Освободить ресурсы (по умолчанию — ничего)."""
+70
View File
@@ -0,0 +1,70 @@
"""Очистка ответа LLM перед озвучкой.
По ТЗ: ИИ должен говорить чистым «человеческим» текстом — без звёздочек,
решёток, списков, ссылок и эмодзи. Первая линия обороны — системный промпт
(запрещает разметку), эта функция — вторая (убирает то, что просочилось).
"""
from __future__ import annotations
import re
# Эмодзи и декоративные символы (TTS их либо молчит, либо читает мусором)
_EMOJI = re.compile(
"["
"\U0001F000-\U0001FAFF" # эмодзи и пиктограммы
"\U00002600-\U000027BF" # misc symbols
"\U0001F1E6-\U0001F1FF" # флаги
"\U00002B00-\U00002BFF" # стрелки/звёзды
"\U0000FE00-\U0000FE0F" # variation selectors
"\U00002190-\U000021FF" # стрелки
"]+"
)
_CODE_FENCE = re.compile(r"```[\s\S]*?(```|$)")
# Служебный маркер многочастных ответов: [ЧАСТЬ 2 ИЗ 5] — вырезается до озвучки
_PART_MARKER = re.compile(r"\s*\[ЧАСТЬ\s+\d+\s+ИЗ\s+\d+\]\s*", re.I)
_IMAGE = re.compile(r"!\[([^\]]*)\]\([^)]*\)")
_LINK = re.compile(r"\[([^\]]*)\]\([^)]*\)")
_HEADER = re.compile(r"^[ \t]{0,3}#{1,6}[ \t]*", re.M)
_BOLD = re.compile(r"(\*\*\*|\*\*|__)(?=\S)(.+?\S)\1")
_ITALIC = re.compile(r"(?<![\w*\\])(\*|_)(?=\S)(.+?\S)\1(?![\w*])")
_STRIKE = re.compile(r"~~(.+?)~~")
_BULLET = re.compile(r"(?:^|\n)[ \t]*[-*+][ \t]+")
_NUMBERED = re.compile(r"(?:^|\n)[ \t]*\d{1,3}[.)][ \t]+")
_BLOCKQUOTE = re.compile(r"(?:^|\n)[ \t]*>[ \t]?")
_HRULE = re.compile(r"(?:^|\n)[ \t]*([-_=*])[ \t]*\1[ \t]*\1[\1 \t=-]*(?:\n|$)")
_INLINE_CODE = re.compile(r"`([^`]*)`")
_MULTI_WS = re.compile(r"[ \t\f\v]+")
_MULTI_NL = re.compile(r"\s*\n\s*")
def clean_for_speech(text: str) -> str:
"""Markdown/эмодзи → одна чистая строка, пригодная для TTS."""
if not text:
return ""
t = _CODE_FENCE.sub(" ", text)
t = _PART_MARKER.sub(" ", t) # служебный маркер части — не для ушей
t = _IMAGE.sub(r"\1", t)
t = _LINK.sub(r"\1", t)
t = _HEADER.sub("", t)
t = _BOLD.sub(r"\2", t)
t = _ITALIC.sub(r"\2", t)
t = _STRIKE.sub(r"\1", t)
t = _BULLET.sub(" ", t)
t = _NUMBERED.sub(" ", t)
t = _BLOCKQUOTE.sub(" ", t)
t = _HRULE.sub(" ", t)
t = _INLINE_CODE.sub(r"\1", t)
t = _EMOJI.sub(" ", t)
# Переносы строк — в пробелы (для озвучки это одна речь)
t = t.replace("\r", "")
t = _MULTI_WS.sub(" ", t)
t = _MULTI_NL.sub(" ", t)
t = t.strip()
# Двойные знаки препинания после чистки («…!» и т.п.)
t = re.sub(r"\s+([,.!?;:])", r"\1", t)
t = re.sub(r"([,.!?;:])\1+", r"\1", t)
return t
+36
View File
@@ -0,0 +1,36 @@
"""Загрузка секретов: переменные окружения + файл .env в корне проекта.
Ключи НЕ хранятся в коде. Порядок:
1) переменная окружения (OPENROUTER_API_KEY=...);
2) файл .env в корне проекта (строка вида OPENROUTER_API_KEY=sk-or-...).
Файл .env.example — шаблон: cp .env.example .env и вписать свой ключ.
"""
from __future__ import annotations
import os
from pathlib import Path
_PROJECT_ROOT = Path(__file__).resolve().parents[2]
def get_secret(name: str) -> str | None:
"""Взять секрет из окружения или .env (первый источник побеждает)."""
value = os.environ.get(name)
if value:
return value.strip()
env_file = _PROJECT_ROOT / ".env"
if env_file.is_file():
for line in env_file.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, val = line.partition("=")
if key.strip() == name:
return val.strip().strip('"').strip("'")
return None
def get_api_key() -> str | None:
return get_secret("OPENROUTER_API_KEY")
+29
View File
@@ -0,0 +1,29 @@
"""Системный промпт ассистента.
Главная линия обороны от «грязного» текста: короткие ответы, разговорный
язык, явный запрет разметки. Экономит и время (меньше токенов), и слух отца.
Многочастные длинные ответы (идея юзера): если произведение не влезает в
один ответ, модель САМА делит его на части и помечает маркерами [ЧАСТЬ i ИЗ n],
отдаёт следующую часть по команде «дальше». Ассистент бдит о полноте не кодом,
а инструкцией модели.
"""
SYSTEM_PROMPT = (
"Ты — голосовой ассистент для пожилого незрячего человека. "
"Обычно отвечай ко��отко: одно-три простых предложения. "
"Говори простым, тёплым и ясным языком, без сложных терминов. "
"Никакой разметки: без списков, звёздочек, заголовков, кода и эмодзи. "
"Только чистый текст, который приятно слушать. "
"Если просят сделать что-то на компьютере — объясняй словами просто и по шагам. "
"НО если просят прочитать стихотворение, сказку, историю или рассказать "
"подробно — дай полный текст целиком, не сокращай и не пересказывай кратко. "
"Стихи читай полностью, с строфами через перенос строки. "
"\n\nЕСЛИ произведение длинное (стихотворение более 30 строк, длинная сказка "
"или история) и может не поместиться в один ответ целиком: дай первую "
"осмысленную часть, в самом кон��е добавь ровно строку '[ЧАСТЬ i ИЗ n]' "
"(например [ЧАСТЬ 1 ИЗ 3]) и на этом остановись. Когда человек скажет "
"«дальше» или «продолжай» — отдай следующую часть с таким же маркером в "
"конце, продолжая ровно с места обрыва, без повторов. В последней части "
"маркер не ставь и спроси, не хочет ли человек послушать ещё что-нибудь. "
"Маркер вслух не читается, он служебный."
)
+138
View File
@@ -0,0 +1,138 @@
"""Провайдер LLM через OpenRouter (openai-совместимый API).
Сеть (проверено 2026-09-11):
- OpenRouter API доступен из РФ напрямую (200), НО часть апстрим-провайдеров
(OpenAI, Google) блокирует запросы с российских IP по своим ToS (403).
- Решение: опциональный прокси (по умолчанию берётся из .env: OPENROUTER_PROXY),
запросы выходят через зарубежный egress — тогда доступны все модели.
Модель — параметр конструктора (задаётся в тесте/конфиге).
"""
from __future__ import annotations
import re
import time
from typing import Optional
from .base import DialogHistory, LlmProvider, LlmResult
from .clean_text import clean_for_speech
from .config import get_api_key, get_secret
from .prompts import SYSTEM_PROMPT
class OpenRouterProvider(LlmProvider):
name = "openrouter"
# Вопросы, требующие свежих данных → модель с суффиксом :online (веб-поиск OpenRouter).
# Онлайн-поиск медленнее (+4-9 c) и чуть дороже, поэтому только для «актуальных» тем.
ONLINE_MODEL_ENV = "OPENROUTER_MODEL_ONLINE"
ONLINE_HINTS = (
"погод", "курс", "доллар", "евро", "рубл", "новост", "сегодняшн",
"сейчас на улице", "за окном", "курс валют", "биткоин", "акци",
"спорт", "счет матча", "результат матч", "последн",
)
def __init__(
self,
# openai/* заблокированы ToS-фильтром OpenAI по аккаунту OpenRouter с РФ-биллингом
# (403 даже через зарубежный egress). Рабочие из РФ (проверено 2026-09-11):
# google/gemma-3-27b-it, deepseek/deepseek-chat-v3-0324, qwen/qwen-2.5-72b-instruct,
# mistralai/mistral-small-3.1-24b-instruct, meta-llama/llama-3.3-70b-instruct,
# z-ai/glm-5.3-flash (reasoning-модель: тратит бюджет max_tokens на размышления).
# None → берётся OPENROUTER_MODEL из .env (там google/gemma-3-27b-it).
model: str | None = None,
online_model: str | None = None, # None → OPENROUTER_MODEL_ONLINE (glm-5.3-flash:online)
api_key: str | None = None,
base_url: str = "https://openrouter.ai/api/v1",
proxy: str | None = None, # http://хост:порт; None → OPENROUTER_PROXY из .env
temperature: float = 0.6,
# Reasoning-модели (z-ai/glm-*) тратят этот бюджет на «размышления» ДО ответа
# (у glm-5.3-flash на простую фразу ушло 65 reasoning-токенов из 78).
# Если бюджет мал, reasoning съедает его и content приходит ПУСТЫМ.
# 2000 хватает на длинное стихотворение/историю; это ~$0.0003–0.0008 на ответ.
max_tokens: int = 2000,
timeout_sec: float = 30.0,
) -> None:
from openai import OpenAI
key = api_key or get_api_key()
if not key:
raise RuntimeError(
"Нет API-ключа. Создай файл .env в корне проекта со строкой\n"
"OPENROUTER_API_KEY=sk-or-v1-...\n"
"(или export OPENROUTER_API_KEY=... перед запуском)"
)
proxy_url = proxy or get_secret("OPENROUTER_PROXY")
http_client = None
if proxy_url:
import httpx
http_client = httpx.Client(proxy=proxy_url, timeout=timeout_sec)
self._model = model or get_secret("OPENROUTER_MODEL") or "google/gemma-3-27b-it"
# Модель с веб-поиском для «актуальных» вопросов (погода/курсы/новости)
self._online_model = (online_model or get_secret(self.ONLINE_MODEL_ENV)
or "z-ai/glm-5.3-flash:online")
self._temperature = temperature
self._max_tokens = max_tokens
self._client = OpenAI(
base_url=base_url,
api_key=key,
timeout=timeout_sec,
http_client=http_client,
)
# --- API -------------------------------------------------------------
def _needs_online(self, question: str, history: Optional[DialogHistory] = None) -> bool:
"""Похоже ли, что вопрос требует свежих данных из интернета.
Смотрим не только на текущую фразу, но и на последние реплики диалога:
уточнение «я поэтому и спрашиваю...» после вопроса про акции должно
пойти в online-модель, хотя в самой фразе нет слов «акции/курс».
"""
q = question.lower()
if any(hint in q for hint in self.ONLINE_HINTS):
return True
# контекст последних 4 реплик (2 вопроса+ответа) тоже проверяем
for turn in (history or [])[-4:]:
text = turn.get("text", "").lower()
if any(hint in text for hint in self.ONLINE_HINTS):
return True
# маркеры продолжения/уточнения («я поэтому и спрашиваю», «а теперь...»)
if history and re.search(r"\b(поэтому и спрашиваю|я же спрашивал|ты не ответила|"
r"а теперь|так вот|продолжай|уточню)\b", q):
return True
return False
def ask(self, question: str, history: Optional[DialogHistory] = None) -> LlmResult:
system = SYSTEM_PROMPT
if self.context_block:
system += "\n\n" + self.context_block
messages = [{"role": "system", "content": system}]
for turn in history or []:
messages.append({"role": turn["role"], "content": turn["text"]})
messages.append({"role": "user", "content": question})
# Вопрос про погоду/курсы/новости (или уточнение к такому) → модель с веб-поиском
used_model = self._online_model if self._needs_online(question, history) else self._model
t0 = time.perf_counter()
response = self._client.chat.completions.create(
model=used_model,
messages=messages,
temperature=self._temperature,
max_tokens=self._max_tokens,
)
elapsed = time.perf_counter() - t0
raw = response.choices[0].message.content or "" if response.choices else ""
# Убираем маркеры источников, которые добавляет веб-плагин (ссылки в конце)
raw = re.sub(r"\s*\b[a-z0-9.-]+\.(?:ru|com|net|org|ai)\b\s*$", "", raw, flags=re.I)
return LlmResult(
question=question,
raw_text=raw.strip(),
cleaned_text=clean_for_speech(raw),
generation_sec=elapsed,
model=used_model,
)
+149
View File
@@ -0,0 +1,149 @@
"""Тест Модуля 4: brain (LLM) + очистка Markdown + полная цепочка.
Запуск (нужен ключ: файл .env со строкой OPENROUTER_API_KEY=sk-or-v1-...):
python modules/brain/test_brain.py clean # тест очистки без сети
python modules/brain/test_brain.py --ask "Привет!" # вопрос → ответ (raw + чистый + время)
python modules/brain/test_brain.py --ask "Привет!" --play # и озвучить ответ
python modules/brain/test_brain.py --model google/gemini-flash-1.5 --ask "..."
python modules/brain/test_brain.py --mic # ПОЛНАЯ ЦЕПЬ: говоришь → слышишь ответ
"""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
try:
from modules.brain import clean_for_speech, get_provider
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from modules.brain import clean_for_speech, get_provider
# ---------------------------------------------------------------------------
def clean_selftest() -> None:
"""Проверка очистки Markdown без сети: грязные строки → чистый текст."""
cases = [
("**Привет!** Как *дела*?", "Привет! Как дела?"),
("## Заголовок\n- пункт один\n- пункт два", "Заголовок пункт один пункт два"),
("1. Включи компьютер\n2. Открой программу", "Включи компьютер Открой программу"),
("Ссылка: [Google](https://google.com)", "Ссылка: Google"),
("Код: `python -m venv` и ```print(1)```", "Код: python -m venv и"),
("Смайл 😊 и стрелка ➡️ тут", "Смайл и стрелка тут"),
("Текст выше\n> Важная цитата\nТекст ниже", "Текст выше Важная цитата Текст ниже"),
("Раздел один\n---\nРаздел два", "Раздел один Раздел два"),
]
ok = True
for dirty, expected in cases:
got = clean_for_speech(dirty)
status = "OK " if got == expected else "FAIL"
ok &= got == expected
print(f"[{status}] {dirty!r}\n → {got!r}" + ("" if got == expected else f"\n ожидалось {expected!r}"))
print("\nОЧИСТКА:", "ПРОЙДЕНА ✅" if ok else "ЕСТЬ ОШИБКИ ❌")
# ---------------------------------------------------------------------------
def ask_once(brain, question: str, play: bool, history=None):
result = brain.ask(question, history=history)
print(f"\nВопрос: {question}")
print(f"Ответ ({result.generation_sec:.2f} c): {result.cleaned_text}")
if result.raw_text != result.cleaned_text:
print(f"(raw был: {result.raw_text!r})")
if play:
from modules.audio_io import Player
from modules.tts import get_provider as get_tts
tts = get_tts("edge")
speech = tts.synthesize(result.cleaned_text)
print(f"TTS: синтез {speech.generation_sec:.2f} c, аудио {speech.duration_sec:.1f} c")
Player().play(speech.audio, speech.samplerate, blocking=True)
return result
def mic_loop(brain) -> None:
"""Полная цепочка: запись с микрофона → STT → LLM → TTS → воспроизведение."""
from modules.audio_io import Player, Recorder
from modules.stt import get_provider as get_stt
from modules.tts import get_provider as get_tts
print("Загружаю STT (faster-whisper small)...")
stt = get_stt("faster-whisper", model_size="small")
tts = get_tts("edge")
player = Player()
rec = Recorder()
history = []
print("\nГОТОВ. [Enter] → говори вопрос → [Enter] → услышишь ответ. Пустой Enter — выход.")
while True:
cmd = input("> ").strip()
if cmd == "":
break
rec.start()
input(" 🔴 запись... Enter = закончить ")
audio = rec.stop()
if len(audio) == 0:
print(" Слишком коротко — повтори.")
continue
stt_res = stt.transcribe(audio)
question = stt_res.text.strip()
if not question:
print(" STT ничего не расслышал. Повтори.")
continue
print(f" Ты: {question}")
llm = brain.ask(question, history=history)
print(f" AI ({llm.generation_sec:.2f} c): {llm.cleaned_text}")
speech = tts.synthesize(llm.cleaned_text)
print(f" TTS ({speech.generation_sec:.2f} c), играю...")
player.play(speech.audio, speech.samplerate, blocking=True)
history.append({"role": "user", "text": question})
history.append({"role": "assistant", "text": llm.cleaned_text})
history = history[-8:] # держим контекст, но ограничиваем
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Тест Модуля 4: brain")
p.add_argument("mode", nargs="?", default="ask", choices=["ask", "clean", "mic"],
help="clean = офлайн-тест очистки; ask = текстовый вопрос; mic = полная цепь")
p.add_argument("--ask", "--question", dest="question", default="Привет! Кто ты и что умеешь?",
help="вопрос к модели (режим ask)")
p.add_argument("--model", default="google/gemma-3-27b-it",
help="модель OpenRouter (deepseek/deepseek-chat-v3-0324, qwen/qwen-2.5-72b-instruct...)")
p.add_argument("--play", action="store_true", help="озвучить ответ (edge-tts)")
p.add_argument("--turns", type=int, default=0,
help="сколько вопросов подряд в режиме ask с общей историей")
return p.parse_args()
def main() -> int:
args = parse_args()
if args.mode == "clean":
clean_selftest()
return 0
brain = get_provider("openrouter", model=args.model)
print(f"Модель: {args.model}")
if args.mode == "mic":
mic_loop(brain)
return 0
questions = [args.question]
if args.turns > 1:
extra = input(f"Введи ещё {args.turns - 1} вопроса через ';' : ")
questions = [q.strip() for q in extra.split(";") if q.strip()] or questions
history = []
for q in questions:
result = ask_once(brain, q, play=args.play, history=history)
history += [{"role": "user", "text": q}, {"role": "assistant", "text": result.cleaned_text}]
return 0
if __name__ == "__main__":
sys.exit(main())
+43
View File
@@ -0,0 +1,43 @@
"""Модуль 5: hotkeys — клавиши управления (событийные бэкенды).
Использование:
from modules.hotkeys import KeyEvents, get_backend
events = KeyEvents(on_talk_down=..., on_talk_up=..., on_stop=...,
on_repeat=..., on_quit=...)
backend = get_backend("console", events) # или "keys" (Windows / root-Linux)
backend.start()
"""
from __future__ import annotations
import os
from .base import KeyEvents
from .console import ConsoleBackend
from .keyboard_hook import KeyboardHookBackend
__all__ = ["KeyEvents", "ConsoleBackend", "KeyboardHookBackend", "get_backend"]
_BACKENDS = {
ConsoleBackend.name: ConsoleBackend,
KeyboardHookBackend.name: KeyboardHookBackend,
}
def get_backend(name: str, events: KeyEvents, **kwargs):
"""Фабрика бэкендов клавиш."""
if name == KeyboardHookBackend.name:
# Клавиши можно переопределить через переменные окружения/.env:
# HOTKEY_DIALOG (пробел-тумблер), HOTKEY_TALK, HOTKEY_STOP, HOTKEY_REPEAT
kwargs.setdefault("dialog", os.environ.get("HOTKEY_DIALOG", "space"))
kwargs.setdefault("talk", os.environ.get("HOTKEY_TALK", "num 0"))
kwargs.setdefault("stop_key", os.environ.get("HOTKEY_STOP", "esc"))
kwargs.setdefault("repeat", os.environ.get("HOTKEY_REPEAT", "enter"))
try:
backend_cls = _BACKENDS[name]
except KeyError:
raise ValueError(
f"Неизвестный бэкенд клавиш: {name!r}. Доступны: {sorted(_BACKENDS)}"
) from None
return backend_cls(events, **kwargs)
+30
View File
@@ -0,0 +1,30 @@
"""События клавиш — контракт между бэкендом ввода и ассистентом.
Бэкенд (console / Windows-хук) только ПЕРЕДАЁТ события; вся логика
реакции — в машине состояний (modules/assistant).
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable, Optional
@dataclass
class KeyEvents:
"""Набор колбэков ассистента.
on_dialog_toggle — ПРОБЕЛ: начать диалог (куи «готов к диалогу») или
завершить его (куи «до новых встреч»);
on_talk_down — клавиша «Слушай» НАЖАТА (push-to-talk, резервный режим);
on_talk_up — клавиша «Слушай» ОТПУЩЕНА;
on_stop — «Замолчи»: заглушить речь / отменить;
on_repeat — «Повтори»: последнюю фразу;
on_quit — выход из программы.
"""
on_talk_down: Callable[[], None]
on_talk_up: Callable[[], None]
on_stop: Callable[[], None]
on_repeat: Callable[[], None]
on_quit: Callable[[], None]
on_dialog_toggle: Optional[Callable[[], None]] = None
+100
View File
@@ -0,0 +1,100 @@
"""Консольный бэкенд клавиш — для разработки и теста в Linux.
Глобальный хук клавиатуры (библиотека keyboard) на Linux требует root,
поэтому здесь клавиши имитируются вводом в терминале:
Enter — нажать/отпустить «Слушай» (первое нажатие = начать запись,
второе = отпустить и обработать; аналог удержания кнопки);
s + Enter — «Замолчи»;
r + Enter — «Повтори»;
q + Enter — выход.
На Windows-ПК отец получит настоящие глобальные клавиши (keyboard_hook.py).
"""
from __future__ import annotations
import threading
from .base import KeyEvents
class ConsoleBackend:
name = "console"
def __init__(self, events: KeyEvents) -> None:
self._events = events
self._talk_held = False
self._running = False
def start(self) -> None:
import threading
self._running = True
thread = threading.Thread(target=self._loop, daemon=True)
thread.start()
def stop(self) -> None:
self._running = False
# --- внутреннее ------------------------------------------------------
def _loop(self) -> None:
# Одиночные клавиши БЕЗ Enter (readchar): работает без root, когда
# окно терминала в фокусе. Пробел = настоящий пробел, как на Windows.
try:
import readchar
use_readchar = True
except ImportError:
use_readchar = False # фолбэк: строки + Enter (старое поведение)
if use_readchar:
self._loop_readchar(readchar)
else:
self._loop_lines()
def _loop_readchar(self, readchar) -> None:
while self._running:
key = readchar.readkey()
if key in ("\r", "\n"):
continue # Enter игнорируем (не part of управления)
if key == " ":
self._events.on_dialog_toggle() # ПРОБЕЛ = тумблер диалога
elif key == "s":
self._events.on_stop()
elif key == "r":
self._events.on_repeat()
elif key in ("q", "\x03"): # q или Ctrl+C
self._events.on_quit()
return
# остальные клавиши игнорируются
def _loop_lines(self) -> None:
"""Фолбэк: строки через Enter (если readchar не установлен)."""
if self._events.on_dialog_toggle:
print(" (readchar не установлен: пробел имитируй 'd' + Enter; "
"pip install readchar — включит настоящий пробел)")
while self._running:
try:
line = input()
except EOFError: # stdin закрылся — выходим
self._events.on_quit()
return
cmd = line.strip().lower()
if cmd == "q":
self._events.on_quit()
return
if cmd in ("d", "space", "пробел"):
if self._events.on_dialog_toggle:
self._events.on_dialog_toggle()
elif cmd == "s":
self._events.on_stop()
elif cmd == "r":
self._events.on_repeat()
elif cmd == "":
# Enter без текста = нажатие/отпускание «Слушай» (push-to-talk)
if self._talk_held:
self._talk_held = False
self._events.on_talk_up()
else:
self._talk_held = True
self._events.on_talk_down()
+57
View File
@@ -0,0 +1,57 @@
"""Бэкенд глобальных клавиш на библиотеке keyboard — целевой для Windows.
Перехватывает клавиши ГЛОБАЛЬНО (окно может быть свёрнуто) — именно то,
что нужно отцу. Требования/ограничения библиотеки keyboard:
- Windows: работает без админ-прав;
- Linux: требует root (поэтому в разработке используем console-бэкенд);
- проект библиотеки не развивается с 2020 — если на Windows появятся
проблемы, заменяем на pynput (интерфейс KeyEvents менять не придётся).
Клавиши задаются в .env (HOTKEY_TALK / HOTKEY_STOP / HOTKEY_REPEAT);
по умолчанию: space / esc / enter. Имена — как в библиотеке keyboard
(numpad-клавиши подберём при настройке на Windows, напр. «num 0», «num enter»).
"""
from __future__ import annotations
from .base import KeyEvents
class KeyboardHookBackend:
name = "keys"
def __init__(self, events: KeyEvents, talk: str = "num 0",
stop_key: str = "esc", repeat: str = "enter",
dialog: str = "space") -> None:
self._events = events
self._talk = talk
self._stop_key = stop_key
self._repeat = repeat
self._dialog = dialog # ПРОБЕЛ: тумблер диалога
self._started = False
def start(self) -> None:
import keyboard # ленивый импорт: на Linux без root упадёт — см. console
if self._dialog and self._events.on_dialog_toggle:
keyboard.add_hotkey(self._dialog, self._events.on_dialog_toggle, suppress=True)
if self._talk:
keyboard.on_press_key(self._talk, self._talk_press)
keyboard.on_release_key(self._talk, self._talk_release)
keyboard.add_hotkey(self._stop_key, self._events.on_stop, suppress=True)
keyboard.add_hotkey(self._repeat, self._events.on_repeat, suppress=True)
self._started = True
def stop(self) -> None:
if self._started:
import keyboard
keyboard.unhook_all()
self._started = False
# --- внутреннее ------------------------------------------------------
def _talk_press(self, event) -> None: # noqa: ANN001
# Библиотека шлёт повторные down-события при удержании; лишние вызовы
# безопасны: машина состояний игнорирует on_talk_down в состоянии listening.
self._events.on_talk_down()
def _talk_release(self, event) -> None: # noqa: ANN001
self._events.on_talk_up()
+57
View File
@@ -0,0 +1,57 @@
# Модуль 1: STT (речь → текст)
Обёртка над **faster-whisper** с интерфейсом «провайдер»: если для ПК отца понадобится
другой распознаватель (vosk), он добавляется новым классом без изменения остального проекта.
## Файлы
| Файл | Назначение |
|---|---|
| `base.py` | Интерфейс `SttProvider`, результат `TranscriptResult` |
| `provider_faster_whisper.py` | Реализация на faster-whisper (CPU, int8) |
| `test_stt.py` | Тест-скрипт: файл / генерация фразы / микрофон |
| `samples/` | Тестовые аудио (создаются при запуске, в git не входят) |
## Использование из кода
```python
from modules.stt import get_provider
stt = get_provider("faster-whisper", model_size="small") # tiny|base|small|medium
result = stt.transcribe("phrase.wav") # путь к файлу (wav/mp3/flac)
result = stt.transcribe(np_array_float32_16k) # или массив с микрофона
print(result.text, result.processing_sec)
```
## Что замеряет тест
- **processing_sec** — время распознавания (без загрузки модели);
- **RTF** (realtime factor) — обработка/длина аудио; RTF < 1 — быстрее реального времени;
- при `--repeat N` — лучший (минимальный) прогон.
## Критерии приёмки Модуля 1
1. Тестовая фраза распознаётся дословно (allow punctuation/case).
2. Фраза ~5 c обрабатывается за <= 3 c (RTF <= 0.6) на `small`/int8/CPU.
3. Если нет — пробуем `base`, фиксируем цифры; если и `base` медленный — добавляем Vosk-провайдер.
## Решение по скорости (замер 2026-09-10, Linux, фраза 8.9 c, edge-tts голос)
| Модель | Обработка | RTF | Точность |
|---|---|---|---|
| **small** | **1.17 c** | **0.13** | дословно ✅ |
| base | 0.54 c | 0.06 | 2 ошибки ("выпить чаю" → "выпечаю") |
| tiny | 0.39 c | 0.04 | 4 ошибки ("тестовы", "выпечаю") |
**Вердикт: faster-whisper `small` (int8, CPU, beam=1, VAD on)** — RTF 0.13 даёт запас ×3
даже на слабом ПК; качество base/tiny недостаточно для пожилого пользователя.
Загрузка модели при старте: ~2–5 c (после скачивания), кэш в `~/.cache/huggingface`.
Память по факту не замерена — при желании: `/usr/bin/time -v .venv/bin/python modules/stt/test_stt.py --make-sample`.
## Известные ограничения
- Первый запуск качает модель с HuggingFace (small ≈ 460 МБ). Если недоступно:
`export HF_ENDPOINT=https://hf-mirror.com` перед запуском.
- Запись с микрофона идёт на 16 кГц: если запись звучит «с ускорением», у устройства
проблемы с ресемплингом — сообщи, добавим ресемплинг явно.
+39
View File
@@ -0,0 +1,39 @@
"""STT-модуль: речь → текст.
Использование из других модулей:
from modules.stt import get_provider
stt = get_provider("faster-whisper", model_size="small")
result = stt.transcribe("phrase.wav")
print(result.text)
"""
from __future__ import annotations
from .base import SAMPLE_RATE, AudioInput, SttProvider, TranscriptResult
from .provider_faster_whisper import FasterWhisperProvider
__all__ = [
"SAMPLE_RATE",
"AudioInput",
"SttProvider",
"TranscriptResult",
"FasterWhisperProvider",
"get_provider",
]
_PROVIDERS = {
FasterWhisperProvider.name: FasterWhisperProvider,
# "vosk" добавим, если faster-whisper окажется тяжёлым для ПК отца (замер в Модуле 1)
}
def get_provider(name: str, **kwargs) -> SttProvider:
"""Фабрика провайдеров по имени (выбор из конфига)."""
try:
provider_cls = _PROVIDERS[name]
except KeyError:
raise ValueError(
f"Неизвестный STT-провайдер: {name!r}. Доступны: {sorted(_PROVIDERS)}"
) from None
return provider_cls(**kwargs)
+61
View File
@@ -0,0 +1,61 @@
"""Базовый интерфейс STT-модуля (речь → текст).
Любой распознаватель (faster-whisper, vosk, ...) реализует SttProvider,
чтобы остальной проект не зависел от конкретной библиотеки.
Замена провайдера = правка одной строки в конфиге.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from typing import Union
import numpy as np
# Что умеет принимать transcribe(): путь к файлу ЛИБО готовый массив float32
AudioInput = Union[str, Path, np.ndarray]
# Все провайдеры ждут 16 кГц mono float32 ([-1 .. 1])
SAMPLE_RATE = 16000
@dataclass
class TranscriptResult:
"""Результат распознавания одной фразы."""
text: str
duration_sec: float # длина аудио в секундах
processing_sec: float # сколько времени считали
language: str = "ru"
language_probability: float = 0.0
@property
def realtime_factor(self) -> float:
"""processing / duration. Значение < 1.0 — быстрее реального времени."""
if self.duration_sec <= 0:
return float("inf")
return self.processing_sec / self.duration_sec
def __str__(self) -> str:
return (
f"[{self.duration_sec:.1f} c аудио → {self.processing_sec:.2f} c, "
f"RTF {self.realtime_factor:.2f}] {self.text!r}"
)
class SttProvider(ABC):
"""Интерфейс провайдера распознавания речи."""
name: str = "base"
@abstractmethod
def transcribe(self, audio: AudioInput, sample_rate: int = SAMPLE_RATE) -> TranscriptResult:
"""Распознать речь.
audio — путь к файлу (wav/mp3/flac, декодирование внутри) ИЛИ
numpy-массив float32 mono с частотой sample_rate.
"""
def close(self) -> None:
"""Освободить ресурсы (по умолчанию — ничего)."""
+83
View File
@@ -0,0 +1,83 @@
"""Провайдер STT на faster-whisper (CTranslate2).
Модель автоматически скачивается с HuggingFace при первом запуске
и кэшируется в ~/.cache/huggingface (small ≈ 460 МБ).
Если скачивание из HuggingFace недоступно/медленное, можно указать зеркало:
export HF_ENDPOINT=https://hf-mirror.com
"""
from __future__ import annotations
import time
from pathlib import Path
from typing import Optional
import numpy as np
from faster_whisper import WhisperModel
from .base import SAMPLE_RATE, AudioInput, SttProvider, TranscriptResult
class FasterWhisperProvider(SttProvider):
name = "faster-whisper"
def __init__(
self,
model_size: str = "small", # tiny | base | small | medium | large-v3
device: str = "cpu", # cpu | cuda
compute_type: str = "int8", # int8 на CPU — быстро и мало памяти
language: str = "ru", # фиксируем язык: детекция не нужна
beam_size: int = 1, # 1 = жадный поиск, заметно быстрее
vad_filter: bool = True, # Silero-VAD внутри: отрезает тишину
cpu_threads: int = 0, # 0 = по умолчанию движка
) -> None:
self._language = language
self._beam_size = beam_size
self._vad_filter = vad_filter
t0 = time.perf_counter()
self._model = WhisperModel(
model_size,
device=device,
compute_type=compute_type,
cpu_threads=cpu_threads,
)
# Время загрузки (и первого скачивания) модели — отдельно от распознавания.
self.load_sec = time.perf_counter() - t0
def transcribe(self, audio: AudioInput, sample_rate: int = SAMPLE_RATE) -> TranscriptResult:
duration: Optional[float] = None
if isinstance(audio, np.ndarray):
samples = np.asarray(audio, dtype=np.float32)
if samples.ndim > 1: # (frames, channels) → mono
samples = samples.mean(axis=1)
duration = len(samples) / sample_rate
source: AudioInput = samples
else:
path = Path(audio)
if not path.is_file():
raise FileNotFoundError(f"Аудио-файл не найден: {path}")
source = str(path)
t0 = time.perf_counter()
segments, info = self._model.transcribe(
source,
language=self._language,
beam_size=self._beam_size,
vad_filter=self._vad_filter,
)
text = "".join(segment.text for segment in segments)
text = " ".join(text.split()) # нормализуем пробелы между сегментами
processing = time.perf_counter() - t0
if duration is None: # для файлов длина известна после transcribe
duration = float(getattr(info, "duration", 0.0) or 0.0)
return TranscriptResult(
text=text,
duration_sec=duration,
processing_sec=processing,
language=info.language or self._language,
language_probability=float(info.language_probability or 0.0),
)
+137
View File
@@ -0,0 +1,137 @@
"""Тест STT-модуля (Модуль 1).
Запуск:
python modules/stt/test_stt.py --make-sample # сгенерировать sample.mp3 (edge-tts) и распознать
python modules/stt/test_stt.py modules/stt/samples/sample.mp3
python modules/stt/test_stt.py --record 5 # 5 секунд с микрофона → распознать
python modules/stt/test_stt.py --model base <file> # лёгкая модель для слабого ПК
python modules/stt/test_stt.py --repeat 3 <file> # замерить скорость несколькими прогонами
Критерий приёмки: фраза распознана верно, обработка <= 3 c на фразе ~5 c.
"""
from __future__ import annotations
import argparse
import asyncio
import sys
import time
from pathlib import Path
from typing import Optional
import numpy as np
try:
from modules.stt import get_provider
except ImportError: # запуск как обычного скрипта: добавляем корень проекта в путь
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from modules.stt import get_provider
SAMPLES_DIR = Path(__file__).resolve().parent / "samples"
SAMPLE_PHRASE = (
"Привет! Это тестовый голосовой ассистент. "
"Сегодня хорошая погода, и я хочу выпить чаю."
)
DEFAULT_VOICE = "ru-RU-SvetlanaNeural"
def make_sample(path: Path, voice: str) -> None:
"""Синтез русской фразы через edge-tts — удобный источник тестового аудио."""
import edge_tts
path.parent.mkdir(parents=True, exist_ok=True)
async def _save() -> None:
await edge_tts.Communicate(SAMPLE_PHRASE, voice).save(str(path))
print(f"Генерирую {path.name} (голос {voice})...")
asyncio.run(_save())
def record(seconds: int, sample_rate: int = 16000) -> np.ndarray:
"""Запись с микрофона по умолчанию → float32 mono, 16 кГц."""
import sounddevice as sd
import soundfile as sf
print(f"\nЗапись {seconds} с. Приготовьтесь:")
for i in (3, 2, 1):
print(f" {i}...")
time.sleep(0.6)
print(" >>> ГОВОРИТЕ <<<")
raw = sd.rec(int(seconds * sample_rate), samplerate=sample_rate, channels=1, dtype="int16")
sd.wait()
audio = raw.astype(np.float32) / 32768.0
audio = audio.mean(axis=1) # (frames, 1) → mono
wav = SAMPLES_DIR / "recording.wav"
sf.write(str(wav), audio, sample_rate)
print(f"Сохранено: {wav}")
return audio
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Тест STT-модуля (faster-whisper)")
p.add_argument("audio", nargs="?", help="аудио-файл (wav/mp3/flac)")
p.add_argument("--make-sample", action="store_true",
help="сгенерировать sample.mp3 (edge-tts) и распознать его")
p.add_argument("--voice", default=DEFAULT_VOICE, help="голос edge-tts для --make-sample")
p.add_argument("--record", type=int, default=0, metavar="SEC",
help="записать SEC секунд с микрофона и распознать")
p.add_argument("--model", default="small", help="tiny | base | small | medium")
p.add_argument("--device", default="cpu", help="cpu | cuda")
p.add_argument("--compute", default="int8", help="int8 | float16 | float32")
p.add_argument("--beam", type=int, default=1, help="beam size (1 = быстрее)")
p.add_argument("--no-vad", action="store_true", help="не отрезать тишину")
p.add_argument("--repeat", type=int, default=1, help="число прогонов для замера")
return p.parse_args()
def main() -> int:
args = parse_args()
audio_array: Optional[np.ndarray] = None
target: Optional[Path] = None
if args.record:
audio_array = record(args.record)
elif args.make_sample:
target = SAMPLES_DIR / "sample.mp3"
if not target.exists():
make_sample(target, args.voice)
else:
print(f"Использую существующий {target}")
elif args.audio:
target = Path(args.audio)
else:
print(__doc__)
return 1
print(f"\nЗагружаю модель {args.model!r} ({args.compute}, {args.device})...")
print("Первый запуск скачивает модель с HuggingFace (small ≈ 460 МБ), это нормально.")
stt = get_provider(
"faster-whisper",
model_size=args.model,
device=args.device,
compute_type=args.compute,
beam_size=args.beam,
vad_filter=not args.no_vad,
)
print(f"Модель готова за {stt.load_sec:.1f} с.\n")
results = []
for i in range(1, args.repeat + 1):
source = audio_array if audio_array is not None else target
res = stt.transcribe(source)
results.append(res)
print(f"[прогон {i}/{args.repeat}] {res}")
best = min(results, key=lambda r: r.processing_sec)
print("\n=== ИТОГ ===")
print(f"Текст: {best.text or '(пусто — нечего распознавать)'}")
print(f"Аудио {best.duration_sec:.1f} c → обработка {best.processing_sec:.2f} c "
f"(RTF {best.realtime_factor:.2f})")
print(f"Язык: {best.language} (уверенность {best.language_probability:.2f})")
passed = bool(best.text.strip())
print("ПРИЁМКА:", "ПРОЙДЕНА ✅" if passed else "ПРОВАЛЕНА ❌ (пустой текст)")
return 0 if passed else 2
if __name__ == "__main__":
sys.exit(main())
+104
View File
@@ -0,0 +1,104 @@
# Модуль 3: TTS (текст → звук)
Два провайдера за общим интерфейсом `TtsProvider`: переключение = выбор в конфиге.
| Провайдер | Сеть | Латентность (110 симв.) | Голоса | Роль |
|---|---|---|---|---|
| **edge** (`provider_edge.py`) | онлайн, бесплатный сервис MS, без ключа | 3.0 c (4 предл.), 1.7 c (короткая) | Svetlana/Dmitry, отличное качество | основной |
| **piper** (`provider_piper.py`) | офлайн, локально на CPU | 0.47 c | dmitri/irina/ruslan, проще | фолбэк без сети |
## Режимы edge-провайдера
- **Один запрос** (по умолчанию): весь текст → один round-trip (~1.7–3 c). Паузы между
предложениями удлиняются многоточиями («точка» → «…» — edge-tts читает как длинную паузу).
- **per_sentence=True**: каждое предложение отдельным запросом + точная тишина
`sentence_pause_sec` между ними. Медленнее (N запросов), нужен для тонкой подгонки пауз.
Замер 2026-09-11 (свежие соединения в каждом процессе): один запрос 110 симв = 3.0 c,
короткая фраза 36 симв = 1.7 c, Piper 110 симв = 0.47 c. В живой программе соединение
переиспользуется — будет быстрее. Цепочка целиком: STT ~1.2 c + LLM ~1–2 c + TTS ~1.7 c ≈ 4–5 c
от отпускания клавиши до голоса (маскируется earcon «думаю»).
## Использование из кода
```python
from modules.tts import get_provider
tts = get_provider("edge") # voice="ru-RU-DmitryNeural" — мужской
result = tts.synthesize("Привет! Как дела?")
# result.audio (float32), result.samplerate, result.generation_sec, result.duration_sec
```
Фолбэк при обрыве сети (switch = одна строка):
```python
tts = get_provider("piper", model_path="models/piper/ru_RU-dmitri-medium.onnx")
```
## Потоковая озвучка (стриминг, 2026-09-11)
`tts.stream(text)` — генератор: отдаёт чанки-предложения **по мере готовности**
(параллельные запросы, отдача в порядке следования). `Assistant` играёт чанк
сразу, не дожидаясь синтеза всего текста:
- первое предложение звучит через ~0.5 c после готовности LLM (один round-trip);
- длинные ответы больше не пропорционально тормозят (2-е/3-е предложение
синтезируются, пока играет 1-е);
- между чанками вставляется пауза sentence_pause_sec;
- перебивание отцом гасит поток и синтез (счётчик поколений).
Плюс в `Player.play()` добавлен префикс тишины 0.1 c (`lead_silence_sec`) —
защита от «съедания» первого слова при открытии аудио-потока (недозаполненный
буфер ALSA/Pulse).
## Настройка из .env (2026-09-11)
```
TTS_PROVIDER=edge # edge | piper
TTS_VOICE=ru-RU-SvetlanaNeural # мужской: ru-RU-DmitryNeural
TTS_RATE=+15 # темп речи, % (+ = быстрее; ≈+30% к исходному −15)
PIPER_MODEL=models/piper/ru_RU-dmitri-medium.onnx
```
Выбранный конфиг юзером: **edge + Svetlana, rate +15** («женский голос лучше и темп
нормальный»). Piper остаётся офлайн-фолбэком (0.22 c синтез против 0.49 c у edge),
при желании ускориться — одна строка в .env.
Фабрика: `make_tts()` из `modules/tts` — тесты ассистента читают эти параметры
автоматически, смену голоса/скорости/провайдера можно делать без правки кода.
## Файлы
| Файл | Назначение |
|---|---|
| `base.py` | Интерфейс `TtsProvider`, результат `SynthResult` |
| `provider_edge.py` | edge-tts: один запрос / сшивание предложений |
| `provider_piper.py` | Piper офлайн (`length_scale` из rate автоматом в config) |
| `config.py` | `make_tts()` — фабрика из .env (провайдер, голос, скорость) |
| `text_split.py` | Разбивка на предложения; защита сокращений («т.д.», «т.к.») |
| `test_tts.py` | Тест-скрипт (синтез / воспроизведение / интерактив / провайдеры) |
## Тест-скрипт
```bash
.venv/bin/python modules/tts/test_tts.py --play # фраза по умолчанию + звук
.venv/bin/python modules/tts/test_tts.py --text "Своё предложение"
.venv/bin/python modules/tts/test_tts.py --voice ru-RU-DmitryNeural --play # мужской
.venv/bin/python modules/tts/test_tts.py --listen # вводишь текст — слышишь
.venv/bin/python modules/tts/test_tts.py --per-sentence # медленный режим (сравнение)
.venv/bin/python modules/tts/test_tts.py --provider piper --model models/piper/ru_RU-dmitri-medium.onnx --play
```
## Критерии приёмки Модуля 3
1. ✅ Русский голос звучит естественно, скорость −15% — принято юзером на слух (Svetlana).
2. ✅ Паузы между предложениями слышны (многоточия / точная тишина).
3. ✅ Синтез короткой фразы ≤ 3 c: 1.7 c (edge), 0.47 c (piper).
## Известные ограничения
- edge-tts — неофициальный API Microsoft: исторически ломался (403). План Б — Piper, он уже в проекте.
- Piper: голос скачивается один раз в `models/piper/` (команда в docstring провайдера);
качество проще edge, зато 0.5 c и полностью офлайн.
- В `provider_edge._run_async` учтён будущий asyncio-контекст (aiogram в Модуле 6):
синтез не сломает чужой event loop.
+44
View File
@@ -0,0 +1,44 @@
"""TTS-модуль: текст → звук.
Использование из других модулей:
from modules.tts import get_provider
tts = get_provider("edge") # онлайн, голос MS
tts = get_provider("piper", model_path="...") # офлайн-фолбэк
result = tts.synthesize("Привет! Как дела?")
# result.audio (float32), result.samplerate, result.generation_sec
"""
from __future__ import annotations
from .base import SynthResult, TtsProvider
from .config import make_tts
from .provider_edge import EdgeTtsProvider
from .provider_piper import PiperTtsProvider
from .text_split import split_sentences
__all__ = [
"SynthResult",
"TtsProvider",
"EdgeTtsProvider",
"PiperTtsProvider",
"make_tts",
"split_sentences",
"get_provider",
]
_PROVIDERS = {
EdgeTtsProvider.name: EdgeTtsProvider,
PiperTtsProvider.name: PiperTtsProvider,
}
def get_provider(name: str, **kwargs) -> TtsProvider:
"""Фабрика провайдеров по имени (выбор из конфига)."""
try:
provider_cls = _PROVIDERS[name]
except KeyError:
raise ValueError(
f"Неизвестный TTS-провайдер: {name!r}. Доступны: {sorted(_PROVIDERS)}"
) from None
return provider_cls(**kwargs)
+49
View File
@@ -0,0 +1,49 @@
"""Базовый интерфейс TTS-модуля (текст → звук).
Провайдеры (edge-tts онлайн, piper офлайн) реализуют TtsProvider,
чтобы остальной проект не зависел от конкретного синтезатора:
если edge-tts отвалится (неофициальный эндпоинт Microsoft), переключение
на Piper — одна строка в конфиге.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
import numpy as np
@dataclass
class SynthResult:
"""Результат синтеза одной фразы."""
text: str
audio: np.ndarray # float32 mono [-1..1]
samplerate: int
generation_sec: float # время генерации (без воспроизведения)
provider: str = ""
sentence_count: int = 1
@property
def duration_sec(self) -> float:
if self.samplerate <= 0:
return 0.0
return len(self.audio) / self.samplerate
def __str__(self) -> str:
return (f"[{self.provider}] текст {len(self.text)} симв. → "
f"аудио {self.duration_sec:.1f} c, синтез {self.generation_sec:.2f} c, "
f"{self.samplerate} Гц, предложений: {self.sentence_count}")
class TtsProvider(ABC):
"""Интерфейс провайдера синтеза речи."""
name: str = "base"
@abstractmethod
def synthesize(self, text: str) -> SynthResult:
"""Синтезировать речь. text — уже очищенный от Markdown текст."""
def close(self) -> None:
"""Освободить ресурсы (по умолчанию — ничего)."""
+43
View File
@@ -0,0 +1,43 @@
"""Настройки TTS из .env — единая точка выбора провайдера и скорости речи.
Параметры .env (все опциональны):
TTS_PROVIDER=edge | piper (по умолчанию edge — лучше качество)
TTS_VOICE=ru-RU-SvetlanaNeural (голос edge; мужской: ru-RU-DmitryNeural)
TTS_RATE=+15 (скорость речи, %; +быстрее, −медленнее)
PIPER_MODEL=models/piper/ru_RU-dmitri-medium.onnx
Про скорость: −15 был «для пожилых медленно» по исходному ТЗ; живой тест показал,
что хочется быстрее (~+30% к этому темпу) → TTS_RATE=+15.
"""
from __future__ import annotations
from ..brain.config import get_secret
from .base import TtsProvider
from .provider_edge import EdgeTtsProvider
from .provider_piper import PiperTtsProvider
def make_tts(provider: str | None = None, **overrides) -> TtsProvider:
"""Создать TTS-провайдера по .env (+ переопределения в коде для тестов)."""
name = (provider or get_secret("TTS_PROVIDER") or "edge").lower()
rate = int(get_secret("TTS_RATE") or -15)
if name == "edge":
kwargs = {
"voice": get_secret("TTS_VOICE") or "ru-RU-SvetlanaNeural",
"rate": rate,
}
kwargs.update(overrides)
return EdgeTtsProvider(**kwargs)
if name == "piper":
# Piper: rate % → length_scale (1.15 ≈ −15%, 0.77 ≈ +30%)
length_scale = round(100.0 / (100.0 + rate), 3)
kwargs = {
"model_path": get_secret("PIPER_MODEL") or "models/piper/ru_RU-dmitri-medium.onnx",
"length_scale": length_scale,
}
kwargs.update(overrides)
return PiperTtsProvider(**kwargs)
raise ValueError(f"Неизвестный TTS_PROVIDER: {name!r} (доступны: edge, piper)")
+208
View File
@@ -0,0 +1,208 @@
"""Провайдер TTS на edge-tts (онлайн-сервис Microsoft, без ключа и бесплатно).
Плюсы: отличные русские голоса (Svetlana/Dmitry), rate/volume/pitch «из коробки».
Риск: неофициальный эндпоинт — исторически ломался (403). Поэтому в проекте
есть офлайн-фолбэк Piper, а провайдер создаётся через фабрику get_provider().
Скорость речи по ТЗ: −15% (rate="-15%").
Латентность: по умолчанию ВЕСЬ текст синтезируется ОДНИМ запросом (1 сетевой
round-trip, ~1–2 c), паузы между предложениями удлиняются многоточиями
(edge-tts читает «…» как длинную паузу). Режим per_sentence=True синтезирует
каждое предложение отдельным запросом и вставляет точную тишину
sentence_pause_sec — медленнее (N запросов), нужен только для тонкой подгонки пауз.
"""
from __future__ import annotations
import asyncio
import concurrent.futures
import time
from typing import List, Optional, Union
import numpy as np
from ..audio_io.codec import decode_audio_bytes
from .base import SynthResult, TtsProvider
from .text_split import insert_pauses, split_sentences
def _fmt_prosody(value: Union[int, str], unit: str) -> str:
"""int → '+15%' / '-15%' (edge-tts требует знак и единицу)."""
if isinstance(value, int):
return f"{value:+d}{unit}"
return str(value)
def _run_async(coro):
"""Выполнить корутину из синхронного кода.
Если event loop уже крутится (например, поток aiogram в Модуле 6) —
выполняем в отдельном потоке, чтобы не ломать чужой цикл.
"""
try:
asyncio.get_running_loop()
except RuntimeError:
return asyncio.run(coro)
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
return ex.submit(asyncio.run, coro).result()
class EdgeTtsProvider(TtsProvider):
name = "edge"
def __init__(
self,
voice: str = "ru-RU-SvetlanaNeural", # мужской: ru-RU-DmitryNeural
rate: Union[int, str] = -15, # −15% скорости по ТЗ
volume: Union[int, str] = 0, # +0% (громкость задаётся в Player)
pitch: Union[int, str] = 0, # +0 Гц
sentence_pause_sec: float = 0.35, # точная пауза (только per_sentence)
per_sentence: bool = False, # False = один запрос (быстро)
) -> None:
self._voice = voice
self._rate = _fmt_prosody(rate, "%")
self._volume = _fmt_prosody(volume, "%")
self._pitch = _fmt_prosody(pitch, "Hz")
self._pause = float(sentence_pause_sec)
self._per_sentence = bool(per_sentence)
self._cache: dict[str, SynthResult] = {} # повторные фразы (прощание и т.п.) — 0 c
# --- API -------------------------------------------------------------
def warmup(self) -> None:
"""Прогрев: короткий синтез (резолв DNS, TLS-сессия, соединение с сервисом).
Вызывается ассистентом в фоне при старте — первый реальный ответ
синтезируется за ~0.5 c вместо 2.5–4 c холодного соединения.
"""
try:
self.synthesize("Слушаю вас.")
except Exception:
pass
def stream(self, text: str):
"""Потоковая озвучка: чанки-предложения по мере готовности.
Yields (audio: np.ndarray, samplerate: int) — первый чанк приходит
через ~один сетевой round-trip (~0.5 c), не дожидаясь синтеза всего
текста. Пауза sentence_pause_sec уже вставлена В КОНЕЦ каждого чанка.
"""
sentences = split_sentences(text) or [text]
if len(sentences) == 1:
audio, sr = decode_audio_bytes(self._generate_one(text))
yield audio, sr
return
# все запросы параллельно; отдаём в порядке следования предложений
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=min(len(sentences), 4)) as pool:
for mp3 in pool.map(self._generate_one, sentences):
audio, sr = decode_audio_bytes(mp3)
yield audio, sr
def synthesize(self, text: str) -> SynthResult:
t0 = time.perf_counter()
cached = self._cache.get(text)
if cached is not None:
return SynthResult(
text=cached.text, audio=cached.audio, samplerate=cached.samplerate,
generation_sec=0.0, provider=self.name,
sentence_count=cached.sentence_count,
)
if self._per_sentence:
result = self._synthesize_stitched(text, t0)
else:
result = self._synthesize_single(text, t0)
if len(text) <= 120: # кэшируем только короткие (прощание, отказы)
self._cache[text] = result
return result
def _synthesize_single(self, text: str, t0: float) -> SynthResult:
"""Минимальная латентность: предложения синтезируются ПАРАЛЛЕЛЬНО
(каждый edge-запрос несёт ~1.3 c сетевого round-trip; параллельно —
суммарное время = самый долгий запрос, а не сумма), затем склеиваются
с тишиной sentence_pause_sec между ними."""
sentences = split_sentences(text) or [text]
if len(sentences) == 1:
audio, samplerate = decode_audio_bytes(self._generate_one(text))
return SynthResult(
text=text,
audio=audio,
samplerate=samplerate,
generation_sec=time.perf_counter() - t0,
provider=self.name,
sentence_count=1,
)
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=min(len(sentences), 4)) as pool:
mp3_list = list(pool.map(self._generate_one, sentences))
pieces: List[np.ndarray] = []
samplerate: Optional[int] = None
pause_samples = int(self._pause * 24000) # уточним после первого декода
for i, mp3 in enumerate(mp3_list):
audio, sr = decode_audio_bytes(mp3)
if samplerate is None:
samplerate = sr
pause_samples = int(self._pause * sr)
pieces.append(audio)
if i < len(mp3_list) - 1 and self._pause > 0:
pieces.append(np.zeros(pause_samples, dtype=np.float32))
combined = np.concatenate(pieces) if pieces else np.zeros(0, dtype=np.float32)
return SynthResult(
text=text,
audio=combined,
samplerate=samplerate or 24000,
generation_sec=time.perf_counter() - t0,
provider=self.name,
sentence_count=len(sentences),
)
def _synthesize_stitched(self, text: str, t0: float) -> SynthResult:
"""Каждое предложение отдельным запросом + точная пауза-тишина (медленно)."""
sentences = split_sentences(text) or [text]
pieces: List[np.ndarray] = []
samplerate: Optional[int] = None
for i, sentence in enumerate(sentences):
audio, sr = decode_audio_bytes(self._generate_one(sentence))
if samplerate is None:
samplerate = sr
pieces.append(audio)
if i < len(sentences) - 1 and self._pause > 0:
pieces.append(np.zeros(int(self._pause * sr), dtype=np.float32))
combined = np.concatenate(pieces) if pieces else np.zeros(0, dtype=np.float32)
return SynthResult(
text=text,
audio=combined,
samplerate=samplerate or 24000,
generation_sec=time.perf_counter() - t0,
provider=self.name,
sentence_count=len(sentences),
)
# --- внутреннее ------------------------------------------------------
def _generate_one(self, sentence: str) -> bytes:
"""mp3-байты одного предложения."""
import edge_tts
async def _inner() -> bytes:
com = edge_tts.Communicate(
sentence,
self._voice,
rate=self._rate,
volume=self._volume,
pitch=self._pitch,
)
chunks: List[bytes] = []
async for item in com.stream():
if item["type"] == "audio":
chunks.append(item["data"])
return b"".join(chunks)
return _run_async(_inner())
+79
View File
@@ -0,0 +1,79 @@
"""Провайдер TTS на Piper — полностью офлайн-фолбэк.
Используется, если edge-tts недоступен (сервис Microsoft ломался в прошлом,
а интернет у отца может пропасть). Piper работает локально на CPU.
Установка и голос (однократно):
pip install piper-tts
python -m piper.download_voices ru_RU-dmitri-medium --download-dir models/piper
Пайпер медленнее edge-tts и голоса проще, но работает без сети.
API: PiperVoice.load(path); voice.synthesize(text) -> чанки AudioChunk
(audio_int16_bytes, sample_rate, sample_width, sample_channels).
"""
from __future__ import annotations
import time
from pathlib import Path
from typing import List, Optional
import numpy as np
from .base import SynthResult, TtsProvider
from .text_split import split_sentences
class PiperTtsProvider(TtsProvider):
name = "piper"
def __init__(
self,
model_path: Union[str, Path],
length_scale: float = 1.15, # 1.0 = обычная скорость; 1.15 ≈ −15%
noise_scale: float = 0.667,
noise_w_scale: float = 0.8,
sentence_pause_sec: float = 0.35,
) -> None:
from piper import PiperVoice, SynthesisConfig # ленивый импорт
path = Path(model_path)
if not path.is_file():
raise FileNotFoundError(
f"Модель Piper не найдена: {path}\n"
f"Скачай голос: python -m piper.download_voices ru_RU-dmitri-medium "
f"--download-dir {path.parent}"
)
self._voice = PiperVoice.load(str(path))
self._config = SynthesisConfig(
length_scale=length_scale,
noise_scale=noise_scale,
noise_w_scale=noise_w_scale,
)
self._pause = float(sentence_pause_sec)
def synthesize(self, text: str) -> SynthResult:
t0 = time.perf_counter()
sentences = split_sentences(text) or [text]
pieces: List[np.ndarray] = []
samplerate: Optional[int] = None
for i, sentence in enumerate(sentences):
chunks = list(self._voice.synthesize(sentence, syn_config=self._config))
for ch in chunks:
if samplerate is None:
samplerate = ch.sample_rate
pcm = np.frombuffer(ch.audio_int16_bytes, dtype=np.int16)
pieces.append(pcm.astype(np.float32) / 32768.0)
if i < len(sentences) - 1 and self._pause > 0 and samplerate:
pieces.append(np.zeros(int(self._pause * samplerate), dtype=np.float32))
audio = np.concatenate(pieces) if pieces else np.zeros(0, dtype=np.float32)
return SynthResult(
text=text,
audio=audio,
samplerate=samplerate or 22050,
generation_sec=time.perf_counter() - t0,
provider=self.name,
sentence_count=len(sentences),
)
+107
View File
@@ -0,0 +1,107 @@
"""Тест TTS-модуля (Модуль 3).
Запуск:
python modules/tts/test_tts.py # синтез фразы по умолчанию, сохранить wav
python modules/tts/test_tts.py --play # то же + воспроизвести
python modules/tts/test_tts.py --text "Своё предложение"
python modules/tts/test_tts.py --voice ru-RU-DmitryNeural --play
python modules/tts/test_tts.py --rate -10 # другая скорость
python modules/tts/test_tts.py --provider piper --model models/piper/ru_RU-dmitri-medium.onnx
python modules/tts/test_tts.py --listen # интерактивно: вводишь текст — слышишь
Критерии приёмки Модуля 3:
1. Русская фраза звучит естественно, скорость на ~15% медленнее обычной.
2. Несколько предложений — с заметными паузами между ними.
3. Синтез короткой фразы (до 100 символов) — <= 3 c.
"""
from __future__ import annotations
import argparse
import sys
import time
from pathlib import Path
try:
from modules.audio_io import Player
from modules.audio_io.codec import decode_audio_bytes # noqa: F401 (использование ниже)
from modules.tts import get_provider
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from modules.audio_io import Player
from modules.audio_io.codec import decode_audio_bytes # noqa: F401
from modules.tts import get_provider
SAMPLES_DIR = Path(__file__).resolve().parent / "samples"
DEFAULT_TEXT = (
"Здравствуйте! Это проверка голоса ассистента. "
"Сейчас проверим, насколько понятна речь. Погода сегодня хорошая."
)
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Тест TTS-модуля")
p.add_argument("--text", default=DEFAULT_TEXT, help="текст для синтеза")
p.add_argument("--provider", default="edge", choices=["edge", "piper"])
p.add_argument("--voice", default="ru-RU-SvetlanaNeural",
help="голос edge-tts (ru-RU-DmitryNeural — мужской)")
p.add_argument("--rate", default=-15, help="скорость речи, %% (−15 по ТЗ)")
p.add_argument("--model", default="models/piper/ru_RU-dmitri-medium.onnx",
help="путь к .onnx модели Piper")
p.add_argument("--pause", type=float, default=0.35, help="пауза между предложениями, с")
p.add_argument("--per-sentence", action="store_true",
help="режим «предложение = запрос» (сравнение латентности)")
p.add_argument("--play", action="store_true", help="воспроизвести результат")
p.add_argument("--listen", action="store_true",
help="интерактивно: вводишь текст — слышишь звук")
p.add_argument("--save", default=str(SAMPLES_DIR / "tts_out.wav"))
return p.parse_args()
def speak_once(tts, text: str, play: bool, save: Path) -> None:
result = tts.synthesize(text)
print(result)
save = Path(save)
save.parent.mkdir(parents=True, exist_ok=True)
import soundfile as sf
sf.write(str(save), result.audio, result.samplerate)
print(f"Сохранено: {save} ({len(result.audio) / result.samplerate:.1f} c)")
if play:
player = Player()
print("Воспроизвожу...")
player.play(result.audio, result.samplerate, blocking=True)
def listen_loop(tts) -> None:
player = Player()
print("Режим «слушать»: вводи текст, Enter — озвучить, пустая строка — выход.")
while True:
text = input("текст> ").strip()
if not text:
return
result = tts.synthesize(text)
print(f" синтез {result.generation_sec:.2f} c, аудио {result.duration_sec:.1f} c")
player.play(result.audio, result.samplerate, blocking=True)
def main() -> int:
args = parse_args()
kwargs = {"sentence_pause_sec": args.pause}
if args.provider == "edge":
kwargs.update(voice=args.voice, rate=int(args.rate), per_sentence=args.per_sentence)
else:
kwargs.update(model_path=args.model)
tts = get_provider(args.provider, **kwargs)
if args.listen:
listen_loop(tts)
else:
speak_once(tts, args.text, args.play, args.save)
return 0
if __name__ == "__main__":
sys.exit(main())
+38
View File
@@ -0,0 +1,38 @@
"""Разбиение текста на предложения — для вставки увеличенных пауз.
Пожилой человек воспринимает речь медленнее: между предложениями
добавляем тишину (задаётся в провайдере как sentence_pause_sec).
Минимальная защита от сокращений («т.д.», «т.п.» и т.п.).
"""
from __future__ import annotations
import re
from typing import List
# Сокращения, где точка — не конец предложения (после точки может быть «д»)
_ABBREV = ["т.д", "т.п", "т.е", "т.к", "т.н", "др", "пр", "г", "ул", "кв", "руб", "мин"]
_SENT_SPLIT = re.compile(r"(?<=[.!?…])\s+")
def split_sentences(text: str) -> List[str]:
"""Разбить текст на предложения (простая эвристика, без NLP)."""
text = text.strip()
if not text:
return []
# Временно прячем точки в сокращениях
protected = text
for abbr in _ABBREV:
protected = protected.replace(f" {abbr}.", f" {abbr}\x01")
parts = [p.strip() for p in _SENT_SPLIT.split(protected) if p.strip()]
# Возвращаем точки на место
for abbr in _ABBREV:
parts = [p.replace(f"{abbr}\x01", f"{abbr}.") for p in parts]
return parts
def insert_pauses(sentences: List[str]) -> str:
"""Склеить предложения с «длинными» точками (… ) для естественных пауз в одном запросе."""
return "… ".join(sentences) if len(sentences) > 1 else (sentences[0] if sentences else "")