82 lines
4.2 KiB
Python
82 lines
4.2 KiB
Python
"""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()}», чтобы начать диалог." |