예제 소스

probe_vlm_coords.py

#!/usr/bin/env python3
"""15×15 VLM 좌표 인식 진단.

실행 중 vlm_pva_server(:9299) 에 OMK3 로 알려진 국면을 보내고
응답 좌표가 빈 칸인지·합리적인지 출력한다.

  .venv/bin/python scripts/probe_vlm_coords.py
  .venv/bin/python scripts/probe_vlm_coords.py --host 127.0.0.1 --port 9299
"""

from __future__ import annotations

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

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

from PIL import Image, ImageDraw  # noqa: E402

MAGIC_OMK3 = b"OMK3"
MAGIC_RPLY = b"RPLY"


def render_png(board: bytes, size: int, cell: int = 28, pad: int = 14) -> bytes:
    side = pad * 2 + cell * (size - 1)
    img = Image.new("RGB", (side, side), (222, 191, 130))
    draw = ImageDraw.Draw(img)
    for i in range(size):
        a = pad + i * cell
        draw.line([(pad, a), (pad + cell * (size - 1), a)], fill=(40, 30, 20), width=1)
        draw.line([(a, pad), (a, pad + cell * (size - 1))], fill=(40, 30, 20), width=1)
    rad = max(4, cell // 3)
    for r in range(size):
        for c in range(size):
            v = board[r * size + c]
            if v == 0:
                continue
            x = pad + c * cell
            y = pad + r * cell
            color = (20, 20, 20) if v == 1 else (245, 245, 245)
            outline = (0, 0, 0)
            draw.ellipse([x - rad, y - rad, x + rad, y + rad], fill=color, outline=outline)
    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return buf.getvalue()


def make_board(size: int, stones: list[tuple[int, int, int]]) -> bytes:
    b = bytearray(size * size)
    for r, c, color in stones:
        b[r * size + c] = color
    return bytes(b)


def ask_move(
    host: str,
    port: int,
    board: bytes,
    size: int,
    ai_side: int = 2,
    timeout: float = 300.0,
) -> tuple[bool, str]:
    png = render_png(board, size)
    body = bytes([size & 0xFF, ai_side & 0xFF]) + struct.pack(">I", len(png)) + png + board
    frame = MAGIC_OMK3 + struct.pack(">I", len(body)) + body
    s = socket.create_connection((host, port), timeout=10)
    s.settimeout(timeout)
    try:
        s.sendall(frame)
        head = _recv_exact(s, 4)
        if head != MAGIC_RPLY:
            return False, f"bad header {head!r}"
        flag = _recv_exact(s, 1)[0]
        if flag == 1:
            rc = _recv_exact(s, 2)
            return True, f"{rc[0]} {rc[1]}"
        if flag == 0:
            n = struct.unpack(">H", _recv_exact(s, 2))[0]
            msg = _recv_exact(s, n).decode("utf-8", errors="replace")
            return False, msg
        return False, f"unknown flag {flag}"
    finally:
        s.close()


def _recv_exact(s: socket.socket, n: int) -> bytes:
    buf = b""
    while len(buf) < n:
        chunk = s.recv(n - len(buf))
        if not chunk:
            raise ConnectionError("eof")
        buf += chunk
    return buf


def main() -> int:
    p = argparse.ArgumentParser(description="Probe VLM coordinate recognition on 15x15")
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--port", type=int, default=9299)
    p.add_argument("--size", type=int, default=15)
    args = p.parse_args()
    size = args.size
    mid = size // 2

    cases = [
        ("empty opening", [], 2),
        ("black center only", [(mid, mid, 1)], 2),
        (
            "known cluster",
            [
                (mid, mid, 1),
                (mid, mid + 1, 1),
                (mid + 1, mid, 2),
                (mid - 1, mid, 2),
            ],
            2,
        ),
        (
            "edge trap (occupied near 13,*)",
            [(13, 12, 2), (13, 13, 2), (13, 14, 2), (7, 7, 1)] if size >= 15 else [(mid, mid, 1)],
            2,
        ),
    ]

    print(f"probe → {args.host}:{args.port} size={size}")
    print("서버 로그의 'VLM raw reply' 도 함께 보세요.\n")

    # 텍스트 보드 포맷 미리보기 (로컬 import)
    try:
        from vlm_pva_server import _board_to_text, parse_two_integers

        sample = make_board(size, [(mid, mid, 1), (mid, mid + 1, 2)])
        print("--- board text sample ---")
        print(_board_to_text(sample, size))
        print("--- parse self-check ---")
        for s in (
            "7 8",
            f"I play {mid} {mid + 2}",
            "Black stones: [(7, 7)]\n8 9",
            "0 1 2 3 ... answer: 10 11",
        ):
            print(f"  {s!r} → {parse_two_integers(s, size)}")
        print()
    except Exception as e:
        print(f"(local import skip: {e})\n")

    ok_n = 0
    for name, stones, ai_side in cases:
        board = make_board(size, stones)
        print(f"== {name} ==")
        print(f"  stones={stones}")
        try:
            ok, msg = ask_move(args.host, args.port, board, size, ai_side=ai_side)
        except Exception as e:
            print(f"  FAIL transport: {e}")
            continue
        if not ok:
            print(f"  server error: {msg}")
            continue
        try:
            r_s, c_s = msg.split()
            r, c = int(r_s), int(c_s)
        except Exception:
            print(f"  bad reply: {msg!r}")
            continue
        empty = board[r * size + c] == 0
        in_b = 0 <= r < size and 0 <= c < size
        mark = "OK" if empty and in_b else "BAD"
        if empty and in_b:
            ok_n += 1
        print(f"  reply=({r},{c}) empty={empty} → {mark}")
    print(f"\nsummary: {ok_n}/{len(cases)} empty legal replies")
    return 0 if ok_n == len(cases) else 1


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