예제 소스

play_two_models.py

#!/usr/bin/env python3
"""모델 목록에서 흑·백을 골라 대국한다.

수 상한 0은 안전 상한 1000수다. 예:

  python scripts/play_two_models.py --black gemma4 --white r5 --games 5 --max-moves 0
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

SAFETY_PLIES = 1000
MOVE_TIMEOUT_S = 360.0


def _load(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"import failed: {path}")
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod


mon = _load("monitor_mariago_web_play", ROOT / "scripts" / "monitor_mariago_web_play.py")
bridge = _load("mariago_ws_bridge", ROOT / "scripts" / "mariago_ws_bridge.py")
mdb = _load("match_db", ROOT / "scripts" / "match_db.py")

from rl_env.board import estimate_area_score  # noqa: E402


def now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def load_models() -> dict:
    path = ROOT / "scripts" / "models.json"
    data = json.loads(path.read_text(encoding="utf-8"))
    return data["models"]


def ask(spec: dict, board: list[int], side: int, epoch: int, rejects: list[tuple[int, int]]) -> dict:
    return bridge.request_move_sync(
        "127.0.0.1",
        int(spec["port"]),
        board,
        side,
        epoch,
        rejects,
        MOVE_TIMEOUT_S,
    )


def play_turn(spec: dict, board: list[int], side: int, epoch: int, ko) -> tuple[str, Optional[int], Optional[int], list[int], object, str]:
    """한 수. 반환: kind, row, col, new_board, new_ko, detail."""
    rejects: list[tuple[int, int]] = []
    last = ""
    for _ in range(2):
        resp = ask(spec, board, side, epoch, rejects)
        if resp.get("type") == "error":
            last = str(resp.get("error") or "error")
            continue
        if resp.get("pass"):
            return "pass", None, None, board, ko, "pass"
        row, col = int(resp["row"]), int(resp["col"])
        ok, reason, nxt, _cap, new_ko = mon.try_play(board, row, col, side, ko)
        if ok:
            return "move", row, col, nxt, new_ko, "ok"
        last = reason
        rejects.append((row, col))
    return "illegal", None, None, board, ko, last


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--black", required=True, help="scripts/models.json 의 모델 이름")
    ap.add_argument("--white", required=True)
    ap.add_argument("--games", type=int, default=5)
    ap.add_argument("--max-moves", type=int, default=0, help="0이면 1000수 안전 상한")
    ap.add_argument("--out-dir", default="logs/model_match")
    args = ap.parse_args()

    models = load_models()
    for name in (args.black, args.white):
        if name not in models:
            raise SystemExit(f"unknown model {name}; have {sorted(models)}")
    black = models[args.black]
    white = models[args.white]
    cap = SAFETY_PLIES if args.max_moves <= 0 else args.max_moves
    out = ROOT / args.out_dir
    out.mkdir(parents=True, exist_ok=True)

    db = mdb.MatchDB(out / "matches.sqlite")
    run_id = db.start_run(
        now(),
        {"black": args.black, "white": args.white, "games": args.games, "max_moves": args.max_moves},
        f"{args.black}_vs_{args.white}",
        None,
    )
    print(f"[match] run={run_id} {args.black}(B:{black['port']}) vs {args.white}(W:{white['port']}) games={args.games} cap={cap}", flush=True)

    judge = None
    try:
        judge = mdb.FinalJudge(visits=64)
    except Exception as exc:
        print(f"[match] katago judge off: {exc}", flush=True)

    for game_id in range(1, args.games + 1):
        board = mon.empty_board()
        ko = None
        epoch = game_id
        db.start_game(game_id, epoch, now())
        passes = 0
        fails = 0
        issues: list[str] = []
        ply = 0
        color_turn = 0
        end = "max_plies"
        print(f"[match] game {game_id} start", flush=True)
        while ply < cap:
            side_name = "B" if color_turn % 2 == 0 else "W"
            side = mon.BLACK if side_name == "B" else mon.WHITE
            spec = black if side_name == "B" else white
            t0 = time.time()
            kind, row, col, nxt, ko, detail = play_turn(spec, board, side, epoch, ko)
            dt = time.time() - t0
            db.add_move(
                game_id=game_id,
                ply=ply,
                color=side_name,
                kind=kind,
                board=board,
                ts=now(),
                row=row,
                col=col,
                dt=dt,
                attempt="first",
                server_rec={"model": args.black if side_name == "B" else args.white, "detail": detail},
            )
            ply += 1
            if kind == "illegal":
                fails += 1
                issues.append(f"{side_name}_illegal:{detail}")
                print(f"[match] g{game_id} ply {ply} {side_name} illegal {detail}", flush=True)
                if fails >= 3:
                    end = "error"
                    break
                continue
            fails = 0
            color_turn += 1
            if kind == "pass":
                passes += 1
                print(f"[match] g{game_id} ply {ply} {side_name} pass ({dt:.0f}s)", flush=True)
                if passes >= 2:
                    end = "double_pass"
                    break
                continue
            passes = 0
            board = nxt
            if ply % 20 == 0:
                print(f"[match] g{game_id} ply {ply} {side_name} ({row},{col}) {dt:.0f}s", flush=True)

        b_area, w_area, _dame = estimate_area_score(bytes(int(v) & 0xFF for v in board), 19)
        lead = None
        if judge is not None and end != "error":
            try:
                lead = judge.lead_black(board)
            except Exception as exc:
                print(f"[match] judge failed: {exc}", flush=True)
        if lead is None:
            margin = b_area - (w_area + 6.5)
            winner = "black" if margin > 0 else "white" if margin < 0 else "draw"
        else:
            winner = "black" if lead > 0 else "white" if lead < 0 else "draw"
        score = {"winner": winner, "black_score": b_area, "white_score": w_area + 6.5, "katago_lead_b": lead}
        db.finish_game(game_id, ts=now(), end_reason=end, plies=ply, score=score, issues=issues, katago_lead_b=lead)
        print(f"[match] game {game_id} {end} plies={ply} winner={winner} area={b_area}-{w_area + 6.5:.1f} lead={lead}", flush=True)

    db.close()
    print(f"[match] done {out / 'matches.sqlite'}", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())