예제 소스
장기로 이해하는 규칙 엔진과 평가 함수 탐색 · 파일 59개
watch.py
"""고수 대국 관전 기록: 로비 '관심 대국' 목록의 첫 대국을 관전하며 수순을 기보 DB에 저장.
python watch.py 계속 반복 (관전 → 대국 끝 → 로비 → 다음 대국)
관전만 함: 판 위는 절대 누르지 않음. 누르는 것은 로비의 관심 대국 첫 칸과 관전 화면 아래 수순 버튼
(처음으로/다음 수/마지막으로 - 들어가자마자 0수부터 한 수씩 넘겨 처음부터 기록)뿐.
초대가 와도 수락하지 않음 (대국 봇은 bot.py play).
저장: data/monomob_kibo.sqlite, games.source = 'watch'
"""
from __future__ import annotations
import subprocess
import sys
import time
import numpy as np
import bot
import recorder
from recorder import Recorder, occ_of, read_board
WATCH_DIALOGS = {"result", "repeat", "notice", "update"}
LIST_ITEM_BELOW_HEADER = 137 # '관심 대국' 제목 가운데 → 첫 대국 칸 가운데
LIST_BOTTOM_Y = 1790 # '최신 정보 갱신/대국신청' 버튼 줄(~1834부터) 위로 여유를 둠
ROW_BELOW_ICON = 32 # 楚 아이콘 가운데 → 대국 칸 가운데
LIST_PAGES = 6 # 목록을 내려 가며 볼 쪽 수
OPEN_WAIT_SEC = 8 # 관전자가 꽉 찬 대국은 눌러도 안 열림 → 다음 칸
STALE_SEC = 300 # 이만큼 판이 안 바뀌면 다른 대국으로
NO_BOARD_SEC = 20 # 관전 화면에서 판이 이만큼 안 보이면 로비로
REPLAY_STEP_SEC = 0.5
REPLAY_MAX = 400
# 관전 화면의 판은 가로 위치/간격은 대국 화면과 같고, 세로 위치는 광고 띠·잡은 기물 칸에 따라 바뀜
SPECTATE_DY = 107.44
GRID_MIN_CONTRAST = 60 # 판 줄 어두움 (판: ~130, 판 없는 화면: ~10)
def fit_grid(img) -> bool:
"""판 가로줄(열 사이 가운데에서 보이는 어두운 줄 10개)로 판 세로 위치를 찾아 bot 좌표에 반영."""
g = bot.cv2.cvtColor(img, bot.cv2.COLOR_BGR2GRAY).astype(np.float32)
x0, dx = bot.CFG["grid"]["x0"], bot.CFG["grid"]["dx"]
xs = [int(x0 + dx / 2 + dx * k) for k in range(8)]
prof = np.mean([g[:, x - 6 : x + 7].mean(axis=1) for x in xs], axis=0)
best_c, best_y = 0.0, 0
for y0 in range(300, 700):
ys = (y0 + np.arange(10) * SPECTATE_DY).round().astype(int)
if ys[-1] + 13 >= len(prof):
break
c = (prof[ys - 12].mean() + prof[ys + 12].mean()) / 2 - prof[ys].mean()
if c > best_c:
best_c, best_y = float(c), y0
if best_c < GRID_MIN_CONTRAST:
return False
bot.CFG["grid"] = {**bot.CFG["grid"], "y0": float(best_y), "dy": SPECTATE_DY}
return True
def replay_button(img, name: str) -> tuple[int, int] | None:
gray = bot.cv2.cvtColor(img, bot.cv2.COLOR_BGR2GRAY)
score, pos = bot._match(gray, name, 1400)
return pos if score >= bot.CFG["guard_threshold"] else None
# 기물 모양(스킨)은 사람마다 달라 종류 인식이 틀릴 수 있음 → 수순은 칸의 진영(초/한/빈칸)만으로 추적하고,
# 기물 종류는 시작 상차림에서 이어받음. 마상 배치는 뒤 수순과 맞는 조합으로 결정.
def explain_occ(board: list[str], target: list[str], first: str) -> list[tuple[int, int]] | None:
"""board에서 first부터 번갈아 둬서 진영 배치가 target이 되는 최단 수순."""
if occ_of(board) == target:
return []
if sum(a != b for a, b in zip(occ_of(board), target)) > recorder.MAX_GAP * 2:
return None
def dfs(b: list[str], side: str, depth: int) -> list[tuple[int, int]] | None:
mism = [i for i in range(90) if (b[i][0] if b[i] else bot.EMPTY) != target[i]]
if not mism:
return []
if depth == 0 or len(mism) > depth * 2:
return None
for f in mism:
if not b[f] or b[f][0] != side:
continue
for t in recorder.rules.targets(b, f):
if b[t] and b[t][1] == "K":
continue
nb = b[:]
nb[t], nb[f] = nb[f], ""
rest = dfs(nb, bot.opposite_side(side), depth - 1)
if rest is not None:
return [(f, t), *rest]
return None
for depth in range(1, recorder.MAX_GAP + 1):
seq = dfs(board, first, depth)
if seq is not None:
return seq
return None
def simulate(board: list[str], occs: list[list[str]], last: str | None) -> tuple[list[tuple], int]:
"""진영 배치 목록을 차례로 설명하는 수들 [(진영, from, to, 기물, 잡은 기물, 둔 뒤 판) 또는 (진영, None) = 한수 쉼]
과 설명된 단계 수."""
b = board[:]
out: list[tuple] = []
for k, occ in enumerate(occs):
if occ_of(b) == occ:
continue
sides = [bot.opposite_side(last)] if last else [bot.CHO, bot.HAN]
if last:
sides.append(last)
seq, first = None, sides[0]
for first in sides:
seq = explain_occ(b, occ, first)
if seq:
break
if not seq:
return out, k
if last and first == last:
out.append((bot.opposite_side(first), None))
for f, t in seq:
piece, cap = b[f], b[t]
b[t], b[f] = piece, ""
out.append((piece[0], f, t, piece, cap, b[:]))
last = piece[0]
return out, len(occs)
def record_moves(rec: Recorder, moves: list[tuple]) -> None:
for mv in moves:
if mv[1] is None:
rec.add(mv[0], "pass", "", "")
print(f"{rec.ply:3d}. {bot.SIDE_KO[mv[0]]} 한수 쉼")
continue
side, f, t, piece, cap, after = mv
rec.board = after
rec.add(side, recorder.sq(f) + recorder.sq(t), piece, cap)
capk = f" x{bot.PIECE_KO[cap[1]]}" if cap else ""
print(f"{rec.ply:3d}. {bot.SIDE_KO[side]} {bot.PIECE_KO[piece[1]]} {recorder.label(f)}→{recorder.label(t)}{capk}")
def start_candidates(occ0: list[str]) -> list[list[str]]:
"""0수 국면의 진영 배치에 맞는 시작판 후보 (마상 배치 16가지, 접장기면 빠진 기물 제거)."""
start = bot.start_occupancy()
if any(o != bot.EMPTY and o != s for o, s in zip(occ0, start)):
return []
out = []
for cw in bot.FORMATIONS:
for hw in bot.FORMATIONS:
b = bot.start_board(cw, hw)
out.append([p if occ0[i] != bot.EMPTY else "" for i, p in enumerate(b)])
return out
def advance(rec: Recorder, obs: list[str]) -> bool:
moves, n = simulate(rec.board, [obs], rec.last_side)
if n < 1:
return False
record_moves(rec, moves)
return True
def replay_from_start(rec: Recorder) -> bool:
"""수순 버튼으로 0수로 돌아가 한 수씩 넘기며 진영 배치를 모은 뒤, 맞는 시작판으로 처음부터 기록.
끝나면 마지막 수(실시간)로."""
img = bot.screencap()
first = replay_button(img, "replay_first")
if not first:
return False
bot.tap(*first)
time.sleep(1.2)
img = bot.screencap()
if not fit_grid(img):
return False
view = bot.detect_view(bot.classify_screen(img))
if view is None:
return False
occs = [view.canonical(bot.classify_screen(img))]
read0 = read_board(img)
cands = start_candidates(occs[0])
if not cands:
if not read0:
return False
cands = [read0[1]]
print(f" 수순 처음(0수)부터 한 수씩 넘기며 읽는 중 (화면 위쪽={'초' if view.flipped else '한'})")
same = 0
for _ in range(REPLAY_MAX):
nxt = replay_button(img, "replay_next")
if not nxt:
break
bot.tap(*nxt)
time.sleep(REPLAY_STEP_SEC)
img = bot.screencap()
if not fit_grid(img):
same += 1
if same >= 3:
break
continue
occ = view.canonical(bot.classify_screen(img))
if occ == occs[-1]:
same += 1
if same >= 3:
break
continue
same = 0
occs.append(occ)
last = replay_button(bot.screencap(), "replay_last")
if last:
bot.tap(*last)
time.sleep(1.2)
def agree(b: list[str]) -> int:
return sum(1 for x, y in zip(b, read0[1]) if x and x == y) if read0 else 0
best = max(((simulate(c, occs[1:], None), agree(c), c) for c in cands), key=lambda r: (r[0][1], r[1]))
(moves, n), _, board = best
rec.open_game(view, board, True, 0)
record_moves(rec, moves)
if n < len(occs) - 1:
print(f" ! {n + 1}번째 단계부터 수순을 못 맞춤 - 여기까지 기록하고 실시간으로")
return True
def lobby_item(img) -> tuple[int, int] | None:
gray = bot.cv2.cvtColor(img, bot.cv2.COLOR_BGR2GRAY)
score, (_, hy) = bot._match(gray, "lobby_watch_header", 300, 2000)
if score < bot.CFG["guard_threshold"]:
return None
return 540, hy + LIST_ITEM_BELOW_HEADER
def is_lobby(img) -> bool:
"""로비 아래 고정 줄의 '최신 정보 갱신' 버튼 (목록을 내려도 보임)."""
gray = bot.cv2.cvtColor(img, bot.cv2.COLOR_BGR2GRAY)
return bot._match(gray, "lobby_refresh_btn", 1700)[0] >= bot.CFG["guard_threshold"]
def list_rows(img) -> list[tuple[int, int]]:
"""로비 목록에서 대국 칸들 (왼쪽 楚 아이콘 기준). 아래 대국신청 버튼 줄은 절대 안 누름."""
gray = bot.cv2.cvtColor(img, bot.cv2.COLOR_BGR2GRAY)[:, :160]
tpl = bot.cv2.imread(str(bot.GUARDS / "list_cho_icon.png"), bot.cv2.IMREAD_GRAYSCALE)
res = bot.cv2.matchTemplate(gray, tpl, bot.cv2.TM_CCOEFF_NORMED)
ys = sorted({int(y) for y in np.where(res >= bot.CFG["guard_threshold"])[0]})
rows: list[tuple[int, int]] = []
for y in ys:
cy = y + tpl.shape[0] // 2 + ROW_BELOW_ICON
if y >= 250 and cy < LIST_BOTTOM_Y and (not rows or cy - rows[-1][1] > 80):
rows.append((540, cy))
return rows
def scroll_list() -> None:
bot.adb("shell", "input", "swipe", "540", "1650", "540", "750", "500")
time.sleep(1.5)
def watch_one(rec: Recorder) -> str:
"""관전 화면에서 판이 끝나거나 멈출 때까지 기록. 끝난 이유를 돌려줌."""
start_occ = bot.start_occupancy()
last_obs = None
last_change = time.time()
last_board_seen = time.time()
fails = 0
polls = 0
if replay_from_start(rec):
last_change = last_board_seen = time.time()
while True:
polls += 1
if polls % 6 == 0:
if bot.is_ad_activity(bot.focused_activity()):
return "ad"
bot.ensure_app_foreground()
img = bot.screencap()
if polls % 3 == 0 and bot.handle_dialog(img, WATCH_DIALOGS):
return "dialog"
if is_lobby(img):
return "lobby"
if not fit_grid(img):
if time.time() - last_board_seen > NO_BOARD_SEC:
return "no_board"
time.sleep(1)
continue
last_board_seen = time.time()
if rec.game_id is None:
read = read_board(img)
if read:
rec.open_game(read[0], read[1], occ_of(read[1]) == start_occ, 0)
last_change = time.time()
time.sleep(1)
continue
obs = rec.view.canonical(bot.classify_screen(img))
if obs != occ_of(rec.board) and obs == last_obs:
if advance(rec, obs):
fails = 0
last_change = time.time()
else:
fails += 1
if fails >= 4:
read = read_board(img)
if read:
print(" ! 수순 복원 실패 - 현재 국면부터 다시 기록")
rec.close_game("resync")
rec.open_game(read[0], read[1], False, 0)
fails = 0
if time.time() - last_change > STALE_SEC:
return "stale"
last_obs = obs
time.sleep(0.3)
def main() -> None:
sys.stdout.reconfigure(line_buffering=True) # type: ignore[union-attr]
bot.adb("shell", "svc", "power", "stayon", "usb")
rec = Recorder(recorder.db(), None, source="watch")
waits = 0
print("관전 기록 시작 (Ctrl+C로 종료)")
try:
while True:
try:
if bot.is_ad_activity(bot.focused_activity()):
bot.handle_ad(use_mirror=False)
continue
bot.ensure_app_foreground()
img = bot.screencap()
if bot.handle_dialog(img, WATCH_DIALOGS):
continue
if not is_lobby(img):
# 로비도 관전 화면도 아님 (끝난 판 화면 등): 뒤로가기로 로비로
print(" 로비가 아님 → 뒤로가기")
bot.adb("shell", "input", "keyevent", "KEYCODE_BACK")
time.sleep(3)
continue
opened = False
for page in range(LIST_PAGES):
if page:
scroll_list()
img = bot.screencap()
if not is_lobby(img):
break
for k, row in enumerate(list_rows(img)):
print(f"관심 대국 {page + 1}쪽 {k + 1}번째 칸 {row} → 관전 들어가기")
bot.tap(*row)
for _ in range(OPEN_WAIT_SEC):
time.sleep(1)
if not is_lobby(bot.screencap()):
opened = True
break
if opened:
break
print(" 안 열림 (관전자가 꽉 찼거나 방금 끝난 대국)")
if opened:
break
if not opened:
waits += 1
pause = min(120, 10 * waits)
print(f" 열리는 대국이 없음 - {pause}초 뒤 다시")
time.sleep(pause)
continue
waits = 0
reason = watch_one(rec)
rec.close_game(reason)
print(f"관전 종료: {reason}")
if reason in ("stale", "no_board"):
bot.adb("shell", "input", "keyevent", "KEYCODE_BACK")
time.sleep(3)
except (subprocess.SubprocessError, OSError, RuntimeError) as e:
print(f" ! 오류 ({type(e).__name__}: {e}) - 5초 뒤 재시도")
time.sleep(5)
except KeyboardInterrupt:
rec.close_game("stopped")
print("중단")
if __name__ == "__main__":
main()