app.py

"""AIRWAVES — Space entrypoint. Conduct a live set with your bare hands.

The static SPA (web/) is served at / and does ALL the work that matters in the
browser: MediaPipe hand tracking + the Web Audio bend engine, zero GPU, camera
never leaves the device. The ONLY server/GPU touch is GET /api/hype, which
synthesizes the VoxCPM2 hype-man voice bank ONCE inside a single @spaces.GPU
window, caches it, and serves base64 WAVs to every visitor instantly.

The "her" pattern: deterministic browser code owns every audio fact (beat,
filter Hz, gains, loop points); the tiny model owns only the hype-man's voice.

Run:       python app.py
Backends:  AIRWAVES_BACKEND = mock | zerogpu   (default mock)
"""
from __future__ import annotations

import asyncio
import base64
import logging
import os
import threading

# ZeroGPU: `spaces` must be imported before torch anywhere in the process, and
# TORCHDYNAMO_DISABLE must precede any torch import. No-op off the GPU backend.
if os.environ.get("AIRWAVES_BACKEND", "").strip().lower() in ("zerogpu", "zero-gpu", "vox", "voxcpm"):
    os.environ.setdefault("TORCHDYNAMO_DISABLE", "1")
    try:
        import spaces  # noqa: F401
    except ImportError:
        pass

from pathlib import Path

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles

from mind import backends as mind_backends

log = logging.getLogger("airwaves")

ROOT = Path(__file__).resolve().parent
WEB_DIR = ROOT / "web"

# The hype-man lines — KEYS MUST MATCH web/hype.js HYPE_LINES.
HYPE_LINES = {
    "hello": "Manos arriba — let's go!",
    "build": "Build it up… here it comes…",
    "drop": "Dale! Drop it!",
    "sweep": "Bring it up!",
    "mix": "Mix it — vamos!",
}
VOICE_ID = "club_mc"


def create_app() -> FastAPI:
    """gr.Server (the proven 'her' idiom): a FastAPI app the sdk:gradio runtime
    serves natively. API routes are registered BEFORE launch; the web/ static
    mount is deferred until AFTER launch (see _mount_static)."""
    try:
        import gradio as gr

        app: FastAPI = gr.Server(title="AIRWAVES")
    except Exception:  # pragma: no cover — gradio always present in prod
        app = FastAPI(title="AIRWAVES", docs_url=None, redoc_url=None)

    app.state.bank = None
    app.state.bank_lock = threading.Lock()

    @app.get("/api/health")
    async def health() -> JSONResponse:
        return JSONResponse(
            {
                "ok": True,
                "backend": os.environ.get("AIRWAVES_BACKEND", "mock"),
                "bank_ready": app.state.bank is not None,
            }
        )

    @app.get("/api/hype")
    async def hype() -> JSONResponse:
        """The VoxCPM2 hype-man bank: generated once (one GPU window), cached
        forever, served instantly thereafter. On any failure return an empty
        bank — the frontend falls back to browser speech, the show goes on."""
        if app.state.bank is None:
            try:
                app.state.bank = await asyncio.to_thread(_build_bank, app)
            except Exception:
                log.exception("hype bank synthesis failed; client falls back to speech")
                return JSONResponse({"bank": [], "error": "voice unavailable"})
        return JSONResponse({"bank": app.state.bank, "voice": "openbmb/VoxCPM2"})

    def _mount_static() -> None:
        """Root static mount, deferred until AFTER launch: a '/' catch-all
        registered before gr.Server.launch() shadows the /gradio_api routes and
        the launch self-check 404s. Evict only gradio's own '/' SPA route."""
        if WEB_DIR.is_dir():
            app.router.routes[:] = [
                r for r in app.router.routes if getattr(r, "path", None) != "/"
            ]
            app.mount("/", StaticFiles(directory=str(WEB_DIR), html=True), name="web")
        else:  # pragma: no cover
            log.warning("web/ not found at %s; serving API only", WEB_DIR)

            @app.get("/", include_in_schema=False)
            async def root_placeholder() -> JSONResponse:
                return JSONResponse({"ok": True, "hint": "frontend not built"})

    app.state.mount_static = _mount_static
    if not hasattr(app, "launch"):  # plain-FastAPI dev path mounts immediately
        _mount_static()
    return app


def _build_bank(app: FastAPI) -> list:
    """Synthesize all hype lines once (thread-safe); return [{key, wav_b64}]."""
    with app.state.bank_lock:
        if app.state.bank is not None:
            return app.state.bank
        wavs = mind_backends.synth_bank(HYPE_LINES, VOICE_ID)
        return [
            {"key": k, "wav_b64": base64.b64encode(v).decode("ascii")}
            for k, v in wavs.items()
        ]


def _serve(application: FastAPI) -> None:
    port = int(os.environ.get("PORT", os.environ.get("GRADIO_SERVER_PORT", 7860)))
    log.info("AIRWAVES listening on http://0.0.0.0:%d", port)
    if hasattr(application, "launch"):
        application.launch(
            server_name="0.0.0.0",
            server_port=port,
            show_error=False,
            prevent_thread_lock=True,
        )
        mount_static = getattr(application.state, "mount_static", None)
        if mount_static is not None:
            mount_static()
        # Warm the voice model off the serving thread so the first /api/hype is fast.
        threading.Thread(target=mind_backends.warm_voice, daemon=True).start()
        import time as _time

        while True:
            _time.sleep(3600)
    else:  # pragma: no cover — gradio-less dev fallback
        import uvicorn

        uvicorn.run(application, host="0.0.0.0", port=port)


if __name__ == "__main__" or os.environ.get("SPACE_ID"):
    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s"
    )
    _serve(create_app())