예제 소스

ensemble_smoke.py

#!/usr/bin/env python3
"""앙상블 프록시 스모크 테스트 — 몇 개 국면을 보내 응답과 판정 로그를 확인.

  .venv/bin/python scripts/ensemble_smoke.py --port 9318
"""

from __future__ import annotations

import argparse
import io
import socket
import sys
import time
from pathlib import Path

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

from scripts.mariago_ensemble_proxy import MoveRequest, gtp, recv_reply, send_request  # noqa: E402
from vlm_server.dataset_baduk import render_board_png  # noqa: E402

N = 19
LETTERS = "ABCDEFGHJKLMNOPQRST"

POSITIONS = {
    "opening_W": ("W", "B:Q16 D4 Q4 W:D16"),
    "midgame_W": (
        "W",
        "B:Q16 D4 Q4 R10 C12 F3 O3 Q7 E16 "
        "W:D16 C14 F17 R14 O17 Q12 D10 C6 R3",
    ),
    "fight_B": (
        "B",
        "B:Q16 D4 Q4 Q10 P10 O11 D10 "
        "W:D16 Q14 R10 R11 P11 O12 R9 C10",
    ),
}


def parse(spec: str) -> bytes:
    board = bytearray(N * N)
    color = 0
    for tok in spec.split():
        if tok.startswith("B:"):
            color, tok = 1, tok[2:]
        elif tok.startswith("W:"):
            color, tok = 2, tok[2:]
        if not tok:
            continue
        c = LETTERS.index(tok[0])
        r = N - int(tok[1:])
        board[r * N + c] = color
    return bytes(board)


def png_bytes(board: bytes) -> bytes:
    buf = io.BytesIO()
    render_board_png(board, N, px=256).save(buf, format="PNG")
    return buf.getvalue()


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=9318)
    ap.add_argument("--timeout-s", type=float, default=240)
    args = ap.parse_args()

    for name, (side, spec) in POSITIONS.items():
        board = parse(spec)
        req = MoveRequest(1 if side == "B" else 2, png_bytes(board), board, 1234, [])
        t0 = time.perf_counter()
        with socket.create_connection(("127.0.0.1", args.port), timeout=10) as s:
            s.settimeout(args.timeout_s)
            send_request(s, req, 1234)
            ok, move, err = recv_reply(s)
        ms = (time.perf_counter() - t0) * 1000
        print(f"{name:10s} {side} → {gtp(move) if ok and move else err}  ({ms:.0f} ms)")
    return 0


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