File size: 5,477 Bytes
97c3bf5
 
 
 
 
 
 
 
7f4a7eb
97c3bf5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7f4a7eb
 
97c3bf5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
import csv
import chess.pgn
import chess.engine
from enum import Enum
from typing import List, Dict
import asyncio
import json
import sys
import os

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())


def load_opening_book(csv_path):
    opening_book = {}
    try:
        with open(csv_path, newline='', encoding='utf-8') as csvfile:
            reader = csv.reader(csvfile)
            next(reader)
            for row in reader:
                if len(row) < 3:
                    continue
                pgn_moves = row[2]
                game = chess.pgn.Game()
                board = game.board()
                for move in pgn_moves.split():
                    if "." in move:
                        continue
                    try:
                        chess_move = board.push_san(move)
                        fen = " ".join(board.fen().split()[:4])
                        opening_book[fen] = chess_move.uci()
                    except ValueError:
                        break
    except Exception as e:
        print(f"Error loading opening book: {e}")
    return opening_book


engine_path = os.path.join(os.getcwd(), "models", "stockfish_14_x64_avx2.exe")
csv_path = os.path.join(os.getcwd(), "assets", "opening_book.csv")
opening_book = load_opening_book(csv_path)


class GamePhase(Enum):
    OPENING = "opening"
    MIDDLEGAME = "middlegame"
    ENDGAME = "endgame"

class Classification(Enum):
    BRILLIANT = "brilliant"
    GREAT = "great"
    BEST = "best"
    EXCELLENT = "excellent"
    GOOD = "good"
    INACCURACY = "inaccuracy"
    MISTAKE = "mistake"
    MISS = "miss"
    BLUNDER = "blunder"
    BOOK = "book"
    FORCED = "forced"

classification_values = {
    Classification.BLUNDER: 0,
    Classification.MISTAKE: 0.2,
    Classification.MISS: 0.3,
    Classification.INACCURACY: 0.4,
    Classification.GOOD: 0.65,
    Classification.EXCELLENT: 0.9,
    Classification.BEST: 1,
    Classification.GREAT: 1,
    Classification.BRILLIANT: 1,
    Classification.BOOK: 1,
    Classification.FORCED: 1,
}

centipawn_classifications = [
    Classification.BEST,
    Classification.EXCELLENT,
    Classification.GOOD,
    Classification.INACCURACY,
    Classification.MISS,
    Classification.MISTAKE,
    Classification.BLUNDER,
]

FORCED_WIN_THRESHOLD = 500
MISS_CENTIPAWN_LOSS = 300
MISS_MATE_THRESHOLD = 3
ENDGAME_MATERIAL_THRESHOLD = 24
QUEEN_VALUE = 9

def detect_game_phase(board: chess.Board, in_opening: bool) -> GamePhase:
    if in_opening:
        return GamePhase.OPENING
    total_material = sum(
        len(board.pieces(p, color)) * {1: 1, 2: 3, 3: 3, 4: 5, 5: QUEEN_VALUE}[p]
        for color in [chess.WHITE, chess.BLACK] for p in chess.PIECE_TYPES if p != chess.KING
    )
    queens = sum(len(board.pieces(chess.QUEEN, color)) for color in [chess.WHITE, chess.BLACK])
    return GamePhase.ENDGAME if total_material <= ENDGAME_MATERIAL_THRESHOLD or (queens == 0 and total_material <= ENDGAME_MATERIAL_THRESHOLD * 2) else GamePhase.MIDDLEGAME

def is_book_move(board, opening_book, max_depth=8):  
    return None if board.fullmove_number > max_depth else opening_book.get(" ".join(board.fen().split()[:4]))

async def analyze_pgn(pgn_file: str):
    with open(pgn_file) as pgn:
        game = chess.pgn.read_game(pgn)
    if not game:
        return json.dumps({"error": "No game found in PGN file."})
    
    results = {
        "moves": [],
        "phases": {},
        "players": {"white": {}, "black": {}}
    }
    
    with chess.engine.SimpleEngine.popen_uci(engine_path) as engine:
        board = game.board()
        classifications = {"white": {p.value: [] for p in GamePhase}, "black": {p.value: [] for p in GamePhase}}
        in_opening = True

        for move_number, node in enumerate(game.mainline(), start=1):
            pre_info = engine.analyse(board, chess.engine.Limit(depth=20))
            pre_eval = pre_info["score"].white().score(mate_score=10000) or 0
            best_move = pre_info.get("pv", [None])[0]
            move = node.move
            board.push(move)
            post_info = engine.analyse(board, chess.engine.Limit(depth=20))
            post_eval = post_info["score"].white().score(mate_score=10000) or 0
            book_move = is_book_move(board, opening_book)
            current_phase = detect_game_phase(board, in_opening)
            if not book_move and in_opening:
                in_opening = False
            eval_loss = abs(pre_eval - post_eval)
            classification = Classification.BOOK if book_move else Classification.BLUNDER
            player = "white" if board.turn == chess.BLACK else "black"
            classifications[player][current_phase.value].append(classification.value)
            results["moves"].append({
                "move_number": move_number,
                "player": player,
                "move": move.uci(),
                "evaluation": post_eval / 100,
                "evaluation_loss": eval_loss / 100,
                "classification": classification.value
            })

        for phase in GamePhase:
            results["phases"][phase.value] = {
                "white": classifications["white"][phase.value],
                "black": classifications["black"][phase.value]
            }
        results["players"]["white"] = classifications["white"]
        results["players"]["black"] = classifications["black"]
    
    return json.dumps(results, indent=4)