게시판바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기예제 소스

예제 소스

바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개

sgf_parser.py

"""아주 작은 SGF 파서.

학습 데이터 변환에 필요한 것만 지원:
- SZ (보드 크기)
- HA (handicap 핸디캡, 정보용)
- KM (덤, 정보용)
- AB / AW (사전 배치 흑/백)
- B / W (수)
- 변형(variation) 의 경우 첫 메인 라인만 사용

좌표는 'a'..'s' (19 알파벳) 로 표기. 'a' = 0, 's' = 18.
B[] (빈 괄호) 또는 B[tt] 는 패스로 본다.
"""

from __future__ import annotations

import os
from dataclasses import dataclass, field
from typing import Iterable, Iterator, List, Optional, Tuple


@dataclass
class SgfGame:
    size: int = 19
    handicap: int = 0
    komi: float = 0.0
    setup_black: List[Tuple[int, int]] = field(default_factory=list)
    setup_white: List[Tuple[int, int]] = field(default_factory=list)
    moves: List[Tuple[str, Optional[Tuple[int, int]]]] = field(default_factory=list)
    """[(color('B'|'W'), (row,col) or None for pass), ...]"""
    rule_summary: str = ""
    result: str = ""              # RE 태그 원문 (예: "B+R", "W+12.5", "Draw")
    winner: Optional[int] = None  # 1=흑, 2=백, None=무승부/미상


def _decode_point(s: str, size: int) -> Optional[Tuple[int, int]]:
    if not s:
        return None
    s = s.strip().lower()
    if not s:
        return None
    # SGF: 'tt' 는 19x19 에서 패스 관용 표기
    if size == 19 and s == "tt":
        return None
    if len(s) < 2:
        return None
    col = ord(s[0]) - ord("a")
    row = ord(s[1]) - ord("a")
    if not (0 <= col < size and 0 <= row < size):
        return None
    return (row, col)


def _tokenize(text: str) -> Iterator[Tuple[str, str]]:
    """매우 단순한 SGF 토크나이저: (PROP_KEY, VALUE) 쌍 yield.

    여러 값이 붙은 경우 같은 KEY 로 여러 번 yield.
    PROP 가 없는 ';' 'newnode' 시그널은 ('NODE', '') 로.
    """
    i = 0
    n = len(text)
    while i < n:
        c = text[i]
        if c == ";":
            yield ("NODE", "")
            i += 1
            continue
        if c.isalpha():
            # property key: 연속 대문자
            j = i
            while j < n and text[j].isalpha():
                j += 1
            key = text[i:j].upper()
            i = j
            # 값들: '[ ... ]' 가 0회 이상 (사실 SGF 는 ≥1)
            while i < n:
                # 공백 스킵
                while i < n and text[i].isspace():
                    i += 1
                if i >= n or text[i] != "[":
                    break
                i += 1  # '[' 통과
                buf = []
                while i < n:
                    ch = text[i]
                    if ch == "\\" and i + 1 < n:
                        buf.append(text[i + 1])
                        i += 2
                        continue
                    if ch == "]":
                        i += 1
                        break
                    buf.append(ch)
                    i += 1
                yield (key, "".join(buf))
            continue
        # 그 외 ('(', ')', 공백 등) 는 무시 (변형 트리는 main line 만 쓴다는 가정)
        i += 1


def parse_sgf_text(text: str) -> Optional[SgfGame]:
    """텍스트에서 첫 게임의 메인 라인을 SgfGame 으로 변환. 실패하면 None."""
    g = SgfGame()
    in_root = False
    seen_first_node = False
    in_variation = 0
    # 변형 (...) 안에 들어가면 무시 (main line 외 가지)
    i = 0
    n = len(text)
    # 변형 무시는 ( ... ) 깊이로 직접 처리하기 위해 토크나이저 대신 단순 스캐닝.
    # 단, 첫 ( 은 게임 트리 자체이므로 변형이 아니다.
    depth = 0
    main_buf: List[str] = []
    while i < n:
        ch = text[i]
        if ch == "(":
            if depth == 0:
                # 게임 트리 시작 자체 → 메인에 포함
                main_buf.append(ch)
            else:
                # 변형 시작 → 다음 ')' 매칭까지 스킵
                pass
            depth += 1
            i += 1
            continue
        if ch == ")":
            depth -= 1
            if depth == 0:
                main_buf.append(ch)
            i += 1
            continue
        if depth <= 1:
            main_buf.append(ch)
        i += 1

    main_text = "".join(main_buf)
    for key, val in _tokenize(main_text):
        if key == "NODE":
            continue
        if key == "SZ":
            try:
                g.size = int(val)
            except ValueError:
                continue
        elif key == "HA":
            try:
                g.handicap = int(val)
            except ValueError:
                continue
        elif key == "KM":
            try:
                g.komi = float(val)
            except ValueError:
                continue
        elif key == "AB":
            p = _decode_point(val, g.size)
            if p is not None:
                g.setup_black.append(p)
        elif key == "AW":
            p = _decode_point(val, g.size)
            if p is not None:
                g.setup_white.append(p)
        elif key == "B":
            p = _decode_point(val, g.size)
            g.moves.append(("B", p))
        elif key == "W":
            p = _decode_point(val, g.size)
            g.moves.append(("W", p))
        elif key == "RE":
            g.result = (val or "").strip()
        # 그 외 메타 무시
    # RE 태그에서 승자 추출. SGF: "B+..." = 흑승, "W+..." = 백승, "Draw"/"0" = 무승부.
    if g.result:
        head = g.result[:1].upper()
        if head == "B":
            g.winner = 1
        elif head == "W":
            g.winner = 2
        # Draw/Void/? 등은 None 유지.
    g.rule_summary = f"SZ={g.size} HA={g.handicap} KM={g.komi} RE={g.result or '?'}"
    return g


def parse_sgf_file(path: str) -> Optional[SgfGame]:
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            return parse_sgf_text(f.read())
    except OSError:
        return None


def iter_sgf_files(root: str) -> Iterator[str]:
    for dirpath, _, filenames in os.walk(root):
        for fn in filenames:
            if fn.lower().endswith(".sgf"):
                yield os.path.join(dirpath, fn)