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, game_browser from ui import player_bar from inventory import personal, vendor, belt, common from game_recovery import GameRecovery from death_manager import DeathManager from run import Pindle, ShenkEld, Trav, Nihlathak, Arcane, Diablo, Vizier, Level from run import ColdPlains 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: # Merc-resurrect circuit breaker. A resurrect NPC that cannot be detected costs a full # NPC hunt (~40s) per attempt, twice per game, in every game — with nothing to show for # it. After this many consecutive failures, stop trying for a while and run mercless. _MERC_RESURRECT_FAIL_LIMIT = 2 _MERC_RESURRECT_SKIP_GAMES = 15 # Set to the live Bot so utils.stealth can log into the same timeline without # importing Bot (which would be a circular import). _tl_active = None @staticmethod def timeline(phase: str, step: str, status: str = "start", detail: str = "") -> None: """Module-safe entry point for the timeline; a no-op if no Bot is running.""" if Bot._tl_active is not None: Bot._tl_active.tl(phase, step, status, detail) 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_cold_plains": Config().routes.get("run_cold_plains"), "run_baal_xp": Config().routes.get("run_baal_xp") and Config().baal_xp.get("enabled", False), "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._cold_plains = ColdPlains(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._tl_starts: dict = {} self._tl_history: list = [] # Rolling window for the periodic Discord timing report. Deliberately NOT reset # per game — it is reset when a report is sent. self._tl_agg: dict = {} # Scheduled-break state. Re-rolled after every break so the interval is # never the same twice. self._next_break_after = None self._next_break_len: float = 0.0 self._tl_window_start: float = time.time() self._tl_games = 0 self._tl_games_failed = 0 self._last_error_shot: str | None = None Bot._tl_active = self self._curr_loc: bool | Location = None # Act the character spawned in this game — detected reliably at spawn (town # markers are visible there). Used as the act-of-record when mid-town marker # detection fails (markers are position-dependent and often off-screen). self._spawn_act: Location | None = 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._maintenance_step: str | None = None 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._disabled_runs: set[str] = set() self._max_run_failures = Config().general.get("disable_run_after_failures", 5) # Restore runs disabled in a prior game this session (GameStats persists across Bot instances). for _run_name in self._game_stats.get_session_disabled_runs(): self._disabled_runs.add(_run_name) 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 # Create State Machine self._states=['initialization','hero_selection', 'town', 'level', 'cold_plains', 'baal_xp', '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_cold_plains', 'source': 'town', 'dest': 'cold_plains', 'before': "on_run_cold_plains"}, { 'trigger': 'run_baal_xp', 'source': 'town', 'dest': 'baal_xp', 'before': "on_run_baal_xp"}, { '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', 'cold_plains', 'baal_xp', '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 _verify_town_location(self, assumed: Location | None = None) -> Location: """Reconcile the tracked town location with the act the character is physically in. Call at every retry/fallback site instead of hardcoding a town start. The char respawns in whatever act it last save+exited from, so assumed locations rot across failed games and poison all subsequent pathing. """ assumed_act = TownManager.get_act_from_location(assumed) if assumed else None detected = self._town_manager.detect_current_act() if detected is None: # Town markers are position-dependent and often off-screen mid-town; the # spawn act (detected reliably at game start) is the act of record then. fallback = assumed_act or self._spawn_act or Location.A5_TOWN_START Logger.warning(f"_verify_town_location: act detection failed — falling back to {fallback}") return fallback if assumed_act != detected: if assumed_act: Logger.warning(f"Town location mismatch: assumed {assumed_act}, detected {detected} — using detected") return detected # Return the act anchor (town start) — recovery paths need a location with # defined pather routes, and `assumed` may be a non-town spot like a WP. return assumed_act 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 tl(self, phase: str, step: str, status: str = "start", detail: str = "") -> None: """One line of the whole-cycle timeline: spawn -> town -> run -> loot. grep "TL>" log/log.txt phase : game | town | run | stlth status : start | ok | skip | fail Every step is timed: "start" stamps the clock, and the terminating line reports took=Ns. That is what turns the timeline from a trace into something you can find the slow phase in. Format is deliberately fixed-width and machine-readable — do not prettify it. """ key = (phase, step) if status == "start": self._tl_starts[key] = time.time() took = "" else: t0 = self._tl_starts.pop(key, None) secs = (time.time() - t0) if t0 is not None else None took = f"took={secs:.1f}s" if secs is not None else "" # Keep a breadcrumb trail for this game so a failure can be explained by what # preceded it rather than just naming the step that happened to blow up. self._tl_history.append((phase, step, status, secs)) if secs is not None: agg = self._tl_agg.setdefault((phase, step), [0, 0.0, 0]) agg[0] += 1 agg[1] += secs if status == "fail": agg[2] += 1 gs = self._game_stats head = (f"TL> g{getattr(gs, '_game_counter', 0)} r{getattr(gs, '_run_counter', 0)} " f"| {phase:<5} | {step:<17} | {status:<5}") bits = [b for b in (took, detail) if b] line = f"{head} | {' | '.join(bits)}" if bits else head (Logger.warning if status == "fail" else Logger.info)(line) Bot._tl_active = self def _maybe_report_timings(self) -> None: """Post a timing + failure digest to Discord on a fixed interval. Aggregated from the same timeline the log uses, so the numbers here and the TL>/FAIL> lines can never drift apart. The window resets on every send, so each report covers exactly the period since the last one. """ try: hours = float(Config().general.get("discord_timing_report_h", 2) or 0) except Exception: hours = 2.0 if hours <= 0: return elapsed = time.time() - self._tl_window_start if elapsed < hours * 3600: return _NL = chr(10) try: games, failed = self._tl_games, self._tl_games_failed ok = games - failed pct = (failed / games * 100) if games else 0.0 lines = [ f"**Timing report** - last {elapsed / 3600:.1f}h", f"Games: {games} ({ok} ok, {failed} failed - {pct:.1f}%)", ] def _avg(key): a = self._tl_agg.get(key) return (a[1] / a[0]) if a and a[0] else 0.0 game_avg = _avg(("town", "maintenance")) + _avg(("run", "approach")) + _avg(("run", "battle")) lines.append( f"Avg town {_avg(('town','maintenance')):.0f}s | " f"approach {_avg(('run','approach')):.0f}s | " f"battle {_avg(('run','battle')):.0f}s | cycle ~{game_avg:.0f}s" ) # Leaf steps only: "maintenance" and the run_name entry are umbrellas that # contain the others. Stealth is excluded too — an AFK break is deliberate # idling, so listing it as a "slow step" buries the real ones. leaves = {k: v for k, v in self._tl_agg.items() if k[0] != "stlth" and not (k[1] == "maintenance" or k[1].startswith("run_"))} slow = sorted(leaves.items(), key=lambda kv: -(kv[1][1] / max(kv[1][0], 1)))[:5] if slow: lines.append(_NL + "__Slowest steps (avg)__") for (ph, st), (cnt, tot, f) in slow: lines.append(f"`{ph + '.' + st:<24}` {tot / cnt:5.0f}s x{cnt}" + (f" ({f} fail)" if f else "")) fails = sorted(((k, v[2]) for k, v in self._tl_agg.items() if v[2]), key=lambda kv: -kv[1]) if fails: lines.append(_NL + "__Failures by step__") for (ph, st), c in fails[:6]: lines.append(f"`{ph + '.' + st:<24}` {c}") else: lines.append(_NL + "No failing steps this window.") stl = {k: v for k, v in self._tl_agg.items() if k[0] == "stlth"} # Report observed against CONFIGURED rate, and call out anything that # never fired. A behaviour missing from the timeline is the exact # signature of the AFK-break bug — silence must be reported, not # left to look like "the roll just hasn't come up". try: from config import Config as _Cfg scfg = _Cfg().stealth expected = {"skip_run": scfg["skip_run_chance"] / 100.0, "afk_break": scfg["afk_break_chance"] / 100.0} except Exception: expected = {} if stl or expected: lines.append(_NL + "__Stealth__") for (ph, st), (cnt, tot, _f) in sorted(stl.items()): exp = f" (exp ~{expected[st] * games:.0f})" if st in expected and games else "" lines.append(f"`{st:<20}` x{cnt}{exp}" + (f" {tot / 60:.0f}m" if tot >= 60 else "")) for name, p_game in expected.items(): if any(k[1] == name for k in stl): continue exp_n = p_game * games flag = "NEVER FIRED" if exp_n >= 3 else "none (too few games to judge)" lines.append(f"`{name:<20}` x0 <- {flag} (exp ~{exp_n:.0f})") if self._messenger.enabled: self._messenger.send_message(_NL.join(lines)) Logger.info(f"Timing report sent ({games} games, {failed} failed, window {elapsed/3600:.1f}h)") except Exception as e: Logger.warning(f"Timing report failed (non-fatal): {e}") finally: self._tl_agg = {} self._tl_window_start = time.time() self._tl_games = 0 self._tl_games_failed = 0 def _report_failure(self, reason: str) -> None: """Emit one self-contained failure record. grep "FAIL>" log/log.txt A bare "Approach failed [step: X]" names the step that blew up but not what led there — and the step that blew up is frequently not the one that cost the time. This prints the reason, where the char ended up, the screenshot, the phase costs, and the breadcrumb trail of the whole game, so a failure can be diagnosed from the log alone without replaying it. """ gs = self._game_stats tag = f"FAIL> g{getattr(gs, '_game_counter', 0)} r{getattr(gs, '_run_counter', 0)}" Logger.error(f"{tag} | {reason}") Logger.error(f"{tag} | at={self._curr_loc} | step={self._maintenance_step or '-'}" + (f" | shot={self._last_error_shot}" if self._last_error_shot else "")) hist = [h for h in self._tl_history if h[3] is not None] if hist: # Rank leaf steps only. "maintenance" and the run_name entry are umbrellas # that contain the others, so they always top the list and say nothing. leaves = [h for h in hist if not (h[1] == "maintenance" or h[1].startswith("run_"))] costly = sorted(leaves or hist, key=lambda h: -h[3])[:3] Logger.error(f"{tag} | slowest: " + ", ".join(f"{p}.{st}={sec:.0f}s" for p, st, _, sec in costly)) trail = " > ".join( f"{p}.{st}{'!' if stat == 'fail' else ''}{f'({sec:.0f}s)' if sec >= 1 else ''}" for p, st, stat, sec in hist[-12:] ) Logger.error(f"{tag} | trail: {trail}") fails = [h for h in self._tl_history if h[2] == "fail"] if len(fails) > 1: Logger.error(f"{tag} | note: {len(fails)} failing steps this game: " + ", ".join(f"{p}.{st}" for p, st, _, _ in fails)) def on_init(self): self._game_stats.log_start_game() self._tl_history = [] self._last_error_shot = None self._town_manager.reset_wp_budget() keyboard.release(Config().char["stand_still"]) active_routes = list(self._do_runs.keys()) difficulty = Config().general.get("difficulty", "unknown") char_type = Config().char.get("type", "unknown") Logger.info(f"=== BOT START === char={char_type} | difficulty={difficulty} | routes={active_routes}") self.tl("game", "start", "start", f"char={char_type} difficulty={difficulty} routes={active_routes}") # Report which stealth behaviours actually have reachable call sites. # A behaviour that is enabled but unreachable prints as UNREACHABLE — # the AFK-break gap would have been visible on session one instead of # after 225 games. try: from utils.stealth import log_manifest log_manifest() except Exception as e: Logger.debug(f"stealth manifest unavailable: {e}") # Force D2R client area to stable position to prevent offset drift from utils.misc import enforce_d2r_window enforce_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 # Same for a cold-plains-only grinder (e.g. the bloodmoor lvl 1 sorc): # the character is already standing in A1 town, so skip the town marker # scan and go straight to the run. if list(self._do_runs.keys()) == ["run_cold_plains"]: Logger.info("Cold Plains run only - skipping town detection, starting directly") self._curr_loc = Location.A1_TOWN_START self._spawn_act = 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: # Stranded mid-town: town markers are position-dependent, so a char left # standing away from the spawn shows NO known marker. If we're in-game, # save+exit to character select and re-detect instead of giving up. if is_visible(ScreenObjects.InGame): Logger.warning("on_init: in-game but no known marker visible — save+exit to recover") view.save_and_exit() wait(2, 3) if (match := template_finder.search_and_wait(list(transition_to_screens.keys()), best_match=True, timeout=20)).valid: self.trigger_or_stop(transition_to_screens[match.name]) return 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() # Never trust the highlighted character blindly — if a char_name is # configured, select it first so the saved template is the right char. char_name = Config().general.get("char_name", "") if char_name and not character_select.select_char_by_name(char_name): Logger.warning(f"Could not select configured character '{char_name}' — saving currently highlighted character instead") 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: # Try a direct marker scan before any blind default — D2R respawns the # char in the act it last save+exited from, which is A4/A5 for this # route, so the old A1 default ran wrong-act pathing immediately. detected = self._town_manager.detect_current_act(timeout=4) if detected: Logger.warning(f"Town spawn templates missed — act detection says {detected}") self._curr_loc = detected else: Logger.warning("Could not detect town spawn — defaulting to A5_TOWN_START (route home act)") self._curr_loc = Location.A5_TOWN_START self._spawn_act = TownManager.get_act_from_location(self._curr_loc) self.tl("game", "spawn", "ok", f"at {self._curr_loc} (act {self._spawn_act})") # 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 — detecting town act from screen") self._curr_loc = self._verify_town_location() # Pause health manager if not already paused set_pause_state(True) # One-time skill bind preflight (like the D2R settings check at start): # press each configured skill hotkey and verify the expected icon appears # on the right slot. Warns loudly on mismatch; never blocks the run. if not getattr(self, "_skill_preflight_done", False): self._skill_preflight_done = True try: from utils.skill_preflight import validate_build_skill_icons if not validate_build_skill_icons(Config()): Logger.warning("Skill preflight found mismatched binds — run tools/set_binds_from_params.py to fix them automatically.") except Exception as e: Logger.warning(f"Skill preflight skipped: {e}") # Hard timeout: if town maintenance is still running after this many seconds, # something is stuck (pathing lost, NPC template mismatch, etc.). Bail out # with save-and-exit so the bot rejoins a clean game rather than spinning forever. _maint_start = time.time() _maint_max = Config().general["max_maintenance_time_s"] def _maint_timed_out(next_step: str) -> bool: elapsed = time.time() - _maint_start if elapsed > _maint_max: reason = f"Maintenance timeout after {elapsed:.0f}s before [{next_step}] — save and exit to rejoin" Logger.error(reason) self._save_error_screenshot("maintenance_timeout", reason) if not self._game_stats.get_failure_reason(): self._game_stats.set_failure_reason(reason) self.trigger_or_stop("end_game", failed=True) return True return False def _step(step: str, status: str = "start", detail: str = "") -> None: """Emit one line of the town-maintenance timeline. Every step reports start and outcome in the same shape so a whole town visit can be read (or grepped) as a sequence: grep "TL>" log/log.txt status is one of: start | ok | skip | fail. Keep the format stable — it is meant to be machine-readable, not prose. """ if status == "start": self._maintenance_step = step self.tl("town", step, status, detail) self._step_fn = _step _step("maintenance", "start", f"at {self._curr_loc}") # 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. # Stealth: an occasional unproductive town action. The bot otherwise # never does anything without a purpose, which is itself distinctive. # Town only, panel check paused, and strictly best-effort — this must # never be able to fail a maintenance step. try: from utils.stealth import should_browse_town, idle_drift if should_browse_town(): set_panel_check_paused(True) keyboard.send(Config().char["inventory_screen"]) wait(0.7, 1.9) idle_drift() wait(0.3, 0.9) keyboard.send(Config().char["inventory_screen"]) wait(0.2, 0.4) set_panel_check_paused(False) except Exception as e: Logger.debug(f"town browse skipped: {e}") try: set_panel_check_paused(False) except Exception: pass _step("town_heal") _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 _step("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]): if _maint_timed_out("identify_items"): return Logger.info("ID items at cain") _step("identify_items", "start", f"items needing id={sum(1 for i in items if getattr(i, 'need_id', False))}") self._curr_loc = self._town_manager.identify(self._curr_loc) if self._curr_loc is True: Logger.warning("identify() returned True (unexpected location) — re-detecting town act") self._curr_loc = self._verify_town_location() if not self._curr_loc: _step("identify_items", "fail", "Cain not available — continuing unidentified") Logger.warning("Could not identify items (Cain not available). Continuing without ID.") self._curr_loc = self._verify_town_location() else: # recheck inventory items = personal.inspect_items(game_stats=self._game_stats) _step("identify_items", "ok", f"at {self._curr_loc}") 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 sell_count = sum(1 for item in items if item.sell) if items else 0 stash_gold = personal.get_inventory_gold_full() _step("inspect_inventory", "ok", f"in pack={len(items) if items else 0} keep={sum(1 for i in items if i.keep) if items else 0} " f"sell={sell_count} gold_full={bool(stash_gold)}") # 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 # Raised 3 -> 8 (2026-08-27) to match the tp threshold on the same trip. # Every rare now consumes an ID scroll (the pickit's rare catch-all), so the # tome drains far faster than when only a handful of item types were picked # up. At 3 the bot ran down to 2 scrolls before restocking, and an un-ID'd # rare is never sold — it is held as need_id and occupies a slot instead. # The A5->A4 vendor trip this can trigger is also much safer now that # detect_current_act refuses to guess the act (Bug 28). consumables.should_buy("id", min_remaining = 8) ) # A vendor trip from A5 means an A5->A4 WP round-trip, the most failure-prone # navigation in the bot. 1-2 junk items aren't worth that risk - let them # accumulate and sell when the trip is needed anyway (or 3+ are pending). if need_refill or sell_count >= 3: if _maint_timed_out("buy_consumables"): return Logger.info("Buy consumables and/or sell items") _step("buy_consumables", "start", f"needs id={consumables.get_needs('id')} tp={consumables.get_needs('tp')} " f"hp={consumables.get_needs('health')} mana={consumables.get_needs('mana')} " f"rejuv={consumables.get_needs('rejuv')} | sell_pending={sell_count}") prev_buy_loc = self._curr_loc # A5 Malah is unreliable in the current patch (wandering NPC, stale body # templates — fails most games). Buy at A4 Jamella instead when in A5. buy_loc = self._curr_loc if TownManager.get_act_from_location(buy_loc) == Location.A5_TOWN_START: Logger.info("Buy consumables: in A5 — traveling to A4 Jamella (Malah unreliable)") a4_loc = self._town_manager.go_to_act(4, buy_loc) if a4_loc: buy_loc = a4_loc self._curr_loc = a4_loc prev_buy_loc = a4_loc else: Logger.warning("Buy consumables: travel to A4 failed — trying A5 Malah anyway") self._curr_loc, result_items = self._town_manager.buy_consumables(buy_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()}") _step("buy_consumables", "ok", f"at {self._curr_loc} | after: {consumables.get_needs()}") else: Logger.warning("Buy consumables failed, retrying in current act") wait(0.5, 0.6) if _maint_timed_out("buy_consumables_retry"): return retry_start = self._verify_town_location(prev_buy_loc) self._curr_loc, result_items = self._town_manager.buy_consumables(retry_start, items = items) if not self._curr_loc: # Travel to an alternate act's vendor. Prefer A4 Jamella: she stands # at a fixed spot and detects reliably, unlike Malah who wanders. alt_act = 4 if TownManager.get_act_from_location(retry_start) != Location.A4_TOWN_START else 5 Logger.warning(f"Buy consumables failed again, retrying via act {alt_act} vendor") wait(0.5, 0.6) if _maint_timed_out("buy_consumables_alt_act_retry"): return alt_loc = self._town_manager.go_to_act(alt_act, self._verify_town_location(retry_start)) if alt_loc: self._curr_loc, result_items = self._town_manager.buy_consumables(alt_loc, items = items) else: # Never run another act's pathing without confirmed travel — that # wanders the char to a random spot and fires false-positive clicks. Logger.error(f"Buy consumables retry: could not navigate to act {alt_act} — skipping alternate retry") 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: # NON-FATAL: buying pots is optional (belt refills from drops), but # STASHING loot is the whole point of the run. Do NOT end the game # here — that skips the stash step and strands picked-up runes/items # in inventory forever. Re-anchor and fall through to stash. reason = "Buy consumables failed (vendor not found) — continuing to stash without buying" _step("buy_consumables", "fail", "vendor not found after retries — continuing to stash") Logger.warning(reason) self._save_error_screenshot("maintenance", reason) self._curr_loc = self._verify_town_location(prev_buy_loc) 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") _step("heal") prev_heal_loc = self._curr_loc 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(self._verify_town_location(prev_heal_loc)) if not self._curr_loc: _step("heal", "fail", "healer not reachable") Logger.warning("Heal failed after retry, continuing without heal") self._curr_loc = self._verify_town_location(prev_heal_loc) # 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 not (keep_items or stash_gold): _step("stash_items", "skip", "nothing kept and gold not full") if keep_items or stash_gold: if _maint_timed_out("stash_items"): return Logger.info("Stashing items") _step("stash_items", "start", f"keep_items={bool(keep_items)} gold_full={bool(stash_gold)}") 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 detected town location") wait(0.5, 0.6) if _maint_timed_out("stash_items_retry"): return self._curr_loc, result_items = self._town_manager.stash(self._verify_town_location(prev_loc), items=items, game_stats=self._game_stats) if not self._curr_loc: reason = "Maintenance failed [step: stash_items] — stash NPC not found after retry" _step("stash_items", "fail", "stash not found after retry — ending game") Logger.error(reason) self._save_error_screenshot("maintenance", reason) if not self._game_stats.get_failure_reason(): self._game_stats.set_failure_reason(reason) self.trigger_or_stop("end_game", failed=True) return sell_items = any([item.sell for item in result_items]) if result_items else None _step("stash_items", "ok", f"at {self._curr_loc} | left in pack={len(result_items) if result_items else 0}") #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 scan_interval = Config().general.get("stash_scan_interval", False) if scan_interval and self._game_stats._run_counter % scan_interval == 0: Logger.info(f"Stash scan triggered (run {self._game_stats._run_counter}, interval {scan_interval})") try: import sys as _sys, os as _os _scripts = _os.path.join(_os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))), "scripts") if _scripts not in _sys.path: _sys.path.insert(0, _scripts) from stash_inventory import scan_page, TOTAL_PAGES, ensure_stash_open from make_stash_csv import make_csv _root = _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))) if ensure_stash_open(): all_items = {} for _p in range(TOTAL_PAGES): all_items[_p] = scan_page(_p) _json_path = _os.path.join(_root, "log", "stash_inventory.json") import json as _json with open(_json_path, "w") as _f: _json.dump(all_items, _f, indent=2) _csv_path = _os.path.join(_root, "stash_list.csv") _order, _ = make_csv(_json_path, _csv_path) Logger.info(f"Stash scan complete: {len(_order)} unique items -> {_csv_path}") from inventory.common import close as _close _close() except Exception as _e: Logger.warning(f"Stash scan failed (non-fatal): {_e}") # 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 not (need_repair or need_routine_repair or need_refill_teleport or sell_items): _step("repair", "skip", "no repair due and nothing to sell") if need_repair or need_routine_repair or need_refill_teleport or sell_items: if _maint_timed_out("repair"): return 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") _reason = ("gear about to break" if need_repair else "routine" if need_routine_repair else "tp charges" if need_refill_teleport else "sell items") _step("repair", "start", f"reason={_reason} sell_pending={bool(sell_items)}") prev_repair_loc = self._curr_loc self._curr_loc, result_items = self._town_manager.repair(self._curr_loc, items) if not self._curr_loc: Logger.warning("Repair failed, retrying from detected town location") wait(0.5, 0.6) self._curr_loc, result_items = self._town_manager.repair(self._verify_town_location(prev_repair_loc), items) if self._curr_loc: items = result_items _step("repair", "ok", f"at {self._curr_loc}") if not self._curr_loc: _step("repair", "fail", "vendor not found — non-fatal, continuing") # Keep maintenance best-effort for stability: avoid killing runs on flaky NPC/vendor detection. # IMPORTANT: re-detect the act — a failed repair trip may have left us in A4, # and assuming A5 here poisons the next run's waypoint pathing AND the act # the character respawns in after save+exit. 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 = self._verify_town_location() else: Logger.warning(f"Repair/vendor interaction failed [step: repair]; skipping maintenance and continuing run.") self._curr_loc = self._verify_town_location() # Check if merc needs to be revived if _maint_timed_out("resurrect_merc"): return _step("resurrect_merc") 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) # ALWAYS toggle 'o' back, whatever it opened. Symmetry is the only safe # dismiss here: # - closing only when MercPanelText matched left the CHRONICLE panel # (what 'o' actually opens on this client) up for the whole game # - "just send esc" is WORSE: with nothing open, esc opens the GAME MENU, # which LeftPanel/RightPanel do not match, so nothing closes that either # and every later template search sees a menu instead of the town # 'o' opened it, so 'o' closes it, and it cannot open something new. keyboard.send("o") wait(0.25, 0.35) if merc_panel_open: merc_visible = True # Record the REAL answer before the breaker below overwrites merc_visible # as a way of suppressing the attempt — otherwise the happy path (merc # alive) is indistinguishable from the suppressed one. merc_alive = merc_visible gs = self._game_stats skip_until = getattr(gs, "_merc_resurrect_skip_until", 0) if not merc_visible and gs._game_counter < skip_until: Logger.debug( f"Skipping merc resurrect until game {skip_until} " f"(resurrect NPC unreachable {getattr(gs, '_merc_resurrect_fail_streak', 0)}x in a row)" ) _step("resurrect_merc", "skip", f"breaker active until game {skip_until}") merc_visible = True # suppress this game's attempt without touching state 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: # Only retry while resurrect still works sometimes. When the NPC simply # cannot be found, the retry is a second guaranteed ~40s hunt. if getattr(gs, "_merc_resurrect_fail_streak", 0) < 1: Logger.warning("Resurrect failed, retrying") wait(0.5, 0.6) new_loc = self._town_manager.resurrect(self._curr_loc) else: Logger.warning("Resurrect failed — skipping the retry, this NPC has been unreachable") if new_loc is False: # Failed to resurrect (can't afford or other error) - don't log death, just continue _step("resurrect_merc", "fail", "NPC not reachable — continuing mercless") Logger.warning("Failed to resurrect merc after retry, continuing without merc") self._game_stats._merc_resurrect_failed = True gs._merc_resurrect_fail_streak = getattr(gs, "_merc_resurrect_fail_streak", 0) + 1 if gs._merc_resurrect_fail_streak >= self._MERC_RESURRECT_FAIL_LIMIT: gs._merc_resurrect_skip_until = gs._game_counter + self._MERC_RESURRECT_SKIP_GAMES Logger.warning( f"Merc resurrect has failed {gs._merc_resurrect_fail_streak}x in a row — " f"skipping it until game {gs._merc_resurrect_skip_until} to stop burning " f"~40s per hunt. Running without a merc until then." ) # resurrect() may have traveled to A4 before failing — re-anchor breadcrumb = getattr(self._town_manager, "last_known_loc", None) if breadcrumb: Logger.info(f"Resurrect traveled before failing — act-of-record now {breadcrumb}") self._curr_loc = breadcrumb self._town_manager.last_known_loc = None elif new_loc is None: _step("resurrect_merc", "skip", "resurrect returned None — staying put") Logger.warning("Resurrect returned None, continuing at current location") else: _step("resurrect_merc", "ok", f"at {new_loc}") self._game_stats.log_merc_death() self._curr_loc = new_loc # It worked — clear the breaker so a transient failure never # permanently disables resurrecting. gs._merc_resurrect_fail_streak = 0 gs._merc_resurrect_skip_until = 0 elif merc_alive: _step("resurrect_merc", "skip", "merc alive") elif self._game_stats._merc_resurrect_failed: _step("resurrect_merc", "skip", "already failed this game") else: _step("resurrect_merc", "skip", "use_merc disabled") # Gamble if needed if _maint_timed_out("gamble"): return if not (vendor.get_gamble_status() and Config().char["gamble_items"]): _step("gamble", "skip", "stash not full / gambling not configured") else: _step("gamble") 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: _step("gamble", "fail", "gamble vendor not reachable") Logger.warning("Gamble failed, skipping gamble") self._curr_loc = self._verify_town_location() 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 = self._verify_town_location() break _step("maintenance", "ok", f"at {self._curr_loc}") # 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.tl("game", "end", "fail" if failed else "ok", (self._game_stats.get_failure_reason() or "") if failed else "") self._tl_games += 1 if failed: self._tl_games_failed += 1 self._report_failure(self._game_stats.get_failure_reason() or "unknown reason") self._maybe_report_timings() 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) # Session budget: stop after roughly N hours (rolled per run), so # consecutive days are not identical in length. Playing the same hours # every night is a stronger signal than any per-click timing. try: from utils.stealth import session_budget_seconds budget = session_budget_seconds() if budget and (time.time() - self._timer) > budget: msg = f"Session budget reached ({hms(time.time() - self._timer)}) — stopping for today." Logger.info(msg) self.tl("stlth", "session_end", "ok", msg) if self._messenger.enabled: self._messenger.send_message(msg) return self.stop() except Exception as e: Logger.debug(f"session budget check skipped: {e}") # Scheduled long break. The interval and duration are RE-ROLLED each # time: a break at exactly 120 minutes every time is still a pattern, # just a slower one than taking no break at all. from utils.stealth import break_schedule, idle_drift if self._next_break_after is None: self._next_break_after, self._next_break_len = break_schedule() if self._next_break_after: elapsed_time = time.time() - self._timer Logger.debug(f'Session length = {math.ceil(elapsed_time/60)} minutes, next break after {self._next_break_after/60:.0f}m.') if elapsed_time > self._next_break_after: break_msg = f'Ran for {hms(elapsed_time)}, taking a break for {hms(self._next_break_len)}.' Logger.info(break_msg) self.tl("stlth", "scheduled_break", "start", break_msg) if self._messenger.enabled: self._messenger.send_message(break_msg) if not self._pausing: self.toggle_pause() idle_drift() wait(self._next_break_len) # Re-roll for the next one and restart the clock. self._next_break_after, self._next_break_len = break_schedule() self._timer = time.time() self.tl("stlth", "scheduled_break", "ok", "resuming") break_msg = f'Break over, will now run for {self._next_break_after/60:.0f}m.' 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): # baal_xp is a self-contained cycle: it leaves the public game and re-enters # its own game inside the handler, so the character is already standing in # town when we get here. Skip the TP-to-town logic and go straight to # maintenance (which re-detects the act and does the town routine). if self.state == "baal_xp": if not self._curr_loc: self._curr_loc = self._verify_town_location() set_pause_state(True) self.trigger_or_stop("maintenance") return if not Config().char["pre_buff_every_run"]: self._pre_buffed = True # No-TP levelling chars (e.g. a pre-clvl-18 sorc): the run object # walks the char home itself — tp_town() would just fail and waste # time on retries. run_obj = getattr(self, "_current_run", None) if run_obj is not None and hasattr(run_obj, "return_to_town"): self._curr_loc = run_obj.return_to_town() if self._curr_loc: set_pause_state(True) # Stealth: same AFK-break roll as the tp_town() path below. This # branch returns early, so without this call a character that walks # home (no teleport / no TP scrolls — e.g. the Pindle red-portal # exit) NEVER takes an AFK break: measured 0 breaks in 225 games # against a configured 5% per game. maybe_afk_break() self.trigger_or_stop("maintenance") return # 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 = self._verify_town_location(tp_loc) return self.trigger_or_stop("maintenance") # TP failed twice Logger.warning("TP to town failed twice, trying to walk back") set_pause_state(True) # No TP at all (e.g. a pre-clvl-18 levelling char): walk back to town. # The next maintenance's open_wp() re-detects the physical act and # traverses from wherever the character actually is, so this works # from anywhere in A1. if not self._char.capabilities.can_teleport_natively and not self._char.capabilities.can_teleport_with_charges: Logger.info("No teleport available - walking back to town") self._walk_back_to_town() self._curr_loc = self._verify_town_location(self._curr_loc if isinstance(self._curr_loc, str) else None) maybe_afk_break() self.trigger_or_stop("maintenance") return self._curr_loc = self._verify_town_location(self._curr_loc if isinstance(self._curr_loc, str) else None) maybe_afk_break() self.trigger_or_stop("maintenance") def _walk_back_to_town(self, max_steps: int = 24): """Walk south (town is south of the A1 outdoor areas) until a town marker is visible. Used when the character has no teleport at all. Town markers are position-dependent — they only match when the char is close to the specific town spot the template was captured at — so a marker match means we're effectively home. If we never get one, fall back to the last known town location: the next maintenance's open_wp() re-detects the physical act and traverses from wherever the char is. """ from screen import convert_abs_to_monitor for i in range(max_steps): if self._town_manager.detect_current_act(timeout=1.5): Logger.info(f"Walk-back: town markers visible after {i + 1} step(s)") return pos_m = convert_abs_to_monitor((random.randint(-40, 40), 120 + random.randint(0, 60))) self._char.walk(pos_m, force_move=True) wait(1.0, 1.6) Logger.warning(f"Walk-back: no town marker after {max_steps} steps - continuing anyway") # 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 self._last_error_shot = 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 across games (stored in GameStats so counts survive Bot instance recreation). After too many failures a run is disabled for the rest of the session and the game-level consecutive-fail counter is reset so the bot keeps running the remaining routes instead of quitting.""" prev_count = self._game_stats.get_session_run_failure_count(run_name) count = self._game_stats.update_session_run_failure(run_name, failed) if not failed: if prev_count > 0: Logger.info(f"{run_name} succeeded — resetting its failure counter") return 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) self._game_stats.disable_session_run(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) # Reset game-level consecutive fail counter so game_controller restarts # the bot (for remaining routes) instead of quitting. self._game_stats.reset_consecutive_fails() # 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") # Clear the current run reference — on_end_run() uses it to decide # whether the run object walks the char home (no-TP levelling chars). self._current_run = None 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) self.tl("run", run_name, "start", f"from {self._curr_loc}") set_pause_state(False) self._current_run = run_obj self.tl("run", "approach", "start") self._curr_loc = run_obj.approach(self._curr_loc, *approach_args) if not self._curr_loc: # Approach failed — couldn't reach the boss step = getattr(run_obj, "approach_fail_step", None) reason = f"Approach failed for {run_name}" + (f" [step: {step}]" if step else "") self.tl("run", "approach", "fail", f"step={step or '?'}") Logger.error(reason) self._game_stats.set_failure_reason(reason) self._save_error_screenshot(run_name, reason) picked = None loot = self._pickit.consume_run_loot() if hasattr(self, "_pickit") and self._pickit else [] if loot: Logger.info(f"Loot from {run_name} (approach failed): {', '.join(loot)}") self.tl("run", "loot", "ok", f"(approach failed) {', '.join(loot)}") self._game_stats.log_run_finished(run_name, True, picked, loot=loot) self._record_run_result(run_name, True) self._ending_run_helper(res) return self.tl("run", "approach", "ok", f"at {self._curr_loc}") self.tl("run", "battle", "start") 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 self.tl("run", "battle", "ok" if battle_succeeded else "fail") 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") # Per-boss loot summary so drops can be value-judged per run loot = self._pickit.consume_run_loot() if hasattr(self, "_pickit") and self._pickit else [] if loot: counted = {} for name in loot: counted[name] = counted.get(name, 0) + 1 summary = ", ".join(f"{n}x {name}" if n > 1 else name for name, n in counted.items()) Logger.info(f"Loot from {run_name}: {summary}") self.tl("run", "loot", "ok", summary) else: Logger.info(f"Loot from {run_name}: nothing picked up") self.tl("run", "loot", "ok", "nothing picked up") self._game_stats.log_run_finished(run_name, not battle_succeeded, picked, loot=loot) 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_cold_plains(self): self._game_stats.update_location("ColdPlains") self._run_wrapper("run_cold_plains", self._cold_plains, (not self._pre_buffed,), ()) def on_run_baal_xp(self): """Baal XP farm: leave own game, join a public game, hide, collect XP, leave. Self-contained cycle — it exits and re-enters the game entirely, so it does NOT use the run_wrapper/approach/battle pattern. The character is back in its own game (town) when this returns; on_end_run() then triggers maintenance. The hide-and-wait phase is a pure wait: no maintenance, no runs, no pathing. The health manager thread keeps running (auto-potions). Death or low HP leaves the game immediately; the game_controller's death handling takes over from there (corpse pickup on respawn).""" self._game_stats.update_location("BaalXP") self._do_runs["run_baal_xp"] = False self._game_stats.log_run_started("run_baal_xp") set_pause_state(False) cfg = Config().baal_xp Logger.info(f"=== BAAL XP FARM === filter='{cfg['game_name_filter']}' max_wait={cfg['max_wait_s']}s xp_target={cfg['xp_threshold']}") try: # --- Phase 1: Leave current game, get to hero selection --- if is_visible(ScreenObjects.InGame): if not view.fast_save_and_exit(): Logger.error("baal_xp: fast_save_and_exit failed") self._game_stats.set_failure_reason("baal_xp: save_and_exit failed") self._save_error_screenshot("baal_xp", "save_exit_failed") self.trigger_or_stop("end_run") return # Wait for hero selection (main menu) if not GameRecovery(DeathManager()).go_to_hero_selection(): Logger.error("baal_xp: could not reach hero selection") self._game_stats.set_failure_reason("baal_xp: hero selection unreachable") self.trigger_or_stop("end_run") return # --- Phase 2: Select character and join a public game --- if not character_select.select_char(): Logger.error("baal_xp: character selection failed") self._game_stats.set_failure_reason("baal_xp: char select failed") self.trigger_or_stop("end_run") return if not game_browser.join_game( name_filter=cfg["game_name_filter"], max_wait_s=cfg["join_timeout_s"], ): Logger.error("baal_xp: failed to join a game") self._game_stats.set_failure_reason("baal_xp: game join failed") self._save_error_screenshot("baal_xp", "join_failed") # Go back to our own game self._recover_to_own_game() self.trigger_or_stop("end_run") return # --- Phase 3: Wait for in-game --- if not game_browser.wait_for_in_game(timeout=60): Logger.error("baal_xp: never entered the joined game") self._game_stats.set_failure_reason("baal_xp: in-game timeout") self._recover_to_own_game() self.trigger_or_stop("end_run") return # --- Phase 4: Enter game setup --- # Handle corpse if we died in this game before if is_visible(ScreenObjects.Corpse): view.pickup_corpse() wait_until_hidden(ScreenObjects.Corpse) belt.fill_up_belt_from_inventory(Config().char["num_loot_columns"]) self._char.discover_capabilities() # /nopickup so we don't grab loot from other players if Config().char["enable_no_pickup"] and not self._game_stats._nopickup_active: if view.enable_no_pickup(): self._game_stats._nopickup_active = True Logger.info("baal_xp: /nopickup active") # Pre-buff self._char.pre_buff() # --- Phase 5: Walk to hide spot and stand still --- # One click to the configured screen position, then zero input while # hiding (anti-cheat stealth: minimal input while AFK). hide_x, hide_y = cfg["hide_x"], cfg["hide_y"] hide_pos = convert_screen_to_monitor((hide_x, hide_y)) Logger.info(f"baal_xp: walking to hide spot ({hide_x}, {hide_y})") mouse.move(*hide_pos) mouse.click("left") wait(2.0, 3.0) # give the character time to walk there # Record starting XP try: start_exp, _ = player_bar.get_experience() except Exception: start_exp = 0 Logger.info(f"baal_xp: starting XP = {start_exp}") # --- Phase 6: Hide and wait --- # Pure wait loop: no maintenance, no runs. The health manager thread # keeps running (auto-potions). Leave on: max timer, XP target, # low HP, death, or being kicked back to the main menu. max_wait = cfg["max_wait_s"] xp_target = cfg["xp_threshold"] min_hp = cfg["min_hp_pct"] start_time = time.time() last_log = 0.0 last_xp_check = 0.0 while True: elapsed = time.time() - start_time if elapsed > max_wait: Logger.info(f"baal_xp: max wait {max_wait}s reached — leaving") break img = grab() if not is_visible(ScreenObjects.InGame, img): # Death screen, kicked to menu, or loading — bail out. # The death manager thread handles the death screen itself. Logger.warning("baal_xp: left in-game state (kicked/died) — leaving") break hp_pct = meters.get_health(img) * 100 if hp_pct < min_hp: Logger.warning(f"baal_xp: HP {hp_pct:.0f}% < {min_hp}% — chickening out") break # Check XP every 30s to avoid OCR spam if time.time() - last_xp_check > 30: last_xp_check = time.time() try: cur_exp, _ = player_bar.get_experience() gained = cur_exp - start_exp if cur_exp > start_exp else 0 Logger.info(f"baal_xp: {elapsed:.0f}s in | XP {cur_exp} (+{gained}) | HP {hp_pct:.0f}%") if gained >= xp_target: Logger.info(f"baal_xp: XP target {xp_target} reached (+{gained}) — leaving") break except Exception as e: Logger.debug(f"baal_xp: XP check failed: {e}") if time.time() - last_log > 60: last_log = time.time() Logger.debug(f"baal_xp: hiding... {elapsed:.0f}s in, HP {hp_pct:.0f}%") wait(3, 5) # --- Phase 7: Leave the game --- Logger.info("baal_xp: leaving game") if is_visible(ScreenObjects.InGame) and not view.fast_save_and_exit(): Logger.error("baal_xp: save_and_exit on leave failed") self._game_stats.set_failure_reason("baal_xp: leave failed") self._save_error_screenshot("baal_xp", "leave_failed") # --- Phase 8: Recover to own game --- self._recover_to_own_game() # Log run result self._game_stats.log_run_finished("run_baal_xp", False, None, loot=[]) self._game_stats.log_exp() self.trigger_or_stop("end_run") except Exception as e: import traceback tb = traceback.format_exc() error_msg = f"baal_xp exception: {type(e).__name__}: {e}" Logger.error(error_msg) Logger.error(tb) self._game_stats.set_failure_reason(error_msg) self._save_error_screenshot("baal_xp", f"exception_{type(e).__name__}") # Best effort: get back to our own game so the bot keeps running self._recover_to_own_game() self._game_stats.log_run_finished("run_baal_xp", True, None, loot=[]) self.trigger_or_stop("end_run") def _recover_to_own_game(self): """After leaving a public game, get back to hero selection and re-enter our own game.""" # Wait for hero selection if not is_visible(ScreenObjects.MainMenu): if not GameRecovery(DeathManager()).go_to_hero_selection(): Logger.error("baal_xp: could not reach hero selection after leaving public game") return # Select character if not character_select.select_char(): Logger.error("baal_xp: char select failed during recovery") return # Start our own game if not main_menu.start_game(): Logger.error("baal_xp: failed to start own game during recovery") return # Wait for town spawn self._curr_loc = self._town_manager.wait_for_town_spawn() if not self._curr_loc: detected = self._town_manager.detect_current_act(timeout=4) self._curr_loc = detected or Location.A5_TOWN_START Logger.info(f"baal_xp: recovered to own game at {self._curr_loc}") 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,), ())