mergedd getFen

.gitattributes

@@ -0,0 +1,2 @@
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.exe filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1 @@
+api_tokens

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.9
+
+# Set the working directory in the container
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy everything from the current directory to the container
+COPY . .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

main.py

@@ -1,12 +1,24 @@
 import io
+import os
+import tempfile
 from fastapi import FastAPI, File, UploadFile, Form
 from fastapi.responses import JSONResponse, StreamingResponse
 from PIL import Image, UnidentifiedImageError
 from routes.segmentation import segment_chess_board
 from routes.detection import detect_pieces
 from routes.fen_generator import gen_fen
+from routes.chess_review import analyze_pgn
 from typing import List, Dict, Any, Union
 from pydantic import BaseModel
+import asyncio
+import sys
+import tracemalloc
+tracemalloc.start()
+
+
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
+
 
 app = FastAPI()
 
@@ -15,76 +27,13 @@ class DetectionResults(BaseModel):
     confidences: list
     classes: list
 
-@app.get("/")
+@app.get("/test")
 async def read_root():
     return {
         "name": "Narendra",
         "age": 20,
         "Gender": "Male"
     }
-
-@app.post("/getSeg")
-async def get_seg(file: UploadFile = File(...)):
-    print(f'Image received: {file.filename}')
-
-    try:
-        image_content = await file.read()
-        if not image_content:
-            return JSONResponse(content={"error": "Empty file uploaded"}, status_code=400)
-
-        try:
-            image = Image.open(io.BytesIO(image_content))
-        except UnidentifiedImageError:
-            return JSONResponse(content={"error": "Invalid image format"}, status_code=400)
-
-        # If segment_chess_board is async, use `await`, otherwise remove `await`
-        segmented_image = await segment_chess_board(image)
-
-        if isinstance(segmented_image, dict):
-            return JSONResponse(content=segmented_image, status_code=400)
-
-        # Save to in-memory bytes
-        img_bytes = io.BytesIO()
-        segmented_image.save(img_bytes, format="PNG")
-        img_bytes.seek(0)
-
-        print("Image successfully processed and returned")
-        return StreamingResponse(
-            img_bytes,
-            media_type="image/png",
-            headers={"Content-Disposition": "inline; filename=output.png"}
-        )
-
-
-    except Exception as e:
-        return JSONResponse(content={"error": str(e)}, status_code=500)
-    
-
-@app.post("/getCoords")
-async def get_coords(file: UploadFile = File(...)):
-    try:
-        image_content = await file.read()
-
-        if not image_content:
-            print("No image found")
-            return JSONResponse(content={"error": "Empty file uploaded"}, status_code=400)
-
-        try:
-            image = Image.open(io.BytesIO(image_content))
-        except UnidentifiedImageError:
-            return JSONResponse(content={"error": "Invalid image format"}, status_code=400)
-
-        detection_results = await detect_pieces(image)
-
-        if "error" in detection_results:
-            return JSONResponse(content=detection_results, status_code=400)
-
-        print("Image successfully processed and returned")
-        return JSONResponse(content={"detections": detection_results}, status_code=200)
-
-    except Exception as e:
-        print(f"Unexpected error: {str(e)}")
-        return JSONResponse(content={"error": "Unexpected error occurred", "details": str(e)}, status_code=500)
     
 
 @app.post("/getFen")
@@ -125,3 +74,30 @@ async def get_fen(file : UploadFile = File(), perspective : str = Form("w"), nex
     
     except Exception as e:
         return JSONResponse(content={"error": "Unexpected error occurred", "details": str(e)}, status_code=500)
+    
+@app.post('/getReview')
+async def getReview(file: UploadFile = File(...)):  
+    print(os.getcwd())
+    print("call recieved")
+
+    if not file.filename.endswith(".pgn"):
+        return JSONResponse(content={"error": "Invalid file format. Please upload a PGN file"}, status_code=400)
+      
+    try:
+        # Save the uploaded file to a temporary file
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".pgn") as tmp_file:
+            tmp_file.write(await file.read())
+            tmp_file_path = tmp_file.name
+
+        # Analyze the PGN file
+        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 analysis_result
+    
+    except Exception as e:
+        return  JSONResponse(content={"error": "Unexpected error occurred", "details": str(e)}, status_code=500)

models/SegModel (1).pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d577a96ac47f8a6222719816b4810cf4b207de17885599e6e068eacc36e252cf
+size 6787281

models/chessDetection3d.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e1b2e9d61b0ea9f1ce81341d1aa7e54ff5721797925dbc960133d578021a0e4c
+size 114439558

models/stockfish-windows-x86-64-avx2.exe

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:0b24c22f7894fa13ab27e32a29763055d0867dfb123d8763579dea5b7a91f419
+size 79811584

requirements.txt

@@ -0,0 +1,68 @@
+annotated-types==0.7.0
+anyio==4.8.0
+certifi==2025.1.31
+charset-normalizer==3.4.1
+chess==1.11.1
+click==8.1.8
+colorama==0.4.6
+contourpy==1.3.1
+cycler==0.12.1
+dnspython==2.7.0
+email_validator==2.2.0
+fastapi==0.115.8
+fastapi-cli==0.0.7
+filelock==3.17.0
+fonttools==4.56.0
+fsspec==2025.2.0
+h11==0.14.0
+httpcore==1.0.7
+httptools==0.6.4
+httpx==0.28.1
+idna==3.10
+Jinja2==3.1.5
+kiwisolver==1.4.8
+markdown-it-py==3.0.0
+MarkupSafe==3.0.2
+matplotlib==3.10.0
+mdurl==0.1.2
+mpmath==1.3.0
+networkx==3.4.2
+numpy==2.1.1
+opencv-python==4.11.0.86
+packaging==24.2
+pandas==2.2.3
+pillow==11.1.0
+psutil==7.0.0
+py-cpuinfo==9.0.0
+pydantic==2.10.6
+pydantic_core==2.27.2
+Pygments==2.19.1
+pyparsing==3.2.1
+python-dateutil==2.9.0.post0
+python-dotenv==1.0.1
+python-multipart==0.0.20
+pytz==2025.1
+PyYAML==6.0.2
+requests==2.32.3
+rich==13.9.4
+rich-toolkit==0.13.2
+scipy==1.15.2
+seaborn==0.13.2
+setuptools==75.8.0
+shellingham==1.5.4
+six==1.17.0
+sniffio==1.3.1
+starlette==0.45.3
+sympy==1.13.1
+torch==2.6.0
+torchvision==0.21.0
+tqdm==4.67.1
+typer==0.15.1
+typing_extensions==4.12.2
+tzdata==2025.1
+ultralytics==8.3.76
+ultralytics-thop==2.0.14
+urllib3==2.3.0
+uvicorn==0.34.0
+watchfiles==1.0.4
+websockets==15.0

routes/chess_review.py

@@ -0,0 +1,287 @@
+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
+from datetime import datetime
+import csv
+import json
+from fastapi.responses import JSONResponse
+
+app = FastAPI()
+
+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,
+]
+
+# Analysis parameters
+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 = 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):  
+    if board.fullmove_number > max_depth:  
+        return None  
+    fen = " ".join(board.fen().split()[:4])
+    return opening_book.get(fen)
+
+
+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 {"error": "No game found in the PGN file."}
+    
+    result = {
+        "move_analysis": [],
+        "phase_analysis": {},
+        "player_summaries": {}
+    }
+    
+    with chess.engine.SimpleEngine.popen_uci(engine_path) as engine:
+        board = game.board()
+        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, limit = chess.engine.Limit(time = 0.3))
+            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)
+            
+            # 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].append(classification)
+            phase_data[current_phase].append(classification)
+
+            # Add move analysis to result
+            result["move_analysis"].append({
+                "move_number": move_number,
+                "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:
+            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 next((c for c, t in rating_order if average >= t), Classification.BLUNDER)

routes/detection.py

@@ -1,6 +1,10 @@
 from ultralytics import YOLO
 from PIL import Image
-detect_model = YOLO(r'D:\venv\chess-vision\models\chessDetection3d.pt')
+import os 
+
+curr = os.getcwd()
+detect_model_path = os.path.join(curr, 'models', 'chessDetection3d.pt')
+detect_model = YOLO(detect_model_path)
 
 async def detect_pieces(image : Image):
     if image is None:

routes/segmentation.py

@@ -1,7 +1,11 @@
 from ultralytics import YOLO
 from PIL import Image
+import os
 
-seg_model = YOLO(r'D:\venv\chess-vision\models\SegModel (1).pt')
+curr = os.getcwd()
+seg_model_path = os.path.join(curr, 'models', 'SegModel (1).pt')
+
+seg_model = YOLO(seg_model_path)
 
 async def segment_chess_board(image : Image):
     if image is None: