added new api getfen which is working percfectly

main.py

@@ -140,6 +140,7 @@ async def get_fen(file : UploadFile = File(), perspective : str = Form("w"), nex
     
 @app.post('/getReview')
 async def getReview(file: UploadFile = File(...)):  
+    print(os.getcwd())
     print("call recieved")
 
     if not file.filename.endswith(".pgn"):
@@ -152,14 +153,14 @@ async def getReview(file: UploadFile = File(...)):
             tmp_file_path = tmp_file.name
 
         # Analyze the PGN file
-        analysis_result = await analyze_pgn(tmp_file_path)
+        analysis_result = analyze_pgn(tmp_file_path)
 
         # Clean up the temporary file
         os.remove(tmp_file_path)
 
         if not analysis_result:
             return JSONResponse(content={"error": "No game found in the PGN file"}, status_code=400)
-        
-        return JSONResponse(content=analysis_result, status_code=200)
+        return analysis_result
+    
     except Exception as e:
         return  JSONResponse(content={"error": "Unexpected error occurred", "details": str(e)}, status_code=500)

routes/chess_review.py

@@ -1,47 +1,16 @@
-import csv
+from fastapi import FastAPI, File, UploadFile, HTTPException
+import os
+import tempfile
 import chess.pgn
 import chess.engine
 from enum import Enum
 from typing import List, Dict
-import asyncio
+from datetime import datetime
+import csv
 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)
+from fastapi.responses import JSONResponse
 
+app = FastAPI()
 
 class GamePhase(Enum):
     OPENING = "opening"
@@ -85,6 +54,7 @@ centipawn_classifications = [
     Classification.BLUNDER,
 ]
 
+# Analysis parameters
 FORCED_WIN_THRESHOLD = 500
 MISS_CENTIPAWN_LOSS = 300
 MISS_MATE_THRESHOLD = 3
@@ -94,64 +64,224 @@ 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
+        
+    total_material = 0
+    queens = 0
+    
+    for color in [chess.WHITE, chess.BLACK]:
+        for piece_type in chess.PIECE_TYPES:
+            if piece_type == chess.KING:
+                continue
+                
+            count = len(board.pieces(piece_type, color))
+            value = {
+                chess.PAWN: 1,
+                chess.KNIGHT: 3,
+                chess.BISHOP: 3,
+                chess.ROOK: 5,
+                chess.QUEEN: QUEEN_VALUE
+            }[piece_type]
+            
+            total_material += count * value
+            if piece_type == chess.QUEEN:
+                queens += count
+
+    endgame_conditions = [
+        total_material <= ENDGAME_MATERIAL_THRESHOLD,
+        queens == 0 and total_material <= ENDGAME_MATERIAL_THRESHOLD * 2,
+    ]
+    
+    return GamePhase.ENDGAME if any(endgame_conditions) else GamePhase.MIDDLEGAME
+
+def get_evaluation_loss_threshold(classif: Classification, prev_eval: float) -> float:
+    prev_eval = abs(prev_eval)
+    if classif == Classification.BEST:
+        return max(0.0001 * prev_eval**2 + 0.0236 * prev_eval - 3.7143, 0)
+    elif classif == Classification.EXCELLENT:
+        return max(0.0002 * prev_eval**2 + 0.1231 * prev_eval + 27.5455, 0)
+    elif classif == Classification.GOOD:
+        return max(0.0002 * prev_eval**2 + 0.2643 * prev_eval + 60.5455, 0)
+    elif classif == Classification.INACCURACY:
+        return max(0.0002 * prev_eval**2 + 0.3624 * prev_eval + 108.0909, 0)
+    elif classif == Classification.MISS:
+        return max(0.00025 * prev_eval**2 + 0.38255 * prev_eval + 166.9541, 0)
+    elif classif == Classification.MISTAKE:
+        return max(0.0003 * prev_eval**2 + 0.4027 * prev_eval + 225.8182, 0)
+    else:
+        return float("inf")
+
+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
 
 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]))
+    if board.fullmove_number > max_depth:  
+        return None  
+    fen = " ".join(board.fen().split()[:4])
+    return opening_book.get(fen)
+
 
-async def analyze_pgn(pgn_file: str):
+engine_path = os.path.join(os.getcwd(), "models", "stockfish-windows-x86-64-avx2.exe")
+book_csv_path = os.path.join(os.getcwd(), "assets", "openings_master.csv")
+
+def analyze_pgn(pgn_file: str) -> Dict:
+    opening_book = load_opening_book(book_csv_path)
+    
     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."})
+        return {"error": "No game found in the PGN file."}
     
-    results = {
-        "moves": [],
-        "phases": {},
-        "players": {"white": {}, "black": {}}
+    result = {
+        "move_analysis": [],
+        "phase_analysis": {},
+        "player_summaries": {}
     }
     
     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}}
+        classifications = {
+            "white": {phase: [] for phase in GamePhase},
+            "black": {phase: [] for phase in GamePhase}
+        }
+        phase_data = {phase: [] for phase in GamePhase}
         in_opening = True
 
         for move_number, node in enumerate(game.mainline(), start=1):
+            # Analyze position before the move
             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]
+            
+            # Make the move
             move = node.move
             board.push(move)
+            
+            # Analyze position after the move
             post_info = engine.analyse(board, chess.engine.Limit(depth=20))
             post_eval = post_info["score"].white().score(mate_score=10000) or 0
+            
+            # Determine game phase
             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
+
+            # Calculate evaluation loss
             eval_loss = abs(pre_eval - post_eval)
-            classification = Classification.BOOK if book_move else Classification.BLUNDER
+            
+            # Initial classification
+            classification = Classification.BOOK if book_move else None
+            if not classification:
+                for classif in centipawn_classifications:
+                    threshold = get_evaluation_loss_threshold(classif, pre_eval)
+                    if eval_loss <= threshold:
+                        classification = classif
+                        break
+                classification = classification or Classification.BLUNDER
+
+            # Check for missed opportunities
+            is_winning = abs(pre_eval) >= FORCED_WIN_THRESHOLD
+            is_forced_win = pre_info["score"].is_mate() and pre_info["score"].relative.mate() <= MISS_MATE_THRESHOLD
+            if is_winning and move != best_move and (eval_loss >= MISS_CENTIPAWN_LOSS or is_forced_win):
+                classification = Classification.MISS
+
+            # Check for brilliant moves
+            if classification == Classification.BEST:
+                if pre_eval < -150 and post_eval >= 150:
+                    classification = Classification.GREAT
+                elif pre_eval < -300 and post_eval >= 300:
+                    classification = Classification.BRILLIANT
+
+            # Track classifications
             player = "white" if board.turn == chess.BLACK else "black"
-            classifications[player][current_phase.value].append(classification.value)
-            results["moves"].append({
+            classifications[player][current_phase].append(classification)
+            phase_data[current_phase].append(classification)
+
+            # Add move analysis to result
+            result["move_analysis"].append({
                 "move_number": move_number,
-                "player": player,
+                "player": "White" if board.turn == chess.BLACK else "Black",
                 "move": move.uci(),
                 "evaluation": post_eval / 100,
                 "evaluation_loss": eval_loss / 100,
                 "classification": classification.value
             })
 
+        # Phase analysis
         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"]
+            moves = phase_data[phase]
+            if moves:
+                rating = get_phase_rating(moves)
+                result["phase_analysis"][phase.value] = {
+                    "rating": rating.value,
+                    "move_count": len(moves)
+                }
+
+        # Player summaries
+        for color in ["white", "black"]:
+            player = game.headers["White" if color == "white" else "Black"]
+            counts = {c.value: 0 for c in Classification}
+            
+            for phase in GamePhase:
+                phase_moves = classifications[color][phase]
+                for m in phase_moves:
+                    counts[m.value] += 1
+            
+            result["player_summaries"][player] = counts
+
+    def convert_enums(obj):
+        if isinstance(obj, Enum):  # Convert Enum to its value
+            return obj.value
+        if isinstance(obj, dict):  # Recursively handle dicts
+            return {k: convert_enums(v) for k, v in obj.items()}
+        if isinstance(obj, list):  # Recursively handle lists
+            return [convert_enums(i) for i in obj]
+        return obj  # Return other types as they are
+
+    json_result = convert_enums(result)
+
+    return JSONResponse(content=json_result)
+
+
+def get_phase_rating(classified_moves: List[Classification]) -> Classification:
+    if not classified_moves:
+        return Classification.GOOD
+        
+    total = sum(classification_values[m] for m in classified_moves)
+    average = total / len(classified_moves)
+    
+    rating_order = [
+        (Classification.BRILLIANT, 0.95),
+        (Classification.GREAT, 0.85),
+        (Classification.BEST, 0.75),
+        (Classification.EXCELLENT, 0.65),
+        (Classification.GOOD, 0.5),
+        (Classification.INACCURACY, 0.35),
+        (Classification.MISS, 0.25),
+        (Classification.MISTAKE, 0.15)
+    ]
     
-    return json.dumps(results, indent=4)
+    return next((c for c, t in rating_order if average >= t), Classification.BLUNDER)