664 lines
30 KiB
Python
664 lines
30 KiB
Python
"""
|
|
Core FSM Bot engine for DirtyLeague.
|
|
Manages screen recognition, game state transitions, and automated inputs.
|
|
Supports resolution-independent relative coordinates and adaptive template scaling.
|
|
"""
|
|
|
|
from enum import Enum
|
|
import logging
|
|
import os
|
|
import re
|
|
import time
|
|
from typing import Dict, Optional, Tuple
|
|
import cv2
|
|
import numpy as np
|
|
from window_utils import WindowManager
|
|
from ocr_utils import ocr_reader
|
|
|
|
|
|
class GameState(Enum):
|
|
UNKNOWN = "UNKNOWN"
|
|
MAIN_MENU = "MAIN_MENU"
|
|
TOWER_LOBBY = "TOWER_LOBBY"
|
|
CHEST_OPEN_SCREEN = "CHEST_OPEN_SCREEN"
|
|
IN_GAME = "IN_GAME"
|
|
VICTORY_SCREEN = "VICTORY_SCREEN"
|
|
DEFEAT_SCREEN = "DEFEAT_SCREEN"
|
|
NO_FREE_SLOTS_SCREEN = "NO_FREE_SLOTS_SCREEN"
|
|
|
|
|
|
class DirtyLeagueBot:
|
|
# Reference design resolution from which templates and coordinate ratios were calibrated
|
|
BASE_DESIGN_WIDTH = 3840
|
|
BASE_DESIGN_HEIGHT = 2050
|
|
|
|
# Normalized relative ROIs (rx1, ry1, rx2, ry2) spanning 0.0 to 1.0
|
|
REL_ROI_EXIT = (0.4167, 0.9268, 0.5000, 0.9854)
|
|
REL_ROI_LEAVE = (0.4948, 0.5610, 0.6250, 0.6829)
|
|
REL_ROI_FIGHT = (0.7812, 0.3659, 0.9245, 0.4878)
|
|
REL_ROI_DEFEAT_OK = (0.2344, 0.7220, 0.3776, 0.8293)
|
|
REL_ROI_DEFEAT_TEXT = (0.1302, 0.6829, 0.3125, 0.8293)
|
|
REL_ROI_VICTORY_COLLECT = (0.1953, 0.7561, 0.3125, 0.8537)
|
|
REL_ROI_VICTORY_BANNER = (0.1042, 0.1463, 0.4167, 0.3415)
|
|
REL_ROI_NO_SLOTS_REMOVE = (0.3516, 0.8195, 0.5078, 0.9024)
|
|
REL_ROI_NO_SLOTS_OPEN = (0.4818, 0.8195, 0.6120, 0.9024)
|
|
REL_ROI_CHEST_TURN_CARDS = (0.4036, 0.6585, 0.5859, 0.7561)
|
|
REL_ROI_CHEST_COLLECT = (0.4036, 0.6585, 0.5859, 0.7561)
|
|
REL_ROI_OFFER_CLOSE = (0.8854, 0.0, 1.0, 0.1463)
|
|
REL_CROWN_CHEST_POS = (0.2279, 0.6146)
|
|
|
|
def __init__(self, config: dict):
|
|
self.config = config
|
|
self.bot_conf = config.get("bot", {})
|
|
self.target_conf = config.get("target_window", {})
|
|
self.assets_dir = config.get("assets_dir", "assets")
|
|
|
|
self.max_wins_limit = self.bot_conf.get("max_wins_limit", 3)
|
|
self.poll_interval = self.bot_conf.get("poll_interval_sec", 0.8)
|
|
self.confidence_threshold = self.bot_conf.get("confidence_threshold", 0.8)
|
|
self.action_delay = self.bot_conf.get("action_delay_sec", 0.5)
|
|
|
|
self.current_wins = 0
|
|
self.total_games = 0
|
|
self.current_trophies: Optional[int] = None
|
|
self.max_trophies: Optional[int] = None
|
|
|
|
self.league_conf = config.get("league_retention", {})
|
|
self.upper_trophies = self.league_conf.get("upper_trophy_threshold", 2125)
|
|
self.lower_trophies = self.league_conf.get("lower_trophy_threshold", 125)
|
|
self.derank_mode: bool = False
|
|
self._result_recorded_for_match: bool = False
|
|
|
|
self.running = False
|
|
self.state = GameState.UNKNOWN
|
|
self._unknown_state_count = 0
|
|
|
|
self.wm = WindowManager(
|
|
window_title=self.target_conf.get("title", "DirtyLeague"),
|
|
process_name=self.target_conf.get("process_name", "DirtyLeague.exe")
|
|
)
|
|
|
|
# Template storage with dynamic scaling cache
|
|
self.base_templates: Dict[str, np.ndarray] = {}
|
|
self.scaled_templates: Dict[str, np.ndarray] = {}
|
|
self._current_scale: float = 1.0
|
|
self._load_templates()
|
|
|
|
@property
|
|
def templates(self) -> Dict[str, np.ndarray]:
|
|
"""Provides access to templates at the current resolution scale."""
|
|
return self.scaled_templates if self.scaled_templates else self.base_templates
|
|
|
|
def _load_templates(self):
|
|
"""Preloads all base PNG template images from the assets directory."""
|
|
if not os.path.exists(self.assets_dir):
|
|
return
|
|
for file_name in os.listdir(self.assets_dir):
|
|
if file_name.endswith(".png"):
|
|
name_without_ext = os.path.splitext(file_name)[0]
|
|
path = os.path.join(self.assets_dir, file_name)
|
|
img = cv2.imread(path)
|
|
if img is not None:
|
|
self.base_templates[name_without_ext] = img
|
|
self.scaled_templates[name_without_ext] = img
|
|
logging.info(f"Loaded {len(self.base_templates)} base template(s) from '{self.assets_dir}'")
|
|
|
|
def get_frame_scale(self, frame: np.ndarray) -> float:
|
|
"""
|
|
Calculates UI scaling factor relative to base design height.
|
|
Unity Canvas Scaler in DirtyLeague scales UI proportionally to screen height.
|
|
"""
|
|
return frame.shape[0] / float(self.BASE_DESIGN_HEIGHT)
|
|
|
|
def get_scaled_template(self, template_name: str, scale: float) -> Optional[np.ndarray]:
|
|
"""
|
|
Retrieves template scaled for the given resolution scale factor (fh / BASE_DESIGN_HEIGHT).
|
|
Caches scaled results so resizing only happens on window resolution changes.
|
|
"""
|
|
if template_name not in self.base_templates:
|
|
path = os.path.join(self.assets_dir, f"{template_name}.png")
|
|
if os.path.exists(path):
|
|
img = cv2.imread(path)
|
|
if img is not None:
|
|
self.base_templates[template_name] = img
|
|
if template_name not in self.base_templates:
|
|
return None
|
|
|
|
# If resolution scale changed, rebuild scaled template cache
|
|
if abs(scale - self._current_scale) > 0.02:
|
|
self._current_scale = scale
|
|
self.scaled_templates.clear()
|
|
interp = cv2.INTER_AREA if scale < 1.0 else cv2.INTER_CUBIC
|
|
for name, base_img in self.base_templates.items():
|
|
if abs(scale - 1.0) <= 0.02:
|
|
self.scaled_templates[name] = base_img
|
|
else:
|
|
self.scaled_templates[name] = cv2.resize(
|
|
base_img, (0, 0), fx=scale, fy=scale, interpolation=interp
|
|
)
|
|
|
|
return self.scaled_templates.get(template_name)
|
|
|
|
def find_template(self, frame: np.ndarray, template_name: str, threshold: Optional[float] = None) -> Optional[Tuple[int, int]]:
|
|
"""
|
|
Full-frame search for template_name in frame.
|
|
Automatically scales template to match the current frame scale.
|
|
Returns client center coordinates (center_x, center_y) if matched, else None.
|
|
"""
|
|
fh, fw = frame.shape[:2]
|
|
scale = self.get_frame_scale(frame)
|
|
tpl = self.get_scaled_template(template_name, scale)
|
|
if tpl is None:
|
|
return None
|
|
|
|
th, tw = tpl.shape[:2]
|
|
if th > fh or tw > fw:
|
|
return None
|
|
|
|
res = cv2.matchTemplate(frame, tpl, cv2.TM_CCOEFF_NORMED)
|
|
_, max_val, _, max_loc = cv2.minMaxLoc(res)
|
|
|
|
target_thresh = threshold if threshold is not None else self.confidence_threshold
|
|
if max_val >= target_thresh:
|
|
center_x = max_loc[0] + tw // 2
|
|
center_y = max_loc[1] + th // 2
|
|
logging.debug(f"Matched '{template_name}' (conf: {max_val:.3f}) at client ({center_x}, {center_y})")
|
|
return center_x, center_y
|
|
|
|
return None
|
|
|
|
def find_template_in_rel_roi(
|
|
self,
|
|
frame: np.ndarray,
|
|
template_name: str,
|
|
rel_roi: Tuple[float, float, float, float],
|
|
threshold: Optional[float] = None
|
|
) -> Optional[Tuple[int, int]]:
|
|
"""
|
|
Fast template search restricted to normalized relative ROI: (rx1, ry1, rx2, ry2).
|
|
Returns client center coordinates (cx, cy) if matched, else None.
|
|
Automatically scales template to match current frame resolution. Runs in ~2-5ms!
|
|
"""
|
|
fh, fw = frame.shape[:2]
|
|
rx1, ry1, rx2, ry2 = rel_roi
|
|
x1 = max(0, min(fw, int(rx1 * fw)))
|
|
y1 = max(0, min(fh, int(ry1 * fh)))
|
|
x2 = max(0, min(fw, int(rx2 * fw)))
|
|
y2 = max(0, min(fh, int(ry2 * fh)))
|
|
|
|
scale = self.get_frame_scale(frame)
|
|
tpl = self.get_scaled_template(template_name, scale)
|
|
if tpl is None:
|
|
return None
|
|
|
|
th, tw = tpl.shape[:2]
|
|
if (x2 - x1) < tw or (y2 - y1) < th:
|
|
return None
|
|
|
|
crop = frame[y1:y2, x1:x2]
|
|
res = cv2.matchTemplate(crop, tpl, cv2.TM_CCOEFF_NORMED)
|
|
_, max_val, _, max_loc = cv2.minMaxLoc(res)
|
|
|
|
target_thresh = threshold if threshold is not None else self.confidence_threshold
|
|
if max_val >= target_thresh:
|
|
cx = x1 + max_loc[0] + tw // 2
|
|
cy = y1 + max_loc[1] + th // 2
|
|
return cx, cy
|
|
return None
|
|
|
|
def find_template_in_roi(
|
|
self,
|
|
frame: np.ndarray,
|
|
template_name: str,
|
|
roi: Tuple[int, int, int, int],
|
|
threshold: Optional[float] = None
|
|
) -> Optional[Tuple[int, int]]:
|
|
"""Backward-compatible pixel ROI search: converts pixel ROI to relative ROI."""
|
|
rx1 = roi[0] / float(self.BASE_DESIGN_WIDTH)
|
|
ry1 = roi[1] / float(self.BASE_DESIGN_HEIGHT)
|
|
rx2 = roi[2] / float(self.BASE_DESIGN_WIDTH)
|
|
ry2 = roi[3] / float(self.BASE_DESIGN_HEIGHT)
|
|
return self.find_template_in_rel_roi(frame, template_name, (rx1, ry1, rx2, ry2), threshold)
|
|
|
|
def click_client_pos(self, x: int, y: int):
|
|
"""Translates client coordinates to screen and performs a click via Win32 API."""
|
|
screen_x, screen_y = self.wm.client_to_screen(x, y)
|
|
logging.info(f"Clicking at client ({x}, {y}) -> screen ({screen_x}, {screen_y})")
|
|
from window_utils import ensure_interactive_desktop, user32
|
|
ensure_interactive_desktop()
|
|
self.wm.focus()
|
|
user32.SetCursorPos(screen_x, screen_y)
|
|
time.sleep(0.08)
|
|
user32.mouse_event(0x0002, 0, 0, 0, 0) # MOUSEEVENTF_LEFTDOWN
|
|
time.sleep(0.08)
|
|
user32.mouse_event(0x0004, 0, 0, 0, 0) # MOUSEEVENTF_LEFTUP
|
|
time.sleep(self.action_delay)
|
|
|
|
def click_client_pos_fast(self, x: int, y: int):
|
|
"""Ultra-fast click without artificial action delays for time-critical reactions (e.g. instant surrender)."""
|
|
screen_x, screen_y = self.wm.client_to_screen(x, y)
|
|
from window_utils import ensure_interactive_desktop, user32
|
|
ensure_interactive_desktop()
|
|
self.wm.focus()
|
|
user32.SetCursorPos(screen_x, screen_y)
|
|
time.sleep(0.02)
|
|
user32.mouse_event(0x0002, 0, 0, 0, 0) # MOUSEEVENTF_LEFTDOWN
|
|
time.sleep(0.02)
|
|
user32.mouse_event(0x0004, 0, 0, 0, 0) # MOUSEEVENTF_LEFTUP
|
|
|
|
def click_template(self, frame: np.ndarray, template_name: str, threshold: Optional[float] = None) -> bool:
|
|
"""Finds and clicks a template in the frame if found."""
|
|
pos = self.find_template(frame, template_name, threshold)
|
|
if pos:
|
|
self.click_client_pos(pos[0], pos[1])
|
|
return True
|
|
return False
|
|
|
|
def detect_state(self, frame: np.ndarray) -> GameState:
|
|
"""Determines current game screen based on visible templates (accelerated with relative ROIs)."""
|
|
# 1. Ultra-fast relative ROI checks (~3-5ms each on any resolution)
|
|
if self.find_template_in_rel_roi(frame, "btn_remove_chest", self.REL_ROI_NO_SLOTS_REMOVE) or \
|
|
self.find_template_in_rel_roi(frame, "btn_open_chest", self.REL_ROI_NO_SLOTS_OPEN):
|
|
return GameState.NO_FREE_SLOTS_SCREEN
|
|
|
|
if self.find_template_in_rel_roi(frame, "btn_ok", self.REL_ROI_DEFEAT_OK) or \
|
|
self.find_template_in_rel_roi(frame, "text_defeat", self.REL_ROI_DEFEAT_TEXT, threshold=0.65):
|
|
return GameState.DEFEAT_SCREEN
|
|
|
|
if self.find_template_in_rel_roi(frame, "btn_exit", self.REL_ROI_EXIT) or \
|
|
self.find_template_in_rel_roi(frame, "btn_leave", self.REL_ROI_LEAVE):
|
|
return GameState.IN_GAME
|
|
|
|
if self.find_template_in_rel_roi(frame, "btn_fight", self.REL_ROI_FIGHT):
|
|
return GameState.TOWER_LOBBY
|
|
|
|
if self.find_template_in_rel_roi(frame, "btn_turn_all_cards", self.REL_ROI_CHEST_TURN_CARDS) or \
|
|
self.find_template_in_rel_roi(frame, "btn_collect", self.REL_ROI_CHEST_COLLECT, threshold=0.65):
|
|
return GameState.CHEST_OPEN_SCREEN
|
|
|
|
if self.find_template_in_rel_roi(frame, "btn_collect_victory", self.REL_ROI_VICTORY_COLLECT) or \
|
|
self.find_template_in_rel_roi(frame, "banner_victory", self.REL_ROI_VICTORY_BANNER, threshold=0.65):
|
|
return GameState.VICTORY_SCREEN
|
|
|
|
# 2. Full-frame fallback checks if ROI didn't trigger:
|
|
if self.find_template(frame, "btn_remove_chest") or self.find_template(frame, "btn_open_chest"):
|
|
return GameState.NO_FREE_SLOTS_SCREEN
|
|
if self.find_template(frame, "text_victory") or self.find_template(frame, "banner_victory") or self.find_template(frame, "btn_collect_victory"):
|
|
return GameState.VICTORY_SCREEN
|
|
if self.find_template(frame, "text_defeat", threshold=0.65) or self.find_template(frame, "btn_ok"):
|
|
return GameState.DEFEAT_SCREEN
|
|
if self.find_template(frame, "btn_leave") or self.find_template(frame, "btn_exit") or self.find_template(frame, "btn_pause"):
|
|
return GameState.IN_GAME
|
|
if self.find_template(frame, "btn_collect") or self.find_template(frame, "btn_turn_all_cards"):
|
|
return GameState.CHEST_OPEN_SCREEN
|
|
if self.find_template(frame, "btn_fight") or self.find_template(frame, "btn_play"):
|
|
return GameState.TOWER_LOBBY
|
|
if self.find_template(frame, "btn_mode_tower"):
|
|
return GameState.MAIN_MENU
|
|
|
|
return GameState.UNKNOWN
|
|
|
|
def scan_bottom_chests(self, frame: np.ndarray) -> dict:
|
|
"""
|
|
Scans chest slots 1..4 in the bottom bar of TOWER_LOBBY.
|
|
Bottom collection bar is anchored to Bottom-Left in Unity Canvas.
|
|
Uses height scaling and centered crops with hybrid template + OCR detection.
|
|
"""
|
|
scale = self.get_frame_scale(frame)
|
|
tpl_open = self.get_scaled_template("status_open", scale)
|
|
tpl_vacant = self.get_scaled_template("status_vacant", scale)
|
|
|
|
base_x = 630
|
|
pitch_x = 213
|
|
base_y = 1885
|
|
half_w = int(90 * scale)
|
|
half_h = int(35 * scale)
|
|
|
|
results = {}
|
|
for i in range(4):
|
|
slot_id = i + 1
|
|
cx = int((base_x + i * pitch_x) * scale)
|
|
cy = int(base_y * scale)
|
|
|
|
y1 = max(0, cy - half_h)
|
|
y2 = min(frame.shape[0], cy + half_h)
|
|
x1 = max(0, cx - half_w)
|
|
x2 = min(frame.shape[1], cx + half_w)
|
|
status_crop = frame[y1:y2, x1:x2]
|
|
|
|
status = "TIMER"
|
|
# 1. Template matching for OPEN
|
|
is_open = False
|
|
if tpl_open is not None and status_crop.shape[0] >= tpl_open.shape[0] and status_crop.shape[1] >= tpl_open.shape[1]:
|
|
res_open = cv2.matchTemplate(status_crop, tpl_open, cv2.TM_CCOEFF_NORMED)
|
|
_, max_v_open, _, _ = cv2.minMaxLoc(res_open)
|
|
if max_v_open >= 0.70:
|
|
is_open = True
|
|
|
|
# 2. OCR fallback/confirmation for OPEN
|
|
txt = ocr_reader.read_text(status_crop).lower()
|
|
if is_open or "open" in txt:
|
|
status = "OPEN"
|
|
else:
|
|
# 3. Check for VACANT
|
|
if tpl_vacant is not None and status_crop.shape[0] >= tpl_vacant.shape[0] and status_crop.shape[1] >= tpl_vacant.shape[1]:
|
|
res_vacant = cv2.matchTemplate(status_crop, tpl_vacant, cv2.TM_CCOEFF_NORMED)
|
|
_, max_v_vacant, _, _ = cv2.minMaxLoc(res_vacant)
|
|
if max_v_vacant >= 0.70:
|
|
status = "VACANT"
|
|
if status != "VACANT" and ("vacan" in txt or txt == ""):
|
|
if not re.search(r"\d", txt):
|
|
status = "VACANT"
|
|
|
|
click_cx = cx
|
|
click_cy = cy - int(80 * scale)
|
|
results[slot_id] = {
|
|
"status": status,
|
|
"click_pos": (click_cx, click_cy)
|
|
}
|
|
|
|
return results
|
|
|
|
def check_and_dismiss_offer_popup(self, frame: np.ndarray) -> bool:
|
|
"""
|
|
Detects and dismisses promotional or purchase offer popups ('X' close button).
|
|
Uses relative ROI for the top-right corner across all resolutions.
|
|
"""
|
|
for tpl_name in ["btn_close", "btn_close_offer"]:
|
|
pos = self.find_template_in_rel_roi(frame, tpl_name, self.REL_ROI_OFFER_CLOSE, threshold=0.55)
|
|
if pos:
|
|
logging.info(f"[OFFER POPUP] Detected close button '{tpl_name}' at client {pos}. Dismissing...")
|
|
self.click_client_pos(pos[0], pos[1])
|
|
time.sleep(1.0)
|
|
return True
|
|
return False
|
|
|
|
def handle_chest_open_screen(self, frame: np.ndarray):
|
|
logging.info("State: CHEST_OPEN_SCREEN.")
|
|
# Step 1: Check if "Turn all cards over" is visible
|
|
turn_pos = self.find_template_in_rel_roi(frame, "btn_turn_all_cards", self.REL_ROI_CHEST_TURN_CARDS) or \
|
|
self.find_template(frame, "btn_turn_all_cards")
|
|
if turn_pos:
|
|
self.click_client_pos(turn_pos[0], turn_pos[1])
|
|
logging.info("Clicked 'Turn all cards over'. Waiting for card flip animation...")
|
|
time.sleep(2.0)
|
|
return
|
|
|
|
# Step 2: Check if "Collect" is visible
|
|
collect_pos = self.find_template_in_rel_roi(frame, "btn_collect", self.REL_ROI_CHEST_COLLECT) or \
|
|
self.find_template(frame, "btn_collect")
|
|
if collect_pos:
|
|
self.click_client_pos(collect_pos[0], collect_pos[1])
|
|
logging.info("Clicked 'Collect'. Rewards claimed, returning to lobby...")
|
|
time.sleep(1.5)
|
|
# Check for promotional offer popup stub
|
|
after_frame = self.wm.capture_frame(client_only=True)
|
|
self.check_and_dismiss_offer_popup(after_frame)
|
|
|
|
def handle_main_menu(self, frame: np.ndarray):
|
|
logging.info("State: MAIN_MENU. Navigating to TOWER mode...")
|
|
if self.click_template(frame, "btn_mode_tower"):
|
|
logging.info("Clicked 'btn_mode_tower'. Transitioning to TOWER...")
|
|
time.sleep(1.5)
|
|
|
|
def update_league_strategy(self):
|
|
"""Updates derank_mode based on current trophies and thresholds."""
|
|
if self.current_trophies is None:
|
|
return
|
|
|
|
if not self.derank_mode and self.current_trophies >= self.upper_trophies:
|
|
self.derank_mode = True
|
|
logging.warning(
|
|
f"[LEAGUE STRATEGY] Upper threshold reached ({self.current_trophies} >= {self.upper_trophies}). "
|
|
f"ACTIVATING FORCED DERANK MODE! (Will surrender matches until <= {self.lower_trophies})"
|
|
)
|
|
elif self.derank_mode and self.current_trophies <= self.lower_trophies:
|
|
self.derank_mode = False
|
|
logging.info(
|
|
f"[LEAGUE STRATEGY] Lower threshold reached ({self.current_trophies} <= {self.lower_trophies}). "
|
|
f"ACTIVATING NORMAL PLAY MODE! (Playing to win until >= {self.upper_trophies})"
|
|
)
|
|
|
|
def fast_surrender_pipeline(self, max_wait_sec: float = 15.0) -> bool:
|
|
"""
|
|
Ultra-fast surrender pipeline for FORCED DERANK.
|
|
Polls Exit button relative ROI (~20ms per check). The millisecond battle finishes loading,
|
|
clicks Exit via fast click, then immediately polls Leave dialog relative ROI and confirms.
|
|
Surrenders the battle within < 0.15s of battle load, before auto-battle can deal lethal damage.
|
|
"""
|
|
logging.info("[FAST DERANK] High-frequency surrender monitor engaged. Waiting for battle screen...")
|
|
t0 = time.time()
|
|
|
|
# Step 1: High-frequency polling for Exit button (check every 20ms)
|
|
exit_clicked = False
|
|
while time.time() - t0 < max_wait_sec:
|
|
if not self.running:
|
|
return False
|
|
frame = self.wm.capture_frame(client_only=True)
|
|
if frame is None:
|
|
time.sleep(0.02)
|
|
continue
|
|
|
|
exit_pos = self.find_template_in_rel_roi(frame, "btn_exit", self.REL_ROI_EXIT, threshold=0.75)
|
|
if exit_pos:
|
|
logging.info(f"[FAST DERANK] Battle loaded! 'Exit' detected at {exit_pos} in {time.time() - t0:.2f}s. Clicking...")
|
|
self.click_client_pos_fast(exit_pos[0], exit_pos[1])
|
|
exit_clicked = True
|
|
break
|
|
|
|
leave_pos = self.find_template_in_rel_roi(frame, "btn_leave", self.REL_ROI_LEAVE, threshold=0.75)
|
|
if leave_pos:
|
|
self.click_client_pos_fast(leave_pos[0], leave_pos[1])
|
|
logging.info("[FAST DERANK] Clicked existing 'Leave' confirmation.")
|
|
return True
|
|
|
|
time.sleep(0.02)
|
|
|
|
if not exit_clicked:
|
|
logging.warning("[FAST DERANK] Exit button not found within timeout.")
|
|
return False
|
|
|
|
# Step 2: High-frequency polling for 'Leave' confirmation dialog (check every 20ms)
|
|
t_leave = time.time()
|
|
while time.time() - t_leave < 4.0:
|
|
if not self.running:
|
|
return False
|
|
frame = self.wm.capture_frame(client_only=True)
|
|
if frame is None:
|
|
time.sleep(0.02)
|
|
continue
|
|
|
|
leave_pos = self.find_template_in_rel_roi(frame, "btn_leave", self.REL_ROI_LEAVE, threshold=0.75)
|
|
if leave_pos:
|
|
self.click_client_pos_fast(leave_pos[0], leave_pos[1])
|
|
logging.info(f"[FAST DERANK] Confirmed 'Leave' in {time.time() - t_leave:.3f}s! Total surrender time: {time.time() - t0:.2f}s.")
|
|
time.sleep(0.5)
|
|
return True
|
|
|
|
time.sleep(0.02)
|
|
|
|
logging.warning("[FAST DERANK] Leave button not found after clicking Exit.")
|
|
return False
|
|
|
|
def handle_lobby(self, frame: np.ndarray):
|
|
logging.info("State: TOWER_LOBBY.")
|
|
self._result_recorded_for_match = False
|
|
fh, fw = frame.shape[:2]
|
|
|
|
# Read trophies counter above FIGHT button
|
|
trophies = ocr_reader.read_trophies(frame)
|
|
if trophies:
|
|
self.current_trophies, self.max_trophies = trophies
|
|
logging.info(f"[TROPHIES] Current rating: {self.current_trophies}/{self.max_trophies} (Goal: {self.max_trophies})")
|
|
self.update_league_strategy()
|
|
|
|
mode_str = "FORCED DERANK (Auto-surrender)" if self.derank_mode else "NORMAL PLAY (Play to win)"
|
|
logging.info(f"[STRATEGY] Current Battle Plan: {mode_str}")
|
|
|
|
# Check crown chest progress (#/# 👑)
|
|
crown_info = ocr_reader.read_crown_chest(frame)
|
|
if crown_info:
|
|
c_current, c_max = crown_info
|
|
logging.info(f"[CROWN CHEST] Progress: {c_current}/{c_max} [CROWN]")
|
|
if c_current >= c_max:
|
|
logging.info(f"[CROWN CHEST] Goal reached ({c_current}/{c_max})! Clicking crown chest...")
|
|
cx = int(self.REL_CROWN_CHEST_POS[0] * fw)
|
|
cy = int(self.REL_CROWN_CHEST_POS[1] * fh)
|
|
self.click_client_pos(cx, cy)
|
|
time.sleep(1.5)
|
|
return
|
|
|
|
# Check bottom chests status
|
|
chests = self.scan_bottom_chests(frame)
|
|
logging.info(f"[CHESTS] Current slots: { {k: v['status'] for k, v in chests.items()} }")
|
|
|
|
# If any chest slot is ready (OPEN), click it to collect rewards
|
|
for slot_id, info in sorted(chests.items()):
|
|
if info["status"] == "OPEN":
|
|
logging.info(f"[CHESTS] Slot #{slot_id} is OPEN! Clicking to claim at {info['click_pos']}...")
|
|
self.click_client_pos(*info["click_pos"])
|
|
time.sleep(1.5)
|
|
return
|
|
|
|
logging.info("Searching for 'btn_fight' / 'btn_play'...")
|
|
fight_pos = self.find_template_in_rel_roi(frame, "btn_fight", self.REL_ROI_FIGHT) or \
|
|
self.find_template(frame, "btn_fight") or \
|
|
self.find_template(frame, "btn_play")
|
|
if fight_pos:
|
|
self.click_client_pos(fight_pos[0], fight_pos[1])
|
|
logging.info("Clicked fight button. Transitioning towards IN_GAME.")
|
|
if self.derank_mode or (self.current_wins >= self.max_wins_limit):
|
|
self.fast_surrender_pipeline()
|
|
else:
|
|
time.sleep(1.5)
|
|
|
|
def handle_in_game(self, frame: np.ndarray):
|
|
should_surrender = self.derank_mode or (self.current_wins >= self.max_wins_limit)
|
|
reason = "Forced Derank" if self.derank_mode else f"Win Streak Limit ({self.current_wins}/{self.max_wins_limit})"
|
|
|
|
if should_surrender:
|
|
logging.warning(f"Surrender condition met [{reason}]. Fast-tracking surrender...")
|
|
self.fast_surrender_pipeline(max_wait_sec=5.0)
|
|
return
|
|
else:
|
|
logging.info("Match in progress (Normal Play, Auto-battle active). Waiting for match results...")
|
|
|
|
def handle_victory(self, frame: np.ndarray):
|
|
if not self._result_recorded_for_match:
|
|
self._result_recorded_for_match = True
|
|
self.current_wins += 1
|
|
self.total_games += 1
|
|
logging.info(f"[WIN] Victory recorded! Streak: {self.current_wins}/{self.max_wins_limit}. Total games: {self.total_games}")
|
|
|
|
collect_pos = self.find_template_in_rel_roi(frame, "btn_collect_victory", self.REL_ROI_VICTORY_COLLECT) or \
|
|
self.find_template(frame, "btn_collect_victory")
|
|
if collect_pos:
|
|
self.click_client_pos(collect_pos[0], collect_pos[1])
|
|
else:
|
|
self.click_template(frame, "btn_continue")
|
|
time.sleep(1.0)
|
|
|
|
def handle_defeat(self, frame: np.ndarray):
|
|
if not self._result_recorded_for_match:
|
|
self._result_recorded_for_match = True
|
|
self.current_wins = 0
|
|
self.total_games += 1
|
|
logging.info(f"[LOSS] Streak reset to 0. Total games: {self.total_games}")
|
|
|
|
ok_pos = self.find_template_in_rel_roi(frame, "btn_ok", self.REL_ROI_DEFEAT_OK) or \
|
|
self.find_template(frame, "btn_ok")
|
|
if ok_pos:
|
|
self.click_client_pos(ok_pos[0], ok_pos[1])
|
|
else:
|
|
self.click_template(frame, "btn_continue")
|
|
time.sleep(1.0)
|
|
|
|
def handle_no_free_slots(self, frame: np.ndarray):
|
|
logging.info("State: NO_FREE_SLOTS_SCREEN ('You have no slot available for this chest').")
|
|
open_pos = self.find_template_in_rel_roi(frame, "btn_open_chest", self.REL_ROI_NO_SLOTS_OPEN) or \
|
|
self.find_template(frame, "btn_open_chest")
|
|
if open_pos:
|
|
logging.debug(f"[NO_FREE_SLOTS] 'Open chest' option detected at {open_pos}.")
|
|
|
|
remove_pos = self.find_template_in_rel_roi(frame, "btn_remove_chest", self.REL_ROI_NO_SLOTS_REMOVE) or \
|
|
self.find_template(frame, "btn_remove_chest")
|
|
if remove_pos:
|
|
self.click_client_pos(remove_pos[0], remove_pos[1])
|
|
logging.info("Clicked 'Remove chest' to discard unslotted chest. Returning to lobby...")
|
|
time.sleep(1.0)
|
|
|
|
def step(self):
|
|
"""Executes a single FSM iteration."""
|
|
frame = self.wm.capture_frame(client_only=True)
|
|
|
|
# Global check: if a promotional / purchase popup is visible, dismiss it first
|
|
if self.check_and_dismiss_offer_popup(frame):
|
|
return
|
|
|
|
detected_state = self.detect_state(frame)
|
|
|
|
if detected_state != self.state:
|
|
logging.info(f"State changed: {self.state.value} -> {detected_state.value}")
|
|
self.state = detected_state
|
|
|
|
if self.state == GameState.MAIN_MENU:
|
|
self.handle_main_menu(frame)
|
|
elif self.state == GameState.TOWER_LOBBY:
|
|
self.handle_lobby(frame)
|
|
elif self.state == GameState.CHEST_OPEN_SCREEN:
|
|
self.handle_chest_open_screen(frame)
|
|
elif self.state == GameState.IN_GAME:
|
|
self.handle_in_game(frame)
|
|
elif self.state == GameState.VICTORY_SCREEN:
|
|
self.handle_victory(frame)
|
|
elif self.state == GameState.DEFEAT_SCREEN:
|
|
self.handle_defeat(frame)
|
|
elif self.state == GameState.NO_FREE_SLOTS_SCREEN:
|
|
self.handle_no_free_slots(frame)
|
|
if self.state == GameState.UNKNOWN:
|
|
self._unknown_state_count += 1
|
|
if self._unknown_state_count % 15 == 0:
|
|
logging.warning(
|
|
f"[STALL WARNING] State has been UNKNOWN for {self._unknown_state_count * self.poll_interval:.1f}s. "
|
|
f"Attempting cursor unpark and recovery..."
|
|
)
|
|
park_x, park_y = self.wm.client_to_screen(50, 50)
|
|
from window_utils import user32
|
|
user32.SetCursorPos(park_x, park_y)
|
|
logging.debug("State: UNKNOWN (waiting or transition in progress)")
|
|
else:
|
|
self._unknown_state_count = 0
|
|
|
|
def start(self, max_duration_sec: Optional[int] = None, max_cycles: Optional[int] = None):
|
|
"""Starts main bot loop with optional duration and battle cycle limits."""
|
|
self.running = True
|
|
self.wm.focus()
|
|
limits_desc = []
|
|
if max_duration_sec:
|
|
limits_desc.append(f"Max time: {max_duration_sec}s")
|
|
if max_cycles:
|
|
limits_desc.append(f"Max cycles: {max_cycles}")
|
|
limit_str = f" ({', '.join(limits_desc)})" if limits_desc else ""
|
|
logging.info(f"DirtyLeague Bot loop started{limit_str}.")
|
|
|
|
start_time = time.time()
|
|
initial_games = self.total_games
|
|
|
|
while self.running:
|
|
if max_duration_sec and (time.time() - start_time) >= max_duration_sec:
|
|
logging.info(f"Execution limit reached: {max_duration_sec}s elapsed. Stopping gracefully...")
|
|
break
|
|
if max_cycles and (self.total_games - initial_games) >= max_cycles:
|
|
logging.info(f"Cycle limit reached: {max_cycles} battle(s) completed. Stopping gracefully...")
|
|
break
|
|
try:
|
|
self.step()
|
|
except Exception as e:
|
|
logging.error(f"Error during bot step: {e}")
|
|
time.sleep(self.poll_interval)
|
|
|
|
def stop(self):
|
|
"""Stops main bot loop."""
|
|
self.running = False
|
|
logging.info("DirtyLeague Bot loop stopped.")
|