update app.py

app.py

@@ -1,121 +1,203 @@
+# ──────────────────────────────────────────────────────────────────────────────
+# space_app.py — HF Gradio Space for Breakthrough move prediction
+#
+# Deploy this file as app.py in your Hugging Face Space.
+#
+# Coordinate system (identical to apps/games/breakthrough_engine.py):
+#   rows 0–7, 0-indexed
+#   row 0 = rank 8 = TOP  (Black / 'B' side)
+#   row 7 = rank 1 = BOTTOM (White / 'W' side)
+#   cols 0–7 → 'a'–'h'
+#   rank formula: rank = 8 − row
+#
+# Piece → MCVS player mapping (matches engine directions):
+#   'B' → PLAYER1 = 1, direction = +1 (moves DOWN, row increases toward rank 1)
+#   'W' → PLAYER2 = 2, direction = −1 (moves UP,   row decreases toward rank 8)
+#
+# move_to_uci is the exact inverse of breakthrough_engine._sq_to_coords:
+#   (fr, fc, tr, tc) → cols[fc] + str(8−fr) + cols[tc] + str(8−tr)
+#
+# Gradio endpoint exposed:
+#   api_name="get_move" → /gradio_api/call/get_move  (Gradio 4 SSE queue)
+#   inputs: [fen: str, player: str]  → POST {"data": [fen, player]}
+#   output: uci_move: str            ← SSE  data: ["e4e3"]
+# ──────────────────────────────────────────────────────────────────────────────
+import os
+
 import gradio as gr
 import numpy as np
 from huggingface_hub import hf_hub_download
-import os
 
 TOKEN = os.environ.get("HF_TOKEN")
 
-
-# MODEL repo (model type)
-model_repo = "test1978/breakthrough-model"
-model_path = hf_hub_download(
-    model_repo,
+# ─────────────────────────────────────────────────────────────────────────────
+# 1.  Load breakthrough_mcvs.py from Hub model repo into this namespace.
+#     After exec() the following names are available globally:
+#       Breakthrough, MCVSSearcher, HilbertOrderedZoneDatabase,
+#       ABCModelDynamic, WeightedMatrixABC, move_to_index, …
+# ─────────────────────────────────────────────────────────────────────────────
+_model_path = hf_hub_download(
+    "test1978/breakthrough-model",
     "breakthrough_mcvs.py",
     repo_type="model",
     token=TOKEN,
 )
-with open(model_path, "r", encoding="utf-8-sig") as f:
-    exec(f.read())
+with open(_model_path, "r", encoding="utf-8-sig") as _fh:
+    exec(_fh.read(), globals())  # noqa: S102 — trusted internal model file
 
-
-# DB dataset (dataset type!)
-db_repo = "test1978/breakthrough-data"
-db_path = hf_hub_download(
-    db_repo,
+# ─────────────────────────────────────────────────────────────────────────────
+# 2.  Load the zone database from Hub dataset repo.
+# ─────────────────────────────────────────────────────────────────────────────
+_db_path = hf_hub_download(
+    "test1978/breakthrough-data",
     "breakthrough_zone_db.npz",
     repo_type="dataset",
     token=TOKEN,
 )
-zonedb_data = np.load(db_path, allow_pickle=True)
-
-
-# Init YOUR zone DB
-zonedb = HilbertOrderedZoneDatabase()
-zonedb.winning_matrices = list(zonedb_data.get("winning", []))
-zonedb.losing_matrices = list(zonedb_data.get("losing", []))
-zonedb.draw_matrices = list(zonedb_data.get("draw", []))
+_db_data = np.load(_db_path, allow_pickle=True)
 
+zonedb = HilbertOrderedZoneDatabase()  # noqa: F821 — defined by exec above
+zonedb.winning_matrices = list(_db_data.get("winning", []))
+zonedb.losing_matrices  = list(_db_data.get("losing",  []))
+zonedb.draw_matrices    = list(_db_data.get("draw",    []))
 
-def parse_board(board_text):
-    print(f"[DEBUG] parse_board raw input:\n{repr(board_text)}")
-    lines = board_text.strip().splitlines()
-    rows = [line.strip() for line in lines if line.strip()]
-    if len(rows) != 8:
-        raise ValueError(f"Board must have exactly 8 rows (got {len(rows)}), raw lines:\n{lines}")
+print(
+    f"[INIT] Zone DB loaded: "
+    f"W={len(zonedb.winning_matrices)} "
+    f"L={len(zonedb.losing_matrices)} "
+    f"D={len(zonedb.draw_matrices)}"
+)
 
-    board = np.zeros((8, 8), dtype=np.int32)
-    mapping = {".": 0, "1": 1, "2": 2}
-    for r, line in enumerate(rows):
-        parts = line.split()
-        if len(parts) != 8:
-            raise ValueError(f"Each row must have 8 space-separated cells (row {r+1}: {line})")
-        for c, cell in enumerate(parts):
-            if cell not in mapping:
-                raise ValueError(f"Use '.', '1', or '2' only (cell {r},{c}: {cell})")
-            board[r, c] = mapping[cell]
-    return board
+# ─────────────────────────────────────────────────────────────────────────────
+# 3.  move_to_uci — exact inverse of breakthrough_engine._sq_to_coords
+#
+# breakthrough_engine._coords_to_sq(row, col):
+#     rank_idx = 7 - row          # row 0 → rank_idx 7 → rank "8"
+#     return f"{_FILES[col]}{_RANKS[rank_idx]}"   # _RANKS = "12345678"
+#   ⟹ rank = rank_idx + 1 = 8 - row
+#
+# Therefore:
+#   move_to_uci((fr, fc, tr, tc))  =  cols[fc] + str(8−fr) + cols[tc] + str(8−tr)
+#
+# Verification:
+#   (4, 1, 3, 1) → 'b' + str(4) + 'b' + str(5) = 'b4b5'
+#   bot-runner: 'b4' = _sq_to_coords('b4') = (4,1)  ✓
+#               'b5' = _sq_to_coords('b5') = (3,1)  ✓  → move (4,1)→(3,1)
+#   (7, 0, 6, 0) → 'a1a2'  (White, row 7→6, rank 1→2)       ✓
+#   (0, 3, 1, 3) → 'd8d7'  (Black, row 0→1, rank 8→7)       ✓
+# ─────────────────────────────────────────────────────────────────────────────
+_COLS = "abcdefgh"
 
 
-# ALIGN: coordinate system with bot_runner
-# Rows are 0-indexed, row 0 = TOP (black side), row 7 = BOTTOM (white side).
-# UCI rank = row + 1  (so row 0 -> rank 1, row 7 -> rank 8).
-def move_to_uci(move):
+def move_to_uci(move: tuple) -> str:
     fr, fc, tr, tc = move
-    cols = "abcdefgh"
-    return cols[fc] + str(fr + 1) + cols[tc] + str(tr + 1)
+    # rank = 8 - row  (row 0 = rank 8, row 7 = rank 1)
+    return _COLS[fc] + str(8 - fr) + _COLS[tc] + str(8 - tr)
 
 
-def fen_to_board(fen):
-    rows_str = fen.strip().split(" ")[0]  # strip off ' w' or ' b'
-    parts = rows_str.split("/")
-    if len(parts) != 8:
-        raise ValueError(f"FEN must have 8 rows, got {len(parts)}")
+# ─────────────────────────────────────────────────────────────────────────────
+# 4.  fen_to_board — FEN string → Breakthrough.board (numpy int32 array, 8×8)
+#
+# FEN example: "BBBBBBBB/BBBBBBBB/8/8/8/8/WWWWWWWW/WWWWWWWW w"
+#   First rank string = row 0 = rank 8 = top (Black side)
+#   Last  rank string = row 7 = rank 1 = bottom (White side)
+#
+# Mapping (must match Breakthrough.get_legal_moves() directions):
+#   'B' → 1 = PLAYER1  (direction=+1, moves DOWN, row increases)
+#   'W' → 2 = PLAYER2  (direction=−1, moves UP,   row decreases)
+#
+# Why B→PLAYER1 and not PLAYER2?
+#   In breakthrough_mcvs.py, PLAYER1 starts at board[0:2,:] (top rows) with
+#   direction=+1, so it naturally represents the top-side piece ('B').
+#   PLAYER2 starts at board[6:8,:] (bottom rows) with direction=−1, matching 'W'.
+# ─────────────────────────────────────────────────────────────────────────────
+def fen_to_board(fen: str) -> np.ndarray:
+    ranks_str = fen.strip().split(" ")[0]
+    rank_parts = ranks_str.split("/")
+    if len(rank_parts) != 8:
+        raise ValueError(f"FEN must have 8 ranks separated by '/'; got {len(rank_parts)}: {fen!r}")
+
     board = np.zeros((8, 8), dtype=np.int32)
-    for r, row in enumerate(parts):
+    for r, rank_str in enumerate(rank_parts):
         c = 0
-        for ch in row:
+        for ch in rank_str:
             if ch == "B":
-                board[r, c] = 2
+                # Black piece at top rows → PLAYER1 (direction=+1 downward)
+                board[r, c] = 1
                 c += 1
             elif ch == "W":
-                board[r, c] = 1
+                # White piece at bottom rows → PLAYER2 (direction=−1 upward)
+                board[r, c] = 2
                 c += 1
             elif ch.isdigit():
-                skip = int(ch)
-                c += skip
+                c += int(ch)   # empty squares
             else:
-                raise ValueError(f"Invalid FEN char {ch} in row {r} ({row})")
+                raise ValueError(f"Unexpected FEN char {ch!r} in rank {r}: {rank_str!r}")
         if c != 8:
-            raise ValueError(f"Row {r} has {c} columns, should be 8")
+            raise ValueError(f"Rank {r} has {c} columns, expected 8: {rank_str!r}")
     return board
 
-# ALIGN: only return legal moves in get_move
-def get_move(board_text, player=None):
-    print(f"[DEBUG] get_move FEN board_text={repr(board_text)}, player={player}")
-    game = Breakthrough()
-    game.board = fen_to_board(board_text)
-    if player == "1":
-        game.move_count = 0
-    elif player == "2":
-        game.move_count = 1
-    else:
-        # Default: infer from FEN suffix or assume white
-        game.move_count = 0 if " w" in board_text.lower() else 1
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 5.  get_move — main function exposed via Gradio
+#
+# Parameters
+# ----------
+# fen    : Breakthrough FEN, e.g. "BBBBBBBB/8/8/8/8/8/8/WWWWWWWW w"
+# player : who moves next; accepted values:
+#            "w" or "2"  → White (PLAYER2, move_count = 1 = odd)
+#            "b" or "1"  → Black (PLAYER1, move_count = 0 = even)
+#            "" or None  → infer from FEN side-to-move suffix
+#                          " w" → White (move_count=1)
+#                          " b" → Black (move_count=0)
+#
+# Returns
+# -------
+# uci : str — a legal UCI move string ("e4e3", "b4b5", …).
+#   If search produces no legal move and no legal move exists at all,
+#   returns "0000" (a safe sentinel the bot-runner recognises as no-op).
+# ─────────────────────────────────────────────────────────────────────────────
+def get_move(fen: str, player: str = "") -> str:
+    print(f"[DEBUG] get_move FEN board_text={fen!r}, player={player!r}")
+
+    # ── Build game state ──────────────────────────────────────────────────────
+    game = Breakthrough()             # noqa: F821
+    game.board = fen_to_board(fen)
     game._cached_matrix = None
-    searcher = MCVSSearcher(None, None, zonedb, lambda_zone=1.0, k_zone=5)
+
+    # Determine which player moves next (move_count parity):
+    #   even → PLAYER1 (Black / 'b'), odd → PLAYER2 (White / 'w')
+    if player in ("w", "2"):
+        game.move_count = 1           # White = PLAYER2 = odd move
+    elif player in ("b", "1"):
+        game.move_count = 0           # Black = PLAYER1 = even move
+    else:
+        # Infer from FEN suffix — " w" means White to move = PLAYER2
+        game.move_count = 1 if " w" in fen.lower() else 0
+
+    # ── Run MCVS search ───────────────────────────────────────────────────────
+    searcher = MCVSSearcher(          # noqa: F821
+        policy_net=None,
+        value_net=None,
+        zone_db=zonedb,
+        lambda_zone=1.0,
+        k_zone=5,
+    )
     visits, _ = searcher.search_with_time_budget(game, 1.0)
 
-    # Debug: show how many legal moves and visits
-    legal_moves = list(game.get_legal_moves())
-    legal_set = set(legal_moves)
-    print(f"[DEBUG] legal moves count: {len(legal_moves)}")
-    if legal_moves:
-        print(f"[DEBUG] sample legal move: {legal_moves[0]}")
+    # ── Collect legal moves for validation and fallback ───────────────────────
+    legal_list = list(game.get_legal_moves())
+    legal_set  = set(legal_list)
+    print(f"[DEBUG] legal moves count: {len(legal_list)}")
+    if legal_list:
+        print(f"[DEBUG] sample legal move: {legal_list[0]}")
     print(f"[DEBUG] visits count: {len(visits)}")
     if visits:
         print(f"[DEBUG] sample visit key: {next(iter(visits))}")
 
-    # Pick the best visited move that is also legal
+    # ── Pick the best visited move that is actually legal ─────────────────────
+    # search_with_time_budget should only return legal moves, but we verify
+    # to guard against any residual coordinate-system bugs.
     best_move = None
     if visits:
         for candidate in sorted(visits, key=visits.get, reverse=True):
@@ -123,39 +205,53 @@ def get_move(board_text, player=None):
                 best_move = candidate
                 break
         if best_move is None:
-            print(f"[WARNING] no visited move is legal; visited={list(visits.keys())[:5]}")
+            print(f"[WARNING] no visited move is legal; top visited={list(visits.keys())[:5]}")
 
-    # Fall back to first legal move if needed
+    # ── Fallback: first legal move from get_legal_moves() ────────────────────
     if best_move is None:
-        if legal_moves:
-            best_move = legal_moves[0]
+        if legal_list:
+            best_move = legal_list[0]
             print(f"[WARNING] using first legal move as fallback: {best_move}")
         else:
-            # No legal moves at all — return a safe dummy
-            print("[WARNING] no legal moves in position; returning dummy move")
-            return move_to_uci((3, 3, 3, 3))
+            print("[WARNING] no legal moves in position — returning sentinel 0000")
+            return "0000"
+
+    uci = move_to_uci(best_move)
+    print(f"[DEBUG] get_move OK: FEN={fen} -> {uci}")
+    return uci
 
-    move = move_to_uci(best_move)
-    print(f"[DEBUG] get_move OK: FEN={board_text} -> {move}")
-    return move
 
+# ─────────────────────────────────────────────────────────────────────────────
+# 6.  Gradio Interface
+#
+# api_name="get_move"  →  Gradio 4 SSE queue endpoint:
+#   POST /gradio_api/call/get_move  body: {"data": [fen, player]}
+#   GET  /gradio_api/call/get_move/{event_id}  (SSE)
+#   → event: complete
+#     data: ["e4e3"]
+#
+# This matches what predict_breakthrough._try_space_api sends.
+# ─────────────────────────────────────────────────────────────────────────────
 demo = gr.Interface(
     fn=get_move,
-    inputs=gr.Textbox(
-        label="Breakthrough Board (1/2/.)",
-        value=(
-            "1 1 1 1 1 1 1 1\n"
-            "1 1 1 1 1 1 1 1\n"
-            ". . . . . . . .\n"
-            ". . . . . . . .\n"
-            ". . . . . . . .\n"
-            ". . . . . . . .\n"
-            "2 2 2 2 2 2 2 2\n"
-            "2 2 2 2 2 2 2 2"
+    inputs=[
+        gr.Textbox(
+            label="Breakthrough FEN",
+            placeholder="BBBBBBBB/BBBBBBBB/8/8/8/8/WWWWWWWW/WWWWWWWW w",
         ),
-    ),
+        gr.Textbox(
+            label="Player to move  (w / b)",
+            placeholder="w",
+            value="",
+        ),
+    ],
     outputs=gr.Textbox(label="UCI Move"),
-    title="🎯 BreakthroughMCVS Secure (Model + Dataset)",
+    title="Breakthrough Move Predictor",
+    description=(
+        "Returns a legal UCI move for the given Breakthrough position. "
+        "FEN format: BBBBBBBB/.../WWWWWWWW followed by  w  or  b."
+    ),
+    api_name="get_move",   # → /gradio_api/call/get_move
 )
 
 demo.launch()