93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Listen for the 'alexa' wakeword via openwakeword. Beep on detection.
|
|
|
|
Pass criteria (per spec):
|
|
- "alexa" at normal volume from ~1 m triggers within ~1 s (>= 8/10 attempts).
|
|
- During ~1 minute of unrelated speech, <= ~1 false positive.
|
|
- Python process < 50% of one core (check with htop).
|
|
"""
|
|
import os
|
|
os.environ.setdefault("PA_ALSA_PLUGHW", "1")
|
|
|
|
import sys
|
|
import time
|
|
|
|
import numpy as np
|
|
import sounddevice as sd
|
|
import openwakeword.utils
|
|
from openwakeword.model import Model
|
|
|
|
SAMPLE_RATE = 16_000
|
|
FRAME_SAMPLES = 1280
|
|
CHANNELS = 1
|
|
DTYPE = "int16"
|
|
WAKEWORD = "alexa"
|
|
THRESHOLD = 0.5
|
|
COOLDOWN_S = 1.0
|
|
BEEP_FREQ_HZ = 880.0
|
|
BEEP_DURATION_S = 0.2
|
|
|
|
|
|
def find_usb_device() -> int:
|
|
for idx, dev in enumerate(sd.query_devices()):
|
|
name = dev["name"].lower()
|
|
if "usb" in name and dev["max_input_channels"] >= 1:
|
|
return idx
|
|
print("USB audio device not found. Devices:", file=sys.stderr)
|
|
print(sd.query_devices(), file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def beep(device: int) -> None:
|
|
t = np.arange(int(SAMPLE_RATE * BEEP_DURATION_S)) / SAMPLE_RATE
|
|
tone = (0.3 * np.sin(2 * np.pi * BEEP_FREQ_HZ * t)).astype(np.float32)
|
|
sd.play(tone, samplerate=SAMPLE_RATE, device=device)
|
|
sd.wait()
|
|
|
|
|
|
def main() -> int:
|
|
print("Ensuring openwakeword onnx models are downloaded (idempotent)...")
|
|
openwakeword.utils.download_models()
|
|
|
|
print("Loading openwakeword model...")
|
|
t0 = time.monotonic()
|
|
model = Model(wakeword_models=[WAKEWORD], inference_framework="onnx")
|
|
print(f"Model loaded in {time.monotonic() - t0:.2f}s.")
|
|
|
|
device = find_usb_device()
|
|
print(
|
|
f"Listening on device {device} ({sd.query_devices(device)['name']!r}). "
|
|
f"Say '{WAKEWORD}'. Ctrl-C to exit."
|
|
)
|
|
|
|
last_trigger = 0.0
|
|
start = time.monotonic()
|
|
|
|
with sd.InputStream(
|
|
samplerate=SAMPLE_RATE,
|
|
channels=CHANNELS,
|
|
dtype=DTYPE,
|
|
device=device,
|
|
blocksize=FRAME_SAMPLES,
|
|
) as stream:
|
|
while True:
|
|
frame, overflowed = stream.read(FRAME_SAMPLES)
|
|
if overflowed:
|
|
print("[status] input overflow", file=sys.stderr)
|
|
mono = frame[:, 0]
|
|
scores = model.predict(mono)
|
|
score = float(scores.get(WAKEWORD, 0.0))
|
|
now = time.monotonic()
|
|
if score >= THRESHOLD and (now - last_trigger) >= COOLDOWN_S:
|
|
print(f"DETECTED {WAKEWORD} score={score:.3f} t={now - start:.1f}s")
|
|
last_trigger = now
|
|
beep(device)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except KeyboardInterrupt:
|
|
print("\nbye")
|
|
sys.exit(0)
|