589 lines
26 KiB
Python
589 lines
26 KiB
Python
"""
|
|
Core FSM Bot engine for DirtyLeague.
|
|
Manages screen recognition, game state transitions, and automated inputs.
|
|
"""
|
|
|
|
from enum import Enum
|
|
import logging
|
|
import os
|
|
import time
|
|
from typing import Dict, Optional, Tuple
|
|
import cv2
|
|
import numpy as np
|
|
import pydirectinput
|
|
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:
|
|
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.wm = WindowManager(
|
|
window_title=self.target_conf.get("title", "DirtyLeague"),
|
|
process_name=self.target_conf.get("process_name", "DirtyLeague.exe")
|
|
)
|
|
self.templates: Dict[str, np.ndarray] = {}
|
|
self._load_templates()
|
|
|
|
def _load_templates(self):
|
|
"""Preloads all 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.templates[name_without_ext] = img
|
|
logging.info(f"Loaded {len(self.templates)} template(s) from '{self.assets_dir}': {list(self.templates.keys())}")
|
|
|
|
def find_template(self, frame: np.ndarray, template_name: str, threshold: Optional[float] = None) -> Optional[Tuple[int, int]]:
|
|
"""
|
|
Searches for template_name in frame.
|
|
Returns client center coordinates (center_x, center_y) if matched above threshold, else None.
|
|
"""
|
|
if template_name not in self.templates:
|
|
# Try reloading in case new template was added
|
|
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.templates[template_name] = img
|
|
if template_name not in self.templates:
|
|
return None
|
|
|
|
tpl = self.templates[template_name]
|
|
th, tw = tpl.shape[:2]
|
|
fh, fw = frame.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_roi(
|
|
self,
|
|
frame: np.ndarray,
|
|
template_name: str,
|
|
roi: Tuple[int, int, int, int],
|
|
threshold: Optional[float] = None
|
|
) -> Optional[Tuple[int, int]]:
|
|
"""
|
|
Fast template search restricted to ROI: (x1, y1, x2, y2).
|
|
Returns client center coordinates (cx, cy) if matched, else None.
|
|
Runs in ~2-5ms instead of ~420ms for full-frame search.
|
|
"""
|
|
if template_name not in self.templates:
|
|
return None
|
|
tpl = self.templates[template_name]
|
|
th, tw = tpl.shape[:2]
|
|
fh, fw = frame.shape[:2]
|
|
|
|
x1, y1, x2, y2 = roi
|
|
x1, y1 = max(0, x1), max(0, y1)
|
|
x2, y2 = min(fw, x2), min(fh, y2)
|
|
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 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 the current game screen based on visible templates (accelerated with ROIs)."""
|
|
# 1. Ultra-fast ROI checks (~3-5ms each instead of ~420ms full-frame)
|
|
if self.find_template_in_roi(frame, "btn_remove_chest", (1350, 1680, 1950, 1850)) or \
|
|
self.find_template_in_roi(frame, "btn_open_chest", (1850, 1680, 2350, 1850)):
|
|
return GameState.NO_FREE_SLOTS_SCREEN
|
|
|
|
if self.find_template_in_roi(frame, "btn_ok", (900, 1480, 1450, 1700)) or \
|
|
self.find_template_in_roi(frame, "text_defeat", (500, 1400, 1200, 1700), threshold=0.65):
|
|
return GameState.DEFEAT_SCREEN
|
|
|
|
if self.find_template_in_roi(frame, "btn_exit", (1600, 1900, 1920, 2020)) or \
|
|
self.find_template_in_roi(frame, "btn_leave", (1900, 1150, 2400, 1400)):
|
|
return GameState.IN_GAME
|
|
|
|
if self.find_template_in_roi(frame, "btn_fight", (3000, 750, 3550, 1000)):
|
|
return GameState.TOWER_LOBBY
|
|
|
|
if self.find_template_in_roi(frame, "btn_turn_all_cards", (1550, 1350, 2250, 1550)) or \
|
|
self.find_template_in_roi(frame, "btn_collect", (1550, 1350, 2250, 1550), threshold=0.65):
|
|
return GameState.CHEST_OPEN_SCREEN
|
|
|
|
if self.find_template_in_roi(frame, "btn_collect_victory", (750, 1550, 1200, 1750)) or \
|
|
self.find_template_in_roi(frame, "banner_victory", (400, 300, 1600, 700), 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, "anchor_main_menu") or 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.
|
|
Grid calibrated: slot 1 starts at x=525, slot pitch = 213px.
|
|
Returns dict of slot_id -> {"status": "OPEN"|"TIMER"|"VACANT", "click_pos": (cx, cy)}
|
|
"""
|
|
results = {}
|
|
slot_pitch = 213
|
|
base_x = 525
|
|
y = 1850
|
|
w = 210
|
|
h = 70
|
|
|
|
for i in range(4):
|
|
slot_id = i + 1
|
|
sx = base_x + i * slot_pitch
|
|
status_crop = frame[y:y + h, sx:sx + w]
|
|
|
|
is_open = False
|
|
if "status_open" in self.templates:
|
|
res_o = cv2.matchTemplate(status_crop, self.templates["status_open"], cv2.TM_CCOEFF_NORMED)
|
|
_, max_o, _, _ = cv2.minMaxLoc(res_o)
|
|
if max_o >= 0.8:
|
|
is_open = True
|
|
|
|
is_vacant = False
|
|
if "status_vacant" in self.templates:
|
|
res_v = cv2.matchTemplate(status_crop, self.templates["status_vacant"], cv2.TM_CCOEFF_NORMED)
|
|
_, max_v, _, _ = cv2.minMaxLoc(res_v)
|
|
if max_v >= 0.8:
|
|
is_vacant = True
|
|
|
|
if is_open:
|
|
status = "OPEN"
|
|
elif is_vacant:
|
|
status = "VACANT"
|
|
else:
|
|
status = "TIMER"
|
|
|
|
results[slot_id] = {
|
|
"status": status,
|
|
"click_pos": (sx + w // 2, y - 70)
|
|
}
|
|
logging.debug(f"[CHESTS] Slot #{slot_id}: {status}")
|
|
|
|
return results
|
|
|
|
def check_and_dismiss_offer_popup(self, frame: np.ndarray) -> bool:
|
|
"""
|
|
Detects and dismisses promotional or purchase offer popups ('X' close button)
|
|
that may appear randomly or after claiming chest rewards/returning to lobby.
|
|
Returns True if a popup was detected and dismissed, False otherwise.
|
|
"""
|
|
# Search for close button in top-right quadrant (ROI: y=0..300, x=3400..3840)
|
|
fh, fw = frame.shape[:2]
|
|
roi_y1, roi_y2 = 0, min(300, fh)
|
|
roi_x1, roi_x2 = max(0, fw - 500), fw
|
|
roi = frame[roi_y1:roi_y2, roi_x1:roi_x2]
|
|
|
|
for tpl_name in ["btn_close", "btn_close_offer"]:
|
|
if tpl_name not in self.templates:
|
|
continue
|
|
tpl = self.templates[tpl_name]
|
|
th, tw = tpl.shape[:2]
|
|
if th > roi.shape[0] or tw > roi.shape[1]:
|
|
continue
|
|
res = cv2.matchTemplate(roi, tpl, cv2.TM_CCOEFF_NORMED)
|
|
_, max_val, _, max_loc = cv2.minMaxLoc(res)
|
|
if max_val >= 0.65:
|
|
cx = roi_x1 + max_loc[0] + tw // 2
|
|
cy = roi_y1 + max_loc[1] + th // 2
|
|
logging.info(f"[OFFER POPUP] Detected close button '{tpl_name}' (conf: {max_val:.2f}) at client ({cx}, {cy}). Dismissing...")
|
|
self.click_client_pos(cx, cy)
|
|
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
|
|
if self.click_template(frame, "btn_turn_all_cards"):
|
|
logging.info("Clicked 'Turn all cards over'. Waiting for card flip animation...")
|
|
time.sleep(2.0)
|
|
return
|
|
|
|
# Step 2: Check if "Collect" is visible
|
|
if self.click_template(frame, "btn_collect"):
|
|
logging.info("Clicked 'Collect'. Rewards claimed, returning to lobby...")
|
|
time.sleep(1.5)
|
|
# Step 3: 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 handle_lobby(self, frame: np.ndarray):
|
|
logging.info("State: TOWER_LOBBY.")
|
|
# 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 at (875, 1260)...")
|
|
self.click_client_pos(875, 1260)
|
|
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
|
|
|
|
def fast_surrender_pipeline(self, max_wait_sec: float = 15.0) -> bool:
|
|
"""
|
|
Ultra-fast surrender pipeline for FORCED DERANK.
|
|
Polls the Exit button ROI (~20ms per check). The millisecond battle finishes loading,
|
|
clicks Exit via fast click, then immediately polls Leave dialog 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()
|
|
exit_roi = (1600, 1900, 1920, 2020)
|
|
leave_roi = (1900, 1150, 2400, 1400)
|
|
|
|
# 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_roi(frame, "btn_exit", exit_roi, 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_roi(frame, "btn_leave", leave_roi, 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_roi(frame, "btn_leave", leave_roi, 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
|
|
# 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 at (875, 1260)...")
|
|
self.click_client_pos(875, 1260)
|
|
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_roi(frame, "btn_fight", (3000, 750, 3550, 1000)) 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}")
|
|
if not self.click_template(frame, "btn_collect_victory"):
|
|
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}")
|
|
if not self.click_template(frame, "btn_ok"):
|
|
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(frame, "btn_open_chest")
|
|
if open_pos:
|
|
logging.debug(f"[NO_FREE_SLOTS] 'Open chest' option detected at {open_pos}.")
|
|
|
|
if self.click_template(frame, "btn_remove_chest"):
|
|
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)
|
|
else:
|
|
logging.debug("State: UNKNOWN (waiting or transition in progress)")
|
|
|
|
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.")
|