import os import chess import numpy as np import onnxruntime as ort from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from huggingface_hub import hf_hub_download from pydantic import BaseModel app = FastAPI() class ChessRequest(BaseModel): fen: str app.mount("/static", StaticFiles(directory="static"), name="static") templates = Jinja2Templates(directory="templates") MODEL_REPO = "GambitFlow/Synapse-Edge" MODEL_FILENAME = "v1/synapse_edge_v1.onnx" try: print("đŸ“Ĩ Downloading flagship model...") model_path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILENAME) session = ort.InferenceSession(model_path) print("✅ Synapse-Edge v1 Loaded.") except Exception as e: print(f"❌ Model Load Error: {e}") # [CRITICAL FIX]: āĻŸā§āϰ⧇āύāĻŋāĻ‚ āϕ⧋āĻĄā§‡āϰ āϏāĻžāĻĨ⧇ ā§§ā§Ļā§Ļ% āĻŽā§āϝāĻžāϚ āĻ•āϰāĻž āĻŸā§‡āύāϏāϰ āϞāϜāĻŋāĻ• def get_tensor(fen): # āĻŸā§āϰ⧇āύāĻŋāĻ‚ āϕ⧋āĻĄ āĻ…āύ⧁āϝāĻžā§Ÿā§€ ⧧⧧⧝ āĻšā§āϝāĻžāύ⧇āϞ, āĻ•āĻŋāĻ¨ā§āϤ⧁ āĻĄā§‡āϟāĻž āĻļ⧁āϧ⧁ ⧧⧍āϟāĻŋāϤ⧇ tensor = np.zeros((1, 119, 8, 8), dtype=np.float32) position = fen.split(' ')[0] # āĻĒāĻŋāϏ āĻŽā§āϝāĻžāĻĒ (āĻŸā§āϰ⧇āύāĻŋāĻ‚ āϕ⧋āĻĄ āĻ…āύ⧁āϝāĻžā§Ÿā§€) piece_to_channel = {'P':0, 'N':1, 'B':2, 'R':3, 'Q':4, 'K':5, 'p':6, 'n':7, 'b':8, 'r':9, 'q':10, 'k':11} rank = 0 file = 0 for char in position: if char == '/': rank += 1 file = 0 elif char.isdigit(): file += int(char) elif char in piece_to_channel: if rank < 8 and file < 8: tensor[0, piece_to_channel[char], rank, file] = 1.0 file += 1 # āĻŦāĻžāĻ•āĻŋ ā§§ā§Ļā§­āϟāĻŋ āĻšā§āϝāĻžāύ⧇āϞ āϏāĻŦ āϜāĻŋāϰ⧋ āĻĨāĻžāĻ•āĻŦ⧇ (āĻŸā§āϰ⧇āύāĻŋāĻ‚ā§Ÿā§‡āϰ āϏāĻŽā§Ÿ āϝ⧇āĻŽāύ āĻ›āĻŋāϞ) return tensor def predict(fen): tensor = get_tensor(fen) return session.run(None, {"input": tensor}) @app.post("/get_move") async def get_move(req: ChessRequest): try: board = chess.Board(req.fen) if board.is_game_over(): return JSONResponse({"error": "Game over"}, status_code=400) # ā§§. āĻŦāĻ°ā§āϤāĻŽāĻžāύ āĻĒāϜāĻŋāĻļāύ⧇ āχāύāĻĢāĻžāϰ⧇āĻ¨ā§āϏ policy, value, tactical, phase = predict(req.fen) legal_moves = list(board.legal_moves) move_candidates = [] # ⧍. āĻ­ā§āϝāĻžāϞ⧁ āĻšā§‡āĻĄ āĻĻāĻŋā§Ÿā§‡ āĻšā§‡āĻ• āĻ•āϰāĻž (āĻŽāĻĄā§‡āϞ⧇āϰ āύāĻŋāĻœā§‡āϰ āύāϞ⧇āϜ āĻ…āύ⧁āϝāĻžā§Ÿā§€) for move in legal_moves: board.push(move) # āĻ•āĻžāϞ⧋āϰ āϚāĻžāϞ āĻšāϞ⧇ āϏāĻžāĻĻāĻžāϰ āĻĒā§Ÿā§‡āĻ¨ā§āϟ āύ⧇āϗ⧇āϟāĻŋāĻ­ āĻšāĻ“ā§ŸāĻž āĻŽāĻžāύ⧇ āĻ•āĻžāϞ⧋ āϜāĻŋāϤāϛ⧇ _, next_v, _, _ = predict(board.fen()) board.pop() v_score = float(next_v[0][0]) # āϏāĻžāĻĻāĻžāϰ āϚāĻžāϞ āĻšāϞ⧇ āĻŦ⧇āĻļāĻŋ āĻ¸ā§āϕ⧋āϰ āĻ­āĻžāϞ⧋, āĻ•āĻžāϞ⧋āϰ āϚāĻžāϞ āĻšāϞ⧇ āĻ•āĻŽ āĻ¸ā§āϕ⧋āϰ āĻ­āĻžāϞ⧋ actual_score = v_score if board.turn == chess.WHITE else -v_score # ā§Š. āϏāĻŋāĻŽā§āĻĒāϞ āĻĒāϞāĻŋāϏāĻŋ āĻĒā§āϰāĻžā§Ÿā§‹āϰāĻŋāϟāĻŋ (āχāĻ¨ā§āĻĄā§‡āĻ•ā§āϏāĻŋāĻ‚ āĻŽāĻŋāϏāĻŽā§āϝāĻžāϚ āĻā§œāĻžāϤ⧇ āφāĻŽāϰāĻž āĻ­ā§āϝāĻžāϞ⧁āϕ⧇ āϗ⧁āϰ⧁āĻ¤ā§āĻŦ āĻĻāĻŋāĻŦ āĻŦ⧇āĻļāĻŋ) move_candidates.append((move, actual_score)) # āϏāĻŦāĻšā§‡ā§Ÿā§‡ āĻ­āĻžāϞ⧋ āĻ­ā§āϝāĻžāϞ⧁āϰ āϚāĻžāϞāϟāĻŋ āĻŦāĻžāĻ›āĻžāχ move_candidates.sort(key=lambda x: x[1], reverse=True) best_move = move_candidates[0][0] # ā§Ē. āϝāĻĻāĻŋ āϏ⧇āϰāĻž āϚāĻžāϞ⧇āĻ“ āϕ⧋āύ⧋ āĻŦā§āϞāĻžāĻ¨ā§āĻĄāĻžāϰ āĻšāĻ“ā§ŸāĻžāϰ āϭ⧟ āĻĨāĻžāϕ⧇, āϤāĻŦ⧇ āϏ⧇āϕ⧇āĻ¨ā§āĻĄ āĻŦ⧇āĻ¸ā§āϟ āĻŸā§āϰāĻžāχ āĻ•āϰāĻž # (v1 āĻāϰ āϜāĻ¨ā§āϝ āφāĻŽāϰāĻž ā§§-āϞ⧇āϭ⧇āϞ āĻ­ā§āϝāĻžāϞ⧁ āϏāĻžāĻ°ā§āϚāϕ⧇āχ āĻĢāĻžāχāύāĻžāϞ āϰāĻžāĻ–āĻ›āĻŋ) return { "move": best_move.uci(), "value": float(value[0][0]), "tactical": float(tactical[0][0]), "phase": int(np.argmax(phase[0])) } except Exception as e: print(f"đŸ”Ĩ Server Error: {e}") return JSONResponse({"error": str(e)}, status_code=400) @app.get("/", response_class=HTMLResponse) async def read_root(request: Request): return templates.TemplateResponse("index.html", {"request": request}) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)