Spaces:
Sleeping
Sleeping
| # chess_engine/api/rest_api.py | |
| from fastapi import FastAPI, HTTPException, Depends | |
| from fastapi.staticfiles import StaticFiles | |
| from pydantic import BaseModel, Field | |
| from typing import Dict, List, Optional, Any | |
| import chess | |
| from enum import Enum | |
| from chess_engine.api.game_controller import GameController, GameOptions | |
| from chess_engine.ai.stockfish_wrapper import DifficultyLevel | |
| # Define API models | |
| class ColorEnum(str, Enum): | |
| WHITE = "white" | |
| BLACK = "black" | |
| class DifficultyEnum(str, Enum): | |
| BEGINNER = "beginner" | |
| EASY = "easy" | |
| MEDIUM = "medium" | |
| HARD = "hard" | |
| EXPERT = "expert" | |
| MASTER = "master" | |
| class NewGameRequest(BaseModel): | |
| player_color: ColorEnum = ColorEnum.WHITE | |
| difficulty: DifficultyEnum = DifficultyEnum.MEDIUM | |
| time_limit: float = Field(default=1.0, ge=0.1, le=10.0) | |
| use_opening_book: bool = True | |
| enable_analysis: bool = True | |
| stockfish_path: Optional[str] = None | |
| class MoveRequest(BaseModel): | |
| move: str = Field(..., description="Move in UCI notation (e.g., 'e2e4', 'e7e8q') or SAN (e.g., 'e4', 'e8=Q')") | |
| class PromotionDetails(BaseModel): | |
| """Details about a required promotion move""" | |
| from_square: str = Field(..., description="Starting square of the promotion move (e.g., 'e7')") | |
| to_square: str = Field(..., description="Destination square of the promotion move (e.g., 'e8')") | |
| available_pieces: List[str] = Field(..., description="Available pieces for promotion: ['queen', 'rook', 'bishop', 'knight']") | |
| class GameResponse(BaseModel): | |
| """Enhanced game response that can include promotion requirements""" | |
| status: str = Field(..., description="Response status: 'success', 'error', 'promotion_required', 'game_over'") | |
| message: Optional[str] = Field(None, description="Response message") | |
| player_move: Optional[str] = Field(None, description="Player's move in UCI notation") | |
| ai_move: Optional[str] = Field(None, description="AI's move in UCI notation") | |
| board_state: Optional[Dict[str, Any]] = Field(None, description="Current board state") | |
| analysis: Optional[Dict[str, Any]] = Field(None, description="Position analysis") | |
| promotion_details: Optional[PromotionDetails] = Field(None, description="Promotion requirements when status is 'promotion_required'") | |
| result: Optional[str] = Field(None, description="Game result when status is 'game_over'") | |
| winner: Optional[str] = Field(None, description="Winner when game is over") | |
| reason: Optional[str] = Field(None, description="Reason for game end") | |
| class UndoRequest(BaseModel): | |
| count: int = Field(default=2, ge=1, le=10, description="Number of half-moves to undo") | |
| class PromotionRequest(BaseModel): | |
| """Request for completing a promotion move""" | |
| from_square: str = Field(..., description="Starting square of the promotion move (e.g., 'e7')") | |
| to_square: str = Field(..., description="Destination square of the promotion move (e.g., 'e8')") | |
| promotion_piece: str = Field(..., description="Piece to promote to: 'queen', 'rook', 'bishop', 'knight', or 'q', 'r', 'b', 'n'") | |
| # Create FastAPI app | |
| app = FastAPI( | |
| title="Chess Engine API", | |
| description="REST API for chess engine with Stockfish integration", | |
| version="1.0.0" | |
| ) | |
| # Game controller instance | |
| game_controller = GameController() | |
| # Helper function to get game controller | |
| def get_game_controller(): | |
| return game_controller | |
| async def new_game(request: NewGameRequest, controller: GameController = Depends(get_game_controller)): | |
| """Start a new game with specified options""" | |
| # Convert enum values to appropriate types | |
| options = GameOptions( | |
| player_color=chess.WHITE if request.player_color == ColorEnum.WHITE else chess.BLACK, | |
| difficulty=DifficultyLevel[request.difficulty.upper()], | |
| time_limit=request.time_limit, | |
| use_opening_book=request.use_opening_book, | |
| enable_analysis=request.enable_analysis, | |
| stockfish_path=request.stockfish_path | |
| ) | |
| result = controller.start_new_game(options) | |
| if result["status"] == "error": | |
| raise HTTPException(status_code=500, detail=result["message"]) | |
| return result | |
| async def make_move(request: MoveRequest, controller: GameController = Depends(get_game_controller)): | |
| """ | |
| Make a move on the board. | |
| This endpoint handles regular moves and promotion moves. When a pawn reaches the back rank | |
| without specifying a promotion piece, it returns a 'promotion_required' status with details | |
| about the required promotion. | |
| **Move Formats:** | |
| - Regular moves: 'e2e4', 'Nf3', 'O-O' | |
| - Promotion moves: 'e7e8q' (UCI) or 'e8=Q' (SAN) | |
| **Promotion Pieces:** | |
| - 'q' or 'queen' for Queen | |
| - 'r' or 'rook' for Rook | |
| - 'b' or 'bishop' for Bishop | |
| - 'n' or 'knight' for Knight | |
| **Response Status Values:** | |
| - 'success': Move completed successfully | |
| - 'promotion_required': Pawn promotion requires piece selection | |
| - 'error': Invalid move or game state error | |
| - 'game_over': Game ended after this move | |
| """ | |
| result = controller.make_player_move(request.move) | |
| # Handle different response statuses | |
| if result["status"] == "error": | |
| raise HTTPException(status_code=400, detail=result["message"]) | |
| elif result["status"] == "promotion_required": | |
| # Return 200 with promotion_required status - this is not an error | |
| return GameResponse( | |
| status="promotion_required", | |
| message=result["message"], | |
| board_state=result["board_state"], | |
| promotion_details=PromotionDetails( | |
| from_square=result["promotion_details"]["from"], | |
| to_square=result["promotion_details"]["to"], | |
| available_pieces=result["promotion_details"]["available_pieces"] | |
| ) | |
| ) | |
| elif result["status"] == "game_over": | |
| return GameResponse( | |
| status="game_over", | |
| message=result.get("message"), | |
| board_state=result["board_state"], | |
| result=result["result"], | |
| winner=result.get("winner"), | |
| reason=result["reason"] | |
| ) | |
| else: | |
| # Success case | |
| return GameResponse( | |
| status="success", | |
| player_move=result.get("player_move"), | |
| ai_move=result.get("ai_move"), | |
| board_state=result["board_state"], | |
| analysis=result.get("analysis") | |
| ) | |
| async def get_game_state(controller: GameController = Depends(get_game_controller)): | |
| """Get current game state""" | |
| return controller.get_game_state() | |
| async def get_hint(controller: GameController = Depends(get_game_controller)): | |
| """Get a hint for the current position""" | |
| result = controller.get_hint() | |
| if result["status"] == "error": | |
| raise HTTPException(status_code=400, detail=result["message"]) | |
| return result | |
| async def undo_move(request: UndoRequest, controller: GameController = Depends(get_game_controller)): | |
| """Undo moves""" | |
| result = controller.undo_move(request.count) | |
| if result["status"] == "error": | |
| raise HTTPException(status_code=400, detail=result["message"]) | |
| return result | |
| async def complete_promotion(request: PromotionRequest, controller: GameController = Depends(get_game_controller)): | |
| """ | |
| Complete a pawn promotion by specifying the promotion piece. | |
| This endpoint is used when a previous move request returned 'promotion_required' status. | |
| It completes the promotion move with the specified piece. | |
| **Promotion Pieces:** | |
| - 'queen' or 'q' for Queen (most common choice) | |
| - 'rook' or 'r' for Rook | |
| - 'bishop' or 'b' for Bishop | |
| - 'knight' or 'n' for Knight | |
| **Example Usage:** | |
| 1. Make move 'e7e8' -> returns promotion_required | |
| 2. Call this endpoint with promotion_piece='queen' to complete the move | |
| """ | |
| from chess_engine.promotion import PromotionMoveHandler | |
| # Create the full promotion move in UCI notation | |
| promotion_move = PromotionMoveHandler.create_promotion_move( | |
| request.from_square, | |
| request.to_square, | |
| request.promotion_piece | |
| ) | |
| # Make the promotion move | |
| result = controller.make_player_move(promotion_move) | |
| # Handle different response statuses | |
| if result["status"] == "error": | |
| raise HTTPException(status_code=400, detail=result["message"]) | |
| elif result["status"] == "promotion_required": | |
| # This shouldn't happen with a properly formed promotion move | |
| raise HTTPException(status_code=400, detail="Invalid promotion move format") | |
| elif result["status"] == "game_over": | |
| return GameResponse( | |
| status="game_over", | |
| message=result.get("message"), | |
| board_state=result["board_state"], | |
| result=result["result"], | |
| winner=result.get("winner"), | |
| reason=result["reason"] | |
| ) | |
| else: | |
| # Success case | |
| return GameResponse( | |
| status="success", | |
| player_move=result.get("player_move"), | |
| ai_move=result.get("ai_move"), | |
| board_state=result["board_state"], | |
| analysis=result.get("analysis") | |
| ) | |
| async def resign_game(controller: GameController = Depends(get_game_controller)): | |
| """Resign the current game""" | |
| result = controller.resign() | |
| if result["status"] == "error": | |
| raise HTTPException(status_code=400, detail=result["message"]) | |
| return result | |
| async def get_promotion_pieces(): | |
| """ | |
| Get available promotion pieces with their details. | |
| Returns information about all pieces that a pawn can be promoted to, | |
| including their names, symbols, UCI notation, and relative values. | |
| """ | |
| from chess_engine.promotion import PromotionMoveHandler | |
| return { | |
| "status": "success", | |
| "promotion_pieces": PromotionMoveHandler.get_available_promotions() | |
| } | |
| async def health_check(): | |
| """Health check endpoint for container monitoring""" | |
| return { | |
| "status": "healthy", | |
| "service": "chess-engine-api", | |
| "version": "1.0.0" | |
| } | |
| # Mount static files AFTER defining all API routes | |
| # Mount UI at /ui path | |
| app.mount("/ui", StaticFiles(directory="./dist", html=True), name="static") | |
| # Mount at root for the index.html, but with a lower priority than API endpoints | |
| app.mount("/", StaticFiles(directory="./dist", html=True), name="root_static") |