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

예제 소스

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

ladder.py

"""축(ladder) 판독기.

모양 규칙을 나열하지 않고 실제로 수를 번갈아 두며 읽는다.
  - 수비: 활로 1 그룹을 들고 (a) 활로로 뻗거나 (b) 단수 걸린 공격 돌을 따낸다.
          착수 후 활로 ≥3 이면 탈출, ≤1 이면 실패, 2 이면 공격 차례.
  - 공격: 활로 2 중 하나로 단수. 한 수라도 수비가 못 벗어나면 잡힘.
이렇게 읽으면 다음이 자동으로 반영된다.
  - 축머리(진행 방향의 수비 측 돌) → 뻗은 뒤 활로 ≥3 → 탈출
  - 공격 돌이 >/< 로 꺾이며 단수가 생김 → 수비가 따내며 탈출
  - 진행 방향의 공격 측 돌·판 가장자리 → 활로가 늘지 않아 잡힘
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

BOARD_SIZE = 19
Point = tuple[int, int]


class _Budget(Exception):
    pass


class _Board:
    __slots__ = ("n", "c")

    def __init__(self, cells: bytes | bytearray, size: int = BOARD_SIZE):
        self.n = size
        self.c = bytearray(cells)

    def copy(self) -> "_Board":
        b = _Board.__new__(_Board)
        b.n = self.n
        b.c = bytearray(self.c)
        return b

    def nbrs(self, idx: int) -> list[int]:
        n = self.n
        r, c = divmod(idx, n)
        out = []
        if r > 0:
            out.append(idx - n)
        if r < n - 1:
            out.append(idx + n)
        if c > 0:
            out.append(idx - 1)
        if c < n - 1:
            out.append(idx + 1)
        return out

    def group(self, idx: int) -> tuple[list[int], list[int]]:
        color = self.c[idx]
        stones = [idx]
        seen = {idx}
        libs: list[int] = []
        lib_seen: set[int] = set()
        i = 0
        while i < len(stones):
            for nb in self.nbrs(stones[i]):
                v = self.c[nb]
                if v == 0:
                    if nb not in lib_seen:
                        lib_seen.add(nb)
                        libs.append(nb)
                elif v == color and nb not in seen:
                    seen.add(nb)
                    stones.append(nb)
            i += 1
        return stones, libs

    def play(self, idx: int, who: int) -> Optional[int]:
        """착수. 따낸 돌 수를 반환, 불법(점유·자살)이면 None 이고 보드 불변."""
        if self.c[idx] != 0:
            return None
        opp = 3 - who
        self.c[idx] = who
        captured = 0
        for nb in self.nbrs(idx):
            if self.c[nb] != opp:
                continue
            stones, libs = self.group(nb)
            if not libs:
                captured += len(stones)
                for s in stones:
                    self.c[s] = 0
        if captured == 0 and not self.group(idx)[1]:
            self.c[idx] = 0
            return None
        return captured


@dataclass
class _Reader:
    target: int
    max_nodes: int
    max_depth: int = 240
    nodes: int = 0
    depth: int = 0

    def __post_init__(self) -> None:
        self.path: set[bytes] = set()

    def _tick(self) -> None:
        self.nodes += 1
        if self.nodes > self.max_nodes or self.depth > self.max_depth:
            raise _Budget()

    def _enter(self, b: _Board) -> bool:
        """같은 국면 반복(패 되따냄)이면 False — 반복시키는 쪽의 수는 성립하지 않는 것으로 본다."""
        key = bytes(b.c)
        if key in self.path:
            return False
        self.path.add(key)
        self.depth += 1
        return True

    def _leave(self, b: _Board) -> None:
        self.path.discard(bytes(b.c))
        self.depth -= 1

    def defender_escapes(self, b: _Board) -> bool:
        """target 그룹 활로 1, 수비 차례. 벗어나면 True."""
        self._tick()
        who = b.c[self.target]
        stones, libs = b.group(self.target)
        if len(libs) != 1:
            return len(libs) >= 2
        moves: list[int] = []
        seen_att: set[int] = set()
        for s in stones:
            for nb in b.nbrs(s):
                if b.c[nb] != 3 - who or nb in seen_att:
                    continue
                a_stones, a_libs = b.group(nb)
                seen_att.update(a_stones)
                if len(a_libs) == 1 and a_libs[0] not in moves:
                    moves.append(a_libs[0])
        if libs[0] not in moves:
            moves.append(libs[0])
        for m in moves:
            b2 = b.copy()
            if b2.play(m, who) is None:
                continue
            n_libs = len(b2.group(self.target)[1])
            if n_libs >= 3:
                return True
            if n_libs <= 1:
                continue
            if not self._enter(b2):
                continue
            try:
                if not self.attacker_captures(b2):
                    return True
            finally:
                self._leave(b2)
        return False

    def attacker_captures(self, b: _Board) -> bool:
        """target 그룹 활로 2, 공격 차례. 축으로 잡히면 True."""
        self._tick()
        who = b.c[self.target]
        _, libs = b.group(self.target)
        if len(libs) <= 1:
            return True
        if len(libs) > 2:
            return False
        for m in libs:
            b2 = b.copy()
            if b2.play(m, 3 - who) is None:
                continue
            if b2.c[self.target] == 0:
                return True
            if not self._enter(b2):
                continue
            try:
                if not self.defender_escapes(b2):
                    return True
            finally:
                self._leave(b2)
        return False


def _target_idx(cells: bytes | bytearray, r: int, c: int, size: int) -> int:
    idx = r * size + c
    if cells[idx] not in (1, 2):
        raise ValueError(f"no stone at ({r},{c})")
    return idx


def defender_can_escape(
    cells: bytes | bytearray, r: int, c: int, *, size: int = BOARD_SIZE, max_nodes: int = 20000
) -> Optional[bool]:
    """(r,c) 그룹(활로 1)의 주인이 둘 차례일 때 살아 나갈 수 있나. 판독 한도 초과면 None."""
    b = _Board(cells, size)
    rd = _Reader(_target_idx(cells, r, c, size), max_nodes)
    try:
        return rd.defender_escapes(b)
    except _Budget:
        return None


def attacker_can_capture(
    cells: bytes | bytearray, r: int, c: int, *, size: int = BOARD_SIZE, max_nodes: int = 20000
) -> Optional[bool]:
    """(r,c) 그룹(활로 ≤2)을 상대가 먼저 두어 축/단수로 잡을 수 있나. 한도 초과면 None."""
    b = _Board(cells, size)
    rd = _Reader(_target_idx(cells, r, c, size), max_nodes)
    try:
        return rd.attacker_captures(b)
    except _Budget:
        return None


def ladder_line(
    cells: bytes | bytearray, r: int, c: int, *, size: int = BOARD_SIZE, max_plies: int = 120
) -> list[tuple[int, Point]]:
    """활로 1 그룹이 계속 뻗고 공격이 잡는 수로 단수칠 때의 진행 (who, (r,c)) 목록.

    학습 데이터 생성·설명용. 수비가 탈출하면 탈출 수까지 포함하고 멈춘다.
    """
    b = _Board(cells, size)
    target = _target_idx(cells, r, c, size)
    who = b.c[target]
    out: list[tuple[int, Point]] = []
    while len(out) < max_plies and b.c[target] == who:
        _, libs = b.group(target)
        if len(libs) != 1:
            break
        m = libs[0]
        if b.play(m, who) is None:
            break
        out.append((who, divmod(m, size)))
        _, libs = b.group(target)
        if len(libs) != 2:
            break
        chosen = None
        for a in libs:
            b2 = b.copy()
            if b2.play(a, 3 - who) is None:
                continue
            rd = _Reader(target, 20000)
            try:
                if b2.c[target] == 0 or not rd.defender_escapes(b2):
                    chosen = a
                    break
            except _Budget:
                continue
        if chosen is None:
            break
        b.play(chosen, 3 - who)
        out.append((3 - who, divmod(chosen, size)))
    return out


@dataclass
class LadderVerdict:
    point: Point
    kind: str  # futile_escape | broken_chase | working_chase
    group_size: int
    group_at: Point


def futile_escapes(cells: bytes | bytearray, who: int, *, size: int = BOARD_SIZE) -> list[LadderVerdict]:
    """who 의 단수 그룹 중 뻗어도 축으로 잡히는 그룹의 '뻗는 점'. 두면 돌만 더 잃는다."""
    b = _Board(cells, size)
    out: list[LadderVerdict] = []
    seen: set[int] = set()
    for idx in range(size * size):
        if b.c[idx] != who or idx in seen:
            continue
        stones, libs = b.group(idx)
        seen.update(stones)
        if len(libs) != 1:
            continue
        lib = libs[0]
        b2 = b.copy()
        if b2.play(lib, who) is None or b2.c[idx] != who:
            continue
        if len(b2.group(idx)[1]) >= 3:
            continue
        r, c = divmod(idx, size)
        if defender_can_escape(b.c, r, c, size=size) is False:
            captures = b.copy().play(lib, who) or 0
            if captures:
                continue
            out.append(LadderVerdict(divmod(lib, size), "futile_escape", len(stones), (r, c)))
    return out


def chase_verdict(
    cells: bytes | bytearray, r: int, c: int, who: int, *, size: int = BOARD_SIZE
) -> Optional[LadderVerdict]:
    """who 가 (r,c) 에 두어 상대 그룹을 단수칠 때 축이 되는지.

    working_chase: 상대가 어떻게 해도 못 벗어남.
    broken_chase : 상대가 뻗으면 벗어남(축머리·꺾임 단수 등) — 쫓으면 약점만 남긴다.
    단수가 아니거나 판독 불가면 None.
    """
    b = _Board(cells, size)
    idx = r * size + c
    if b.play(idx, who) is None:
        return None
    best: Optional[LadderVerdict] = None
    for nb in b.nbrs(idx):
        if b.c[nb] != 3 - who:
            continue
        stones, libs = b.group(nb)
        if len(libs) != 1:
            continue
        nr, nc = divmod(nb, size)
        esc = defender_can_escape(b.c, nr, nc, size=size)
        if esc is None:
            continue
        kind = "broken_chase" if esc else "working_chase"
        v = LadderVerdict((r, c), kind, len(stones), (nr, nc))
        if best is None or (v.kind == "working_chase" and best.kind != "working_chase") or (
            v.kind == best.kind and v.group_size > best.group_size
        ):
            best = v
    return best