File size: 4,842 Bytes
e4d7d50
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
"""
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()