"""Entry point. python -m wsai # mock pipeline (no deps, no keys) — runs a demo python -m wsai --voice # eyes-free voice loop demo (STT -> Brain -> TTS) python -m wsai --dashboard # live status website + a short voice demo, then idle python -m wsai --live # capture this screen + Claude eyes/brain python -m wsai --env # build from WSAI_* environment variables The mock run is bounded (a few frames + a scripted conversation) so it exits on its own; --dashboard keeps the site open after a short demo; --live/--env run until Ctrl-C. """ from __future__ import annotations import argparse import asyncio import logging import os import socket from .config import Settings from .factory import build from .monitor import Monitor def _lan_ip() -> str: try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except OSError: return "127.0.0.1" async def _run( settings: Settings, demo: bool, monitor: Monitor | None, *, demo_loop: bool = False, keep_dashboard_open: bool = False, ) -> None: if demo: # Bounded demo so CI / a quick check terminates. from .backends.mock import MockFrameSource, MockSTT pipe = build(settings, monitor=monitor) if pipe.source is not None: # keep eyes-free configs eyes-free pipe.source = MockFrameSource(interval=0.3, limit=4) if pipe.stt is not None: pipe.stt = MockSTT(interval=2.0 if demo_loop else 0.4, loop=demo_loop) await pipe.run() if keep_dashboard_open: # Keep the status page alive without generating fake conversations # forever. The previous default looped mock STT indefinitely, which # made the dashboard look like it had heard thousands of real users. if monitor is not None: monitor.set_status(running=True, listening=False) monitor.log("info", "mock 데모 완료 — 실제 음성 파이프라인 연결 대기") await asyncio.Event().wait() return await build(settings, monitor=monitor).run() def _run_stt_test(host: str, port: int) -> None: """Serve the dashboard with a real GPU STT backend so a human can test recognition from the browser (record mic or upload an audio file). No mock conversation loop — the page just hosts the recognition test.""" import time from .backends.whisper import WhisperSTT from .dashboard import Dashboard from .monitor import Monitor monitor = Monitor() stt = WhisperSTT() dash = Dashboard(monitor, host=host, port=port, stt=stt) dash.start() monitor.set_components({"source": "none", "vision": "none", "stt": "whisper", "brain": "none", "tts": "none"}) monitor.set_status(running=True, listening=False) monitor.log("info", "STT 인식 테스트 서버 시작 — GPU 워밍업 중…") print("\n STT 워밍업 중… (모델 로드 + CUDA 예열)") dash.warm() # load + warm the GPU worker so the first recognition is instant dev = getattr(stt, "resolved_device", None) or "?" monitor.log("info", f"STT 준비 완료 (device={dev}). 녹음/파일 업로드로 인식하세요.") shown = host if host not in ("0.0.0.0", "") else _lan_ip() print(f"\n 음성 인식 테스트 사이트: http://{shown}:{port} (STT device: {dev})") print(f" (로컬: http://127.0.0.1:{port} )\n") try: while True: time.sleep(3600) except KeyboardInterrupt: pass finally: dash.stop() def _run_voice_server(host: str, port: int) -> None: """Serve the STT+TTS voice-turn endpoint that the Discord bot (dave/bot.mjs) calls: it POSTs a captured utterance wav and gets back the reply wav to play into the voice channel. Both STT and TTS run on the GPU and are pre-warmed. The same page also shows the live turn feed. Echo mode for now (the reply is what was heard); the Claude brain can be added as the next slice.""" import time from .backends.melo import MeloTTS from .backends.whisper import WhisperSTT from .dashboard import Dashboard from .monitor import Monitor # Real Claude brain (think + reply). If it can't be constructed (no anthropic # package / no Claude auth), fall back to echo so the loop still works. brain = None brain_name = "echo" if os.environ.get("WSAI_BRAIN", "claude").lower() not in ("none", "echo"): try: from .backends.claude import ClaudeBrain model = os.environ.get("WSAI_BRAIN_MODEL", "claude-sonnet-4-5") brain = ClaudeBrain(model=model) brain_name = "claude" except Exception as exc: # noqa: BLE001 logging.getLogger("wsai").warning("brain disabled (echo fallback): %s", exc) monitor = Monitor() stt = WhisperSTT() tts = MeloTTS() dash = Dashboard(monitor, host=host, port=port, stt=stt, tts=tts, brain=brain) dash.start() monitor.set_components({"source": "none", "vision": "none", "stt": "whisper", "brain": brain_name, "tts": "melo"}) monitor.set_status(running=True, listening=False) monitor.log("info", "디스코드 음성 서버 시작 — STT+TTS GPU 워밍업 중…") print("\n STT+TTS 워밍업 중… (모델 로드 + CUDA 예열)") dash.warm() sdev = getattr(stt, "resolved_device", None) or "?" monitor.set_status(listening=True) monitor.log("info", f"음성 서버 준비 완료 (STT device={sdev}). 디스코드 봇 연결 대기.") shown = host if host not in ("0.0.0.0", "") else _lan_ip() print(f"\n 음성 서버 준비 완료 (STT device: {sdev}, 두뇌: {brain_name})") print(f" 대시보드/상태: http://{shown}:{port}") print(f" 봇 연결 엔드포인트: http://127.0.0.1:{port}/api/voice-turn\n") try: while True: time.sleep(3600) except KeyboardInterrupt: pass finally: dash.stop() def main() -> None: ap = argparse.ArgumentParser(prog="wsai") ap.add_argument("--voice", action="store_true", help="eyes-free voice loop (STT -> Brain -> TTS)") ap.add_argument("--dashboard", action="store_true", help="serve the live status website") ap.add_argument("--dashboard-loop-demo", action="store_true", help="keep generating mock demo utterances forever (off by default)") ap.add_argument("--stt-test", action="store_true", help="serve the dashboard with a live GPU STT recognition test") ap.add_argument("--voice-server", action="store_true", help="serve STT+TTS voice-turn endpoint for the Discord bot (dave/bot.mjs)") ap.add_argument("--live", action="store_true", help="capture screen + Claude backends") ap.add_argument("--env", action="store_true", help="build from WSAI_* env vars") ap.add_argument("--port", type=int, default=int(os.environ.get("WSAI_DASHBOARD_PORT", "8787")), help="dashboard port (default 8787, or WSAI_DASHBOARD_PORT)") ap.add_argument("--host", default=os.environ.get("WSAI_DASHBOARD_HOST", "0.0.0.0"), help="dashboard bind host (default 0.0.0.0)") ap.add_argument("-v", "--verbose", action="store_true") args = ap.parse_args() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(levelname)s %(name)s: %(message)s", ) if args.stt_test: _run_stt_test(args.host, args.port) return if args.voice_server: _run_voice_server(args.host, args.port) return if args.dashboard: # Default to the eyes-free voice preset for the demo; env can override. settings = Settings.from_env() if args.env else Settings.voice() demo = not args.env elif args.voice: settings, demo = Settings.voice(), True elif args.live: settings, demo = Settings.live(), False elif args.env: settings, demo = Settings.from_env(), False else: settings, demo = Settings.mock(), True monitor: Monitor | None = None dash = None if args.dashboard: from .dashboard import Dashboard monitor = Monitor() dash = Dashboard(monitor, host=args.host, port=args.port) dash.start() shown = args.host if args.host not in ("0.0.0.0", "") else _lan_ip() print(f"\n 실시간 상태 사이트: http://{shown}:{args.port}") print(f" (로컬: http://127.0.0.1:{args.port} )\n") try: asyncio.run( _run( settings, demo, monitor, demo_loop=args.dashboard and args.dashboard_loop_demo, keep_dashboard_open=args.dashboard and demo and not args.dashboard_loop_demo, ) ) except KeyboardInterrupt: pass finally: if dash is not None: dash.stop() if __name__ == "__main__": main()