Plan 3: main.py composition root (audio + state + wakeword + session + playback wired)

client/main.py

@@ -1,11 +1,166 @@
-"""Smart Assistant client entry point.
+"""Smart Assistant Pi client entry point.
 
-The composition root lives in main(); see Plan 3 Task 11 for the full wiring.
+CRITICAL: set PA_ALSA_PLUGHW BEFORE importing sounddevice (see findings.md §1).
 """
+import os
+os.environ.setdefault("PA_ALSA_PLUGHW", "1")
+
+import asyncio
+import queue
+import signal
+import sys
+import threading
+from pathlib import Path
+
+import numpy as np
+import sounddevice as sd  # safe to import now
+
+from client.audio import (
+    AudioStreams,
+    FRAME_SAMPLES,
+    resample_24k_to_16k,
+)
+from client.config import ConfigMissingError, load_config
+from client.log import get_logger
+from client.playback import PlaybackWorker
+from client.session import Session
+from client.state import StateMachine
+from client.wakeword import WakewordDetector
+
+_log = get_logger("main")
+
+ASSISTANT_STATE_DIR = Path(os.environ["HOME"]) / "assistant" / "state"
+WAKE_WAV = ASSISTANT_STATE_DIR / "wake.wav"
+SLEEP_WAV = ASSISTANT_STATE_DIR / "sleep.wav"
+
+
+def _build_bridge_thread(
+    mic_queue: queue.Queue,
+    sm: StateMachine,
+    wakeword: WakewordDetector,
+    session: Session,
+    playback: PlaybackWorker,
+    stop_event: threading.Event,
+) -> threading.Thread:
+    def run():
+        while not stop_event.is_set():
+            try:
+                pcm16_bytes = mic_queue.get(timeout=0.25)
+            except queue.Empty:
+                continue
+            if sm.wakeword_enabled:
+                frame = np.frombuffer(pcm16_bytes, dtype=np.int16)
+                if frame.size != FRAME_SAMPLES:
+                    continue
+                feed = resample_24k_to_16k(frame)
+                if wakeword.predict(feed):
+                    if sm.wake_fired():
+                        if WAKE_WAV.exists():
+                            playback.enqueue_wav(WAKE_WAV, "wake")
+                        else:
+                            # Even with no wav, fire the done marker so the
+                            # playback_done(for=wake) envelope is sent.
+                            playback.enqueue_done_marker("wake")
+                        session.call_soon_threadsafe_send_wake()
+            elif sm.uplink_enabled:
+                session.call_soon_threadsafe_send_binary(pcm16_bytes)
+            # else: ASSISTANT_SPEAKING / WAKE_PENDING / IDLE_PENDING -> drop
+    t = threading.Thread(target=run, name="bridge", daemon=True)
+    return t
+
+
+def _make_callbacks(sm: StateMachine, playback: PlaybackWorker):
+    class _CB:
+        def on_session_started(self, conversation_id: str) -> None:
+            _log.info("session_started conversation_id=%s", conversation_id)
+            sm.session_started()
+
+        def on_assistant_audio(self, pcm16: bytes) -> None:
+            sm.assistant_audio_arrived()
+            playback.enqueue_audio(pcm16)
+
+        def on_assistant_done(self) -> None:
+            _log.info("assistant_done")
+            sm.assistant_done()
+
+        def on_session_ended(self, reason: str) -> None:
+            _log.info("session_ended reason=%s", reason)
+            sm.session_ended()
+            if SLEEP_WAV.exists():
+                playback.enqueue_wav(SLEEP_WAV, "sleep")
+            else:
+                playback.enqueue_done_marker("sleep")
+    return _CB()
 
 
 def main() -> None:
-    raise SystemExit("client.main: not yet implemented (see Plan 3 Task 11).")
+    try:
+        cfg = load_config()
+    except ConfigMissingError as exc:
+        raise SystemExit(str(exc)) from exc
+
+    _log.info("device_id=%s backend=%s", cfg.device_id, cfg.backend_ws)
+
+    mic_queue: queue.Queue = queue.Queue(maxsize=64)
+    streams = AudioStreams.open(sd, mic_queue=mic_queue)
+    streams.start()
+
+    sm = StateMachine()
+
+    session_ref: list[Session] = []
+
+    def on_playback_done(label: str) -> None:
+        _log.info("playback_done for=%s", label)
+        if not session_ref:
+            return
+        session = session_ref[0]
+        session.call_soon_threadsafe_send_playback_done(label)
+        if label == "sleep":
+            sm.sleep_played()
+
+    playback = PlaybackWorker(streams, on_done=on_playback_done)
+    playback.start()
+
+    wakeword = WakewordDetector()
+
+    callbacks = _make_callbacks(sm, playback)
+    session = Session(
+        backend_ws=cfg.backend_ws,
+        device_token=cfg.device_token,
+        device_id=cfg.device_id,
+        callbacks=callbacks,
+    )
+    session_ref.append(session)
+
+    def set_volume_handler(args: dict) -> dict:
+        level = int(args.get("level", -1))
+        playback.set_volume(level)
+        return {"ok": True, "level": level}
+
+    session.register_tool("set_volume", set_volume_handler)
+
+    stop_event = threading.Event()
+    bridge = _build_bridge_thread(mic_queue, sm, wakeword, session, playback, stop_event)
+    bridge.start()
+
+    def _shutdown(signum, _frame):
+        _log.info("signal %d received; shutting down", signum)
+        stop_event.set()
+        try:
+            playback.stop()
+        finally:
+            streams.close()
+        sys.exit(0)
+
+    signal.signal(signal.SIGTERM, _shutdown)
+    signal.signal(signal.SIGINT, _shutdown)
+
+    try:
+        asyncio.run(session.run_forever())
+    finally:
+        stop_event.set()
+        playback.stop()
+        streams.close()
 
 
 if __name__ == "__main__":