97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""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.
|
|
|
|
Through `pacat`, not sounddevice: PortAudio only sees raw ALSA devices,
|
|
so a Bluetooth or PipeWire sink is invisible to it and the audio goes
|
|
somewhere nobody can hear. pacat talks to the audio server, so it
|
|
follows the system's default sink like every other player.
|
|
"""
|
|
import shutil
|
|
import subprocess
|
|
|
|
if shutil.which("pacat") is None:
|
|
return _play_sounddevice(chunks, sample_rate, should_stop)
|
|
|
|
proc = subprocess.Popen(
|
|
["pacat", "--rate", str(sample_rate), "--channels", "1",
|
|
"--format", "s16le", "--latency-msec", "60"],
|
|
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
try:
|
|
for chunk in chunks:
|
|
if should_stop():
|
|
return False
|
|
proc.stdin.write(chunk)
|
|
proc.stdin.close()
|
|
proc.wait(timeout=10)
|
|
except (BrokenPipeError, OSError):
|
|
return False
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
return True
|
|
|
|
|
|
def _play_sounddevice(chunks: Iterable[bytes], sample_rate: int,
|
|
should_stop: Callable[[], bool]) -> bool:
|
|
"""Fallback for a machine without pacat."""
|
|
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)])
|