Always-on wake-word listener (local mic → Talk).
Backends:
- phrase_gate — energy/VAD + tiny Whisper STT + regex phrases (default)
- openwakeword — ML scores on PCM (pre-trained hey_jarvis or custom ONNX)
- auto — openWakeWord if installed, else phrase_gate
Default action: emit talk_open on local_trigger (open Talk orb).
Classes
WakeWordListener(
on_wake: Callable[[str], None],
*,
cfg: WakeListenConfig | None = None,
once: bool = False,
device: int | None = None,
debug: bool = False,
transcribe: TranscribeFn | None = None,
)
Mic loop: openWakeWord ML scores and/or STT phrase gate → callback.
Source code in src/diapason/speech/wakeword.py
| def __init__(
self,
on_wake: Callable[[str], None],
*,
cfg: WakeListenConfig | None = None,
once: bool = False,
device: int | None = None,
debug: bool = False,
transcribe: TranscribeFn | None = None,
) -> None:
self._on_wake = on_wake
self._cfg = cfg or WakeListenConfig()
self._once = once
self._device = device
self._debug = debug
self._transcribe = transcribe or default_transcribe_pcm
self._pattern = build_wake_pattern(self._cfg.phrases)
self._stop = threading.Event()
self._thread: Optional[threading.Thread] = None
self._fired = False
self._last_fire = 0.0
self._resolved_backend = "phrase_gate"
|
Methods:
check_text(text: str) -> bool
Unit-test / manual path: return True and invoke callback if wake.
Source code in src/diapason/speech/wakeword.py
| def check_text(self, text: str) -> bool:
"""Unit-test / manual path: return True and invoke callback if wake."""
if not has_wake_word(text, pattern=self._pattern):
return False
return self._fire(text)
|
Functions:
default_transcribe_pcm(pcm: bytes, sample_rate: int) -> str
Best-effort local STT (faster-whisper tiny). Returns '' if unavailable.
Source code in src/diapason/speech/wakeword.py
| def default_transcribe_pcm(pcm: bytes, sample_rate: int) -> str:
"""Best-effort local STT (faster-whisper tiny). Returns '' if unavailable."""
if not pcm:
return ""
try:
import io
import wave
from faster_whisper import WhisperModel
buf = io.BytesIO()
with wave.open(buf, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(sample_rate)
wf.writeframes(pcm)
buf.seek(0)
model = getattr(default_transcribe_pcm, "_model", None)
if model is None:
model = WhisperModel("tiny", device="cpu", compute_type="int8")
setattr(default_transcribe_pcm, "_model", model)
segments, _info = model.transcribe(buf, language=None, beam_size=1)
return " ".join(s.text for s in segments).strip()
except Exception:
logger.debug("wake STT unavailable", exc_info=True)
return ""
|
wake_config_from_toml(
*,
cooldown: float | None = None,
min_rms: float | None = None,
phrases: str | None = None,
backend: str | None = None,
) -> WakeListenConfig
Build WakeListenConfig merging CLI overrides with [speech.wakeword].
Source code in src/diapason/speech/wakeword.py
| def wake_config_from_toml(
*,
cooldown: float | None = None,
min_rms: float | None = None,
phrases: str | None = None,
backend: str | None = None,
) -> WakeListenConfig:
"""Build WakeListenConfig merging CLI overrides with [speech.wakeword]."""
cfg = WakeListenConfig()
try:
from diapason.core.config import load_config
w = load_config().speech.wakeword
cfg.cooldown_s = float(w.cooldown_s)
cfg.phrases = phrases_from_config(w.phrases)
cfg.action = (w.action or "talk").strip() or "talk"
cfg.backend = (w.backend or "phrase_gate").strip() or "phrase_gate"
cfg.sensitivity = float(w.sensitivity)
cfg.model_path = (w.model_path or "").strip()
except Exception:
logger.debug("wake config from toml unavailable", exc_info=True)
if cooldown is not None:
cfg.cooldown_s = float(cooldown)
if min_rms is not None:
cfg.min_rms = float(min_rms)
if phrases:
cfg.phrases = phrases_from_config(phrases)
if backend:
cfg.backend = backend.strip()
return cfg
|