Bolt Assistant: wake-phrase voice assistant for Bolt

bolt_assistant/audio.py

@@ -0,0 +1,61 @@
+"""The default microphone in, the default speaker out, and two short chimes.
+
+sounddevice (PortAudio) for both directions; MIC_DEVICE / SPEAKER_DEVICE in
+.env pick something other than the system default. The chimes are made
+here from sine waves — no sound files to ship.
+"""
+
+from __future__ import annotations
+
+import array
+import math
+from typing import Callable, Iterable
+
+from . import config
+
+
+def open_mic(on_frame: Callable[[bytes], None]):
+    """Start the microphone: 16 kHz mono int16, one 80 ms frame per call to
+    *on_frame*, from PortAudio's thread. Returns the stream (stop()/close())."""
+    import sounddevice as sd
+
+    stream = sd.RawInputStream(
+        samplerate=config.SAMPLE_RATE, channels=1, dtype="int16",
+        blocksize=config.FRAME_SAMPLES, device=config.MIC_DEVICE,
+        callback=lambda data, _frames, _time, _status: on_frame(bytes(data)),
+    )
+    stream.start()
+    return stream
+
+
+def play(chunks: Iterable[bytes], sample_rate: int = config.TTS_SAMPLE_RATE,
+         should_stop: Callable[[], bool] = lambda: False) -> bool:
+    """Play PCM16 mono chunks as they arrive. False if stopped early."""
+    import sounddevice as sd
+
+    with sd.RawOutputStream(samplerate=sample_rate, channels=1, dtype="int16",
+                            device=config.SPEAKER_DEVICE) as out:
+        for chunk in chunks:
+            if should_stop():
+                out.abort()
+                return False
+            out.write(chunk)
+    return True
+
+
+def tone(notes, sample_rate: int = config.TTS_SAMPLE_RATE, volume: float = 0.18) -> bytes:
+    """(frequency Hz, seconds) notes as PCM16, each with a quick fade in and
+    out so it doesn't click."""
+    samples = array.array("h")
+    for freq, seconds in notes:
+        count = int(sample_rate * seconds)
+        fade = max(1, int(sample_rate * 0.012))
+        for i in range(count):
+            envelope = min(1.0, i / fade, (count - i) / fade)
+            samples.append(int(32767 * volume * envelope * math.sin(2 * math.pi * freq * i / sample_rate)))
+    return samples.tobytes()
+
+
+# Rising two notes: "I'm listening". Falling: "didn't work".
+WAKE_CHIME = tone([(880, 0.07), (1320, 0.09)])
+ERROR_CHIME = tone([(520, 0.10), (390, 0.14)])