A3_text_to_speach/backend/main.py

"""Application entry point.

    python backend/main.py      ->  http://localhost:8000

Serves the dashboard, the JSON API and the WebSocket from one process, so there
is nothing to orchestrate on the demo machine.
"""

from __future__ import annotations

import logging
import os
import sys
from contextlib import asynccontextmanager
from pathlib import Path

# Allow `python backend/main.py` as well as `python -m backend.main`.
if __package__ in (None, ""):  # pragma: no cover
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from fastapi import FastAPI, Request  # noqa: E402
from fastapi.middleware.cors import CORSMiddleware  # noqa: E402
from fastapi.responses import JSONResponse  # noqa: E402
from fastapi.staticfiles import StaticFiles  # noqa: E402

from backend.api.routes import router as api_router  # noqa: E402
from backend.api.websocket import ws_router  # noqa: E402
from backend.config.settings import PROJECT_ROOT, get_settings  # noqa: E402
from backend.core.events import get_event_bus  # noqa: E402
from backend.core.logging import configure_logging  # noqa: E402
from backend.robot.base import (  # noqa: E402
    RobotBusy,
    RobotError,
    RobotNotConfigured,
    RobotTimeout,
    RobotUnreachable,
    SpeechFailed,
    TransportNotAvailable,
)
from backend.robot.manager import RobotManager  # noqa: E402
from backend.services.speech_service import SpeechService, ValidationError  # noqa: E402

logger = logging.getLogger("agibot.app")

FRONTEND_DIR = PROJECT_ROOT / "frontend"

# HTTP status for each robot-layer failure. Anything unmapped becomes 500.
STATUS_FOR_ERROR = {
    ValidationError: 400,
    RobotBusy: 409,
    TransportNotAvailable: 501,
    SpeechFailed: 502,
    RobotNotConfigured: 503,
    RobotUnreachable: 503,
    RobotTimeout: 504,
}


def _status_for(exc: RobotError) -> int:
    for error_type, status in STATUS_FOR_ERROR.items():
        if isinstance(exc, error_type):
            return status
    return 500


@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = app.state.settings
    manager: RobotManager = app.state.robot_manager

    banner(settings)
    await manager.start()
    try:
        yield
    finally:
        await app.state.speech_service.shutdown()
        await manager.stop()
        logger.info("shutdown complete")


def create_app() -> FastAPI:
    settings = get_settings()
    configure_logging(settings.server.log_level)

    app = FastAPI(
        title="AGIBOT A3 Voice Control",
        description="Local dashboard for sending speech to an AGIBOT A3 humanoid.",
        version="1.0.0",
        lifespan=lifespan,
        docs_url="/api/docs",
        openapi_url="/api/openapi.json",
    )

    bus = get_event_bus()
    manager = RobotManager(settings, bus)
    app.state.settings = settings
    app.state.event_bus = bus
    app.state.robot_manager = manager
    app.state.speech_service = SpeechService(settings, manager, bus)

    if settings.server.cors_origins:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=settings.server.cors_origins,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    @app.exception_handler(RobotError)
    async def robot_error_handler(request: Request, exc: RobotError) -> JSONResponse:
        status = _status_for(exc)
        if status >= 500:
            logger.warning("%s -> %s (%s)", request.url.path, exc.code, exc)
        return JSONResponse(
            status_code=status,
            content={"success": False, "error": exc.user_message, "errorCode": exc.code},
        )

    app.include_router(api_router)
    app.include_router(ws_router)

    if FRONTEND_DIR.is_dir():
        # Mounted last so /api/* and /ws win.
        app.mount("/", StaticFiles(directory=str(FRONTEND_DIR), html=True), name="frontend")
    else:  # pragma: no cover
        logger.error("frontend directory missing: %s", FRONTEND_DIR)

    return app


def banner(settings) -> None:
    mode = settings.robot.mode.upper()
    target = "local simulation" if settings.is_mock else "{0} via {1}".format(
        settings.robot.address, settings.a3.transport
    )
    url = "http://{0}:{1}".format(
        "localhost" if settings.server.host in ("0.0.0.0", "127.0.0.1") else settings.server.host,
        settings.server.port,
    )
    lines = [
        "",
        "  AGIBOT A3 - Voice Control",
        "  " + "-" * 46,
        "  Mode      : {0}".format(mode),
        "  Robot     : {0}".format(target),
        "  Dashboard : {0}".format(url),
        "  Config    : {0}".format(settings.env_file),
    ]
    for issue in settings.issues:
        lines.append("  {0:<10}: [{1}] {2}".format("Config", issue.level.upper(), issue.message))
    if settings.is_mock:
        lines.append("  Note      : running against the MOCK robot - no hardware needed.")
    lines.append("")
    print("\n".join(lines))


app = create_app()


def main() -> None:
    import uvicorn

    settings = get_settings()
    uvicorn.run(
        "backend.main:app",
        host=settings.server.host,
        port=settings.server.port,
        reload=os.environ.get("DEV_RELOAD", "").lower() in ("1", "true", "yes"),
        log_level=settings.server.log_level,
        access_log=False,
    )


if __name__ == "__main__":
    main()