app.py

# ──────────────────────────────────────────────────────────────────────────────
# 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

TOKEN = os.environ.get("HF_TOKEN")

# ─────────────────────────────────────────────────────────────────────────────
# 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 _fh:
    exec(_fh.read(), globals())  # noqa: S102 — trusted internal model file

# ─────────────────────────────────────────────────────────────────────────────
# 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,
)
_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",    []))

print(
    f"[INIT] Zone DB loaded: "
    f"W={len(zonedb.winning_matrices)} "
    f"L={len(zonedb.losing_matrices)} "
    f"D={len(zonedb.draw_matrices)}"
)

# ─────────────────────────────────────────────────────────────────────────────
# 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"


def move_to_uci(move: tuple) -> str:
    fr, fc, tr, tc = move
    # rank = 8 - row  (row 0 = rank 8, row 7 = rank 1)
    return _COLS[fc] + str(8 - fr) + _COLS[tc] + str(8 - tr)


# ─────────────────────────────────────────────────────────────────────────────
# 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, rank_str in enumerate(rank_parts):
        c = 0
        for ch in rank_str:
            if ch == "B":
                # Black piece at top rows → PLAYER1 (direction=+1 downward)
                board[r, c] = 1
                c += 1
            elif ch == "W":
                # White piece at bottom rows → PLAYER2 (direction=−1 upward)
                board[r, c] = 2
                c += 1
            elif ch.isdigit():
                c += int(ch)   # empty squares
            else:
                raise ValueError(f"Unexpected FEN char {ch!r} in rank {r}: {rank_str!r}")
        if c != 8:
            raise ValueError(f"Rank {r} has {c} columns, expected 8: {rank_str!r}")
    return board


# ─────────────────────────────────────────────────────────────────────────────
# 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

    # 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)

    # ── 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 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):
            if candidate in legal_set:
                best_move = candidate
                break
        if best_move is None:
            print(f"[WARNING] no visited move is legal; top visited={list(visits.keys())[:5]}")

    # ── Fallback: first legal move from get_legal_moves() ────────────────────
    if best_move is None:
        if legal_list:
            best_move = legal_list[0]
            print(f"[WARNING] using first legal move as fallback: {best_move}")
        else:
            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


# ─────────────────────────────────────────────────────────────────────────────
# 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 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="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()