예제 소스
장기로 이해하는 규칙 엔진과 평가 함수 탐색 · 파일 59개
bot.py
#!/usr/bin/env python3
"""모노몹 장기(com.monomob.jangki) 화면을 adb로 읽어서 janggi_engine으로 두는 봇.
python bot.py show [--png FILE] 현재 화면 인식 결과 + 추천 수 출력 (탭 안 함)
python bot.py calib-friendly 나의길 계정과의 친선 대국 화면에서 1회: 판별 기준 저장
python bot.py play 자동 대국 (친선 대국 + 상대=나의길 일 때만 둠)
python bot.py play --weights data/weights_learned.txt 학습 가중치로 두기
python bot.py play --dry-run 추천 수만 출력, 수는 사람이 직접 둠 (같은 제한)
python bot.py ad 지금 떠 있는 광고만 처리 (닫기 대기/미러 창)
play 는 판이 끝나면 광고를 처리하고 다음 판 시작을 기다려 계속 둔다 (--once 로 한 판만).
광고 중에는 맥 미러 창(scrcpy)에서 X를 누르거나, 터미널에서 Enter = 뒤로가기.
python bot.py make-templates FILE --top EHEH --bottom HEHE
시작 화면에서 마/상 템플릿 생성
(--top/--bottom: 화면 왼쪽부터 2,3,7,8번째 줄 기물)
좌표는 엔진(src/Types.h) 기준: file 0..8, rank 0..9 (rank 0 = 한 궁성 쪽).
"""
from __future__ import annotations
import argparse
import json
import queue
import random
import re
import shutil
import subprocess
import sys
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
import cv2
import numpy as np
ROOT = Path(__file__).resolve().parent
ENGINE = ROOT / "janggi_engine"
TEMPLATES = ROOT / "templates"
SHOTS = ROOT / "shots"
CFG = json.loads((ROOT / "config.json").read_text())
EMPTY, CHO, HAN = ".", "c", "h"
WING_FILES = (1, 2, 6, 7)
FORMATIONS = {"HEEH": "inner", "EHHE": "outer", "EHEH": "left", "HEHE": "right"}
PIECE_KO = {"K": "궁", "A": "사", "R": "차", "C": "포", "H": "마", "E": "상", "P": "졸"}
SIDE_KO = {CHO: "초", HAN: "한"}
def opposite_side(side: str) -> str:
return HAN if side == CHO else CHO
# ---------------------------------------------------------------- adb
def adb(*args: str, binary: bool = False, timeout: float = 15) -> bytes | str:
out = subprocess.run(["adb", *args], capture_output=True, timeout=timeout, check=True).stdout
return out if binary else out.decode(errors="replace")
def screencap(retries: int = 10) -> np.ndarray:
"""무압축 screencap (PNG보다 3배 이상 빠름). 헤더: width, height, format[, colorspace]."""
for attempt in range(retries):
try:
data = adb("exec-out", "screencap", binary=True, timeout=8)
assert isinstance(data, bytes)
if len(data) < 16:
raise ValueError("빈 데이터")
w, h = (int(v) for v in np.frombuffer(data[:8], dtype=np.uint32))
if len(data) < 12 + w * h * 4:
raise ValueError(f"잘린 데이터 {len(data)}바이트")
pixels = np.frombuffer(data[len(data) - w * h * 4 :], np.uint8).reshape(h, w, 4)
return cv2.cvtColor(pixels, cv2.COLOR_RGBA2BGR)
except (subprocess.TimeoutExpired, subprocess.CalledProcessError, ValueError) as e:
print(f" ! 화면 캡처 실패({type(e).__name__}: {e}), 재시도 {attempt + 1}/{retries}")
time.sleep(1)
raise RuntimeError("adb 화면 캡처가 계속 실패합니다. 폰 연결을 확인하세요.")
def tap(x: float, y: float) -> None:
adb("shell", "input", "tap", str(int(round(x))), str(int(round(y))))
def focused_activity() -> str:
out = adb("shell", "dumpsys window | grep mCurrentFocus")
m = re.search(r"mCurrentFocus=Window\{\S+ \S+ ([^}]+)\}", str(out))
return m.group(1) if m else ""
def unlock_game_booster() -> bool:
"""삼성 게임 부스터의 '게임 중 잠금' 화면(어두워지고 잠금 아이콘을 끌어야 풀림). 잠금 중엔 터치가 앱에 가지 않음."""
out = str(adb("shell", "dumpsys window windows | grep -c 'GameBooster Lock Screen' || true")).strip()
if not out.isdigit() or int(out) == 0:
return False
print(" [자동] 게임 부스터 잠금 화면 → 잠금 아이콘 끌어서 해제")
tap(540, 1113)
time.sleep(1)
adb("shell", "input", "swipe", "540", "1113", "540", "1550", "600")
time.sleep(1.5)
return True
def ensure_app_foreground() -> None:
"""화면 꺼짐/잠금 화면/홈 화면/게임 부스터 잠금(대기 화면)이면 풀고 장기 앱으로 돌아감."""
power = str(adb("shell", "dumpsys power | grep mWakefulness="))
if "Awake" not in power:
print(" [자동] 화면이 꺼져 있음 → 깨우기")
adb("shell", "input", "keyevent", "KEYCODE_WAKEUP")
time.sleep(1.5)
unlock_game_booster()
act = focused_activity()
if "NotificationShade" in act or "Keyguard" in act or "LockScreen" in act:
print(" [자동] 잠금 화면 → 위로 밀어 해제")
adb("shell", "input", "swipe", "540", "1800", "540", "500", "300")
time.sleep(1.5)
act = focused_activity()
if act and CFG["app_package"] not in act and "com.android.vending" not in act:
print(f" [자동] 장기 앱이 아님 ({act}) → 장기 앱으로 돌아가기")
adb("shell", "monkey", "-p", CFG["app_package"], "-c", "android.intent.category.LAUNCHER", "1")
time.sleep(3)
def is_ad_activity(activity: str) -> bool:
low = activity.lower()
return any(k.lower() in low for k in CFG["ad_activity_keywords"])
# ---------------------------------------------------------------- 친선 대국 확인
GUARDS = ROOT / "guards"
def text_mask(img: np.ndarray, band: list[int]) -> np.ndarray:
"""글자 영역 회색조. 차례에 따라 글자가 흰색(255)/회색(~120)이라 이진화하지 않고 NCC로 비교."""
x0, y0, x1, y1 = band
return cv2.cvtColor(img[y0:y1, x0:x1], cv2.COLOR_BGR2GRAY)
def words(gray: np.ndarray, gap: int = 10) -> list[np.ndarray]:
"""밝은 글자 영역을 가로 간격으로 단어 단위로 자름 (배경 ~60보다 밝은 픽셀 기준)."""
mask = gray > 90
cols = np.where(mask.sum(axis=0) > 0)[0]
if len(cols) == 0:
return []
out, start, prev = [], cols[0], cols[0]
for c in cols[1:]:
if c - prev > gap:
out.append((start, prev))
start = c
prev = c
out.append((start, prev))
crops = []
h, w = gray.shape
for a, b in out:
rows = np.where(mask[:, a : b + 1].sum(axis=1) > 0)[0]
y0, y1 = max(0, rows[0] - 2), min(h, rows[-1] + 3)
crops.append(gray[y0:y1, max(0, a - 2) : min(w, b + 3)])
return crops
def find_word(mask: np.ndarray, name: str) -> float:
path = GUARDS / f"{name}.png"
if not path.exists():
return 0.0
tpl = cv2.imread(str(path), cv2.IMREAD_GRAYSCALE)
if tpl.shape[0] > mask.shape[0] or tpl.shape[1] > mask.shape[1]:
return 0.0
return float(cv2.matchTemplate(mask, tpl, cv2.TM_CCOEFF_NORMED).max())
def friendly_guard(img: np.ndarray) -> tuple[bool, str]:
"""친선 대국이고 상대가 등록된 계정일 때만 True."""
th = CFG["guard_threshold"]
title = text_mask(img, CFG["title_band"])
if find_word(title, "title_ranked") >= th:
return False, "승강급 대국 화면입니다 (자동 착수 안 함)"
if not (GUARDS / "title_friendly.png").exists():
return False, "친선 대국 기준이 없습니다. 친선 대국 화면에서 calib-friendly 먼저 실행"
if find_word(title, "title_friendly") < th:
return False, "친선 대국 화면이 아닙니다"
opp = max(
find_word(text_mask(img, CFG["name_band_left"]), "opponent"),
find_word(text_mask(img, CFG["name_band_right"]), "opponent"),
)
if opp < th:
return False, "상대가 등록된 계정(나의길)이 아닙니다"
return True, "친선 대국 확인"
# ---------------------------------------------------------------- 앱 창 자동 처리
# (종류, 제목 템플릿, 누를 버튼 템플릿, 추가로 있어야 하는 템플릿)
# 로비의 대국신청 같은 버튼은 목록에 없으므로 절대 누르지 않음
DIALOGS = [
("invite", "dlg_invite_title", "btn_accept", "dlg_invite_opp"),
("formation", "dlg_formation_title", "result_ok", None),
("result", "dlg_result_title", "result_ok", None),
("result", "dlg_result_kimul", "result_ok", None), # '楚 기물패' 같은 판정 결과 창
("repeat", "repeat_title", "result_ok", None),
("notice", "dlg_notice_title", "result_ok", None),
("update", "dlg_update_title", "btn_update_get", None),
]
DIALOG_KO = {
"invite": "나의길 친선 대국 초대 → 수락",
"formation": "상차림 선택 → 확인",
"result": "대국 결과 → 확인",
"repeat": "동일수 반복 경고 → 확인",
"notice": "공지사항 → 확인",
"update": "최신버전 다운로드",
}
_tpl_cache: dict[str, np.ndarray] = {}
def _match(gray: np.ndarray, name: str, y0: int = 0, y1: int | None = None) -> tuple[float, tuple[int, int]]:
if name not in _tpl_cache:
_tpl_cache[name] = cv2.imread(str(GUARDS / f"{name}.png"), cv2.IMREAD_GRAYSCALE)
tpl = _tpl_cache[name]
roi = gray[y0:y1]
if tpl is None or roi.shape[0] < tpl.shape[0] or roi.shape[1] < tpl.shape[1]:
return 0.0, (0, 0)
_, score, _, loc = cv2.minMaxLoc(cv2.matchTemplate(roi, tpl, cv2.TM_CCOEFF_NORMED))
return float(score), (loc[0] + tpl.shape[1] // 2, y0 + loc[1] + tpl.shape[0] // 2)
def find_dialog(img: np.ndarray) -> tuple[str, tuple[int, int], tuple[int, int]] | None:
"""(종류, 누를 버튼 위치, 제목 위치)"""
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
th = CFG["guard_threshold"]
for kind, title, button, extra in DIALOGS:
score, tpos = _match(gray, title, 200, 1900)
if score < th:
continue
if extra and _match(gray, extra, 200, 1900)[0] < th:
continue
bscore, pos = _match(gray, button, tpos[1], min(gray.shape[0], tpos[1] + 1100))
if bscore >= th:
return kind, pos, tpos
return None
# 상차림 선택 창: 제목 "상차림 선택" 가운데로부터 각 칸의 위치 (초 기준 왼쪽→오른쪽 배치)
FORMATION_OPTION_OFFSETS = {
"right": (48, 268), # 마상마상 (앱 기본 선택)
"left": (390, 268), # 상마상마
"inner": (48, 391), # 마상상마
"outer": (390, 391), # 상마마상
}
FORMATION_KO = {"right": "마상마상", "left": "상마상마", "inner": "마상상마", "outer": "상마마상"}
FORMATION_CHOICE = "auto"
def pick_formation() -> str:
if FORMATION_CHOICE != "auto":
return FORMATION_CHOICE
return random.choice(list(FORMATION_OPTION_OFFSETS))
def handle_dialog(img: np.ndarray, allowed: set[str] | None = None) -> str | None:
found = find_dialog(img)
if not found:
return None
kind, pos, tpos = found
if allowed is not None and kind not in allowed:
return None
if kind == "update":
update_app()
return kind
if kind == "formation":
name = pick_formation()
dx, dy = FORMATION_OPTION_OFFSETS[name]
print(f" [자동] 상차림 고르기: {FORMATION_KO[name]} ({name})")
tap(tpos[0] + dx, tpos[1] + dy)
time.sleep(0.7)
print(f" [자동] {DIALOG_KO[kind]} {pos}")
tap(*pos)
time.sleep(1.5)
if kind == "result" and not is_ad_activity(focused_activity()):
print(" [자동] 결과 확인 후 뒤로가기 (대국 화면 나가기 → 광고)")
adb("shell", "input", "keyevent", "KEYCODE_BACK")
time.sleep(1.5)
return kind
UPDATE_RECHECK_SEC = 300
UPDATE_INSTALL_SEC = 900
def ui_texts() -> dict[str, tuple[int, int]]:
"""uiautomator로 화면의 글자 → 가운데 좌표."""
adb("shell", "uiautomator", "dump", "/sdcard/janggy_ui.xml", timeout=15)
xml = str(adb("exec-out", "cat", "/sdcard/janggy_ui.xml"))
out: dict[str, tuple[int, int]] = {}
for node in re.findall(r"<node [^>]*>", xml):
attr = dict(re.findall(r'([\w-]+)="([^"]*)"', node))
b = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", attr.get("bounds", ""))
if b and attr.get("text"):
x1, y1, x2, y2 = map(int, b.groups())
out.setdefault(attr["text"].strip(), ((x1 + x2) // 2, (y1 + y2) // 2))
return out
def app_version() -> int:
out = str(adb("shell", f"dumpsys package {CFG['app_package']} | grep versionCode="))
m = re.search(r"versionCode=(\d+)", out)
return int(m.group(1)) if m else 0
def launch_app() -> None:
adb("shell", "monkey", "-p", CFG["app_package"], "-c", "android.intent.category.LAUNCHER", "1")
time.sleep(5)
def update_app() -> bool:
"""앱의 '최신버전 다운로드' 창: 스토어 장기 페이지에서 업데이트 → 설치 끝나면 앱 다시 실행.
스토어에 아직 업데이트가 없으면 앱으로 돌아가 UPDATE_RECHECK_SEC 뒤 다시 확인."""
before = app_version()
print(f" [자동] 최신버전 다운로드 창 → 스토어에서 업데이트 확인 (지금 버전 {before})")
adb("shell", "am", "start", "-a", "android.intent.action.VIEW", "-p", "com.android.vending",
"-d", f"market://details?id={CFG['app_package']}")
time.sleep(6)
pressed = False
deadline = time.time() + UPDATE_INSTALL_SEC
while time.time() < deadline:
if app_version() > before:
print(f" [자동] 업데이트 설치 완료 ({before} → {app_version()}) → 장기 앱 다시 실행")
launch_app()
return True
try:
texts = ui_texts()
except (subprocess.SubprocessError, OSError):
texts = {}
btn = texts.get("업데이트") or texts.get("Update")
if btn and btn[1] < 1000 and not pressed:
print(f" [자동] 스토어 '업데이트' 누름 {btn}")
tap(*btn)
pressed = True
elif not pressed and any(t in texts for t in ("플레이", "열기", "Play", "Open")):
print(f" 스토어에 아직 업데이트가 없음 → 앱으로 돌아가 {UPDATE_RECHECK_SEC // 60}분 뒤 다시 확인")
adb("shell", "input", "keyevent", "KEYCODE_BACK")
time.sleep(UPDATE_RECHECK_SEC)
launch_app()
return False
time.sleep(10)
print(" ! 업데이트가 제한 시간 안에 끝나지 않음 → 앱 다시 실행")
notify("장기 앱 업데이트가 끝나지 않았습니다")
launch_app()
return False
# ---------------------------------------------------------------- 광고
class GameInterrupted(Exception):
pass
_keys: queue.Queue[str] = queue.Queue()
def _stdin_reader() -> None:
for line in sys.stdin:
_keys.put(line.strip())
def notify(msg: str) -> None:
subprocess.run(
["osascript", "-e", f'display notification "{msg}" with title "장기 봇" sound name "Glass"'],
check=False,
capture_output=True,
)
_scrcpy: subprocess.Popen | None = None
def open_mirror() -> None:
global _scrcpy
if _scrcpy is not None and _scrcpy.poll() is None:
return
if not shutil.which("scrcpy"):
return
_scrcpy = subprocess.Popen(
["scrcpy", "--window-title", "장기 봇 - 광고 X를 눌러 주세요", "--max-size", "1024", "--no-audio"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
def find_close_button() -> tuple[int, int] | None:
"""광고 화면에서 닫기/건너뛰기로 보이는 버튼 좌표 (uiautomator)."""
try:
adb("shell", "uiautomator", "dump", "/sdcard/janggy_ui.xml", timeout=10)
xml = str(adb("exec-out", "cat", "/sdcard/janggy_ui.xml"))
except (subprocess.SubprocessError, OSError):
return None
words = ("close", "skip", "닫기", "건너뛰기", "dismiss")
for node in re.findall(r"<node [^>]*>", xml):
attr = dict(re.findall(r'([\w-]+)="([^"]*)"', node))
b = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", attr.get("bounds", ""))
if not b:
continue
x1, y1, x2, y2 = map(int, b.groups())
if x2 <= x1 or y2 <= y1:
continue
label = " ".join(attr.get(k, "") for k in ("resource-id", "content-desc", "text")).lower()
if any(wd in label for wd in words):
return (x1 + x2) // 2, (y1 + y2) // 2
return None
def _x_templates() -> list[np.ndarray]:
tpls = []
for size, thick in ((26, 3), (34, 4), (44, 5), (56, 6)):
t = np.zeros((size + 12, size + 12), np.uint8)
cv2.line(t, (6, 6), (size + 5, size + 5), 255, thick, cv2.LINE_AA)
cv2.line(t, (size + 5, 6), (6, size + 5), 255, thick, cv2.LINE_AA)
tpls.append(t)
return tpls
X_TEMPLATES = _x_templates()
def find_corner_x(img: np.ndarray) -> tuple[int, int] | None:
"""광고 위쪽 두 모서리에서 X 모양 닫기 아이콘을 찾음. 설치 버튼 등 가운데는 보지 않음."""
h, w = img.shape[:2]
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
best: tuple[float, int, int] | None = None
for x0 in (0, w - 280):
roi = gray[40:320, x0 : x0 + 280]
for src in (roi, 255 - roi):
for t in X_TEMPLATES:
res = cv2.matchTemplate(src, t, cv2.TM_CCOEFF_NORMED)
_, score, _, loc = cv2.minMaxLoc(res)
if best is None or score > best[0]:
best = (score, x0 + loc[0] + t.shape[1] // 2, 40 + loc[1] + t.shape[0] // 2)
if best and best[0] >= CFG.get("ad_x_threshold", 0.65):
return best[1], best[2]
return None
AD_TOUCH_SPOTS = [(540, 1100), (540, 1960), (270, 262), (540, 1380)]
def find_google_play_badge(img: np.ndarray) -> tuple[int, int] | None:
"""광고 왼쪽 위의 ▷| 아이콘('Google Play' 글자가 붙기도 함). 누르면 스토어가 열림."""
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)[0:200, 0:300]
for name in ("play_icon", "google_play"):
score, pos = _match(gray, name)
if score >= CFG["guard_threshold"]:
return pos
return None
def find_skip_icon(img: np.ndarray) -> tuple[int, int] | None:
"""플레이형 광고 왼쪽 위의 ⏭ (건너뛰기)."""
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)[0:200, 0:300]
score, pos = _match(gray, "skip_icon")
return pos if score >= CFG["guard_threshold"] else None
AD_STATS = ROOT / "data" / "ad_stats.json"
AD_X_SPOT = (999, 71) # 스토어에서 돌아온 검은 화면의 X 자리
AD_GIVEUP_SEC = 480 # 이만큼 안 닫히면 장기 앱을 강제 종료 후 다시 실행
class AdPractice:
"""광고 한 번의 기록: 단계별 화면, 행동/결과. 터치 위치별 스토어 열림 성공률을 누적해 다음 광고에 반영."""
def __init__(self) -> None:
self.dir = SHOTS / "ads" / time.strftime("%m%d_%H%M%S")
self.dir.mkdir(parents=True, exist_ok=True)
self.t0 = time.time()
self.n = 0
self.events: list[dict] = []
self.last_action = "없음"
try:
self.stats = json.loads(AD_STATS.read_text())
except (OSError, ValueError):
self.stats = {"spots": {}, "ads": []}
def shot(self, img: np.ndarray, step: str, mark: tuple[int, int] | None = None) -> None:
if self.n >= 40:
return
if mark:
img = img.copy()
cv2.circle(img, mark, 40, (0, 0, 255), 5)
cv2.imwrite(str(self.dir / f"{self.n:02d}_{step}.png"), img)
self.n += 1
def event(self, what: str, **kw: object) -> None:
self.events.append({"t": round(time.time() - self.t0, 1), "what": what, **kw})
self.last_action = what
def spot_order(self) -> list[tuple[int, int]]:
def rate(spot: tuple[int, int]) -> float:
s = self.stats["spots"].get(f"{spot[0]},{spot[1]}", {})
return (s.get("store", 0) + 1) / (s.get("tries", 0) + 2)
return sorted(AD_TOUCH_SPOTS, key=rate, reverse=True)
def spot_result(self, spot: tuple[int, int], opened: bool) -> None:
s = self.stats["spots"].setdefault(f"{spot[0]},{spot[1]}", {"tries": 0, "store": 0})
s["tries"] += 1
s["store"] += int(opened)
self.event("touch", spot=list(spot), store_opened=opened)
def finish(self) -> None:
secs = round(time.time() - self.t0)
summary = {"dir": self.dir.name, "secs": secs, "closed_by": self.last_action, "events": self.events}
self.stats["ads"].append({k: v for k, v in summary.items() if k != "events"})
AD_STATS.parent.mkdir(exist_ok=True)
AD_STATS.write_text(json.dumps(self.stats, ensure_ascii=False, indent=1))
(self.dir / "events.json").write_text(json.dumps(summary, ensure_ascii=False, indent=1))
print(f"광고 종료 ({secs}초, 마지막 행동: {self.last_action}, 기록: shots/ads/{self.dir.name})")
def handle_ad(use_mirror: bool) -> None:
"""광고 화면이 끝날 때까지 대기. 닫기 버튼이 보이면 누르고, 아니면 사용자에게 맡김."""
print("광고 감지: 닫기(X)가 나오면 자동으로 누릅니다. (Enter = 뒤로가기)")
if use_mirror:
open_mirror()
while not _keys.empty():
_keys.get_nowait()
practice = AdPractice()
practice.shot(screencap(), "start")
tries = 0
clear_streak = 0
started = time.time()
touches = 0
last_touch = 0.0
last_shot = time.time()
from_store = False
store_back_at = 0.0
blind_tapped = False
store_via: str | None = None
blind_ok = True
skip_at = 0.0
tried: list[tuple[int, int]] = []
while clear_streak < 3:
if time.time() - started > AD_GIVEUP_SEC:
print(f" ! 광고가 {AD_GIVEUP_SEC // 60}분 넘게 안 닫힘 → 장기 앱 강제 종료 후 다시 실행")
practice.shot(screencap(), "giveup")
practice.event("force_restart")
adb("shell", "am", "force-stop", CFG["app_package"])
time.sleep(2)
launch_app()
break
act = focused_activity()
if "com.android.vending" in act:
if practice.last_action == "blind_x_tap" and blind_ok:
blind_ok = False
print(" X 자리 직접 탭이 스토어를 엶 → 이 광고에선 더 안 누름")
print(" Play 스토어 열림 → 뒤로가기 2번 (검은 화면의 X 찾기)")
adb("shell", "input", "keyevent", "KEYCODE_BACK")
time.sleep(1)
adb("shell", "input", "keyevent", "KEYCODE_BACK")
practice.event("store_back")
clear_streak = 0
from_store = True
store_back_at = time.time()
blind_tapped = False
time.sleep(2)
continue
if act and not is_ad_activity(act):
clear_streak += 1
time.sleep(1)
continue
clear_streak = 0
img = screencap()
pos = find_corner_x(img)
if pos and is_ad_activity(focused_activity()):
print(f" 광고 X 발견 {pos} → 탭")
practice.shot(img, "x_tap", pos)
practice.event("x_tap", pos=list(pos))
tap(*pos)
time.sleep(1.5)
continue
skip = find_skip_icon(img)
if skip and time.time() - skip_at >= 8:
print(f" 건너뛰기(⏭) {skip} → 탭")
practice.shot(img, "skip", skip)
practice.event("skip", pos=list(skip))
tap(*skip)
skip_at = time.time()
time.sleep(2)
continue
# 화면 터치로 간 스토어에서 돌아오면 왼쪽 위 ▷| 배지가 잠깐 떴다 사라짐 → 바로 눌러야 함
# (배지로 간 스토어에서 돌아온 뒤엔 검은 화면 X를 기다림)
badge = find_google_play_badge(img)
if badge and (store_via != "badge" or time.time() - store_back_at >= 20):
print(f" Google Play 배지 {badge} → 탭 (스토어 열기)")
practice.shot(img, "gp_badge", badge)
practice.event("gp_badge", pos=list(badge))
tap(*badge)
store_via = "badge"
time.sleep(2)
continue
if from_store:
print(" 스토어에서 돌아옴: 오른쪽 위 X가 뜨기를 기다림")
practice.shot(img, "after_store")
from_store = False
# 스토어에서 돌아온 뒤엔 검은 화면 오른쪽 위에 X가 늦게 뜸: 다른 곳은 누르지 않고 X만 봄
since_store = time.time() - store_back_at
if since_store < 20:
if since_store >= 12 and not blind_tapped and blind_ok:
blind_tapped = True
print(f" X가 안 보여 오른쪽 위 X 자리 {AD_X_SPOT} 직접 탭")
practice.shot(img, "blind_x", AD_X_SPOT)
practice.event("blind_x_tap", pos=list(AD_X_SPOT))
tap(*AD_X_SPOT)
time.sleep(1)
continue
if time.time() - last_shot >= 15:
practice.shot(img, "waiting")
last_shot = time.time()
# 광고마다 다르지만 대개: 아무 곳 터치 → 스토어 → 뒤로가기 → 검은 화면의 X
# 터치 위치는 지금까지 스토어가 잘 열렸던 순서로 시도
now = time.time()
if now - started >= 4 and now - last_touch >= 10 and touches < 12:
untried = [s for s in practice.spot_order() if s not in tried]
if not untried:
tried.clear()
untried = practice.spot_order()
spot = untried[0]
tried.append(spot)
touches += 1
last_touch = now
print(f" 광고 화면 터치 {spot} → 스토어 열기 ({touches}회)")
practice.shot(img, "touch", spot)
tap(*spot)
store_via = "touch"
opened = False
for _ in range(6):
time.sleep(0.5)
if "com.android.vending" in focused_activity():
opened = True
break
practice.spot_result(spot, opened)
continue
try:
_keys.get(timeout=2)
print(" 뒤로가기 전송")
adb("shell", "input", "keyevent", "KEYCODE_BACK")
practice.event("user_back")
continue
except queue.Empty:
pass
tries += 1
if tries == 45:
practice.shot(img, "stuck")
print(f" ! 광고가 오래 안 닫힘 (기록: shots/ads/{practice.dir.name})")
notify("광고가 자동으로 안 닫힙니다. X를 눌러 주세요")
if tries % 3 == 0:
pos = find_close_button()
if pos and is_ad_activity(focused_activity()):
print(f" 닫기 버튼 발견 {pos} → 탭")
practice.event("ui_close", pos=list(pos))
tap(*pos)
if tries % 10 == 0 and is_ad_activity(focused_activity()):
practice.event("back_key")
adb("shell", "input", "keyevent", "KEYCODE_BACK")
practice.finish()
# ---------------------------------------------------------------- 화면 인식
def screen_xy(col: int, row: int) -> tuple[float, float]:
g = CFG["grid"]
return g["x0"] + col * g["dx"], g["y0"] + row * g["dy"]
def classify_screen(img: np.ndarray) -> list[str]:
"""화면 행/열 순서(row*9+col)로 빈칸/초(파랑)/한(빨강) 판별."""
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV).astype(np.int16)
r = CFG["sample_radius"]
yy, xx = np.mgrid[-r : r + 1, -r : r + 1]
disk = (xx**2 + yy**2) <= r * r
out: list[str] = []
for row in range(10):
for col in range(9):
x, y = (int(round(v)) for v in screen_xy(col, row))
p = hsv[y - r : y + r + 1, x - r : x + r + 1][disk]
h, s, v = p[:, 0], p[:, 1], p[:, 2]
white = ((s < 50) & (v > 160)).mean()
blue = ((h >= 100) & (h <= 135) & (s > 90)).mean()
red = (((h <= 8) | (h >= 170)) & (s > 90)).mean()
if white + blue + red < 0.5 or max(blue, red) < 0.1:
out.append(EMPTY)
else:
out.append(CHO if blue > red else HAN)
return out
@dataclass
class View:
flipped: bool # True면 화면 위쪽이 초 (엔진 좌표를 180도 돌려서 표시)
def to_screen(self, file: int, rank: int) -> tuple[int, int]:
return (8 - file, 9 - rank) if self.flipped else (file, rank)
def canonical(self, screen_occ: list[str]) -> list[str]:
occ = [EMPTY] * 90
for rank in range(10):
for file in range(9):
col, row = self.to_screen(file, rank)
occ[rank * 9 + file] = screen_occ[row * 9 + col]
return occ
def xy(self, file: int, rank: int) -> tuple[float, float]:
return screen_xy(*self.to_screen(file, rank))
def detect_view(screen_occ: list[str]) -> View | None:
top = [screen_occ[i] for i in range(0, 4 * 9)]
bottom = [screen_occ[i] for i in range(6 * 9, 90)]
if top.count(CHO) > top.count(HAN) and bottom.count(HAN) > bottom.count(CHO):
return View(flipped=True)
if top.count(HAN) > top.count(CHO) and bottom.count(CHO) > bottom.count(HAN):
return View(flipped=False)
return None
def patch_gray(img: np.ndarray, col: int, row: int) -> np.ndarray:
k = CFG["template_half"]
x, y = (int(round(v)) for v in screen_xy(col, row))
return cv2.cvtColor(img[y - k : y + k, x - k : x + k], cv2.COLOR_BGR2GRAY)
def load_templates() -> dict[str, list[np.ndarray]]:
out: dict[str, list[np.ndarray]] = {}
for path in sorted(TEMPLATES.glob("*.png")):
key = "_".join(path.stem.split("_")[:2]) # cho_horse_1 -> cho_horse
out.setdefault(key, []).append(cv2.imread(str(path), cv2.IMREAD_GRAYSCALE))
return out
def match_score(patch: np.ndarray, templates: list[np.ndarray]) -> float:
best = -1.0
for t in templates:
m = 13
inner = t[m:-m, m:-m]
best = max(best, float(cv2.matchTemplate(patch, inner, cv2.TM_CCOEFF_NORMED).max()))
return best
def detect_wings(img: np.ndarray, view: View, occ: list[str], side: str) -> str | None:
"""엔진 좌표 file 1,2,6,7의 마/상 배치 (예: HEEH). 빈 칸은 짝(1-2, 6-7)으로 추론."""
tpl = load_templates()
name = "cho" if side == CHO else "han"
horse, elephant = tpl.get(f"{name}_horse"), tpl.get(f"{name}_elephant")
if not horse or not elephant:
raise RuntimeError("templates/ 에 마/상 템플릿이 없습니다. make-templates 먼저 실행")
rank = 9 if side == CHO else 0
found: dict[int, str] = {}
for f in WING_FILES:
if occ[rank * 9 + f] != side:
continue
col, row = view.to_screen(f, rank)
p = patch_gray(img, col, row)
found[f] = "H" if match_score(p, horse) >= match_score(p, elephant) else "E"
flip = {"H": "E", "E": "H"}
for a, b in ((1, 2), (6, 7)):
if a not in found and b in found:
found[a] = flip[found[b]]
if b not in found and a in found:
found[b] = flip[found[a]]
if a in found and b in found and found[a] == found[b]:
return None
if len(found) != 4:
return None
return "".join(found[f] for f in WING_FILES)
# ---------------------------------------------------------------- 국면
def start_occupancy() -> list[str]:
occ = [EMPTY] * 90
for side, back, king, cannon, pawn in ((HAN, 0, 1, 2, 3), (CHO, 9, 8, 7, 6)):
for f in (0, 1, 2, 3, 5, 6, 7, 8):
occ[back * 9 + f] = side
occ[king * 9 + 4] = side
occ[cannon * 9 + 1] = occ[cannon * 9 + 7] = side
for f in range(0, 9, 2):
occ[pawn * 9 + f] = side
return occ
def start_board(cho_wings: str, han_wings: str) -> list[str]:
"""엔진 좌표의 기물 배열. 각 칸은 '' 또는 'cR', 'hK' 같은 (진영+종류)."""
board = [""] * 90
for side, wings, back, king, cannon, pawn in (
(HAN, han_wings, 0, 1, 2, 3),
(CHO, cho_wings, 9, 8, 7, 6),
):
for f, t in ((0, "R"), (8, "R"), (3, "A"), (5, "A")):
board[back * 9 + f] = side + t
for f, t in zip(WING_FILES, wings):
board[back * 9 + f] = side + t
board[king * 9 + 4] = side + "K"
board[cannon * 9 + 1] = board[cannon * 9 + 7] = side + "C"
for f in range(0, 9, 2):
board[pawn * 9 + f] = side + "P"
return board
def parse_move(mv: str) -> tuple[int, int] | None:
if mv == "pass":
return None
return int(mv[1]) * 9 + int(mv[0]), int(mv[3]) * 9 + int(mv[2])
def apply_occ(occ: list[str], mv: str) -> list[str]:
ft = parse_move(mv)
if ft is None:
return occ
out = occ[:]
out[ft[1]] = out[ft[0]]
out[ft[0]] = EMPTY
return out
@dataclass
class State:
view: View
cho_wings: str
han_wings: str
moves: list[str] = field(default_factory=list)
# 대국 중간에 시작한 경우: 화면에서 읽은 국면과 그때 둘 차례 (moves는 그 이후 수)
base_board: list[str] | None = None
base_side: str = CHO
def board(self) -> list[str]:
b = self.base_board[:] if self.base_board else start_board(self.cho_wings, self.han_wings)
for mv in self.moves:
ft = parse_move(mv)
if ft:
b[ft[1]], b[ft[0]] = b[ft[0]], ""
return b
def occ(self) -> list[str]:
return [p[0] if p else EMPTY for p in self.board()]
def describe(self, mv: str) -> str:
if mv == "pass":
return "한수 쉼"
b = self.board()
f, t = parse_move(mv) # type: ignore[misc]
piece = b[f]
cap = f" x{PIECE_KO[b[t][1]]}" if b[t] else ""
lbl = lambda i: f"{i % 9 + 1}{i // 9 + 1}" # noqa: E731
return f"{SIDE_KO[piece[0]]} {PIECE_KO[piece[1]]} {lbl(f)}→{lbl(t)}{cap}"
ENGINE_WEIGHTS: str | None = None
def engine(state: State, depth: int, avoid: list[str] | None = None) -> dict:
# 반복 금지는 앱이 판정 (앱 규칙: 같은 수 3회 금지). 엔진 규칙(같은 국면 3회)은 끔
cmd = [
str(ENGINE),
"--cho", FORMATIONS[state.cho_wings],
"--han", FORMATIONS[state.han_wings],
"--depth", str(depth),
"--moves", ",".join(state.moves),
"--no-repetition",
]
if avoid:
cmd += ["--avoid", ",".join(avoid)]
if state.base_board:
cmd += [
"--board", ",".join(p or "." for p in state.base_board),
"--side", "cho" if state.base_side == CHO else "han",
]
if ENGINE_WEIGHTS:
cmd += ["--weights", ENGINE_WEIGHTS]
out = json.loads(subprocess.run(cmd, capture_output=True, text=True, check=False).stdout)
if "error" in out:
raise RuntimeError(f"engine: {out['error']}")
return out
def init_state(img: np.ndarray) -> State | None:
"""시작 국면이면 상차림부터, 초반 몇 수면 수순 복원, 그 외엔 지금 국면 그대로 시작."""
import recorder # 순환 import 방지
def valid(state: State) -> State | None:
try:
engine(state, 0)
except RuntimeError:
return None
return state
screen_occ = classify_screen(img)
view = detect_view(screen_occ)
if view is not None:
occ = view.canonical(screen_occ)
cho_w = detect_wings(img, view, occ, CHO)
han_w = detect_wings(img, view, occ, HAN)
if cho_w in FORMATIONS and han_w in FORMATIONS:
state = State(view, cho_w, han_w)
if occ == start_occupancy():
return state
read = recorder.read_board(img)
if read and read[0].flipped == view.flipped:
seq = recorder.explain(state.board(), read[1], CHO)
if seq:
state.moves = [f"{f % 9}{f // 9}{t % 9}{t // 9}" for f, t in seq]
return valid(state)
if view is not None and view.canonical(screen_occ) == start_occupancy():
return None # 시작 국면인데 상차림을 못 읽음: 다음 화면에서 다시
read = recorder.read_board(img)
if read is None:
return None
state = State(read[0], "HEEH", "HEEH", base_board=read[1], base_side=side_to_move_ui(img))
return valid(state)
def side_to_move_ui(img: np.ndarray) -> str:
"""아래 이름 줄: 왼쪽 = 초, 오른쪽 = 한. 차례인 쪽 글자가 흰색(255), 아닌 쪽은 회색."""
left = int(text_mask(img, CFG["name_band_left"]).max())
right = int(text_mask(img, CFG["name_band_right"]).max())
return CHO if left > right else HAN
# ---------------------------------------------------------------- 대국 루프
def save_shot(img: np.ndarray, tag: str) -> Path:
SHOTS.mkdir(exist_ok=True)
path = SHOTS / f"{time.strftime('%H%M%S')}_{tag}.png"
cv2.imwrite(str(path), img)
return path
def wait_for_move(state: State, legal: list[str], who: str) -> str:
occ = state.occ()
start = start_occupancy()
last_obs: list[str] | None = None
mismatch = 0
polls = 0
while True:
polls += 1
if polls % 8 == 0 and is_ad_activity(focused_activity()):
raise GameInterrupted("광고 화면")
img = screencap()
if polls % 2 == 0:
kind = handle_dialog(img)
if kind == "result":
raise GameInterrupted("대국 종료 (결과 창)")
if kind:
last_obs = None
continue
obs = state.view.canonical(classify_screen(img))
if (len(state.moves) >= 2 or occ != start) and obs == start and obs == last_obs:
raise GameInterrupted("새 대국 시작")
if obs != occ:
hits = [mv for mv in legal if apply_occ(occ, mv) == obs]
if hits and obs == last_obs:
return hits[0]
if not hits:
mismatch += 1
if mismatch == 10:
print(f" ! {who} 수를 판독 못함 (화면 저장: {save_shot(img, 'mismatch')})")
else:
mismatch = 0
last_obs = obs
time.sleep(CFG["poll_sec"])
def play_move(state: State, mv: str, side: str) -> bool:
if mv == "pass":
tap(*CFG["pass_button"])
time.sleep(1.5)
return True
f, t = parse_move(mv) # type: ignore[misc]
before = state.occ()
expected = apply_occ(before, mv)
img = screencap()
if state.view.canonical(classify_screen(img)) != before:
print(f" ! 화면이 알고 있는 국면과 달라 두지 않음 (화면 저장: {save_shot(img, 'diff')})")
return False
empty = next((i for i, p in enumerate(before) if p == EMPTY), None)
for attempt in range(4):
if empty is not None:
tap(*state.view.xy(empty % 9, empty // 9))
time.sleep(CFG["tap_gap_sec"])
tap(*state.view.xy(f % 9, f // 9))
time.sleep(CFG["tap_gap_sec"])
tap(*state.view.xy(t % 9, t // 9))
deadline = time.time() + 5
while time.time() < deadline:
time.sleep(CFG["poll_sec"])
try:
img = screencap(retries=2)
except RuntimeError:
continue
kind = handle_dialog(img)
if kind == "result":
raise GameInterrupted("대국 종료 (결과 창)")
if kind:
deadline = time.time() + 5
continue
obs = state.view.canonical(classify_screen(img))
if obs == expected or (obs[f] == EMPTY and obs[t] == side):
return True
print(f" ! 착수가 반영되지 않아 다시 시도 ({attempt + 1}/4)")
print(f" (화면 저장: {save_shot(screencap(), 'move_fail')})")
return False
def wait_for_start(use_mirror: bool) -> State:
print("시작 국면을 기다리는 중...")
polls = 0
prev_key = None
last_diag = 0.0
while True:
polls += 1
if polls % 3 == 0 and is_ad_activity(focused_activity()):
handle_ad(use_mirror)
print("시작 국면을 기다리는 중...")
if polls % 10 == 0:
ensure_app_foreground()
img = screencap()
if handle_dialog(img):
prev_key = None
continue
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
if _match(gray, "replay_first", 1400)[0] >= CFG["guard_threshold"]:
print(" 관전 화면 → 뒤로가기 (로비에서 초대를 기다림)")
adb("shell", "input", "keyevent", "KEYCODE_BACK")
time.sleep(2)
continue
state = init_state(img)
if state is None and time.time() - last_diag > 30 and \
find_word(text_mask(img, CFG["title_band"]), "title_friendly") >= CFG["guard_threshold"]:
last_diag = time.time()
print(f" ! 친선 대국 화면인데 국면을 못 읽음 (화면 저장: {save_shot(img, 'start_unknown')})")
if state is not None:
ok, msg = friendly_guard(img)
key = (state.board(), state.base_side, len(state.moves))
if ok and key == prev_key: # 화면 전환 중 오인식 방지: 두 번 연속 같아야 시작
return state
prev_key = key if ok else None
if not ok and polls % 10 == 1:
print(f" 대기: {msg}")
else:
prev_key = None
time.sleep(1)
def wait_game_end(timeout: float = 90) -> None:
"""엔진이 끝났다고 본 뒤: 결과 창이 뜨면 확인을 누르고 끝. 광고가 뜨면 바로 넘김."""
deadline = time.time() + timeout
while time.time() < deadline:
if is_ad_activity(focused_activity()):
return
if handle_dialog(screencap()) == "result":
return
time.sleep(1)
def wait_for_guard() -> None:
"""내 차례마다 친선 대국/상대 확인. 아니면 탭하지 않고 기다림."""
warned = ""
while True:
img = screencap()
ok, msg = friendly_guard(img)
if ok:
return
if handle_dialog(img) == "result":
raise GameInterrupted("대국 종료 (결과 창)")
if msg != warned:
print(f" ! {msg} - 확인될 때까지 두지 않음")
warned = msg
if is_ad_activity(focused_activity()):
raise GameInterrupted("광고 화면")
time.sleep(2)
def cmd_play(args: argparse.Namespace) -> None:
global ENGINE_WEIGHTS, FORMATION_CHOICE
FORMATION_CHOICE = args.formation
if args.weights:
ENGINE_WEIGHTS = str(Path(args.weights).resolve())
threading.Thread(target=_stdin_reader, daemon=True).start()
# USB 연결 중에는 화면이 꺼지지 않게 (대기 화면 방지). 원래대로: adb shell svc power stayon false
adb("shell", "svc", "power", "stayon", "usb")
ensure_app_foreground()
game_no = 0
while True:
game_no += 1
state = wait_for_start(not args.no_mirror)
print(f"===== {game_no}번째 대국 =====")
try:
play_game(state, args)
except GameInterrupted as e:
print(f"대국 중단: {e} (화면 저장: {save_shot(screencap(), 'end')})")
if args.once:
return
def refresh_wings(state: State) -> None:
"""첫 수 전에 상차림을 다시 읽음 (상차림 고르기 전 기본 배치를 읽었을 수 있음)."""
img = screencap()
occ = state.view.canonical(classify_screen(img))
if occ != start_occupancy():
return
cho_w = detect_wings(img, state.view, occ, CHO)
han_w = detect_wings(img, state.view, occ, HAN)
if cho_w in FORMATIONS and han_w in FORMATIONS and (cho_w, han_w) != (state.cho_wings, state.han_wings):
state.cho_wings, state.han_wings = cho_w, han_w
print(f" 상차림 다시 인식: 초={FORMATIONS[cho_w]}, 한={FORMATIONS[han_w]}")
ENDGAME_PIECES = 12 # 이하이면 깊이 +1 (궁 안전+잡기 연장 평가로 깊이 5: 기물 12개 ~5초, 17개 ~30초)
def repeated_moves(state: State, me: str) -> list[str]:
"""두면 이미 나왔던 국면이 되는 내 수. 앱은 같은 국면 2번째에 경고, 3번째는 착수를 막음."""
b = state.base_board[:] if state.base_board else start_board(state.cho_wings, state.han_wings)
seen = {tuple(b)}
for mv in state.moves:
ft = parse_move(mv)
if ft:
b[ft[1]], b[ft[0]] = b[ft[0]], ""
seen.add(tuple(b))
avoid = set()
for mv in set(state.moves):
ft = parse_move(mv)
if not ft or not b[ft[0]] or b[ft[0]][0] != me or (b[ft[1]] and b[ft[1]][0] == me):
continue
nb = b[:]
nb[ft[1]], nb[ft[0]] = nb[ft[0]], ""
if tuple(nb) in seen:
avoid.add(mv)
return sorted(avoid)
class KiboWriter:
"""봇이 둔 대국을 기보 DB(recorder.py와 같은 형식)에 저장. 첫 수가 생길 때 대국 행을 만듦."""
def __init__(self, state: State, me: str) -> None:
self.state = state
self.me = me
self.gid: int | None = None
self.ply = 0
def _open(self) -> None:
import recorder
s = self.state
self.conn = recorder.db()
self.board = s.base_board[:] if s.base_board else start_board(s.cho_wings, s.han_wings)
self.side = s.base_side
cur = self.conn.execute(
"INSERT INTO games (source, started_at, from_start, start_ply, start_board, flipped, my_side, note)"
" VALUES (?,?,?,?,?,?,?,?)",
("bot", recorder.now(), int(s.base_board is None), 0, json.dumps(self.board), int(s.view.flipped),
self.me, f"bot.py play (AI={SIDE_KO[self.me]})"),
)
self.gid = cur.lastrowid
self.now = recorder.now
def sync(self) -> None:
if self.gid is None:
self._open()
for mv in self.state.moves[self.ply:]:
self.ply += 1
piece = captured = None
if mv != "pass":
f, t = parse_move(mv) # type: ignore[misc]
piece, captured = self.board[f], self.board[t] or None
self.board[t], self.board[f] = self.board[f], ""
self.conn.execute(
"INSERT INTO moves (game_id, ply, side, move, piece, captured, board_after, created_at)"
" VALUES (?,?,?,?,?,?,?,?)",
(self.gid, self.ply, self.side, mv, piece, captured, json.dumps(self.board), self.now()),
)
self.side = opposite_side(self.side)
self.conn.execute("UPDATE games SET move_count=? WHERE id=?", (self.ply, self.gid))
self.conn.commit()
def close(self, reason: str) -> None:
if self.gid is None:
return
self.conn.execute(
"UPDATE games SET ended_at=?, move_count=?, end_reason=? WHERE id=?",
(self.now(), self.ply, reason, self.gid),
)
self.conn.commit()
print(f" 기보 저장: #{self.gid} ({self.ply}수, {reason})")
def play_game(state: State, args: argparse.Namespace) -> None:
me = (HAN if state.view.flipped else CHO) if args.side == "auto" else (CHO if args.side == "cho" else HAN)
writer = KiboWriter(state, me)
try:
_play_game(state, args, writer)
except GameInterrupted as e:
writer.close(str(e))
raise
writer.close("엔진 판정 종료")
def _play_game(state: State, args: argparse.Namespace, writer: KiboWriter) -> None:
if args.side == "auto":
me = HAN if state.view.flipped else CHO # 이 폰 계정의 말이 화면 아래쪽
else:
me = CHO if args.side == "cho" else HAN
print(f"AI 진영: {SIDE_KO[me]}")
print(
f"인식 완료: 화면 위쪽={'초' if state.view.flipped else '한'}, "
f"초 상차림={FORMATIONS[state.cho_wings]}, 한 상차림={FORMATIONS[state.han_wings]}"
)
for mv in state.moves:
print(f" (이미 둔 수) {state.describe(mv)}")
while True:
if not state.moves and state.base_board is None:
refresh_wings(state)
# 기물이 줄면 탐색이 빨라짐: 끝내기에선 한 수 더 깊게 (궁 근처 공격을 놓치지 않게)
pieces = sum(1 for p in state.board() if p)
depth = min(5, args.depth + 1) if pieces <= ENDGAME_PIECES else args.depth
info = engine(state, depth, repeated_moves(state, me))
if info["result"] != "playing":
print(f"대국 종료: {info['result']}")
wait_game_end()
return
side = CHO if info["side"] == "cho" else HAN
ply = len(state.moves) + 1
if side == me:
wait_for_guard()
mv = info["best"]
print(f"{ply:3d}. [AI] {state.describe(mv)}" + (" (장군 받는 중)" if info["check"] else ""))
if args.dry_run:
mv = wait_for_move(state, info["legal"], "내")
elif not play_move(state, mv, me):
print(" ! 자동 착수 실패 - 같은 수를 다시 둡니다")
if not play_move(state, mv, me):
print(" ! 착수가 계속 실패 - 화면에서 수가 바뀌면 이어서 진행합니다")
mv = wait_for_move(state, info["legal"], "내")
else:
mv = wait_for_move(state, info["legal"], "상대")
print(f"{ply:3d}. [상대] {state.describe(mv)}")
state.moves.append(mv)
writer.sync()
def cmd_ad_review(args: argparse.Namespace) -> None:
pngs = args.pngs or sorted((SHOTS / "ads").glob("*/*.png"))
out_dir = SHOTS / "ad_review"
out_dir.mkdir(parents=True, exist_ok=True)
for path in pngs:
img = cv2.imread(str(path))
if img is None:
continue
pos = find_corner_x(img)
print(f"{path.parent.name}/{path.name}: X {'발견 ' + str(pos) if pos else '없음'}")
if pos:
cv2.circle(img, pos, 45, (0, 0, 255), 6)
cv2.imwrite(str(out_dir / f"{path.parent.name}_{path.name}"), img)
try:
stats = json.loads(AD_STATS.read_text())
except (OSError, ValueError):
return
print("\n터치 위치별 스토어 열림:")
for spot, s in sorted(stats["spots"].items(), key=lambda kv: -kv[1]["store"] / max(1, kv[1]["tries"])):
print(f" {spot}: {s['store']}/{s['tries']}")
for ad in stats["ads"][-10:]:
print(f" 광고 {ad['dir']}: {ad['secs']}초, 마지막 행동 {ad['closed_by']}")
def cmd_ad(args: argparse.Namespace) -> None:
threading.Thread(target=_stdin_reader, daemon=True).start()
act = focused_activity()
if not is_ad_activity(act):
print(f"광고 화면이 아닙니다: {act}")
return
handle_ad(not args.no_mirror)
def cmd_show(args: argparse.Namespace) -> None:
img = cv2.imread(args.png) if args.png else screencap()
screen_occ = classify_screen(img)
for row in range(10):
print(" ".join(screen_occ[row * 9 : row * 9 + 9]))
state = init_state(img)
if state is None:
print("시작 국면이 아니거나 인식 실패 (방향/상차림 판별 불가)")
return
print(
f"화면 위쪽={'초' if state.view.flipped else '한'}, "
f"초={FORMATIONS[state.cho_wings]}({state.cho_wings}), "
f"한={FORMATIONS[state.han_wings]}({state.han_wings}), 이미 둔 수={state.moves}"
)
ok, msg = friendly_guard(img)
if not ok and not args.png:
print(f"추천 수 생략: {msg}")
return
info = engine(state, args.depth)
print(f"차례={SIDE_KO[CHO if info['side'] == 'cho' else HAN]}, 추천: {state.describe(info['best'])}")
def cmd_calib_friendly(args: argparse.Namespace) -> None:
"""나의길 계정과의 친선 대국 화면에서 실행: 제목 첫 단어와 상대 이름을 기준 이미지로 저장.
이름 줄: 왼쪽 = "점수 급수 이름", 오른쪽 = "급수 이름 점수"."""
img = screencap()
title = text_mask(img, CFG["title_band"])
if find_word(title, "title_ranked") >= CFG["guard_threshold"]:
sys.exit("지금은 승강급 대국 화면입니다. 친선 대국 화면에서 실행해 주세요.")
tw = words(title)
if args.opponent_side == "right":
ow = words(text_mask(img, CFG["name_band_right"]))
name = ow[-2] if len(ow) >= 2 else None
else:
ow = words(text_mask(img, CFG["name_band_left"]))
name = ow[-1] if ow else None
if not tw or name is None:
sys.exit("제목/상대 이름을 찾지 못했습니다.")
GUARDS.mkdir(exist_ok=True)
cv2.imwrite(str(GUARDS / "title_friendly.png"), tw[0])
cv2.imwrite(str(GUARDS / "opponent.png"), name)
preview = save_shot(img, "calib_friendly")
print(f"저장: guards/title_friendly.png {tw[0].shape}, guards/opponent.png {name.shape}")
print(f"확인용 화면: {preview}")
print("검사 결과:", friendly_guard(img))
def cmd_make_templates(args: argparse.Namespace) -> None:
img = cv2.imread(args.png)
screen_occ = classify_screen(img)
TEMPLATES.mkdir(exist_ok=True)
names = {"H": "horse", "E": "elephant"}
for row, wings in ((0, args.top), (9, args.bottom)):
for i, (col, t) in enumerate(zip(WING_FILES, wings.upper())):
side = screen_occ[row * 9 + col]
if side == EMPTY:
sys.exit(f"화면 ({col},{row}) 이 비어 있습니다")
name = f"{'cho' if side == CHO else 'han'}_{names[t]}_{row}{i}.png"
cv2.imwrite(str(TEMPLATES / name), patch_gray(img, col, row))
print("saved", name)
def main() -> None:
sys.stdout.reconfigure(line_buffering=True) # type: ignore[union-attr]
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("play")
p.add_argument("--side", choices=["auto", "cho", "han"], default="auto")
p.add_argument("--depth", type=int, default=4)
p.add_argument("--weights", default=str(ROOT / "data" / "weights_bot.txt"),
help="평가 가중치 파일 (기본: 자가대국으로 검증한 data/weights_bot.txt)")
p.add_argument("--dry-run", action="store_true")
p.add_argument("--once", action="store_true", help="한 판만 두고 종료")
p.add_argument("--no-mirror", action="store_true", help="광고 때 scrcpy 미러 창을 열지 않음")
p.add_argument("--formation", choices=["auto", "inner", "outer", "left", "right"], default="auto",
help="상차림 선택 창에서 고를 포진 (auto = 매 판 무작위)")
p.set_defaults(func=cmd_play)
p = sub.add_parser("calib-friendly")
p.add_argument("--opponent-side", choices=["left", "right"], default="right")
p.set_defaults(func=cmd_calib_friendly)
p = sub.add_parser("ad")
p.add_argument("--no-mirror", action="store_true")
p.set_defaults(func=cmd_ad)
p = sub.add_parser("ad-review", help="저장된 광고 화면에서 X 탐지 결과 확인 + 터치 위치 통계")
p.add_argument("pngs", nargs="*", type=Path)
p.set_defaults(func=cmd_ad_review)
p = sub.add_parser("show")
p.add_argument("--png")
p.add_argument("--depth", type=int, default=4)
p.set_defaults(func=cmd_show)
p = sub.add_parser("make-templates")
p.add_argument("png")
p.add_argument("--top", required=True)
p.add_argument("--bottom", required=True)
p.set_defaults(func=cmd_make_templates)
args = ap.parse_args()
try:
args.func(args)
except KeyboardInterrupt:
print("\n중단")
if __name__ == "__main__":
main()