From 8662dbe8a793bcb2f1de1eadf43199600f1b6697 Mon Sep 17 00:00:00 2001 From: Assistant builder Date: Thu, 11 Jun 2026 22:52:01 +0000 Subject: [PATCH] Plan 3: main.py composition root (audio + state + wakeword + session + playback wired) --- client/main.py | 161 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 158 insertions(+), 3 deletions(-) diff --git a/client/main.py b/client/main.py index e4e76c7..951840d 100644 --- a/client/main.py +++ b/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__":