Spaces:
Runtime error
Runtime error
| import json | |
| import chess | |
| import chess.engine | |
| from http.server import BaseHTTPRequestHandler, HTTPServer | |
| ENGINE_PATH = "/usr/games/stockfish" | |
| engine = chess.engine.SimpleEngine.popen_uci(ENGINE_PATH) | |
| def analyse_position(fen, depth): | |
| board = chess.Board(fen) | |
| info = engine.analyse(board, chess.engine.Limit(depth=depth)) | |
| score = info["score"].pov(chess.WHITE).score(mate_score=10000) | |
| best_move = info["pv"][0].uci() if "pv" in info and info["pv"] else None | |
| return score, best_move | |
| def classify(cp_loss): | |
| if cp_loss >= 200: | |
| return "blunder" | |
| elif cp_loss >= 100: | |
| return "mistake" | |
| elif cp_loss >= 50: | |
| return "inaccuracy" | |
| return None | |
| class Handler(BaseHTTPRequestHandler): | |
| def do_POST(self): | |
| length = int(self.headers.get("Content-Length", 0)) | |
| body = json.loads(self.rfile.read(length)) | |
| fens = body.get("fens", [])[:10] | |
| moves = body.get("moves", []) # SAN or UCI, len = len(fens) - 1 | |
| depth = body.get("depth", 12) | |
| if len(fens) < 2: | |
| self._respond({"error": "need at least 2 fens"}, 400) | |
| return | |
| # analyse every position once, reuse for before/after pairs | |
| analysed = [analyse_position(f, depth) for f in fens] | |
| flagged = [] | |
| for i in range(len(fens) - 1): | |
| board_before = chess.Board(fens[i]) | |
| mover_is_white = board_before.turn == chess.WHITE | |
| eval_before, best_move = analysed[i] | |
| eval_after, _ = analysed[i + 1] | |
| swing = (eval_after - eval_before) if mover_is_white else (eval_before - eval_after) | |
| cp_loss = max(0, -swing) | |
| severity = classify(cp_loss) | |
| # figure out what move was actually played | |
| played_move_uci = None | |
| if i < len(moves): | |
| try: | |
| played_move_uci = board_before.push_san(moves[i]).uci() | |
| except Exception: | |
| played_move_uci = None | |
| missed_best = ( | |
| best_move is not None | |
| and played_move_uci is not None | |
| and played_move_uci != best_move | |
| ) | |
| # flag if it's a real mistake tier OR a missed-best-move with non-trivial cost | |
| if severity or (missed_best and cp_loss >= 30): | |
| flagged.append({ | |
| "index": i, | |
| "fen_before": fens[i], | |
| "move_played": moves[i] if i < len(moves) else None, | |
| "best_move": best_move, | |
| "eval_before": eval_before, | |
| "eval_after": eval_after, | |
| "cp_loss": cp_loss, | |
| "severity": severity or "missed_chance", | |
| "missed_best_move": missed_best | |
| }) | |
| self._respond({"flagged_positions": flagged}) | |
| def _respond(self, data, status=200): | |
| self.send_response(status) | |
| self.send_header("Content-Type", "application/json") | |
| self.end_headers() | |
| self.wfile.write(json.dumps(data).encode()) | |
| def do_GET(self): | |
| self.send_response(200) | |
| self.end_headers() | |
| self.wfile.write(b"Stockfish engine is running") | |
| if __name__ == "__main__": | |
| server = HTTPServer(("0.0.0.0", 7860), Handler) | |
| server.serve_forever() |