예제 소스

watch_human_vs_vlm.py

#!/usr/bin/env python3
"""사람 vs VLM 실전 모니터 — serve jsonl + baduk 터미널만 따라간다."""

from __future__ import annotations

import json
import time
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
JSONL = ROOT / "logs/vlm_eval_day/play_r5.jsonl"
OUT_DIR = ROOT / "logs/live_human_vs_vlm"
LIVE = OUT_DIR / "LIVE.md"
EVENTS = OUT_DIR / "events.jsonl"
TERM_DIR = Path.home() / ".cursor/projects/Volumes-MY-WORK-mywork-rl-go/terminals"


def iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def coord(row, col, is_pass=False) -> str:
    if is_pass or row is None or col is None:
        return "PASS"
    letters = "ABCDEFGHJKLMNOPQRST"
    try:
        return f"{letters[int(col)]}{19 - int(row)} ({row},{col})"
    except Exception:
        return f"({row},{col})"


def side_name(ai_side) -> str:
    return "흑(B)" if int(ai_side) == 1 else "백(W)" if int(ai_side) == 2 else f"side={ai_side}"


def parse_term_header(path: Path) -> dict:
    meta = {}
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return meta
    for line in text.splitlines()[:20]:
        if line.startswith("command:"):
            meta["command"] = line.split(":", 1)[1].strip().strip('"')
        if line.startswith("last_command:"):
            meta["last_command"] = line.split(":", 1)[1].strip()
        if line.startswith("title:"):
            meta["title"] = line.split(":", 1)[1].strip()
    return meta


def is_baduk_terminal(path: Path) -> bool:
    meta = parse_term_header(path)
    blob = " ".join(str(v) for v in meta.values())
    if "watch_human_vs_vlm" in blob or "daemon_text_client" in blob or "vlm_server serve" in blob:
        return False
    return "baduk" in blob.lower()


def write_live(moves: list[dict], qt_tail: list[str], note: str) -> None:
    lines = [
        f"# 사람 vs VLM 실전 — {iso()}",
        "",
        note,
        "",
        f"- VLM 응답 수(이번 세션): **{len(moves)}**",
        f"- jsonl: `{JSONL}`",
        "",
        "## VLM 수순",
        "",
    ]
    if not moves:
        lines.append("_아직 VLM 수가 없습니다. Qt에서 [시작] 후 두면 여기 쌓입니다._")
    else:
        lines.append("| # | 시각 | 색 | 좌표 | retry | illegal | urgent | Δ | prefer |")
        lines.append("|---|------|----|------|-------|---------|--------|---|---------|")
        for i, m in enumerate(moves, 1):
            ts = (m.get("ts") or "")[11:19]
            lines.append(
                "| {i} | {ts} | {side} | {c} | {retry} | {ill} | {urg} | {d} | {pref} |".format(
                    i=i,
                    ts=ts,
                    side=side_name(m.get("ai_side", 0)),
                    c=coord(m.get("row"), m.get("col"), bool(m.get("pass"))),
                    retry=m.get("retry_count", 0),
                    ill=m.get("illegal_reason") or "-",
                    urg="Y" if m.get("urgent_atari") else "-",
                    d=m.get("score_delta", "-"),
                    pref=m.get("vlm_prefer") or "-",
                )
            )
    lines += ["", "## Qt 새 로그", ""]
    if qt_tail:
        lines += [f"- `{x}`" for x in qt_tail[-24:]]
    else:
        lines.append(
            "_baduk 터미널 대기 중. Cursor 터미널에서 "
            "`./build/baduk.app/Contents/MacOS/baduk` 실행 후 "
            "흑=사람 / 백=VLM → [시작]._"
        )
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    LIVE.write_text("\n".join(lines) + "\n", encoding="utf-8")


def main() -> int:
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    baseline = 0
    if JSONL.exists():
        with JSONL.open(encoding="utf-8") as f:
            baseline = sum(1 for _ in f)
    (OUT_DIR / "baseline.txt").write_text(str(baseline), encoding="utf-8")
    print(f"[watch] baseline jsonl lines={baseline}", flush=True)
    print("[watch] serve 유지 — 사람 vs VLM 대기 (baduk 터미널만 추적)", flush=True)

    moves: list[dict] = []
    qt_offsets: dict[str, int] = {}
    qt_recent: list[str] = []
    last_size = JSONL.stat().st_size if JSONL.exists() else 0
    write_live(moves, [], f"- **state**: waiting  baseline=`{baseline}`  mode=`흑=휴리스틱 / 백=VLM`")

    while True:
        qt_path = None
        if TERM_DIR.is_dir():
            for p in sorted(TERM_DIR.glob("*.txt"), key=lambda x: x.stat().st_mtime, reverse=True):
                if is_baduk_terminal(p):
                    qt_path = p
                    break

        if qt_path is not None:
            key = str(qt_path)
            data = qt_path.read_bytes()
            off = qt_offsets.get(key)
            if off is None:
                qt_offsets[key] = len(data)
            elif len(data) < off:
                qt_offsets[key] = 0
                off = 0
            if off is not None and len(data) > off:
                chunk = data[off:].decode("utf-8", errors="replace")
                qt_offsets[key] = len(data)
                for raw in chunk.splitlines():
                    s = raw.strip()
                    if not s or s.startswith("---"):
                        continue
                    interesting = (
                        "[VLM]" in s
                        or "VLM " in s
                        or "거부" in s
                        or "PASS" in s
                        or "시작" in s
                        or "계가" in s
                        or "기권" in s
                        or "패(" in s
                        or "atari" in s.lower()
                        or "low-line" in s
                        or "selfFill" in s
                        or "아다리" in s
                    )
                    if interesting:
                        qt_recent.append(s)
                        if len(qt_recent) > 80:
                            qt_recent = qt_recent[-80:]
                        print(f"[qt] {s}", flush=True)

        if JSONL.exists():
            size = JSONL.stat().st_size
            if size < last_size:
                last_size = size
            if size > last_size:
                with JSONL.open(encoding="utf-8") as f:
                    f.seek(last_size)
                    chunk = f.read()
                last_size = size
                for line in chunk.splitlines():
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        rec = json.loads(line)
                    except json.JSONDecodeError:
                        continue
                    if rec.get("event") != "move":
                        continue
                    moves.append(rec)
                    with EVENTS.open("a", encoding="utf-8") as ef:
                        ef.write(json.dumps(rec, ensure_ascii=False) + "\n")
                    print(
                        "[vlm] {side} {c} retry={r} illegal={ill} urgent={u} Δ={d} prefer={p}".format(
                            side=side_name(rec.get("ai_side", 0)),
                            c=coord(rec.get("row"), rec.get("col"), bool(rec.get("pass"))),
                            r=rec.get("retry_count", 0),
                            ill=rec.get("illegal_reason") or "-",
                            u="Y" if rec.get("urgent_atari") else "-",
                            d=rec.get("score_delta", "-"),
                            p=rec.get("vlm_prefer") or "-",
                        ),
                        flush=True,
                    )

        write_live(
            moves,
            qt_recent,
            f"- **state**: watching  baseline=`{baseline}`  qt=`{qt_path.name if qt_path else '-'}`  mode=`흑=휴리스틱 / 백=VLM`",
        )
        time.sleep(0.8)


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