fa72da56e3
In nightmare the merc dies most games, so resurrect_merc runs constantly — and Qual-Kehk
detection was failing 100% of the time (5 timeouts in 5 attempts, 107 hover attempts over
12 games). Each failed hunt costs ~40s and the code retried once, so a dead merc cost
~80s in EVERY game. Game length blew out to 185-250s against a normal ~60s.
GameStats._merc_resurrect_failed did not help: log_start_game resets it, so it only ever
suppressed a second attempt within one game. Nothing carried across games.
The name tag template is degenerate rather than merely stale — every grid-sweep "hit"
reported the identical score at unrelated positions:
found name tag at (255, 227) (score 0.424)
found name tag at (1110, 100) (score 0.424)
found name tag at (930, 310) (score 0.424)
so "found" is meaningless; it is matching uniform background.
Fix is cost containment, not detection: a cross-game circuit breaker on GameStats that
log_start_game deliberately does NOT reset — _merc_resurrect_fail_streak and
_merc_resurrect_skip_until, with Bot._MERC_RESURRECT_FAIL_LIMIT=2 and
_MERC_RESURRECT_SKIP_GAMES=15. The retry is also skipped once the streak is >=1, since
that is a second guaranteed-futile 40s hunt. Both counters clear on any successful
resurrect so a transient failure cannot permanently disable resurrecting.
Simulated over 30 games with an undetectable NPC: 60 hunts -> 4 (~40 min -> ~2.7 min).
Measured live: the breaker engaged on game 2 and game times went 250s / 185s -> 14s, 14s,
43s, 71s, 111s.
Still open: recapturing qual_name_tag_white.png is the actual fix for detection.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1146 lines
61 KiB
Python
1146 lines
61 KiB
Python
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:
|
|
|
|
# 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
|
|
|
|
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
|
|
# 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', '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 _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 on_init(self):
|
|
self._game_stats.log_start_game()
|
|
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}")
|
|
# 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
|
|
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)
|
|
|
|
# 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
|
|
|
|
# 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.
|
|
self._maintenance_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
|
|
self._maintenance_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")
|
|
self._maintenance_step = "identify_items"
|
|
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:
|
|
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)
|
|
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()
|
|
|
|
# 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)
|
|
)
|
|
# 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")
|
|
self._maintenance_step = "buy_consumables"
|
|
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()}")
|
|
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"
|
|
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")
|
|
self._maintenance_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:
|
|
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 keep_items or stash_gold:
|
|
if _maint_timed_out("stash_items"): return
|
|
Logger.info("Stashing items")
|
|
self._maintenance_step = "stash_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 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"
|
|
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
|
|
#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 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")
|
|
self._maintenance_step = "repair"
|
|
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
|
|
if not self._curr_loc:
|
|
# 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
|
|
self._maintenance_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)
|
|
if merc_panel_open:
|
|
# Merc is alive - panel opened, close it
|
|
keyboard.send("o")
|
|
wait(0.2, 0.3)
|
|
merc_visible = True
|
|
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)"
|
|
)
|
|
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
|
|
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:
|
|
Logger.warning("Resurrect returned None, continuing at current location")
|
|
pass
|
|
else:
|
|
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
|
|
|
|
# Gamble if needed
|
|
if _maint_timed_out("gamble"): return
|
|
self._maintenance_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:
|
|
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
|
|
|
|
# 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 = 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)
|
|
self._curr_loc = self._verify_town_location(self._curr_loc if isinstance(self._curr_loc, str) else None)
|
|
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 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")
|
|
|
|
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
|
|
step = getattr(run_obj, "approach_fail_step", None)
|
|
reason = f"Approach failed for {run_name}" + (f" [step: {step}]" if step else "")
|
|
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._game_stats.log_run_finished(run_name, True, picked, loot=loot)
|
|
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")
|
|
# 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}")
|
|
else:
|
|
Logger.info(f"Loot from {run_name}: 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_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,), ())
|
|
|
|
|
|
|