Download handler.py from test1978/chess-model: direct link, hf CLI and curl.
- Browser
- Download file 20.7 kB
-
https://huggingface.co/test1978/chess-model/resolve/2a04987c7b54e4018043b842e96d6ea8ea53f6f6/handler.py
- Command line
-
hf download hf://test1978/chess-model@2a04987c7b54e4018043b842e96d6ea8ea53f6f6/handler.py
-
curl -L -o handler.py https://huggingface.co/test1978/chess-model/resolve/2a04987c7b54e4018043b842e96d6ea8ea53f6f6/handler.py
20.7 kB
| # ββββββββββββββββββββββββββββββββββββββββββββββ | |
| # handler.py | |
| # | |
| # Custom handler for HF Inference Endpoints. | |
| # | |
| # When HF Inference Endpoints detects a handler.py | |
| # in the repo root, it uses this class to serve | |
| # predictions instead of the default Transformers | |
| # pipeline. | |
| # | |
| # Expected repo layout: | |
| # | |
| # your-username/chess-model/ | |
| # βββ handler.py β this file | |
| # βββ config_model.json β lists your Python modules + main | |
| # βββ config_data.json β lists data files + zone_db key | |
| # βββ chess_mcvs.py β your UCT search / eval module | |
| # βββ requirements.txt β runtime deps (numpy, huggingface_hub) | |
| # | |
| # config_model.json example: | |
| # | |
| # { | |
| # "files": ["matrix_model.py", "abc_model.py", "chess_mcvs.py"], | |
| # "main": "chess_mcvs" | |
| # } | |
| # | |
| # config_data.json example: | |
| # | |
| # { | |
| # "files": ["chess_zone_db.npz"], | |
| # "zone_db": "chess_zone_db.npz" | |
| # } | |
| # | |
| # Set the HF_DATA_REPO_ID environment variable to | |
| # the HF dataset repo holding your data files. | |
| # | |
| # Request format (sent by Agladiator): | |
| # | |
| # POST / | |
| # { | |
| # "inputs": { | |
| # "fen": "BBBBBBBB/BBBBBBBB/8/8/8/8/WWWWWWWW/WWWWWWWW w", | |
| # "player": "w", | |
| # "game_type": "chess" | |
| # } | |
| # } | |
| # | |
| # Response format (what you must return): | |
| # | |
| # {"move": "a2a3"} | |
| # | |
| # ββββββββββββββββββββββββββββββββββββββββββββββ | |
| from __future__ import annotations | |
| import importlib.util | |
| import json | |
| import logging | |
| import os | |
| import random | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| log = logging.getLogger(__name__) | |
| # βββββββββββββββββββββββββββββββββββββββββββββ | |
| # Chess board constants (used by the minimal legal-move generator) | |
| # βββββββββββββββββββββββββββββββββββββββββββββ | |
| BOARD_SIZE = 8 | |
| _FILES = "abcdefgh" | |
| _RANKS = "12345678" | |
| WHITE, BLACK = "w", "b" | |
| EMPTY = "." | |
| class EndpointHandler: | |
| """Custom handler for chess AI on HF Inference Endpoints. | |
| Lifecycle: | |
| 1. ``__init__`` is called once when the endpoint container starts. | |
| Use it to load your model artifacts (modules, zone DB, etc.). | |
| 2. ``__call__`` is called for every inference request. | |
| It receives the parsed JSON body and must return a dict. | |
| """ | |
| def __init__(self, path: str): | |
| """Load model artefacts from *path* (the cloned repo directory). | |
| Steps: | |
| 1. Read ``config_model.json`` to discover module files and main module. | |
| 2. Dynamically import each listed module. | |
| 3. Read ``config_data.json``, download data files, load zone DB. | |
| 4. Initialise the search engine. | |
| """ | |
| self.path = Path(path) | |
| self.predict_fn = None | |
| self.zone_db = None | |
| self.module = None | |
| # ββ 1. Read config_model.json βββββββββββ | |
| model_config_path = self.path / "config_model.json" | |
| if not model_config_path.exists(): | |
| log.warning("No config_model.json found in %s", path) | |
| model_config: dict = {} | |
| else: | |
| with open(model_config_path) as fh: | |
| model_config = json.load(fh) | |
| main_module_name = model_config.get("main", "") | |
| # ββ 2. Import listed modules ββββββββββββ | |
| # Add repo dir to sys.path so relative imports work. | |
| if str(self.path) not in sys.path: | |
| sys.path.insert(0, str(self.path)) | |
| for mod_file in model_config.get("files", []): | |
| mod_name = mod_file.removesuffix(".py") | |
| mod_path = self.path / mod_file | |
| if not mod_path.exists(): | |
| log.warning("Module file %s not found β skipping.", mod_file) | |
| continue | |
| spec = importlib.util.spec_from_file_location(mod_name, str(mod_path)) | |
| mod = importlib.util.module_from_spec(spec) | |
| sys.modules[mod_name] = mod | |
| spec.loader.exec_module(mod) | |
| log.info("Loaded module: %s", mod_name) | |
| # Mark the module matching "main" as the primary module. | |
| if mod_name == main_module_name: | |
| self.module = mod | |
| # ββ 3. Read config_data.json & load zone DB ββ | |
| data_config_path = self.path / "config_data.json" | |
| if not data_config_path.exists(): | |
| log.warning("No config_data.json found in %s", path) | |
| data_config: dict = {} | |
| else: | |
| with open(data_config_path) as fh: | |
| data_config = json.load(fh) | |
| data_repo = os.environ.get("HF_DATA_REPO_ID", "") | |
| downloaded_files: dict[str, str] = {} | |
| if data_repo: | |
| try: | |
| from huggingface_hub import hf_hub_download | |
| except Exception: | |
| hf_hub_download = None | |
| log.warning( | |
| "huggingface_hub not installed β cannot download data files. Add it to requirements.txt" | |
| ) | |
| if hf_hub_download is not None: | |
| for data_file in data_config.get("files", []): | |
| try: | |
| local_path = hf_hub_download( | |
| repo_id=data_repo, | |
| filename=data_file, | |
| repo_type="dataset", | |
| ) | |
| downloaded_files[data_file] = local_path | |
| log.info("Downloaded %s from %s", data_file, data_repo) | |
| except Exception: | |
| log.warning( | |
| "Failed to download %s from %s β skipping.", | |
| data_file, | |
| data_repo, | |
| ) | |
| else: | |
| log.warning( | |
| "HF_DATA_REPO_ID not set β skipping data file downloads." | |
| ) | |
| zone_db_filename = data_config.get("zone_db", "") | |
| if zone_db_filename and zone_db_filename in downloaded_files: | |
| self.zone_db = np.load( | |
| downloaded_files[zone_db_filename], allow_pickle=True | |
| ) | |
| log.info("Zone DB loaded from data repo %s", data_repo) | |
| elif zone_db_filename: | |
| local_zone = self.path / zone_db_filename | |
| if local_zone.exists(): | |
| self.zone_db = np.load(str(local_zone), allow_pickle=True) | |
| log.info("Zone DB loaded from local file %s", local_zone) | |
| else: | |
| log.warning("No zone DB found for '%s'.", zone_db_filename) | |
| # ββ 4. Initialise the predictor βββββββββ | |
| self._init_predictor() | |
| log.info("EndpointHandler initialised successfully.") | |
| # ββββββββββββββββββββββββββββββββββββββββββ | |
| # Predictor initialisation | |
| # ββββββββββββββββββββββββββββββββββββββββββ | |
| def _init_predictor(self): | |
| """Discover and configure the prediction function from the | |
| loaded module. Tries several common patterns: | |
| 1. ``module.get_move(fen, player, zone_db=...)`` | |
| 2. ``module.UCTSearcher(zone_db=...).search(fen, player)`` | |
| 3. ``module.predict(fen, player)`` | |
| """ | |
| if self.module is None: | |
| log.warning("No module loaded β predictions will use random fallback.") | |
| return | |
| # Pattern 1: module exposes a top-level get_move() | |
| if hasattr(self.module, "get_move"): | |
| raw_fn = self.module.get_move | |
| self.predict_fn = lambda fen, player: raw_fn( | |
| fen, player, zone_db=self.zone_db | |
| ) | |
| log.info("Using module.get_move() for predictions.") | |
| return | |
| # Pattern 2: module exposes a UCTSearcher class | |
| if hasattr(self.module, "UCTSearcher"): | |
| searcher_cls = self.module.UCTSearcher | |
| kwargs = {} | |
| if self.zone_db is not None: | |
| kwargs["zone_db"] = self.zone_db | |
| searcher = searcher_cls(**kwargs) | |
| self.predict_fn = lambda fen, player: searcher.search(fen, player) | |
| log.info("Using module.UCTSearcher for predictions.") | |
| return | |
| # Pattern 3: module exposes a predict() | |
| if hasattr(self.module, "predict"): | |
| raw_fn = self.module.predict | |
| self.predict_fn = lambda fen, player: raw_fn(fen, player) | |
| log.info("Using module.predict() for predictions.") | |
| return | |
| log.warning( | |
| "Module %s has no get_move / UCTSearcher / predict β " | |
| "predictions will use random fallback.", | |
| self.module.__name__, | |
| ) | |
| # ββββββββββββββββββββββββββββββββββββββββββ | |
| # Inference entry point | |
| # ββββββββββββββββββββββββββββββββββββββββββ | |
| def __call__(self, data: dict) -> dict: | |
| """Handle an inference request. | |
| Expected *data*:: | |
| {"inputs": {"fen": "...", "player": "w"}} | |
| Returns:: | |
| {"move": "a2a3"} | |
| Falls back to a random legal move if the model fails. | |
| """ | |
| inputs = data.get("inputs", data) | |
| fen = inputs.get("fen", "") | |
| player = inputs.get("player", WHITE) | |
| if not fen: | |
| return {"error": "Missing 'fen' in request."} | |
| # Try model prediction | |
| if self.predict_fn is not None: | |
| try: | |
| move = self.predict_fn(fen, player) | |
| if move and isinstance(move, str): | |
| return {"move": move} | |
| except Exception: | |
| log.exception("Model prediction failed for FEN=%s", fen) | |
| # Fallback: random legal move | |
| moves = self._legal_moves(fen) | |
| if not moves: | |
| return {"error": "No legal moves available.", "move": "0000"} | |
| # Prefer captures (diagonal moves in chess) | |
| captures = [m for m in moves if m[0] != m[2]] | |
| move = random.choice(captures) if captures else random.choice(moves) | |
| return {"move": move} | |
| # ββββββββββββββββββββββββββββββββββββββββββ | |
| # Minimal legal-move generator (self-contained) | |
| # ββββββββββββββββββββββββββββββββββββββββββ | |
| def _legal_moves(fen: str) -> list[str]: | |
| """Return a list of pseudo-legal UCI moves for a chess FEN. | |
| This minimal generator supports pawn moves (including promotions | |
| and en-passant), knights, bishops, rooks, queens, kings and | |
| basic castling availability. It does not perform check detection | |
| and therefore may return some pseudo-legal moves in checked positions. | |
| """ | |
| parts = fen.strip().split() | |
| board_part = parts[0] if parts else "" | |
| turn = parts[1] if len(parts) > 1 else WHITE | |
| castling = parts[2] if len(parts) > 2 else "-" | |
| ep_square = parts[3] if len(parts) > 3 else "-" | |
| # Parse FEN β grid (8 ranks, top (rank 8) β bottom (rank 1)) | |
| grid: list[list[str]] = [] | |
| for rank_str in board_part.split("/"): | |
| row: list[str] = [] | |
| for ch in rank_str: | |
| if ch.isdigit(): | |
| row.extend([EMPTY] * int(ch)) | |
| else: | |
| row.append(ch) | |
| # normalize row length | |
| if len(row) != BOARD_SIZE: | |
| row = (row + [EMPTY] * BOARD_SIZE)[:BOARD_SIZE] | |
| grid.append(row) | |
| while len(grid) < BOARD_SIZE: | |
| grid.append([EMPTY] * BOARD_SIZE) | |
| side_white = turn == WHITE | |
| def is_enemy(ch: str) -> bool: | |
| return ch != EMPTY and ch.isupper() != side_white | |
| def is_own(ch: str) -> bool: | |
| return ch != EMPTY and ch.isupper() == side_white | |
| moves: list[str] = [] | |
| # En-passant target square (if any) | |
| ep_r = ep_c = None | |
| if ep_square and ep_square != "-": | |
| try: | |
| ep_c = _FILES.index(ep_square[0]) | |
| ep_r = BOARD_SIZE - int(ep_square[1]) | |
| except Exception: | |
| ep_r = ep_c = None | |
| start_row = 6 if side_white else 1 | |
| last_row = 0 if side_white else 7 | |
| direction = -1 if side_white else 1 | |
| for r in range(BOARD_SIZE): | |
| for c in range(BOARD_SIZE): | |
| ch = grid[r][c] | |
| if ch == EMPTY: | |
| continue | |
| if ch.isupper() != side_white: | |
| continue | |
| piece = ch.lower() | |
| from_sq = f"{_FILES[c]}{_RANKS[BOARD_SIZE - 1 - r]}" | |
| if piece == "p": | |
| nr = r + direction | |
| # forward one | |
| if 0 <= nr < BOARD_SIZE and grid[nr][c] == EMPTY: | |
| # promotion | |
| if nr == last_row: | |
| for promo in ("q", "r", "b", "n"): | |
| moves.append( | |
| f"{from_sq}{_FILES[c]}{_RANKS[BOARD_SIZE - 1 - nr]}{promo}" | |
| ) | |
| else: | |
| moves.append( | |
| f"{from_sq}{_FILES[c]}{_RANKS[BOARD_SIZE - 1 - nr]}" | |
| ) | |
| # double move from start | |
| if r == start_row: | |
| nr2 = r + 2 * direction | |
| if 0 <= nr2 < BOARD_SIZE and grid[nr2][c] == EMPTY: | |
| moves.append( | |
| f"{from_sq}{_FILES[c]}{_RANKS[BOARD_SIZE - 1 - nr2]}" | |
| ) | |
| # captures (including en-passant) | |
| for dc in (-1, 1): | |
| nc = c + dc | |
| if nc < 0 or nc >= BOARD_SIZE: | |
| continue | |
| if 0 <= nr < BOARD_SIZE: | |
| target = grid[nr][nc] | |
| if is_enemy(target): | |
| if nr == last_row: | |
| for promo in ("q", "r", "b", "n"): | |
| moves.append( | |
| f"{from_sq}{_FILES[nc]}{_RANKS[BOARD_SIZE - 1 - nr]}{promo}" | |
| ) | |
| else: | |
| moves.append( | |
| f"{from_sq}{_FILES[nc]}{_RANKS[BOARD_SIZE - 1 - nr]}" | |
| ) | |
| # en-passant capture | |
| if ep_r is not None and ep_r == nr and ep_c == nc: | |
| cap_r = r | |
| cap_c = nc | |
| if ( | |
| 0 <= cap_r < BOARD_SIZE | |
| and grid[cap_r][cap_c].lower() == "p" | |
| and is_enemy(grid[cap_r][cap_c]) | |
| ): | |
| moves.append( | |
| f"{from_sq}{_FILES[nc]}{_RANKS[BOARD_SIZE - 1 - nr]}" | |
| ) | |
| elif piece == "n": | |
| for dr, dc in ( | |
| (-2, -1), | |
| (-2, 1), | |
| (-1, -2), | |
| (-1, 2), | |
| (1, -2), | |
| (1, 2), | |
| (2, -1), | |
| (2, 1), | |
| ): | |
| nr = r + dr | |
| nc = c + dc | |
| if 0 <= nr < BOARD_SIZE and 0 <= nc < BOARD_SIZE: | |
| target = grid[nr][nc] | |
| if not is_own(target): | |
| moves.append( | |
| f"{from_sq}{_FILES[nc]}{_RANKS[BOARD_SIZE - 1 - nr]}" | |
| ) | |
| elif piece in ("b", "r", "q"): | |
| directions = [] | |
| if piece in ("b", "q"): | |
| directions += [(-1, -1), (-1, 1), (1, -1), (1, 1)] | |
| if piece in ("r", "q"): | |
| directions += [(-1, 0), (1, 0), (0, -1), (0, 1)] | |
| for dr, dc in directions: | |
| nr = r + dr | |
| nc = c + dc | |
| while 0 <= nr < BOARD_SIZE and 0 <= nc < BOARD_SIZE: | |
| target = grid[nr][nc] | |
| if target == EMPTY: | |
| moves.append( | |
| f"{from_sq}{_FILES[nc]}{_RANKS[BOARD_SIZE - 1 - nr]}" | |
| ) | |
| else: | |
| if is_enemy(target): | |
| moves.append( | |
| f"{from_sq}{_FILES[nc]}{_RANKS[BOARD_SIZE - 1 - nr]}" | |
| ) | |
| break | |
| nr += dr | |
| nc += dc | |
| elif piece == "k": | |
| for dr in (-1, 0, 1): | |
| for dc in (-1, 0, 1): | |
| if dr == 0 and dc == 0: | |
| continue | |
| nr = r + dr | |
| nc = c + dc | |
| if 0 <= nr < BOARD_SIZE and 0 <= nc < BOARD_SIZE: | |
| target = grid[nr][nc] | |
| if not is_own(target): | |
| moves.append( | |
| f"{from_sq}{_FILES[nc]}{_RANKS[BOARD_SIZE - 1 - nr]}" | |
| ) | |
| # Basic castling checks (does not verify check conditions) | |
| if side_white: | |
| if "K" in castling and r == 7 and c == 4: | |
| if grid[7][5] == EMPTY and grid[7][6] == EMPTY and grid[7][7] != EMPTY and grid[7][7].isupper(): | |
| moves.append(f"{from_sq}g1") | |
| if "Q" in castling and r == 7 and c == 4: | |
| if ( | |
| grid[7][3] == EMPTY | |
| and grid[7][2] == EMPTY | |
| and grid[7][1] == EMPTY | |
| and grid[7][0] != EMPTY | |
| and grid[7][0].isupper() | |
| ): | |
| moves.append(f"{from_sq}c1") | |
| else: | |
| if "k" in castling and r == 0 and c == 4: | |
| if grid[0][5] == EMPTY and grid[0][6] == EMPTY and grid[0][7] != EMPTY and grid[0][7].islower(): | |
| moves.append(f"{from_sq}g8") | |
| if "q" in castling and r == 0 and c == 4: | |
| if ( | |
| grid[0][3] == EMPTY | |
| and grid[0][2] == EMPTY | |
| and grid[0][1] == EMPTY | |
| and grid[0][0] != EMPTY | |
| and grid[0][0].islower() | |
| ): | |
| moves.append(f"{from_sq}c8") | |
| return moves | |