import threading import time import cv2 from input_layer import keyboard from utils.auto_settings import check_settings from utils.log_rotation import safe_imwrite from bot import Bot from config import Config from death_manager import DeathManager from game_recovery import GameRecovery from game_stats import GameStats from health_manager import HealthManager from logger import Logger from messages import Messenger from screen import grab, get_offset_state, find_and_set_window_position from utils.restart import safe_exit from utils.misc import cooperative_shutdown, set_d2r_always_on_top, restore_d2r_window_visibility, wait, enforce_d2r_window class GameController: def __init__(self): self.is_running = False self.health_monitor_thread = None self.health_manager = None self.death_manager = None self.death_monitor_thread = None self.game_recovery = None self.game_stats = GameStats() self.game_controller_thread = None self.bot_thread = None self.bot = None def run_bot(self): # Start bot thread self.bot = Bot(self.game_stats) self.bot_thread = threading.Thread(target=self.bot.start) self.bot_thread.daemon = True self.bot_thread.start() # Register that thread to the death and health manager so they can stop the bot thread if needed self.death_manager.set_callback(lambda: self.bot.stop()) self.health_manager.set_callback(lambda: self.bot.stop()) do_restart = False messenger = Messenger() force_stopped = False while 1: did_chicken = self.health_manager.did_chicken() max_game_length_reached = self.game_stats.get_current_game_length() > Config().general["max_game_length_s"] max_consecutive_fails_reached = False if not Config().general["max_consecutive_fails"] else self.game_stats.get_consecutive_runs_failed() >= Config().general["max_consecutive_fails"] if max_game_length_reached or max_consecutive_fails_reached or self.death_manager.died() or did_chicken or (force_stopped := self.bot._stopping): # Some debug and logging if max_game_length_reached: Logger.info(f"Max game length reached. Attempting to restart {Config().general['name']}!") if not self.game_stats.get_failure_reason(): self.game_stats.set_failure_reason("Max game length reached (stuck)") if Config().general["info_screenshots"]: safe_imwrite("./log/screenshots/info/info_max_game_length_reached_" + time.strftime("%Y%m%d_%H%M%S") + ".png", grab()) elif self.death_manager.died(): self.game_stats.log_death(self.death_manager._last_death_screenshot) elif did_chicken: self.game_stats.log_chicken(self.health_manager._last_chicken_screenshot) self.bot._stash_mutex.acquire() #Grab mutex to ensure stashing is not occuring self.bot.stop() cooperative_shutdown( self.bot_thread, bot=self.bot, timeout=8.0, allow_force_kill=False, ) self.bot._stash_mutex.release() # clean up key presses that might be pressed in the bot_thread after killing keyboard.release(Config().char["stand_still"]) keyboard.release(Config().char["show_items"]) keyboard.release("shift") #stashing/inventory management keyboard.release("ctrl") #stashing/inventory management # Try to recover from whatever situation we are and go back to hero selection if max_consecutive_fails_reached: # Check whether active (non-disabled) runs still exist. # If so, reset the counter and keep going — the bot runs as long as possible. # Only quit if truly no routes remain. all_configured = {k for k, v in { "run_level": Config().routes.get("run_level"), "run_trav": Config().routes.get("run_trav"), "run_pindle": Config().routes.get("run_pindle"), "run_shenk": Config().routes.get("run_eldritch") or Config().routes.get("run_eldritch_shenk"), "run_nihlathak": Config().routes.get("run_nihlathak"), "run_arcane": Config().routes.get("run_arcane"), "run_diablo": Config().routes.get("run_diablo"), "run_vizier": Config().routes.get("run_vizier"), "run_baal": Config().routes.get("run_baal"), "run_mephisto": Config().routes.get("run_mephisto"), "run_andariel": Config().routes.get("run_andariel"), "run_countess": Config().routes.get("run_countess"), }.items() if v} active_runs = all_configured - self.game_stats.get_session_disabled_runs() if active_runs: Logger.warning( f"Consecutive fails {self.game_stats.get_consecutive_runs_failed()} reached limit " f"but active runs remain {active_runs} — resetting counter and continuing." ) if messenger.enabled: messenger.send_message( f"Consecutive fails limit reached but {active_runs} still active — resetting and continuing." ) self.game_stats.reset_consecutive_fails() do_restart = self.game_recovery.go_to_hero_selection() else: reason_str = self.game_stats.get_failure_reason() msg = f"Consecutive fails {self.game_stats.get_consecutive_runs_failed()} >= Max {Config().general['max_consecutive_fails']}" if reason_str: msg += f". Reason: {reason_str}" msg += ". No active runs remain. Quitting botty." Logger.error(msg) if messenger.enabled: messenger.send_message(msg) safe_exit(1) elif self.death_manager.died() and Config().general["hardcore"]: # Hardcore: character is gone, no point restarting msg = "Hardcore character died permanently. Stopping botty." Logger.error(msg) if messenger.enabled: messenger.send_message(msg) safe_exit(1) else: do_restart = self.game_recovery.go_to_hero_selection() break wait(0.5, 0.75) self.bot_thread.join() if do_restart: # Re-check the chicken flag BEFORE resetting it: it may have been set just after # the loop sampled it, in which case force_stopped alone would mislabel the failure. did_chicken = did_chicken or self.health_manager.did_chicken() # Reset flags before running a new bot self.death_manager.reset_death_flag() self.health_manager.reset_chicken_flag() if (max_game_length_reached or did_chicken or force_stopped) and not self.game_stats.get_failure_reason(): if did_chicken: self.game_stats.set_failure_reason("Health chicken triggered") elif force_stopped: self.game_stats.set_failure_reason("Bot stopped (F12 or crash)") else: self.game_stats.set_failure_reason("Max game length reached (stuck)") self.game_stats.log_end_game(failed = (max_game_length_reached or did_chicken or force_stopped)) return self.run_bot() else: if Config().general["info_screenshots"]: safe_imwrite("./log/screenshots/info/info_could_not_recover_" + time.strftime("%Y%m%d_%H%M%S") + ".png", grab()) if Config().general['restart_d2r_when_stuck']: Logger.error("Could not recover. Restarting the bot — game kept running.") if messenger.enabled: messenger.send_message("Got stuck, restarting bot (game kept running)") if max_game_length_reached and not self.game_stats.get_failure_reason(): self.game_stats.set_failure_reason("Max game length reached (stuck)") self.game_stats.log_end_game(failed=max_game_length_reached) import subprocess, sys, os as _os subprocess.Popen([sys.executable, _os.path.abspath(sys.argv[0])]) _os._exit(0) else: Logger.error("Could not recover from a max game length violation. Quitting botty.") if messenger.enabled: messenger.send_message("Got stuck and will now quit botty") safe_exit(1) def start(self): # Check if we user should update the d2r settings diff = check_settings() if len(diff) > 0: Logger.warning("Your D2R settings differ from the requiered ones. Please use Auto Settings to adjust them. The differences are:") Logger.warning(f"{diff}") set_d2r_always_on_top() enforce_d2r_window(5, 98) find_and_set_window_position(force=True) self.setup_screen() self.start_health_manager_thread() self.start_death_manager_thread() self.game_recovery = GameRecovery(self.death_manager) self.start_game_controller_thread() self.is_running = True def stop(self): restore_d2r_window_visibility() if self.death_monitor_thread: cooperative_shutdown( self.death_monitor_thread, death_manager=self.death_manager, ) if self.health_monitor_thread: cooperative_shutdown( self.health_monitor_thread, health_manager=self.health_manager, ) if self.bot_thread: cooperative_shutdown( self.bot_thread, bot=self.bot, health_manager=self.health_manager, death_manager=self.death_manager, timeout=8.0, allow_force_kill=False, ) if self.game_controller_thread: # game_controller_thread runs run_bot() which checks self.bot._stopping cooperative_shutdown( self.game_controller_thread, bot=self.bot, health_manager=self.health_manager, death_manager=self.death_manager, timeout=8.0, allow_force_kill=False, ) self.is_running = False def setup_screen(self): if get_offset_state(): return True return False def start_health_manager_thread(self): # Run health monitor thread self.health_manager = HealthManager() self.health_monitor_thread = threading.Thread(target=self.health_manager.start_monitor) self.health_monitor_thread.daemon = True self.health_monitor_thread.start() def start_death_manager_thread(self): # Run death monitor thread self.death_manager = DeathManager() self.death_monitor_thread = threading.Thread(target=self.death_manager.start_monitor) self.death_monitor_thread.daemon = True self.death_monitor_thread.start() def start_game_controller_thread(self): # Run game controller thread self.game_controller_thread = threading.Thread(target=self.run_bot) self.game_controller_thread.daemon = False self.game_controller_thread.start() def toggle_pause_bot(self): if self.bot: self.bot.toggle_pause()