suvasis's picture
code add
e4d7d50
"""
ChessEcon Backend — Chess Engine
Wraps python-chess to manage game state, validate moves, and detect outcomes.
"""
from __future__ import annotations
import uuid
import chess
import chess.pgn
from typing import Dict, Optional, List
from shared.models import GameState, GameOutcome, GameStatus, NewGameResponse
class ChessEngine:
"""Thread-safe chess game manager. Stores all active games in memory."""
def __init__(self):
self._games: Dict[str, chess.Board] = {}
# ── Game lifecycle ────────────────────────────────────────────────────────
def new_game(self, game_id: Optional[str] = None) -> NewGameResponse:
gid = game_id or str(uuid.uuid4())
board = chess.Board()
self._games[gid] = board
return NewGameResponse(
game_id=gid,
fen=board.fen(),
legal_moves=[m.uci() for m in board.legal_moves],
status=GameStatus.ACTIVE,
)
def get_state(self, game_id: str) -> GameState:
board = self._get_board(game_id)
return GameState(
game_id=game_id,
fen=board.fen(),
legal_moves=[m.uci() for m in board.legal_moves],
outcome=self._outcome(board),
move_number=board.fullmove_number,
move_history=[m.uci() for m in board.move_stack],
status=GameStatus.FINISHED if board.is_game_over() else GameStatus.ACTIVE,
)
def make_move(self, game_id: str, move_uci: str) -> GameState:
board = self._get_board(game_id)
if board.is_game_over():
raise ValueError(f"Game {game_id} is already over")
try:
move = chess.Move.from_uci(move_uci)
except ValueError:
raise ValueError(f"Invalid UCI move format: {move_uci}")
if move not in board.legal_moves:
legal = [m.uci() for m in board.legal_moves]
raise ValueError(
f"Illegal move {move_uci} in position {board.fen()}. "
f"Legal moves: {legal[:10]}{'...' if len(legal) > 10 else ''}"
)
board.push(move)
return self.get_state(game_id)
def delete_game(self, game_id: str) -> None:
self._games.pop(game_id, None)
def list_games(self) -> List[str]:
return list(self._games.keys())
# ── Position analysis ─────────────────────────────────────────────────────
def get_legal_moves(self, game_id: str) -> List[str]:
board = self._get_board(game_id)
return [m.uci() for m in board.legal_moves]
def get_fen(self, game_id: str) -> str:
return self._get_board(game_id).fen()
def is_game_over(self, game_id: str) -> bool:
return self._get_board(game_id).is_game_over()
def complexity_features(self, game_id: str) -> dict:
"""Return raw features used by the complexity analyzer."""
board = self._get_board(game_id)
legal = list(board.legal_moves)
return {
"num_legal_moves": len(legal),
"is_check": board.is_check(),
"has_captures": any(board.is_capture(m) for m in legal),
"num_pieces": len(board.piece_map()),
"fullmove_number": board.fullmove_number,
"material_balance": self._material_balance(board),
}
# ── Private helpers ───────────────────────────────────────────────────────
def _get_board(self, game_id: str) -> chess.Board:
if game_id not in self._games:
raise KeyError(f"Game {game_id} not found")
return self._games[game_id]
@staticmethod
def _outcome(board: chess.Board) -> GameOutcome:
if not board.is_game_over():
return GameOutcome.ONGOING
result = board.result()
if result == "1-0":
return GameOutcome.WHITE_WIN
elif result == "0-1":
return GameOutcome.BLACK_WIN
return GameOutcome.DRAW
@staticmethod
def _material_balance(board: chess.Board) -> float:
"""Positive = white advantage."""
piece_values = {
chess.PAWN: 1, chess.KNIGHT: 3, chess.BISHOP: 3,
chess.ROOK: 5, chess.QUEEN: 9, chess.KING: 0,
}
balance = 0.0
for piece_type, value in piece_values.items():
balance += value * len(board.pieces(piece_type, chess.WHITE))
balance -= value * len(board.pieces(piece_type, chess.BLACK))
return balance
# Singleton instance
chess_engine = ChessEngine()