""" Nexus-Nano Inference API 2.8M parameter ultra-fast engine """ from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field import time import logging from typing import Optional, List from engine import NexusNanoEngine # Logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # FastAPI app = FastAPI( title="Nexus-Nano Inference API", description="Ultra-fast 2.8M parameter chess engine", version="1.0.0" ) # CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Global engine = None # Models class MoveRequest(BaseModel): fen: str = Field(..., description="FEN notation") depth: Optional[int] = Field(4, ge=1, le=6, description="Search depth (1-6)") time_limit: Optional[int] = Field(2000, ge=500, le=10000, description="Time in ms") class MoveResponse(BaseModel): best_move: str evaluation: float depth_searched: int seldepth: int nodes_evaluated: int time_taken: int nps: int pv: List[str] tt_hit_rate: Optional[float] = None class HealthResponse(BaseModel): status: str model_loaded: bool version: str model_size_mb: Optional[float] = None # Startup @app.on_event("startup") async def startup_event(): global engine logger.info("⚡ Starting Nexus-Nano API v1.0...") try: engine = NexusNanoEngine( model_path="/app/models/nexus_nano.onnx", num_threads=1 # Single-threaded for speed ) logger.info("✅ Engine loaded") except Exception as e: logger.error(f"❌ Failed: {e}") raise # Health @app.get("/health", response_model=HealthResponse) async def health_check(): return { "status": "healthy" if engine else "unhealthy", "model_loaded": engine is not None, "version": "1.0.0", "model_size_mb": engine.get_model_size() if engine else None } # Main @app.post("/get-move", response_model=MoveResponse) async def get_move(request: MoveRequest): if engine is None: raise HTTPException(status_code=503, detail="Engine not loaded") if not engine.validate_fen(request.fen): raise HTTPException(status_code=400, detail="Invalid FEN") start_time = time.time() try: result = engine.get_best_move( fen=request.fen, depth=request.depth, time_limit=request.time_limit ) time_taken = int((time.time() - start_time) * 1000) logger.info( f"⚡ Move: {result['best_move']} | " f"Eval: {result['evaluation']:+.2f} | " f"Depth: {result['depth_searched']} | " f"Time: {time_taken}ms | " f"NPS: {result['nps']}" ) return MoveResponse( best_move=result['best_move'], evaluation=result['evaluation'], depth_searched=result['depth_searched'], seldepth=result['seldepth'], nodes_evaluated=result['nodes_evaluated'], time_taken=time_taken, nps=result['nps'], pv=result['pv'], tt_hit_rate=result['tt_stats']['hit_rate'] ) except Exception as e: logger.error(f"Error: {e}") raise HTTPException(status_code=500, detail=str(e)) # Root @app.get("/") async def root(): return { "name": "Nexus-Nano Inference API", "version": "1.0.0", "model": "2.8M parameters (Lightweight CNN)", "tagline": "Ultra-fast chess inference", "search": "Alpha-Beta + Quiescence", "endpoints": { "POST /get-move": "Get best move (fast)", "GET /health": "Health check", "GET /docs": "API docs" } } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info")