예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
test_pass_guard.py
from __future__ import annotations
import sys
import threading
import types
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
try:
import PIL # noqa: F401
except ImportError:
pil = types.ModuleType("PIL")
pil.Image = types.SimpleNamespace(Image=object)
sys.modules["PIL"] = pil
from vlm_server.server import ( # noqa: E402
ServerConfig,
_GoTcpHandler,
_Logger,
_MiniGo,
_position_scoped_reject_bans,
)
N = 19
BLACK, WHITE = 1, 2
def make_board(black: list[tuple[int, int]], white: list[tuple[int, int]]) -> bytes:
cells = bytearray(N * N)
for r, c in black:
cells[r * N + c] = BLACK
for r, c in white:
cells[r * N + c] = WHITE
return bytes(cells)
class _PassModel:
def predict_move(self, *args, **kwargs):
return None, None, True
class _QuietMoveModel:
def __init__(self, move: tuple[int, int]) -> None:
self.move = move
def predict_move(self, *args, **kwargs):
return self.move[0], self.move[1], False
class _Socket:
def __init__(self) -> None:
self.sent = bytearray()
def sendall(self, data: bytes) -> None:
self.sent.extend(data)
class PassGuardTest(unittest.TestCase):
def _move(
self, board: bytes, side: int, model=None
) -> tuple[bool, tuple[int, int] | None]:
cfg = ServerConfig(
host="127.0.0.1",
port=0,
model_id="test",
max_new_tokens=1,
temperature=0.0,
load_model=True,
log_jsonl_path=None,
)
server = types.SimpleNamespace(
cfg=cfg,
model=model or _PassModel(),
policy=None,
policy_lock=threading.Lock(),
logger=_Logger(None),
)
handler = _GoTcpHandler.__new__(_GoTcpHandler)
handler.server = server
sock = _Socket()
return handler._handle_one(
sock,
None,
board,
N,
side,
[],
None,
)
def test_model_pass_is_replaced_when_meaningful_move_exists(self) -> None:
board = make_board(
[(3, 3), (3, 15), (9, 3), (15, 3)],
[(15, 15), (9, 15), (3, 9), (15, 9)],
)
is_pass, move = self._move(board, WHITE)
self.assertFalse(is_pass)
self.assertIsNotNone(move)
def test_sparse_position_is_not_mistaken_for_endgame(self) -> None:
board = make_board(
[],
[(3, 3), (3, 9), (3, 15), (9, 3), (9, 15), (15, 3), (15, 9), (15, 15)],
)
is_pass, move = self._move(board, WHITE)
self.assertFalse(is_pass)
self.assertIsNotNone(move)
def test_model_pass_is_replaced_by_urgent_capture(self) -> None:
board = make_board(
[(5, 5), (3, 3), (15, 3), (3, 15)],
[(4, 5), (5, 4), (6, 5), (15, 15)],
)
is_pass, move = self._move(board, WHITE)
self.assertFalse(is_pass)
self.assertEqual(move, (5, 6))
def test_quiet_model_move_is_not_turned_into_midgame_pass(self) -> None:
# 모델이 합법 좌표를 냈지만 모든 후보 Δ가 임계값(0.75) 미만인 중반.
# 이전에는 PASS 제안일 때만 임계 완화를 해서, 이 경우 서버가 PASS를 보냈다.
board = make_board(
[(3, 3), (3, 15), (9, 3), (15, 3), (5, 9), (13, 13)],
[(15, 15), (9, 15), (3, 9), (15, 9), (13, 5), (6, 13)],
)
original = _MiniGo.pick_by_score
def below_threshold(self, *args, minimum_delta=0.75, **kwargs):
if minimum_delta is not None:
return None
return original(self, *args, minimum_delta=None, **kwargs)
_MiniGo.pick_by_score = below_threshold
try:
is_pass, move = self._move(board, WHITE, _QuietMoveModel((9, 9)))
finally:
_MiniGo.pick_by_score = original
self.assertFalse(is_pass)
self.assertIsNotNone(move)
def test_lone_stone_inside_shallow_enemy_area_is_hopeless(self) -> None:
# 흑이 오른쪽 아래 귀 (16..18, 16..18) 빈 구역을 완전히 둘러쌌다.
# 면적 계가는 백 돌 하나로 이 구역을 공배로 바꿔 큰 이득처럼 보지만 살 수 없다.
wall = [(15, c) for c in range(15, 19)] + [(r, 15) for r in range(16, 19)]
board = make_board(wall + [(3, 3), (3, 15)], [(9, 9), (9, 3), (3, 9), (15, 9)])
mini = _MiniGo(board, N)
self.assertTrue(mini.is_hopeless_invasion(17, 17, WHITE))
self.assertEqual(mini.self_damage_reason(17, 17, WHITE, min_line=1), "hopeless_invasion")
picked = mini.pick_by_score(
WHITE, [(17, 17), (9, 12)], min_line=1, minimum_delta=None
)
self.assertIsNotNone(picked)
self.assertEqual(picked[:2], (9, 12))
def test_open_area_and_weak_wall_are_not_hopeless(self) -> None:
wall = [(15, c) for c in range(15, 19)] + [(r, 15) for r in range(16, 19)]
board = make_board(wall, [(9, 9)])
mini = _MiniGo(board, N)
self.assertFalse(mini.is_hopeless_invasion(9, 12, WHITE))
# 백 돌이 맞닿은 구역은 침입이 아니라 접전이다.
contact = make_board([(17, 17), (18, 16)], [(16, 17), (17, 16), (9, 9)])
self.assertFalse(_MiniGo(contact, N).is_hopeless_invasion(18, 17, WHITE))
# 흑 벽이 단수에 걸리는 수는 공격이므로 남긴다.
thin = make_board(
[(17, 17), (17, 18), (18, 16)], [(16, 17), (16, 18), (17, 16), (9, 9)]
)
thin_mini = _MiniGo(thin, N)
self.assertTrue(thin_mini.is_legal(18, 18, WHITE))
self.assertFalse(thin_mini.is_hopeless_invasion(18, 18, WHITE))
def test_dead_group_area_fill_is_hopeless(self) -> None:
# 백 (17,17) 한 돌이 흑 벽에 둘러싸여 활로 2개뿐이다.
# 이 영역의 빈점들을 메우는 것은 절망적인 수다.
wall = [(16, c) for c in range(16, 19)] + [(r, 16) for r in range(17, 19)]
board = make_board(wall + [(3, 3)], [(17, 17), (9, 9)])
mini = _MiniGo(board, N)
stones, libs = mini._group(17, 17)
self.assertEqual(len(stones), 1)
self.assertLessEqual(libs, 2)
self.assertTrue(mini.is_hopeless_invasion(18, 18, WHITE))
self.assertEqual(mini.self_damage_reason(18, 18, WHITE, min_line=1), "hopeless_invasion")
def test_shallow_one_eye_with_many_liberties_is_hopeless(self) -> None:
# 백 (17,17)의 활로는 4개지만 전부 흑이 가둔 얕은 귀 안이다.
# 활로 2 이하만 죽은 돌로 보던 조건은 이 숨구멍을 메우게 둔다.
# 메운 돌은 집과 포로로 두 번 세어진다.
wall = [(15, c) for c in range(15, 19)] + [(r, 15) for r in range(16, 19)]
board = make_board(wall + [(3, 3)], [(17, 17), (9, 9)])
mini = _MiniGo(board, N)
_stones, libs = mini._group(17, 17)
self.assertGreater(libs, 2)
self.assertTrue(mini.is_hopeless_invasion(18, 18, WHITE))
self.assertTrue(mini.is_hopeless_invasion(16, 16, WHITE))
self.assertEqual(
mini.self_damage_reason(18, 18, WHITE, min_line=1), "hopeless_invasion"
)
picked = mini.pick_by_score(
WHITE, [(18, 18), (9, 12)], min_line=1, minimum_delta=None
)
self.assertIsNotNone(picked)
self.assertEqual(picked[:2], (9, 12))
def test_pass_remains_when_board_has_no_playable_point(self) -> None:
cells = bytes(BLACK if (r + c) % 2 == 0 else WHITE for r in range(N) for c in range(N))
is_pass, move = self._move(cells, WHITE)
self.assertTrue(is_pass)
self.assertIsNone(move)
def test_diagonal_advance_that_allows_double_atari_is_penalized(self) -> None:
# 백 (5,5), (7,7) 사이 (6,6) 진출 뒤 흑 (5,6)이 두 백 그룹을
# 동시에 단수로 만든다. 각 그룹의 남은 활로가 달라 둘 다 구할 수 없다.
board = make_board(
[(4, 5), (6, 5), (7, 6)],
[(5, 5), (7, 7)],
)
mini = _MiniGo(board, N)
threat = mini.double_atari_threat_after(6, 6, WHITE)
picked = mini.pick_by_score(
WHITE,
[(6, 6), (5, 4)],
min_line=1,
minimum_delta=None,
)
self.assertIsNotNone(threat)
self.assertEqual(threat[0], 2)
self.assertEqual(threat[2], (5, 6))
self.assertIsNotNone(picked)
self.assertEqual(picked[:2], (5, 4))
def test_rejected_coordinate_is_released_after_position_changes(self) -> None:
target = (9, 9)
before = bytes(N * N)
state = {
"reject_bans": [[target[0], target[1]]],
"reject_bans_board": before,
"reject_bans_side": WHITE,
"last_offers": {
str(WHITE): [target[0], target[1], before],
},
}
same_bans, same_inferred = _position_scoped_reject_bans(
state, before, N, WHITE, []
)
self.assertIn(target, same_bans)
self.assertEqual(same_inferred, target)
# 그 수가 실제로 진행된 뒤 따내져 target이 다시 빈 점이 된 국면.
# 다른 교차점의 상태가 바뀌었으므로 과거 좌표 금지는 더 이상 유효하지 않다.
after_capture = bytearray(before)
after_capture[8 * N + 9] = BLACK
changed_bans, changed_inferred = _position_scoped_reject_bans(
state, bytes(after_capture), N, WHITE, [target]
)
self.assertNotIn(target, changed_bans)
self.assertIsNone(changed_inferred)
repeated_bans, _ = _position_scoped_reject_bans(
state, bytes(after_capture), N, WHITE, [target]
)
self.assertNotIn(target, repeated_bans)
# 반대로 같은 국면의 재요청에서 새로 추가된 좌표는 정상적으로 금지한다.
fresh_state: dict = {}
_position_scoped_reject_bans(fresh_state, before, N, WHITE, [])
fresh_bans, _ = _position_scoped_reject_bans(
fresh_state, before, N, WHITE, [target]
)
self.assertIn(target, fresh_bans)
if __name__ == "__main__":
unittest.main()