예제 소스

vlm_text_client.py

#!/usr/bin/env python3
"""OMK3 VLM 텍스트 클라이언트 — 좌표 인식 / 수 품질 수동·자동 테스트.

대화형:
  .venv/bin/python scripts/vlm_text_client.py
  .venv/bin/python scripts/vlm_text_client.py --size 15 --host 127.0.0.1 --port 9299

한 번에 좌표 스모크:
  .venv/bin/python scripts/vlm_text_client.py --probe
  .venv/bin/python scripts/vlm_text_client.py --probe --size 15

명령 (대화형):
  show                 보드 출력
  put R C [B|W]        돌 놓기 (기본: 사람 색)
  del R C              돌 제거
  clear                빈 판
  side B|W|1|2         AI 색 (VLM이 둘 쪽)
  size 13|15|19        보드 크기 (판 초기화)
  ask                  VLM에게 한 수 요청 (돌은 자동으로 안 둠)
  play                 ask 후 빈 칸이면 AI 돌 착수
  auto N               play 를 N수 반복 (사람 랜덤 + AI)
  ping                 PING/PONG
  help / quit
"""

from __future__ import annotations

import argparse
import io
import random
import socket
import struct
import sys
import time
from pathlib import Path
from typing import Optional

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

from PIL import Image, ImageDraw  # noqa: E402

MAGIC_OMK3 = b"OMK3"
MAGIC_PING = b"PING"
MAGIC_RPLY = b"RPLY"
MAGIC_PONG = b"PONG"


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


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)
            draw.ellipse([x - rad, y - rad, x + rad, y + rad], fill=color, outline=(0, 0, 0))
    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return buf.getvalue()


def board_text(board: bytes, size: int, mark: Optional[tuple[int, int]] = None) -> str:
    col_hdr = "   " + " ".join(f"{c:2d}" for c in range(size))
    lines = [col_hdr]
    for r in range(size):
        cells = []
        for c in range(size):
            if mark == (r, c):
                cells.append(" *")
                continue
            v = board[r * size + c]
            cells.append(" ." if v == 0 else (" X" if v == 1 else " O"))
        lines.append(f"{r:2d} " + "".join(cells))
    return "\n".join(lines)


class VlmTextClient:
    def __init__(self, host: str, port: int, size: int = 15, ai_side: int = 2, timeout: float = 300.0):
        self.host = host
        self.port = port
        self.size = size
        self.ai_side = ai_side  # 1=B 2=W
        self.timeout = timeout
        self.board = bytearray(size * size)
        self.history: list[tuple[int, int, int]] = []  # r,c,color

    @property
    def human_side(self) -> int:
        return 1 if self.ai_side == 2 else 2

    def reset(self, size: Optional[int] = None) -> None:
        if size is not None:
            self.size = size
        self.board = bytearray(self.size * self.size)
        self.history.clear()

    def put(self, r: int, c: int, color: int) -> None:
        if not (0 <= r < self.size and 0 <= c < self.size):
            raise ValueError(f"범위 밖: ({r},{c})")
        if self.board[r * self.size + c] != 0:
            raise ValueError(f"이미 돌 있음: ({r},{c})")
        self.board[r * self.size + c] = color
        self.history.append((r, c, color))

    def remove(self, r: int, c: int) -> None:
        self.board[r * self.size + c] = 0

    def empty_cells(self) -> list[tuple[int, int]]:
        s = self.size
        return [(r, c) for r in range(s) for c in range(s) if self.board[r * s + c] == 0]

    def ping(self) -> float:
        t0 = time.perf_counter()
        with socket.create_connection((self.host, self.port), timeout=5) as sock:
            sock.settimeout(5)
            sock.sendall(MAGIC_PING)
            head = recv_exact(sock, 4)
            if head != MAGIC_PONG:
                raise RuntimeError(f"bad ping reply {head!r}")
        return time.perf_counter() - t0

    def ask(self) -> tuple[bool, str, float]:
        """(ok, 'r c' or error, elapsed_sec)."""
        png = render_png(bytes(self.board), self.size)
        body = (
            bytes([self.size & 0xFF, self.ai_side & 0xFF])
            + struct.pack(">I", len(png))
            + png
            + bytes(self.board)
        )
        frame = MAGIC_OMK3 + struct.pack(">I", len(body)) + body
        t0 = time.perf_counter()
        with socket.create_connection((self.host, self.port), timeout=10) as sock:
            sock.settimeout(self.timeout)
            sock.sendall(frame)
            head = recv_exact(sock, 4)
            if head != MAGIC_RPLY:
                return False, f"bad header {head!r}", time.perf_counter() - t0
            flag = recv_exact(sock, 1)[0]
            if flag == 1:
                rc = recv_exact(sock, 2)
                return True, f"{rc[0]} {rc[1]}", time.perf_counter() - t0
            if flag == 0:
                n = struct.unpack(">H", recv_exact(sock, 2))[0]
                msg = recv_exact(sock, n).decode("utf-8", errors="replace")
                return False, msg, time.perf_counter() - t0
            return False, f"unknown flag {flag}", time.perf_counter() - t0

    def judge(self, r: int, c: int) -> str:
        if not (0 <= r < self.size and 0 <= c < self.size):
            return "OUT_OF_RANGE"
        if self.board[r * self.size + c] != 0:
            return "OCCUPIED"
        return "LEGAL_EMPTY"


def parse_rc(parts: list[str], start: int = 0) -> tuple[int, int]:
    if len(parts) < start + 2:
        raise ValueError("행 열 필요")
    return int(parts[start]), int(parts[start + 1])


def parse_color(tok: str, default: int) -> int:
    t = tok.strip().upper()
    if t in ("B", "X", "1", "BLACK"):
        return 1
    if t in ("W", "O", "2", "WHITE"):
        return 2
    raise ValueError(f"색 모름: {tok}")


def run_probe(cli: VlmTextClient) -> int:
    size = cli.size
    mid = size // 2
    cases: list[tuple[str, list[tuple[int, int, int]]]] = [
        ("empty", []),
        ("black@center", [(mid, mid, 1)]),
        (
            "cluster",
            [
                (mid, mid, 1),
                (mid, mid + 1, 1),
                (mid + 1, mid, 2),
                (mid - 1, mid, 2),
            ],
        ),
    ]
    if size >= 15:
        cases.append(
            (
                "edge_trap_13",
                [(13, 12, 2), (13, 13, 2), (13, 14, 2), (7, 7, 1)],
            )
        )

    print(f"=== probe {cli.host}:{cli.port} size={size} ai_side={cli.ai_side} ===")
    try:
        ms = cli.ping() * 1000
        print(f"PING ok ({ms:.1f} ms)\n")
    except Exception as e:
        print(f"PING FAIL: {e}")
        return 2

    ok_n = 0
    suspect_n = 0
    for name, stones in cases:
        cli.reset()
        for r, c, color in stones:
            cli.put(r, c, color)
        print(f"-- {name} --")
        print(board_text(bytes(cli.board), size))
        try:
            ok, msg, elapsed = cli.ask()
        except Exception as e:
            print(f"  transport FAIL: {e}\n")
            continue
        if not ok:
            print(f"  server error ({elapsed:.2f}s): {msg}\n")
            continue
        r_s, c_s = msg.split()
        r, c = int(r_s), int(c_s)
        verdict = cli.judge(r, c)
        # 재추론 3회(~0.4s×3)면 보통 1초+. RPLY는 보정 후 좌표라 LEGAL_EMPTY여도
        # 모델 raw가 점유 칸이었을 수 있음 → 서버 로그 '불법 좌표' 확인.
        slow = elapsed >= 0.9
        if verdict == "LEGAL_EMPTY":
            ok_n += 1
            tag = "OK"
            if slow:
                suspect_n += 1
                tag = "OK(slow≈retry/fallback?)"
        else:
            tag = "BAD"
        print(f"  reply=({r},{c}) {verdict} → {tag}  ({elapsed:.2f}s)")
        print(board_text(bytes(cli.board), size, mark=(r, c) if verdict != "OUT_OF_RANGE" else None))
        print()

    print(f"summary: {ok_n}/{len(cases)} LEGAL_EMPTY  (slow/suspect={suspect_n})")
    print("note: RPLY는 서버 보정 후 좌표. raw는 VLM 터미널 '불법 좌표'/'VLM raw reply' 로그 확인.")
    return 0 if ok_n == len(cases) else 1


def repl(cli: VlmTextClient) -> int:
    print(
        f"VLM text client → {cli.host}:{cli.port}  size={cli.size}  "
        f"AI={'W' if cli.ai_side == 2 else 'B'}  human={'B' if cli.human_side == 1 else 'W'}"
    )
    print("help 입력. Ctrl-D / quit 종료.\n")
    try:
        ms = cli.ping() * 1000
        print(f"PING ok ({ms:.1f} ms)")
    except Exception as e:
        print(f"PING FAIL: {e}  (서버 켠 뒤 다시 ping)")

    print(board_text(bytes(cli.board), cli.size))

    while True:
        try:
            line = input("> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            return 0
        if not line:
            continue
        parts = line.split()
        cmd = parts[0].lower()

        try:
            if cmd in ("q", "quit", "exit"):
                return 0
            if cmd in ("h", "help", "?"):
                print(__doc__)
                continue
            if cmd in ("show", "board", "b"):
                side = "W" if cli.ai_side == 2 else "B"
                print(f"size={cli.size} AI={side} stones={sum(1 for x in cli.board if x)}")
                print(board_text(bytes(cli.board), cli.size))
                continue
            if cmd == "clear":
                cli.reset()
                print(board_text(bytes(cli.board), cli.size))
                continue
            if cmd == "size":
                n = int(parts[1])
                if n not in (13, 15, 19):
                    raise ValueError("size는 13/15/19")
                cli.reset(n)
                print(board_text(bytes(cli.board), cli.size))
                continue
            if cmd == "side":
                cli.ai_side = parse_color(parts[1], cli.ai_side)
                print(f"AI={'W' if cli.ai_side == 2 else 'B'}")
                continue
            if cmd in ("put", "p", "place"):
                r, c = parse_rc(parts, 1)
                color = parse_color(parts[3], cli.human_side) if len(parts) > 3 else cli.human_side
                cli.put(r, c, color)
                print(board_text(bytes(cli.board), cli.size, mark=(r, c)))
                continue
            if cmd in ("del", "rm"):
                r, c = parse_rc(parts, 1)
                cli.remove(r, c)
                print(board_text(bytes(cli.board), cli.size))
                continue
            if cmd == "ping":
                ms = cli.ping() * 1000
                print(f"PONG {ms:.1f} ms")
                continue
            if cmd == "ask":
                print("VLM 요청 중…")
                ok, msg, elapsed = cli.ask()
                if not ok:
                    print(f"FAIL ({elapsed:.2f}s): {msg}")
                    continue
                r, c = map(int, msg.split())
                verdict = cli.judge(r, c)
                print(f"reply=({r},{c}) {verdict}  ({elapsed:.2f}s)")
                print(board_text(bytes(cli.board), cli.size, mark=(r, c) if verdict != "OUT_OF_RANGE" else None))
                continue
            if cmd == "play":
                print("VLM 요청 중…")
                ok, msg, elapsed = cli.ask()
                if not ok:
                    print(f"FAIL ({elapsed:.2f}s): {msg}")
                    continue
                r, c = map(int, msg.split())
                verdict = cli.judge(r, c)
                print(f"reply=({r},{c}) {verdict}  ({elapsed:.2f}s)")
                if verdict == "LEGAL_EMPTY":
                    cli.put(r, c, cli.ai_side)
                    print("(착수됨)")
                else:
                    print("(착수 안 함 — 불법 좌표)")
                print(board_text(bytes(cli.board), cli.size, mark=(r, c) if verdict != "OUT_OF_RANGE" else None))
                continue
            if cmd == "auto":
                n = int(parts[1]) if len(parts) > 1 else 3
                rng = random.Random()
                for i in range(n):
                    empties = cli.empty_cells()
                    if not empties:
                        print("빈 칸 없음")
                        break
                    # 사람이 근처 랜덤
                    hr, hc = rng.choice(empties)
                    cli.put(hr, hc, cli.human_side)
                    print(f"\n[{i+1}/{n}] human ({hr},{hc})")
                    ok, msg, elapsed = cli.ask()
                    if not ok:
                        print(f"  FAIL: {msg}")
                        break
                    r, c = map(int, msg.split())
                    verdict = cli.judge(r, c)
                    print(f"  AI ({r},{c}) {verdict} ({elapsed:.2f}s)")
                    if verdict != "LEGAL_EMPTY":
                        print(board_text(bytes(cli.board), cli.size, mark=(r, c) if verdict != "OUT_OF_RANGE" else None))
                        print("중단: 불법 좌표")
                        break
                    cli.put(r, c, cli.ai_side)
                print(board_text(bytes(cli.board), cli.size))
                continue
            if cmd == "probe":
                return run_probe(cli)
            # 단축: "7 8" = put human
            if len(parts) == 2 and parts[0].lstrip("-").isdigit():
                r, c = int(parts[0]), int(parts[1])
                cli.put(r, c, cli.human_side)
                print(board_text(bytes(cli.board), cli.size, mark=(r, c)))
                continue
            print(f"모르는 명령: {cmd}  (help)")
        except Exception as e:
            print(f"! {e}")


def main() -> int:
    p = argparse.ArgumentParser(description="OMK3 VLM text client")
    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, choices=(13, 15, 19))
    p.add_argument("--ai-side", type=int, default=2, choices=(1, 2), help="1=Black 2=White")
    p.add_argument("--timeout", type=float, default=300.0)
    p.add_argument("--probe", action="store_true", help="좌표 스모크 테스트 후 종료")
    args = p.parse_args()

    cli = VlmTextClient(
        host=args.host,
        port=args.port,
        size=args.size,
        ai_side=args.ai_side,
        timeout=args.timeout,
    )
    if args.probe:
        return run_probe(cli)
    return repl(cli)


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