from transitions import Machine import time from input_layer import keyboard import time import os import sys import random import cv2 import math import numpy as np from copy import copy from collections import OrderedDict from health_manager import set_pause_state from transmute import Transmute from utils.misc import wait, hms from utils.log_rotation import safe_imwrite from utils.restart import safe_exit from game_stats import GameStats from logger import Logger from config import Config from screen import grab, convert_monitor_to_screen, convert_screen_to_abs, convert_abs_to_monitor, convert_screen_to_monitor import template_finder from char import IChar from item.pickit import PickIt from item import consumables from pather import Pather, Location from char.sorceress import LightSorc, BlizzSorc, BlizzorbSorc, NovaSorc, HydraSorc from char.trapsin import Trapsin from char.paladin.hammerdin import Hammerdin from char.paladin import FoHdin from char.warlock import FireLock, AbyssLock, EchoLock from char.amazon.javazon import Javazon from char.barbarian import Barbarian from char.necro import Necro from char.poison_necro import Poison_Necro from char.bone_necro import Bone_Necro from char.basic import Basic from char.basic_ranged import Basic_Ranged from ui_manager import wait_until_hidden, wait_until_visible, ScreenObjects, is_visible, detect_screen_object from ui import meters, skills, view, character_select, main_menu from inventory import personal, vendor, belt, common from run import Pindle, ShenkEld, Trav, Nihlathak, Arcane, Diablo, Vizier, Level from run.baal import Baal from run.mephisto import Mephisto from run.andariel import Andariel from run.countess import Countess from town import TownManager, A1, A2, A3, A4, A5, town_manager from messages import Messenger from threading import Lock from utils.stealth import maybe_afk_break, should_skip_run class Bot: def __init__(self, game_stats: GameStats): self._game_stats = game_stats self._messenger = Messenger() self._pather = Pather() self._pickit = PickIt() self._stash_mutex = Lock() # Create Character match Config().char["type"]: case "sorceress" | "light_sorc": self._char: IChar = LightSorc(Config().light_sorc, self._pather) case "blizz_sorc": self._char: IChar = BlizzSorc(Config().blizz_sorc, self._pather) case "blizzorb_sorc": self._char: IChar = BlizzorbSorc(Config().blizzorb_sorc, self._pather) case "nova_sorc": self._char: IChar = NovaSorc(Config().nova_sorc, self._pather) case "hydra_sorc": self._char: IChar = HydraSorc(Config().hydra_sorc, self._pather) case "hammerdin" | "paladin": self._char: IChar = Hammerdin(Config().hammerdin, self._pather, self._pickit) #pickit added for diablo case "fohdin": self._char: IChar = FoHdin(Config().fohdin, self._pather, self._pickit) case "abyss_lock" | "warlock": self._char: IChar = AbyssLock(Config().abyss_lock, self._pather) case "fire_lock": self._char: IChar = FireLock(Config().fire_lock, self._pather) case "echo_lock": self._char: IChar = EchoLock(Config().echo_lock, self._pather) case "amazon" | "javazon": self._char: IChar = Javazon(Config().javazon, self._pather) case "trapsin": self._char: IChar = Trapsin(Config().trapsin, self._pather) case "barbarian": self._char: IChar = Barbarian(Config().barbarian, self._pather) case "poison_necro": self._char: IChar = Poison_Necro(Config().poison_necro, self._pather) case "bone_necro": self._char: IChar = Bone_Necro(Config().bone_necro, self._pather) case "necro": self._char: IChar = Necro(Config().necro, self._pather) case "basic": self._char: IChar = Basic(Config().basic, self._pather) case "basic_ranged": self._char: IChar = Basic_Ranged(Config().basic_ranged, self._pather) case _: Logger.error(f'{Config().char["type"]} is not supported! Closing down bot.') os._exit(1) # Create Town Manager a5 = A5(self._pather, self._char) a4 = A4(self._pather, self._char) a3 = A3(self._pather, self._char) a2 = A2(self._pather, self._char) a1 = A1(self._pather, self._char) self._town_manager = TownManager(a1, a2, a3, a4, a5) # Create runs self._do_runs = { "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"), } # Adapt order to the config self._do_runs = OrderedDict((k, self._do_runs[k]) for k in Config().routes_order if k in self._do_runs and self._do_runs[k]) self._do_runs_reset = copy(self._do_runs) Logger.info(f"Doing runs: {self._do_runs_reset.keys()}") if Config().general["randomize_runs"]: self.shuffle_runs() self._level = Level(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._pindle = Pindle(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._shenk = ShenkEld(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._trav = Trav(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._nihlathak = Nihlathak(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._arcane = Arcane(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._diablo = Diablo(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._vizier = Vizier(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._baal = Baal(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._mephisto = Mephisto(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._andariel = Andariel(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) self._countess = Countess(self._pather, self._town_manager, self._char, self._pickit, self._do_runs) # Create member variables self._picked_up_items = False self._curr_loc: bool | Location = None self._use_id_tome = True self._use_keys = True self._pre_buffed = False self._stopping = False self._pausing = False self._current_threads = [] self._ran_no_pickup = False self._previous_run_failed = False self._timer = time.time() # Per-run recovery: track consecutive failures per run so a single broken # run (bad path, missing WP, etc.) gets disabled instead of killing the bot. self._run_failure_counts: dict[str, int] = {} self._disabled_runs: set[str] = set() self._max_run_failures = Config().general.get("disable_run_after_failures", 5) # Create State Machine self._states=['initialization','hero_selection', 'town', 'level', 'pindle', 'shenk', 'trav', 'nihlathak', 'arcane', 'diablo', 'vizier', 'baal', 'mephisto', 'andariel', 'countess'] self._transitions = [ { 'trigger': 'init', 'source': 'initialization', 'dest': '=','before': "on_init"}, { 'trigger': 'skip_to_level', 'source': 'initialization', 'dest': 'level', 'before': "on_skip_to_level"}, { 'trigger': 'select_character', 'source': 'initialization', 'dest': 'hero_selection', 'before': "on_select_character"}, { 'trigger': 'start_from_town', 'source': ['initialization', 'hero_selection'], 'dest': 'town', 'before': "on_start_from_town"}, { 'trigger': 'create_game', 'source': 'hero_selection', 'dest': '=', 'before': "on_create_game"}, # Tasks within town { 'trigger': 'maintenance', 'source': 'town', 'dest': 'town', 'before': "on_maintenance"}, # Different runs { 'trigger': 'run_level', 'source': 'town', 'dest': 'level', 'before': "on_run_level"}, { 'trigger': 'run_pindle', 'source': 'town', 'dest': 'pindle', 'before': "on_run_pindle"}, { 'trigger': 'run_shenk', 'source': 'town', 'dest': 'shenk', 'before': "on_run_shenk"}, { 'trigger': 'run_trav', 'source': 'town', 'dest': 'trav', 'before': "on_run_trav"}, { 'trigger': 'run_nihlathak', 'source': 'town', 'dest': 'nihlathak', 'before': "on_run_nihlathak"}, { 'trigger': 'run_arcane', 'source': 'town', 'dest': 'arcane', 'before': "on_run_arcane"}, { 'trigger': 'run_diablo', 'source': 'town', 'dest': 'diablo', 'before': "on_run_diablo"}, { 'trigger': 'run_vizier', 'source': 'town', 'dest': 'vizier', 'before': "on_run_vizier"}, { 'trigger': 'run_baal', 'source': 'town', 'dest': 'baal', 'before': "on_run_baal" }, { 'trigger': 'run_mephisto', 'source': 'town', 'dest': 'mephisto', 'before': "on_run_mephisto" }, { 'trigger': 'run_andariel', 'source': 'town', 'dest': 'andariel', 'before': "on_run_andariel" }, { 'trigger': 'run_countess', 'source': 'town', 'dest': 'countess', 'before': "on_run_countess" }, # End run / game { 'trigger': 'end_run', 'source': ['level', 'shenk', 'pindle', 'nihlathak', 'trav', 'arcane', 'diablo','vizier', 'baal', 'mephisto', 'andariel', 'countess'], 'dest': 'town', 'before': "on_end_run"}, { 'trigger': 'end_game', 'source': ['town', 'shenk', 'pindle', 'nihlathak', 'trav', 'arcane', 'diablo','vizier','end_run', 'baal', 'mephisto', 'andariel', 'countess'], 'dest': 'initialization', 'before': "on_end_game"}, ] self.machine = Machine(model=self, states=self._states, initial="initialization", transitions=self._transitions, queued=True) self._transmute = Transmute(self._game_stats) def draw_graph(self): # Draw the whole graph, graphviz binaries must be installed and added to path for this! from transitions.extensions import GraphMachine self.machine = GraphMachine(model=self, states=self._states, initial="initialization", transitions=self._transitions, queued=True) self.machine.get_graph().draw('my_state_diagram.png', prog='dot') def get_curr_location(self): return self._curr_loc def start(self): from utils.misc import register_stop_condition, unregister_stop_condition import threading register_stop_condition(threading.get_ident(), lambda: self._stopping) try: self.trigger_or_stop('init') finally: unregister_stop_condition(threading.get_ident()) def stop(self): self._stopping = True def toggle_pause(self): self._pausing = not self._pausing if self._pausing: Logger.info(f"Pause at next state change...") else: Logger.info(f"Resume") self._game_stats.resume_timer() def trigger_or_stop(self, name: str, **kwargs): if self._pausing: Logger.info(f"{Config().general['name']} is now pausing") self._game_stats.pause_timer() while self._pausing: from utils.misc import wait as _wait _wait(0.2, 0.24) if not self._stopping: self.trigger(name, **kwargs) def restart_or_exit(self, message: str =""): if message: Logger.error(message) if not self._game_stats.get_failure_reason(): self._game_stats.set_failure_reason(message) if Config().general["restart_d2r_when_stuck"]: Logger.info("Restarting bot — game kept running") import subprocess subprocess.Popen([sys.executable, os.path.abspath(sys.argv[0])]) os._exit(0) else: Logger.info("Shut down botty") safe_exit() def current_game_length(self): return self._game_stats.get_current_game_length() def shuffle_runs(self): tmp = list(self._do_runs.items()) random.shuffle(tmp) #We clear() and update() to avoid assignment on _do_runs as this list is referenced #by other classes and we want updates persisted everywhere. self._do_runs.clear() self._do_runs.update(OrderedDict(tmp)) def is_last_run(self): found_unfinished_run = False for key in self._do_runs: if self._do_runs[key]: found_unfinished_run = True break return not found_unfinished_run def _rebuild_as_asset_to_trigger(trigger_to_assets: dict): result = {} for key in trigger_to_assets.keys(): for asset in trigger_to_assets[key]: result[asset] = key return result def on_init(self): self._game_stats.log_start_game() keyboard.release(Config().char["stand_still"]) # Force D2R client area to stable position to prevent offset drift from utils.misc import move_d2r_window move_d2r_window(5, 98) wait(0.3) # If we're only doing run_level and character is already in-game, skip # the town detection and go straight to the run if list(self._do_runs.keys()) == ["run_level"]: Logger.info("Level run only - skipping town detection, starting directly") self._curr_loc = Location.A1_TOWN_START self.trigger_or_stop("skip_to_level") return transition_to_screens = Bot._rebuild_as_asset_to_trigger({ "select_character": main_menu.MAIN_MENU_MARKERS, "start_from_town": town_manager.TOWN_MARKERS, }) if (match := template_finder.search_and_wait(list(transition_to_screens.keys()), best_match=True)).valid: self.trigger_or_stop(transition_to_screens[match.name]) else: self.restart_or_exit(f"Failed to detect {list(transition_to_screens.keys())}.") def on_select_character(self): # Make sure the correct char is selected if not character_select.has_char_template_saved(): character_select.save_char_online_status() character_select.save_char_template() else: if not character_select.select_char(): if Config().general["info_screenshots"]: timestamp = time.strftime("%Y%m%d_%H%M%S") safe_imwrite("./log/screenshots/info/info_failed_character_select_" + timestamp + ".png", grab()) if character_select.has_char_template_saved(): saved_char_img = character_select.get_saved_char_template() safe_imwrite("./log/screenshots/info/info_failed_character_select_saved_template_" + timestamp + ".png", saved_char_img) self._game_stats.set_failure_reason("Character selection failed at hero screen") self.restart_or_exit() self.trigger_or_stop("create_game") def on_create_game(self): # Start a game from hero selection if (m := wait_until_visible(ScreenObjects.MainMenu)).valid: if "DARK" in m.name: keyboard.send("esc") main_menu.start_game() view.move_to_corpse() else: self._game_stats.set_failure_reason("Could not detect main menu to start game") self.restart_or_exit() self.trigger_or_stop("start_from_town") def on_start_from_town(self): self._curr_loc = self._town_manager.wait_for_town_spawn() if not self._curr_loc: Logger.warning("Could not detect town spawn — defaulting to A1_TOWN_START. Run will waypoint to correct act.") self._curr_loc = Location.A1_TOWN_START # Handle picking up corpse in case of death if (corpse_present := is_visible(ScreenObjects.Corpse)): self._previous_run_failed = True view.pickup_corpse() wait_until_hidden(ScreenObjects.Corpse) belt.fill_up_belt_from_inventory(Config().char["num_loot_columns"]) self._char.discover_capabilities() if corpse_present and self._char.capabilities.can_teleport_with_charges and not self._char.select_tp(): keybind = Config().char["teleport"] Logger.info(f"Teleport keybind is lost upon death. Rebinding teleport to '{keybind}'") self._char.remap_right_skill_hotkey("TELE_ACTIVE", Config().char["teleport"]) # Run /nopickup command to avoid picking up stuff on accident if Config().char["enable_no_pickup"] and (not self._ran_no_pickup and not self._game_stats._nopickup_active): self._ran_no_pickup = True if view.enable_no_pickup(): self._game_stats._nopickup_active = True Logger.info("Activated /nopickup") else: Logger.error("Failed to detect if /nopickup command was applied or not") self._game_stats.log_exp() self.trigger_or_stop("maintenance") def on_maintenance(self): # Defensive: ensure _curr_loc is valid before any town manager call if not self._curr_loc: Logger.warning("No current location set — defaulting to A1_TOWN_START") self._curr_loc = Location.A1_TOWN_START # Pause health manager if not already paused set_pause_state(True) # Dismiss skill/quest/help/stats icon if they are on screen if not view.dismiss_skills_icon(): view.return_to_play() # Top up health with belt potions before doing any town work. # The health manager is paused in town so it won't auto-pot for us. # We aim for ~95% HP so the character enters the next run at full health. # Drinks health first (cheaper), then rejuv as fallback if health is unavailable. # Happens before update_pot_needs so the reduced belt is accurately counted # and the buy step will restock whatever was drunk. _town_heal_threshold = 0.95 _town_heal_max_drinks = 5 for _drink_attempt in range(_town_heal_max_drinks): _hp_img = grab() _hp = meters.get_health(_hp_img) if _hp >= _town_heal_threshold: break Logger.info(f"Town heal: HP at {_hp * 100:.0f}% — drinking potion to recover (attempt {_drink_attempt + 1}/{_town_heal_max_drinks})") _mp = meters.get_mana(_hp_img) _drank = belt.drink_potion("health", stats=[_hp, _mp]) if not _drank: _drank = belt.drink_potion("rejuv", stats=[_hp, _mp]) if not _drank: Logger.debug("Town heal: no health or rejuv potions in belt — cannot recover HP in town") break wait(2.0, 2.5) # wait for potion to tick (town regen is slow) # Look at belt to figure out how many pots need to be picked up belt.update_pot_needs() # If we are doing trav_runs and juvs are full we will remove extra runs if ("run_trav" == next(iter(self._do_runs))) and (consumables.get_needs("rejuv") <= 1): while (len(self._do_runs) > 1): skipped_run = self._do_runs.popitem()[0] Logger.info(f"We are running trav and have full rejuvs so skipping {skipped_run}") # Inspect inventory items = None need_inspect = self._picked_up_items or self._previous_run_failed if Config().char["runs_per_stash"]: need_inspect |= (self._game_stats._run_counter - 1) % Config().char["runs_per_stash"] == 0 if need_inspect: img = personal.open_inventory() # Update TP, ID, key needs if self._game_stats._game_counter == 1: self._use_id_tome = common.tome_state(img, 'id')[0] is not None self._use_keys = is_visible(ScreenObjects.Key, img) if (self._game_stats._run_counter - 1) % 4 == 0 or self._previous_run_failed: personal.update_tome_key_needs(img, item_type = 'tp') if self._use_id_tome: personal.update_tome_key_needs(img, item_type = 'id') if self._use_keys: # if keys run out then refilling will be unreliable :( self._use_keys = personal.update_tome_key_needs(img, item_type = 'key') # Check inventory items if personal.inventory_has_items(img): Logger.debug("Inspecting inventory items") items = personal.inspect_items(img, game_stats=self._game_stats, close_window=False) common.close() Logger.debug(f"Needs: {consumables.get_needs()}") #Cast town buffs (ie burst of speed etc) if not self._pre_buffed: self._char.cast_town_buffs(self._curr_loc) if items: # if there are still items that need identifying, go to cain to identify them if any([item.need_id for item in items]): Logger.info("ID items at cain") self._curr_loc = self._town_manager.identify(self._curr_loc) if not self._curr_loc: Logger.warning("Could not identify items (Cain not available). Continuing without ID.") self._curr_loc = Location.A1_TOWN_START else: # recheck inventory items = personal.inspect_items(game_stats=self._game_stats) keep_items = any([item.keep for item in items]) if items else None sell_items = any([item.sell for item in items]) if items else None stash_gold = personal.get_inventory_gold_full() # Check if should need some healing img = grab() need_refill = ( consumables.should_buy("health", min_needed = 3) or consumables.should_buy("mana", min_needed = 3) or (self._use_keys and consumables.should_buy("key", min_remaining = 4)) or consumables.should_buy("tp", min_remaining = 8) or consumables.should_buy("id", min_remaining = 3) ) if need_refill or sell_items: Logger.info("Buy consumables and/or sell items") self._curr_loc, result_items = self._town_manager.buy_consumables(self._curr_loc, items = items) if self._curr_loc: items = result_items sell_items = any([item.sell for item in items]) if items else None Logger.debug(f"Needs: {consumables.get_needs()}") else: Logger.warning("Buy consumables failed, retrying") wait(0.5, 0.6) self._curr_loc, result_items = self._town_manager.buy_consumables(Location.A1_TOWN_START, items = items) if self._curr_loc: items = result_items sell_items = any([item.sell for item in items]) if items else None Logger.debug(f"Needs: {consumables.get_needs()}") else: Logger.warning("Buy consumables failed after retry, continuing without buying") self._curr_loc = Location.A1_TOWN_START elif meters.get_health(img) <= Config().char["take_rejuv_potion_health"] or meters.get_mana(img) <= Config().char["take_rejuv_potion_mana"]: Logger.info("Healing at next possible Vendor") self._curr_loc = self._town_manager.heal(self._curr_loc) if not self._curr_loc: Logger.warning("Heal failed, retrying") wait(0.5, 0.6) self._curr_loc = self._town_manager.heal(Location.A1_TOWN_START) if not self._curr_loc: Logger.warning("Heal failed after retry, continuing without heal") self._curr_loc = Location.A1_TOWN_START # After buying (or failing to buy), pull any pots already sitting in inventory into the belt. # This covers the "out of gold" case where we couldn't buy new pots but have leftover ones. if consumables.get_needs("health") > 0 or consumables.get_needs("mana") > 0 or consumables.get_needs("rejuv") > 0: Logger.debug("Filling belt from inventory pots before stash") belt.fill_up_belt_from_inventory(Config().char["num_loot_columns"]) belt.update_pot_needs() # Stash stuff if keep_items or stash_gold: Logger.info("Stashing items") prev_loc = self._curr_loc self._curr_loc, result_items = self._town_manager.stash(self._curr_loc, items=items, game_stats=self._game_stats) if not self._curr_loc: Logger.warning("Stash failed, retrying from A5 town start") wait(0.5, 0.6) self._curr_loc, result_items = self._town_manager.stash(Location.A5_TOWN_START, items=items, game_stats=self._game_stats) if not self._curr_loc: Logger.warning("Stash failed after retry, skipping stash and keeping items") self._curr_loc = prev_loc return sell_items = any([item.sell for item in result_items]) if result_items else None #Acquire mutex to prevent controller from killing thread during transmutes self._stash_mutex.acquire() Logger.info("Running transmutes") self._transmute.run_transmutes(force=False) common.close() self._stash_mutex.release() self._picked_up_items = False # Check if we are out of tps or need repairing need_repair = is_visible(ScreenObjects.NeedRepair) need_routine_repair = False if not Config().char["runs_per_repair"] else self._game_stats._run_counter % Config().char["runs_per_repair"] == 0 need_refill_teleport = self._char.capabilities.can_teleport_with_charges and (not self._char.select_tp() or self._char.is_low_on_teleport_charges()) if need_repair or need_routine_repair or need_refill_teleport or sell_items: if need_repair: Logger.info("Repair needed. Gear is about to break") elif need_routine_repair: Logger.info(f"Routine repair. Run count={self._game_stats._run_counter}, runs_per_repair={Config().char['runs_per_repair']}") elif need_refill_teleport: Logger.info("Teleport charges ran out. Need to repair") elif sell_items: Logger.info("Selling items at repair vendor") self._curr_loc, result_items = self._town_manager.repair(self._curr_loc, items) if not self._curr_loc: Logger.warning("Repair failed, retrying from A4 town start") wait(0.5, 0.6) self._curr_loc, result_items = self._town_manager.repair(Location.A4_TOWN_START, items) if self._curr_loc: items = result_items if not self._curr_loc: # Keep maintenance best-effort for stability: avoid killing runs on flaky NPC/vendor detection. if need_refill_teleport and self._char.capabilities.can_teleport_with_charges and not self._char.capabilities.can_teleport_natively: Logger.warning("Repair failed and teleport charges are required, trying to buy TP from vendor") consumables.set_needs("tp", 20) self._curr_loc = Location.A5_TOWN_START else: Logger.warning("Repair/vendor interaction failed; skipping maintenance and continuing run.") self._curr_loc = Location.A5_TOWN_START # Check if merc needs to be revived if Config().char["use_merc"]: merc_visible = False try: merc_visible = is_visible(ScreenObjects.MercIcon) except Exception: Logger.debug("Merc icon detection error, skipping resurrect check") merc_visible = True # Confirm with 'O' key - opens merc panel if alive, does nothing if dead if not merc_visible and not self._game_stats._merc_resurrect_failed: Logger.debug("Merc icon not visible, confirming with merc panel check") keyboard.send("o") wait(0.3, 0.4) merc_panel_open = is_visible(ScreenObjects.MercPanelText) if merc_panel_open: # Merc is alive - panel opened, close it keyboard.send("o") wait(0.2, 0.3) merc_visible = True if not merc_visible and not self._game_stats._merc_resurrect_failed: Logger.info("Resurrect merc") new_loc = self._town_manager.resurrect(self._curr_loc) if new_loc is False: # Retry once - sometimes resurrect fails due to timing Logger.warning("Resurrect failed, retrying") wait(0.5, 0.6) new_loc = self._town_manager.resurrect(self._curr_loc) if new_loc is False: # Failed to resurrect (can't afford or other error) - don't log death, just continue Logger.warning("Failed to resurrect merc after retry, continuing without merc") self._game_stats._merc_resurrect_failed = True elif new_loc is None: Logger.warning("Resurrect returned None, continuing at current location") pass else: self._game_stats.log_merc_death() self._curr_loc = new_loc # Gamble if needed while vendor.get_gamble_status() and Config().char["gamble_items"]: Logger.debug("Head to gamble") self._curr_loc = self._town_manager.gamble(self._curr_loc) if not self._curr_loc: Logger.warning("Gamble failed, skipping gamble") self._curr_loc = Location.A1_TOWN_START break items = vendor.gamble() if items: self._curr_loc, _ = self._town_manager.stash(self._curr_loc, items = items) common.close() if not self._curr_loc: Logger.warning("Gamble stash failed") self._curr_loc = Location.A1_TOWN_START break # Start a new run started_run = False self._previous_run_failed = False for key in self._do_runs: if self._do_runs[key]: if should_skip_run(): self._do_runs[key] = False continue self.trigger_or_stop(key) started_run = True break if not started_run: self.trigger_or_stop("end_game") def on_end_game(self, failed: bool = False): if failed and not self._game_stats.get_failure_reason(): self._game_stats.set_failure_reason("Game ended without completing runs") if Config().general["info_screenshots"] and failed: safe_imwrite("./log/screenshots/info/info_failed_game_" + time.strftime("%Y%m%d_%H%M%S") + ".png", grab()) self._curr_loc = False self._pre_buffed = False if view.save_and_exit() == False: Logger.error("Normal save_and_exit failed! Attempting fast_save_and_exit and resetting bot.") img = grab() if not self._game_stats.get_failure_reason(): self._game_stats.set_failure_reason("save_and_exit failed") view.fast_save_and_exit() self.stop() if Config().general["info_screenshots"]: self._last_chicken_screenshot = "./log/screenshots/info/info_failed_exit_" + time.strftime("%Y%m%d_%H%M%S") + ".png" safe_imwrite(self._last_chicken_screenshot, img) set_pause_state(True) self._game_stats.log_end_game(failed=failed) if Config().general["max_runtime_before_break_m"] and Config().general["break_length_m"]: elapsed_time = time.time() - self._timer Logger.debug(f'Session length = {math.ceil(elapsed_time/60)} minutes, max_runtime_before_break_m {Config().general["max_runtime_before_break_m"]}.') if elapsed_time > (Config().general["max_runtime_before_break_m"]*60): break_msg = f'Ran for {hms(elapsed_time)}, taking a break for {hms(Config().general["break_length_m"]*60)}.' Logger.info(break_msg) if self._messenger.enabled: self._messenger.send_message(break_msg) if not self._pausing: self.toggle_pause() wait(Config().general["break_length_m"]*60) break_msg = f'Break over, will now run for {hms(Config().general["max_runtime_before_break_m"]*60)}.' Logger.info(break_msg) if self._messenger.enabled: self._messenger.send_message(break_msg) if self._pausing: self.toggle_pause() self._timer = time.time() #We clear() and update() to avoid assignment on _do_runs as this list is referenced #by other classes and we want updates persisted everywhere. self._do_runs.clear() self._do_runs.update(self._do_runs_reset) if Config().general["randomize_runs"]: self.shuffle_runs() if Config().stealth.get("reshuffle_each_rotation"): tmp = list(self._do_runs_reset.items()) random.shuffle(tmp) self._do_runs_reset = OrderedDict(tmp) self._do_runs.clear() self._do_runs.update(self._do_runs_reset) self.trigger_or_stop("init") def on_end_run(self): if not Config().char["pre_buff_every_run"]: self._pre_buffed = True # Try TP to town, with retries success = self._char.tp_town() if not success: if not skills.has_tps(): consumables.set_needs("tp", 20) Logger.warning("No TP charges left — ending game to restock on next game start") set_pause_state(True) self.trigger_or_stop("end_game") return # Retry TP once Logger.warning("TP to town failed, retrying") wait(0.5, 0.6) success = self._char.tp_town() if success: tp_loc = self._curr_loc if self._curr_loc else Location.A5_TOWN_START self._curr_loc = self._town_manager.wait_for_tp(tp_loc) if self._curr_loc: set_pause_state(True) # Stealth: chance of unscheduled AFK break after returning to town maybe_afk_break() return self.trigger_or_stop("maintenance") # wait_for_tp failed - try to recover Logger.warning("wait_for_tp failed, attempting recovery") set_pause_state(True) self._curr_loc = Location.A5_TOWN_START return self.trigger_or_stop("maintenance") # TP failed twice Logger.warning("TP to town failed twice, trying to walk back") set_pause_state(True) self._curr_loc = Location.A5_TOWN_START self.trigger_or_stop("maintenance") # All the runs go here # ================================== def _save_error_screenshot(self, run_name: str, reason: str): """Save a timestamped screenshot when a run fails, so logs and visuals can be cross-referenced later. Saved to ./log/screenshots/error/ with the run name, sanitized reason, game/run counters and timestamp baked into the filename. If Discord logging of errors is enabled, also sends the message + screenshot.""" saved_path = None try: enabled = Config().general.get("error_screenshots", Config().general.get("info_screenshots", 1)) if enabled: import re as _re safe_reason = _re.sub(r"[^A-Za-z0-9]+", "_", (reason or "unknown")).strip("_")[:60] timestamp = time.strftime("%Y%m%d_%H%M%S") path = ( f"./log/screenshots/error/error_{run_name}_{safe_reason}" f"_g{self._game_stats._game_counter}_r{self._game_stats._run_counter}_{timestamp}.png" ) if safe_imwrite(path, grab()): saved_path = path Logger.info(f"Saved error screenshot for {run_name} ({reason}): {path}") except Exception as e: Logger.warning(f"Failed to save error screenshot for {run_name}: {e}") # Send the error (and screenshot, if Discord) to the configured messenger. try: if self._messenger.enabled and Config().general.get("discord_log_errors", 1): self._messenger.send_error(run_name, reason, saved_path) except Exception as e: Logger.warning(f"Failed to send error notification for {run_name}: {e}") def _record_run_result(self, run_name: str, failed: bool): """Track consecutive failures per run. After too many consecutive failures a run is disabled for the rest of the session so the bot keeps doing the other runs instead of stopping. A success resets that run's counter.""" if not failed: if self._run_failure_counts.get(run_name, 0) > 0: Logger.info(f"{run_name} succeeded — resetting its failure counter") self._run_failure_counts[run_name] = 0 return self._run_failure_counts[run_name] = self._run_failure_counts.get(run_name, 0) + 1 count = self._run_failure_counts[run_name] Logger.warning(f"{run_name} failed {count}/{self._max_run_failures} consecutive times") if count >= self._max_run_failures and run_name not in self._disabled_runs: self._disabled_runs.add(run_name) # Disable for the current cycle and for all future games this session. if run_name in self._do_runs: self._do_runs[run_name] = False if run_name in self._do_runs_reset: self._do_runs_reset[run_name] = False msg = ( f"DISABLING {run_name} for this session: it failed {count} consecutive times. " f"Continuing with the remaining runs. Check the logs/screenshots in " f"log/screenshots/error/ and restart the bot once fixed to re-enable it." ) Logger.error(msg) if self._messenger.enabled: self._messenger.send_message(msg) # If everything has now been disabled there are no routes left to run, # so stop the bot cleanly (no point restarting D2R into empty games). if all(not v for v in self._do_runs_reset.values()): crit = "All runs have been disabled due to repeated failures — no routes left. Stopping bot." Logger.error(crit) if not self._game_stats.get_failure_reason(): self._game_stats.set_failure_reason(crit) if self._messenger.enabled: self._messenger.send_message(crit) # Persist a session report for review, then shut down. try: self._game_stats._save_session_report() except Exception as e: Logger.warning(f"Failed to save session report on shutdown: {e}") self.stop() safe_exit() def _ending_run_helper(self, res: bool | tuple[Location, bool]): self._game_stats._run_counter += 1 self._game_stats.log_exp() self._game_stats.log_run_completed() # either fill member variables with result data or mark run as failed failed_run = True if res: failed_run = False self._curr_loc, self._picked_up_items = res if failed_run: self._previous_run_failed = True # in case its the last run, end game. If run failed but more runs remain, skip to next run. if self.is_last_run(): self.trigger_or_stop("end_game", failed=failed_run) elif failed_run: Logger.warning(f"Run failed, skipping to next run") self.trigger_or_stop("end_run") else: self.trigger_or_stop("end_run") def _run_wrapper(self, run_name: str, run_obj, approach_args, battle_args): """Wrapper for run handlers that catches exceptions and stores the failure reason.""" res = False self._do_runs[run_name] = False self._game_stats.log_run_started(run_name) set_pause_state(False) self._curr_loc = run_obj.approach(self._curr_loc, *approach_args) if not self._curr_loc: # Approach failed — couldn't reach the boss self._game_stats.set_failure_reason(f"Approach failed for {run_name}") self._save_error_screenshot(run_name, "approach_failed") picked = None self._game_stats.log_run_finished(run_name, True, picked) self._record_run_result(run_name, True) self._ending_run_helper(res) return try: res = run_obj.battle(*battle_args) except Exception as e: import traceback tb = traceback.format_exc() error_msg = f"{type(e).__name__}: {e}" Logger.error(f"Exception during {run_name}: {error_msg}") Logger.error(tb) self._game_stats.set_failure_reason(error_msg) self._save_error_screenshot(run_name, f"exception_{type(e).__name__}") res = False picked = res[1] if isinstance(res, tuple) and len(res) > 1 else None # Safely check if battle succeeded - handle numpy arrays, None, tuples battle_succeeded = False if isinstance(res, tuple): battle_succeeded = len(res) > 0 and bool(res[0]) elif isinstance(res, (bool, np.ndarray)): battle_succeeded = bool(res) if not isinstance(res, np.ndarray) else bool(res.any()) else: battle_succeeded = bool(res) if res is not None else False if not battle_succeeded: # Battle failed — boss fight didn't complete if not self._game_stats.get_failure_reason(): self._game_stats.set_failure_reason(f"Battle failed for {run_name}") self._save_error_screenshot(run_name, "battle_failed") self._game_stats.log_run_finished(run_name, not battle_succeeded, picked) self._record_run_result(run_name, not battle_succeeded) self._ending_run_helper(res) def on_skip_to_level(self): """Skip town detection and go straight to level run.""" pass def on_run_level(self): self._game_stats.update_location("Level") self._run_wrapper("run_level", self._level, (not self._pre_buffed,), ()) def on_run_pindle(self): self._game_stats.update_location("Pindle") self._run_wrapper("run_pindle", self._pindle, (not self._pre_buffed,), ()) def on_run_shenk(self): self._game_stats.update_location("Shenk") self._run_wrapper("run_shenk", self._shenk, (), (Config().routes.get("run_eldritch_shenk"), not self._pre_buffed, self._game_stats)) def on_run_trav(self): self._game_stats.update_location("Travincal") self._run_wrapper("run_trav", self._trav, (), (not self._pre_buffed,)) def on_run_nihlathak(self): self._game_stats.update_location("Nihlathak") self._run_wrapper("run_nihlathak", self._nihlathak, (), (True,)) def on_run_arcane(self): self._game_stats.update_location("Arcane") self._run_wrapper("run_arcane", self._arcane, (), (not self._pre_buffed,)) def on_run_diablo(self): self._game_stats.update_location("Diablo") self._run_wrapper("run_diablo", self._diablo, (), (not self._pre_buffed,)) def on_run_vizier(self): self._game_stats.update_location("Vizier") self._run_wrapper("run_vizier", self._vizier, (), (not self._pre_buffed,)) def on_run_baal(self): self._game_stats.update_location("Baal") self._run_wrapper("run_baal", self._baal, (not self._pre_buffed,), ()) def on_run_mephisto(self): self._game_stats.update_location("Mephisto") self._run_wrapper("run_mephisto", self._mephisto, (not self._pre_buffed,), ()) def on_run_andariel(self): self._game_stats.update_location("Andariel") self._run_wrapper("run_andariel", self._andariel, (not self._pre_buffed,), ()) def on_run_countess(self): self._game_stats.update_location("Countess") self._run_wrapper("run_countess", self._countess, (not self._pre_buffed,), ())