watch_sceen_ai/wsai/__main__.py

"""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 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("--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.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()