예제 소스

match_db.py

"""휴리스틱 vs R5 대국 SQLite 저장소.

대국 모니터(monitor_mariago_web_play.py)가 매 수를 기록하고, 백(R5) 응답마다
서버 jsonl(--log-jsonl)에서 같은 국면(board_hash)의 move 레코드를 찾아 붙인다.
PASS 원인은 moves.pass_reason / server_rec(JSON)로 조회한다.

  sqlite3 logs/<run>/matches.sqlite \
    "select pass_reason, count(*) from moves where color='W' and kind='pass' group by 1"
"""

from __future__ import annotations

import json
import os
import sqlite3
import zlib
from pathlib import Path
from typing import Optional, Sequence

SCHEMA = """
create table if not exists runs (
    id integer primary key autoincrement,
    started_at text not null,
    args_json text,
    heuristic_version text,
    server_log text
);
create table if not exists games (
    run_id integer not null,
    game_id integer not null,
    epoch integer,
    started_at text,
    finished_at text,
    end_reason text,
    plies integer,
    winner text,
    black_area real,
    white_area real,
    score_json text,
    issues_json text,
    katago_lead_b real,
    katago_winner text,
    primary key (run_id, game_id)
);
create table if not exists moves (
    id integer primary key autoincrement,
    run_id integer not null,
    game_id integer not null,
    ply integer not null,
    color text not null,
    attempt text not null default 'first',
    kind text not null,
    row integer,
    col integer,
    dt real,
    empty integer,
    board_hash text,
    board text,
    ts text,
    pass_reason text,
    pass_requested integer,
    late_endgame integer,
    unsettled_empty integer,
    legal_n integer,
    server_rec text
);
create index if not exists moves_game on moves(run_id, game_id, ply);
create index if not exists moves_pass on moves(kind, pass_reason);
"""


def board_hash(board: Sequence[int], side: int) -> str:
    """vlm_server.server.board_hash 와 같은 값."""
    return f"{zlib.crc32(bytes(int(v) for v in board) + bytes([int(side) & 0xFF])) & 0xFFFFFFFF:08x}"


def board_text(board: Sequence[int]) -> str:
    return "".join(".XO"[int(v)] for v in board)


class ServerLogTail:
    """서버 jsonl 을 이어 읽어 board_hash 가 같은 최신 move 레코드를 돌려준다."""

    def __init__(self, path: Optional[Path]):
        self.path = path
        self.offset = 0
        if path is not None and path.exists():
            self.offset = path.stat().st_size

    def grab(self, bh: str) -> Optional[dict]:
        if self.path is None or not self.path.exists():
            return None
        size = self.path.stat().st_size
        if size < self.offset:
            self.offset = 0
        if size == self.offset:
            return None
        with self.path.open("rb") as f:
            f.seek(self.offset)
            chunk = f.read(size - self.offset)
        cut = chunk.rfind(b"\n")
        if cut < 0:
            return None
        self.offset += cut + 1
        found = None
        for raw in chunk[: cut + 1].splitlines():
            try:
                rec = json.loads(raw)
            except ValueError:
                continue
            if rec.get("event") == "move" and rec.get("board_hash") == bh:
                found = rec
        return found


class MatchDB:
    def __init__(self, path: Path):
        path.parent.mkdir(parents=True, exist_ok=True)
        self.con = sqlite3.connect(str(path))
        self.con.execute("pragma journal_mode=wal")
        self.con.executescript(SCHEMA)
        self.con.commit()
        self.run_id: Optional[int] = None

    def start_run(self, ts: str, args: dict, heuristic_version: str, server_log: Optional[Path]) -> int:
        cur = self.con.execute(
            "insert into runs(started_at, args_json, heuristic_version, server_log) values (?,?,?,?)",
            (ts, json.dumps(args, ensure_ascii=False), heuristic_version, str(server_log) if server_log else None),
        )
        self.con.commit()
        self.run_id = int(cur.lastrowid)
        return self.run_id

    def start_game(self, game_id: int, epoch: int, ts: str) -> None:
        self.con.execute(
            "insert or replace into games(run_id, game_id, epoch, started_at) values (?,?,?,?)",
            (self.run_id, game_id, epoch, ts),
        )
        self.con.commit()

    def add_move(
        self,
        *,
        game_id: int,
        ply: int,
        color: str,
        kind: str,
        board: Sequence[int],
        ts: str,
        row: Optional[int] = None,
        col: Optional[int] = None,
        dt: Optional[float] = None,
        attempt: str = "first",
        server_rec: Optional[dict] = None,
    ) -> None:
        side = 1 if color == "B" else 2
        rec = server_rec or {}
        self.con.execute(
            """insert into moves(run_id, game_id, ply, color, attempt, kind, row, col, dt, empty,
                   board_hash, board, ts, pass_reason, pass_requested, late_endgame,
                   unsettled_empty, legal_n, server_rec)
               values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
            (
                self.run_id,
                game_id,
                ply,
                color,
                attempt,
                kind,
                row,
                col,
                None if dt is None else round(dt, 3),
                sum(1 for v in board if int(v) == 0),
                board_hash(board, side),
                board_text(board),
                ts,
                rec.get("pass_reason"),
                None if "pass_requested" not in rec else int(bool(rec.get("pass_requested"))),
                None if "late_endgame" not in rec else int(bool(rec.get("late_endgame"))),
                rec.get("unsettled_empty"),
                rec.get("legal_n"),
                json.dumps(rec, ensure_ascii=False) if rec else None,
            ),
        )
        self.con.commit()

    def finish_game(
        self,
        game_id: int,
        *,
        ts: str,
        end_reason: str,
        plies: int,
        score: Optional[dict],
        issues: list,
        katago_lead_b: Optional[float] = None,
    ) -> None:
        sc = score or {}
        kw = None
        if katago_lead_b is not None:
            kw = "black" if katago_lead_b > 0 else ("white" if katago_lead_b < 0 else "draw")
        self.con.execute(
            """update games set finished_at=?, end_reason=?, plies=?, winner=?, black_area=?,
                   white_area=?, score_json=?, issues_json=?, katago_lead_b=?, katago_winner=?
               where run_id=? and game_id=?""",
            (
                ts,
                end_reason,
                plies,
                sc.get("winner"),
                sc.get("black_score"),
                sc.get("white_score"),
                json.dumps(sc, ensure_ascii=False),
                json.dumps(issues, ensure_ascii=False),
                katago_lead_b,
                kw,
                self.run_id,
                game_id,
            ),
        )
        self.con.commit()

    def close(self) -> None:
        try:
            self.con.close()
        except Exception:
            pass


class FinalJudge:
    """종국 판을 KataGo로 계가(흑 기준 scoreLead). 사석 제거가 없는 모니터 계가 보정용."""

    def __init__(self, visits: int = 400, komi: float = 6.5):
        from vlm_server.dataset_ladder import _KataGo

        self.kg = _KataGo(visits, komi)

    def lead_black(self, board: Sequence[int]) -> Optional[float]:
        from vlm_server.dataset_ladder import _gtp

        n = 19
        stones = [["B" if int(v) == 1 else "W", _gtp(divmod(i, n))] for i, v in enumerate(board) if int(v)]
        try:
            res = self.kg.j._query(
                {
                    "initialStones": stones,
                    "initialPlayer": "B",
                    "moves": [],
                    "rules": "korean",
                    "komi": self.kg.komi,
                    "boardXSize": n,
                    "boardYSize": n,
                    "analyzeTurns": [0],
                    "maxVisits": self.kg.visits,
                }
            )
        except Exception:
            return None
        lead = res.get("rootInfo", {}).get("scoreLead")
        return None if lead is None else round(float(lead), 2)


def default_server_log(root: Path) -> Optional[Path]:
    env = os.environ.get("MARIAGO_VLM_LOG_JSONL")
    if env:
        return Path(env)
    p = root / "logs" / "vlm_eval_day" / "play_r5.jsonl"
    return p