diff --git a/client/playback.py b/client/playback.py new file mode 100644 index 0000000..f1eacff --- /dev/null +++ b/client/playback.py @@ -0,0 +1,117 @@ +"""Playback worker thread + wake/sleep cue + software volume. + +Items pushed to the queue: + ("audio", bytes) - PCM16 LE 24 kHz mono data to write + ("wav", (Path, label)) - WAV file to read and play; triggers on_done(label) after + ("done_marker", str) - synthetic marker: after this is popped, call on_done(label) + ("stop", None) - shut the worker down + +Volume is applied as a software gain in [0..1.0] on int16 buffers because the +USB Speaker Phone has no hardware playback volume control (only `PCM Playback +Switch`, a mute). +""" +import queue +import threading +import wave +from pathlib import Path +from typing import Callable, Tuple + +import numpy as np + +from client.audio import SAMPLE_RATE +from client.log import get_logger + +_log = get_logger("playback") + + +class PlaybackWorker: + def __init__(self, audio_streams, on_done: Callable[[str], None]): + """ + `audio_streams.write(pcm16_bytes)` is the blocking write into the + persistent OutputStream. `on_done(label)` is invoked when a tagged + cue (e.g. "wake", "sleep") finishes playing. + """ + self._streams = audio_streams + self._on_done = on_done + self._q: "queue.Queue[Tuple[str, object]]" = queue.Queue() + self._volume = 1.0 + self._thread = threading.Thread(target=self._run, name="playback", daemon=True) + self._running = False + + def start(self) -> None: + self._running = True + self._thread.start() + + def stop(self) -> None: + self._running = False + self._q.put(("stop", None)) + self._thread.join(timeout=2.0) + + def set_volume(self, level: int) -> None: + """Set software gain. `level` is 0..100.""" + if not 0 <= level <= 100: + raise ValueError("level must be 0..100") + self._volume = level / 100.0 + _log.info("volume set to %d%%", level) + + def enqueue_audio(self, pcm16_bytes: bytes) -> None: + self._q.put(("audio", pcm16_bytes)) + + def enqueue_wav(self, path: Path, label: str) -> None: + """Queue a WAV cue. When fully drained, on_done(label) fires.""" + self._q.put(("wav", (path, label))) + + def enqueue_done_marker(self, label: str) -> None: + """Fire on_done(label) once this marker is processed (after all prior audio).""" + self._q.put(("done_marker", label)) + + def _apply_gain(self, pcm16: bytes) -> bytes: + if self._volume == 1.0: + return pcm16 + arr = np.frombuffer(pcm16, dtype=np.int16).astype(np.int32) + arr = (arr * int(self._volume * 1024)) >> 10 + arr = np.clip(arr, -32768, 32767).astype(np.int16) + return arr.tobytes() + + def _play_wav(self, path: Path, label: str) -> None: + if not path.exists(): + _log.info("wav %s not present at %s; skipping", label, path) + self._on_done(label) + return + try: + with wave.open(str(path), "rb") as w: + if w.getframerate() != SAMPLE_RATE or w.getnchannels() != 1 or w.getsampwidth() != 2: + _log.warning( + "wav %s is %dHz/%dch/%dB; expected %dHz/mono/16-bit", + path, w.getframerate(), w.getnchannels(), w.getsampwidth() * 8, SAMPLE_RATE, + ) + self._on_done(label) + return + data = w.readframes(w.getnframes()) + except wave.Error as exc: + _log.warning("could not read wav %s: %s", path, exc) + self._on_done(label) + return + chunk = 3840 + for i in range(0, len(data), chunk): + self._streams.write(self._apply_gain(data[i:i + chunk])) + self._on_done(label) + + def _run(self) -> None: + while self._running: + try: + kind, payload = self._q.get(timeout=0.5) + except queue.Empty: + continue + try: + if kind == "stop": + return + if kind == "audio": + self._streams.write(self._apply_gain(payload)) + elif kind == "wav": + path, label = payload + self._play_wav(path, label) + elif kind == "done_marker": + self._on_done(payload) + except Exception as exc: # pragma: no cover - hardware path + _log.exception("playback worker error: %s", exc)