alexandClaude Sonnet 4.6
6c8da72bf4
- i_char.py: rewrite _pre_buff_cta() to verify each weapon switch via BC skill-bar template; detects wrong slot on game start (leftover from interrupted buff cycle), retries failed switches once, logs clearly when stuck on CTA slot to prevent dying with wrong weapon in combat - personal.py: protect needed consumables from sell/drop in inspect_items; check get_needs() before marking a pot for discard — if the belt needs that pot type, skip it so fill_up_belt_from_inventory can restock later - bot.py: add fill_up_belt_from_inventory + update_pot_needs after buy_consumables so inventory pots reach the belt even when out of gold; add town-heal loop at start of on_maintenance to drink health/rejuv pots until HP >= 95% before the next run (health manager is paused in town) - game_stats.py: add rolling 10-game pickup health check; warns in log and sends Discord alert when zero item pickups occur across 10 games while chickens or merc deaths are present - params.ini: fix show_belt=n -> show_belt=k (belt key was wrong, causing 1.5s wasted recovery attempt every first game in a session) Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
810 lines
40 KiB
Python
810 lines
40 KiB
Python
from transitions import Machine
|
|
import time
|
|
from input_layer import keyboard
|
|
import time
|
|
import os
|
|
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, restart_game
|
|
from game_stats import GameStats
|
|
from logger import Logger
|
|
from config import Config
|
|
from screen import grab, convert_monitor_to_screen, convert_screen_to_abs, convert_abs_to_monitor, convert_screen_to_monitor
|
|
import template_finder
|
|
from char import IChar
|
|
from item.pickit import PickIt
|
|
from item import consumables
|
|
from pather import Pather, Location
|
|
from char.sorceress import LightSorc, BlizzSorc, BlizzorbSorc, NovaSorc, HydraSorc
|
|
from char.trapsin import Trapsin
|
|
from char.paladin.hammerdin import Hammerdin
|
|
from char.paladin import FoHdin
|
|
from char.warlock import FireLock, AbyssLock, EchoLock
|
|
from char.amazon.javazon import Javazon
|
|
from char.barbarian import Barbarian
|
|
from char.necro import Necro
|
|
from char.poison_necro import Poison_Necro
|
|
from char.bone_necro import Bone_Necro
|
|
from char.basic import Basic
|
|
from char.basic_ranged import Basic_Ranged
|
|
from ui_manager import wait_until_hidden, wait_until_visible, ScreenObjects, is_visible, detect_screen_object
|
|
from ui import meters, skills, view, character_select, main_menu
|
|
from inventory import personal, vendor, belt, common
|
|
|
|
from run import Pindle, ShenkEld, Trav, Nihlathak, Arcane, Diablo, Vizier, Level
|
|
from run.baal import Baal
|
|
from run.mephisto import Mephisto
|
|
from run.andariel import Andariel
|
|
from run.countess import Countess
|
|
from town import TownManager, A1, A2, A3, A4, A5, town_manager
|
|
|
|
from messages import Messenger
|
|
from threading import Lock
|
|
from utils.stealth import maybe_afk_break, should_skip_run
|
|
|
|
class Bot:
|
|
|
|
def __init__(self, game_stats: GameStats):
|
|
self._game_stats = game_stats
|
|
self._messenger = Messenger()
|
|
self._pather = Pather()
|
|
self._pickit = PickIt()
|
|
self._stash_mutex = Lock()
|
|
|
|
# Create Character
|
|
match Config().char["type"]:
|
|
case "sorceress" | "light_sorc":
|
|
self._char: IChar = LightSorc(Config().light_sorc, self._pather)
|
|
case "blizz_sorc":
|
|
self._char: IChar = BlizzSorc(Config().blizz_sorc, self._pather)
|
|
case "blizzorb_sorc":
|
|
self._char: IChar = BlizzorbSorc(Config().blizzorb_sorc, self._pather)
|
|
case "nova_sorc":
|
|
self._char: IChar = NovaSorc(Config().nova_sorc, self._pather)
|
|
case "hydra_sorc":
|
|
self._char: IChar = HydraSorc(Config().hydra_sorc, self._pather)
|
|
case "hammerdin" | "paladin":
|
|
self._char: IChar = Hammerdin(Config().hammerdin, self._pather, self._pickit) #pickit added for diablo
|
|
case "fohdin":
|
|
self._char: IChar = FoHdin(Config().fohdin, self._pather, self._pickit)
|
|
case "abyss_lock" | "warlock":
|
|
self._char: IChar = AbyssLock(Config().abyss_lock, self._pather)
|
|
case "fire_lock":
|
|
self._char: IChar = FireLock(Config().fire_lock, self._pather)
|
|
case "echo_lock":
|
|
self._char: IChar = EchoLock(Config().echo_lock, self._pather)
|
|
case "amazon" | "javazon":
|
|
self._char: IChar = Javazon(Config().javazon, self._pather)
|
|
case "trapsin":
|
|
self._char: IChar = Trapsin(Config().trapsin, self._pather)
|
|
case "barbarian":
|
|
self._char: IChar = Barbarian(Config().barbarian, self._pather)
|
|
case "poison_necro":
|
|
self._char: IChar = Poison_Necro(Config().poison_necro, self._pather)
|
|
case "bone_necro":
|
|
self._char: IChar = Bone_Necro(Config().bone_necro, self._pather)
|
|
case "necro":
|
|
self._char: IChar = Necro(Config().necro, self._pather)
|
|
case "basic":
|
|
self._char: IChar = Basic(Config().basic, self._pather)
|
|
case "basic_ranged":
|
|
self._char: IChar = Basic_Ranged(Config().basic_ranged, self._pather)
|
|
case _:
|
|
Logger.error(f'{Config().char["type"]} is not supported! Closing down bot.')
|
|
os._exit(1)
|
|
|
|
# Create Town Manager
|
|
a5 = A5(self._pather, self._char)
|
|
a4 = A4(self._pather, self._char)
|
|
a3 = A3(self._pather, self._char)
|
|
a2 = A2(self._pather, self._char)
|
|
a1 = A1(self._pather, self._char)
|
|
self._town_manager = TownManager(a1, a2, a3, a4, a5)
|
|
|
|
# Create runs
|
|
self._do_runs = {
|
|
"run_level": Config().routes.get("run_level"),
|
|
"run_trav": Config().routes.get("run_trav"),
|
|
"run_pindle": Config().routes.get("run_pindle"),
|
|
"run_shenk": Config().routes.get("run_eldritch") or Config().routes.get("run_eldritch_shenk"),
|
|
"run_nihlathak": Config().routes.get("run_nihlathak"),
|
|
"run_arcane": Config().routes.get("run_arcane"),
|
|
"run_diablo": Config().routes.get("run_diablo"),
|
|
"run_vizier": Config().routes.get("run_vizier"),
|
|
"run_baal": Config().routes.get("run_baal"),
|
|
"run_mephisto": Config().routes.get("run_mephisto"),
|
|
"run_andariel": Config().routes.get("run_andariel"),
|
|
"run_countess": Config().routes.get("run_countess"),
|
|
}
|
|
# Adapt order to the config
|
|
self._do_runs = OrderedDict((k, self._do_runs[k]) for k in Config().routes_order if k in self._do_runs and self._do_runs[k])
|
|
|
|
self._do_runs_reset = copy(self._do_runs)
|
|
Logger.info(f"Doing runs: {self._do_runs_reset.keys()}")
|
|
if Config().general["randomize_runs"]:
|
|
self.shuffle_runs()
|
|
self._level = Level(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._pindle = Pindle(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._shenk = ShenkEld(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._trav = Trav(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._nihlathak = Nihlathak(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._arcane = Arcane(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._diablo = Diablo(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._vizier = Vizier(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
|
|
self._baal = Baal(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._mephisto = Mephisto(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._andariel = Andariel(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
self._countess = Countess(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
|
|
# Create member variables
|
|
self._picked_up_items = False
|
|
self._curr_loc: bool | Location = None
|
|
self._use_id_tome = True
|
|
self._use_keys = True
|
|
self._pre_buffed = False
|
|
self._stopping = False
|
|
self._pausing = False
|
|
self._current_threads = []
|
|
self._ran_no_pickup = False
|
|
self._previous_run_failed = False
|
|
self._timer = time.time()
|
|
|
|
# Create State Machine
|
|
self._states=['initialization','hero_selection', 'town', 'level', 'pindle', 'shenk', 'trav', 'nihlathak', 'arcane', 'diablo', 'vizier', 'baal', 'mephisto', 'andariel', 'countess']
|
|
self._transitions = [
|
|
{ 'trigger': 'init', 'source': 'initialization', 'dest': '=','before': "on_init"},
|
|
{ 'trigger': 'skip_to_level', 'source': 'initialization', 'dest': 'level', 'before': "on_skip_to_level"},
|
|
{ 'trigger': 'select_character', 'source': 'initialization', 'dest': 'hero_selection', 'before': "on_select_character"},
|
|
{ 'trigger': 'start_from_town', 'source': ['initialization', 'hero_selection'], 'dest': 'town', 'before': "on_start_from_town"},
|
|
{ 'trigger': 'create_game', 'source': 'hero_selection', 'dest': '=', 'before': "on_create_game"},
|
|
# Tasks within town
|
|
{ 'trigger': 'maintenance', 'source': 'town', 'dest': 'town', 'before': "on_maintenance"},
|
|
# Different runs
|
|
{ 'trigger': 'run_level', 'source': 'town', 'dest': 'level', 'before': "on_run_level"},
|
|
{ 'trigger': 'run_pindle', 'source': 'town', 'dest': 'pindle', 'before': "on_run_pindle"},
|
|
{ 'trigger': 'run_shenk', 'source': 'town', 'dest': 'shenk', 'before': "on_run_shenk"},
|
|
{ 'trigger': 'run_trav', 'source': 'town', 'dest': 'trav', 'before': "on_run_trav"},
|
|
{ 'trigger': 'run_nihlathak', 'source': 'town', 'dest': 'nihlathak', 'before': "on_run_nihlathak"},
|
|
{ 'trigger': 'run_arcane', 'source': 'town', 'dest': 'arcane', 'before': "on_run_arcane"},
|
|
{ 'trigger': 'run_diablo', 'source': 'town', 'dest': 'diablo', 'before': "on_run_diablo"},
|
|
{ 'trigger': 'run_vizier', 'source': 'town', 'dest': 'vizier', 'before': "on_run_vizier"},
|
|
{ 'trigger': 'run_baal', 'source': 'town', 'dest': 'baal', 'before': "on_run_baal" },
|
|
{ 'trigger': 'run_mephisto', 'source': 'town', 'dest': 'mephisto', 'before': "on_run_mephisto" },
|
|
{ 'trigger': 'run_andariel', 'source': 'town', 'dest': 'andariel', 'before': "on_run_andariel" },
|
|
{ 'trigger': 'run_countess', 'source': 'town', 'dest': 'countess', 'before': "on_run_countess" },
|
|
# End run / game
|
|
{ 'trigger': 'end_run', 'source': ['level', 'shenk', 'pindle', 'nihlathak', 'trav', 'arcane', 'diablo','vizier', 'baal', 'mephisto', 'andariel', 'countess'], 'dest': 'town', 'before': "on_end_run"},
|
|
{ 'trigger': 'end_game', 'source': ['town', 'shenk', 'pindle', 'nihlathak', 'trav', 'arcane', 'diablo','vizier','end_run', 'baal', 'mephisto', 'andariel', 'countess'], 'dest': 'initialization', 'before': "on_end_game"},
|
|
]
|
|
self.machine = Machine(model=self, states=self._states, initial="initialization", transitions=self._transitions, queued=True)
|
|
self._transmute = Transmute(self._game_stats)
|
|
|
|
|
|
def draw_graph(self):
|
|
# Draw the whole graph, graphviz binaries must be installed and added to path for this!
|
|
from transitions.extensions import GraphMachine
|
|
self.machine = GraphMachine(model=self, states=self._states, initial="initialization", transitions=self._transitions, queued=True)
|
|
self.machine.get_graph().draw('my_state_diagram.png', prog='dot')
|
|
|
|
def get_curr_location(self):
|
|
return self._curr_loc
|
|
|
|
def start(self):
|
|
self.trigger_or_stop('init')
|
|
|
|
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("Restart botty")
|
|
restart_game(Config().general["d2r_path"], Config().advanced_options["launch_options"])
|
|
self.stop()
|
|
else:
|
|
Logger.info("Shut down botty")
|
|
safe_exit()
|
|
|
|
def current_game_length(self):
|
|
return self._game_stats.get_current_game_length()
|
|
|
|
def shuffle_runs(self):
|
|
tmp = list(self._do_runs.items())
|
|
random.shuffle(tmp)
|
|
#We clear() and update() to avoid assignment on _do_runs as this list is referenced
|
|
#by other classes and we want updates persisted everywhere.
|
|
self._do_runs.clear()
|
|
self._do_runs.update(OrderedDict(tmp))
|
|
|
|
def is_last_run(self):
|
|
found_unfinished_run = False
|
|
for key in self._do_runs:
|
|
if self._do_runs[key]:
|
|
found_unfinished_run = True
|
|
break
|
|
return not found_unfinished_run
|
|
|
|
def _rebuild_as_asset_to_trigger(trigger_to_assets: dict):
|
|
result = {}
|
|
for key in trigger_to_assets.keys():
|
|
for asset in trigger_to_assets[key]:
|
|
result[asset] = key
|
|
return result
|
|
|
|
def on_init(self):
|
|
self._game_stats.log_start_game()
|
|
keyboard.release(Config().char["stand_still"])
|
|
# Force D2R client area to stable position to prevent offset drift
|
|
from utils.misc import move_d2r_window
|
|
move_d2r_window(5, 98)
|
|
wait(0.3)
|
|
# If we're only doing run_level and character is already in-game, skip
|
|
# the town detection and go straight to the run
|
|
if list(self._do_runs.keys()) == ["run_level"]:
|
|
Logger.info("Level run only - skipping town detection, starting directly")
|
|
self._curr_loc = Location.A1_TOWN_START
|
|
self.trigger_or_stop("skip_to_level")
|
|
return
|
|
transition_to_screens = Bot._rebuild_as_asset_to_trigger({
|
|
"select_character": main_menu.MAIN_MENU_MARKERS,
|
|
"start_from_town": town_manager.TOWN_MARKERS,
|
|
})
|
|
if (match := template_finder.search_and_wait(list(transition_to_screens.keys()), best_match=True)).valid:
|
|
self.trigger_or_stop(transition_to_screens[match.name])
|
|
else:
|
|
self.restart_or_exit(f"Failed to detect {list(transition_to_screens.keys())}.")
|
|
|
|
def on_select_character(self):
|
|
# Make sure the correct char is selected
|
|
if not character_select.has_char_template_saved():
|
|
character_select.save_char_online_status()
|
|
character_select.save_char_template()
|
|
else:
|
|
if not character_select.select_char():
|
|
if Config().general["info_screenshots"]:
|
|
timestamp = time.strftime("%Y%m%d_%H%M%S")
|
|
safe_imwrite("./log/screenshots/info/info_failed_character_select_" + timestamp + ".png", grab())
|
|
if character_select.has_char_template_saved():
|
|
saved_char_img = character_select.get_saved_char_template()
|
|
safe_imwrite("./log/screenshots/info/info_failed_character_select_saved_template_" + timestamp + ".png", saved_char_img)
|
|
self._game_stats.set_failure_reason("Character selection failed at hero screen")
|
|
self.restart_or_exit()
|
|
self.trigger_or_stop("create_game")
|
|
|
|
def on_create_game(self):
|
|
# Start a game from hero selection
|
|
if (m := wait_until_visible(ScreenObjects.MainMenu)).valid:
|
|
if "DARK" in m.name:
|
|
keyboard.send("esc")
|
|
main_menu.start_game()
|
|
view.move_to_corpse()
|
|
else:
|
|
self._game_stats.set_failure_reason("Could not detect main menu to start game")
|
|
self.restart_or_exit()
|
|
self.trigger_or_stop("start_from_town")
|
|
|
|
def on_start_from_town(self):
|
|
self._curr_loc = self._town_manager.wait_for_town_spawn()
|
|
if not self._curr_loc:
|
|
Logger.warning("Could not detect town spawn — defaulting to A1_TOWN_START. Run will waypoint to correct act.")
|
|
self._curr_loc = Location.A1_TOWN_START
|
|
|
|
# Handle picking up corpse in case of death
|
|
if (corpse_present := is_visible(ScreenObjects.Corpse)):
|
|
self._previous_run_failed = True
|
|
view.pickup_corpse()
|
|
wait_until_hidden(ScreenObjects.Corpse)
|
|
belt.fill_up_belt_from_inventory(Config().char["num_loot_columns"])
|
|
self._char.discover_capabilities()
|
|
if corpse_present and self._char.capabilities.can_teleport_with_charges and not self._char.select_tp():
|
|
keybind = Config().char["teleport"]
|
|
Logger.info(f"Teleport keybind is lost upon death. Rebinding teleport to '{keybind}'")
|
|
self._char.remap_right_skill_hotkey("TELE_ACTIVE", Config().char["teleport"])
|
|
|
|
# Run /nopickup command to avoid picking up stuff on accident
|
|
if Config().char["enable_no_pickup"] and (not self._ran_no_pickup and not self._game_stats._nopickup_active):
|
|
self._ran_no_pickup = True
|
|
if view.enable_no_pickup():
|
|
self._game_stats._nopickup_active = True
|
|
Logger.info("Activated /nopickup")
|
|
else:
|
|
Logger.error("Failed to detect if /nopickup command was applied or not")
|
|
|
|
self._game_stats.log_exp()
|
|
|
|
self.trigger_or_stop("maintenance")
|
|
|
|
def on_maintenance(self):
|
|
# Defensive: ensure _curr_loc is valid before any town manager call
|
|
if not self._curr_loc:
|
|
Logger.warning("No current location set — defaulting to A1_TOWN_START")
|
|
self._curr_loc = Location.A1_TOWN_START
|
|
# Pause health manager if not already paused
|
|
set_pause_state(True)
|
|
|
|
# Dismiss skill/quest/help/stats icon if they are on screen
|
|
if not view.dismiss_skills_icon():
|
|
view.return_to_play()
|
|
|
|
# Top up health with belt potions before doing any town work.
|
|
# The health manager is paused in town so it won't auto-pot for us.
|
|
# We aim for ~95% HP so the character enters the next run at full health.
|
|
# Drinks health first (cheaper), then rejuv as fallback if health is unavailable.
|
|
# Happens before update_pot_needs so the reduced belt is accurately counted
|
|
# and the buy step will restock whatever was drunk.
|
|
_town_heal_threshold = 0.95
|
|
_town_heal_max_drinks = 5
|
|
for _drink_attempt in range(_town_heal_max_drinks):
|
|
_hp_img = grab()
|
|
_hp = meters.get_health(_hp_img)
|
|
if _hp >= _town_heal_threshold:
|
|
break
|
|
Logger.info(f"Town heal: HP at {_hp * 100:.0f}% — drinking potion to recover (attempt {_drink_attempt + 1}/{_town_heal_max_drinks})")
|
|
_mp = meters.get_mana(_hp_img)
|
|
_drank = belt.drink_potion("health", stats=[_hp, _mp])
|
|
if not _drank:
|
|
_drank = belt.drink_potion("rejuv", stats=[_hp, _mp])
|
|
if not _drank:
|
|
Logger.debug("Town heal: no health or rejuv potions in belt — cannot recover HP in town")
|
|
break
|
|
wait(2.0, 2.5) # wait for potion to tick (town regen is slow)
|
|
|
|
# Look at belt to figure out how many pots need to be picked up
|
|
belt.update_pot_needs()
|
|
|
|
# If we are doing trav_runs and juvs are full we will remove extra runs
|
|
if ("run_trav" == next(iter(self._do_runs))) and (consumables.get_needs("rejuv") <= 1):
|
|
while (len(self._do_runs) > 1):
|
|
skipped_run = self._do_runs.popitem()[0]
|
|
Logger.info(f"We are running trav and have full rejuvs so skipping {skipped_run}")
|
|
|
|
# Inspect inventory
|
|
items = None
|
|
need_inspect = self._picked_up_items or self._previous_run_failed
|
|
if Config().char["runs_per_stash"]:
|
|
need_inspect |= (self._game_stats._run_counter - 1) % Config().char["runs_per_stash"] == 0
|
|
if need_inspect:
|
|
img = personal.open_inventory()
|
|
# Update TP, ID, key needs
|
|
if self._game_stats._game_counter == 1:
|
|
self._use_id_tome = common.tome_state(img, 'id')[0] is not None
|
|
self._use_keys = is_visible(ScreenObjects.Key, img)
|
|
if (self._game_stats._run_counter - 1) % 4 == 0 or self._previous_run_failed:
|
|
personal.update_tome_key_needs(img, item_type = 'tp')
|
|
if self._use_id_tome:
|
|
personal.update_tome_key_needs(img, item_type = 'id')
|
|
if self._use_keys:
|
|
# if keys run out then refilling will be unreliable :(
|
|
self._use_keys = personal.update_tome_key_needs(img, item_type = 'key')
|
|
# Check inventory items
|
|
if personal.inventory_has_items(img):
|
|
Logger.debug("Inspecting inventory items")
|
|
items = personal.inspect_items(img, game_stats=self._game_stats, close_window=False)
|
|
common.close()
|
|
Logger.debug(f"Needs: {consumables.get_needs()}")
|
|
|
|
#Cast town buffs (ie burst of speed etc)
|
|
if not self._pre_buffed:
|
|
self._char.cast_town_buffs(self._curr_loc)
|
|
|
|
if items:
|
|
# if there are still items that need identifying, go to cain to identify them
|
|
if any([item.need_id for item in items]):
|
|
Logger.info("ID items at cain")
|
|
self._curr_loc = self._town_manager.identify(self._curr_loc)
|
|
if not self._curr_loc:
|
|
Logger.warning("Could not identify items (Cain not available). Continuing without ID.")
|
|
self._curr_loc = Location.A1_TOWN_START
|
|
else:
|
|
# recheck inventory
|
|
items = personal.inspect_items(game_stats=self._game_stats)
|
|
keep_items = any([item.keep for item in items]) if items else None
|
|
sell_items = any([item.sell for item in items]) if items else None
|
|
stash_gold = personal.get_inventory_gold_full()
|
|
|
|
# Check if should need some healing
|
|
img = grab()
|
|
need_refill = (
|
|
consumables.should_buy("health", min_needed = 3) or
|
|
consumables.should_buy("mana", min_needed = 3) or
|
|
(self._use_keys and consumables.should_buy("key", min_remaining = 4)) or
|
|
consumables.should_buy("tp", min_remaining = 8) or
|
|
consumables.should_buy("id", min_remaining = 3)
|
|
)
|
|
if need_refill or sell_items:
|
|
Logger.info("Buy consumables and/or sell items")
|
|
self._curr_loc, result_items = self._town_manager.buy_consumables(self._curr_loc, items = items)
|
|
if self._curr_loc:
|
|
items = result_items
|
|
sell_items = any([item.sell for item in items]) if items else None
|
|
Logger.debug(f"Needs: {consumables.get_needs()}")
|
|
else:
|
|
Logger.warning("Buy consumables failed, retrying")
|
|
wait(0.5, 0.6)
|
|
self._curr_loc, result_items = self._town_manager.buy_consumables(Location.A1_TOWN_START, items = items)
|
|
if self._curr_loc:
|
|
items = result_items
|
|
sell_items = any([item.sell for item in items]) if items else None
|
|
Logger.debug(f"Needs: {consumables.get_needs()}")
|
|
else:
|
|
Logger.warning("Buy consumables failed after retry, continuing without buying")
|
|
self._curr_loc = Location.A1_TOWN_START
|
|
elif meters.get_health(img) <= Config().char["take_rejuv_potion_health"] or meters.get_mana(img) <= Config().char["take_rejuv_potion_mana"]:
|
|
Logger.info("Healing at next possible Vendor")
|
|
self._curr_loc = self._town_manager.heal(self._curr_loc)
|
|
if not self._curr_loc:
|
|
Logger.warning("Heal failed, retrying")
|
|
wait(0.5, 0.6)
|
|
self._curr_loc = self._town_manager.heal(Location.A1_TOWN_START)
|
|
if not self._curr_loc:
|
|
Logger.warning("Heal failed after retry, continuing without heal")
|
|
self._curr_loc = Location.A1_TOWN_START
|
|
|
|
# After buying (or failing to buy), pull any pots already sitting in inventory into the belt.
|
|
# This covers the "out of gold" case where we couldn't buy new pots but have leftover ones.
|
|
if consumables.get_needs("health") > 0 or consumables.get_needs("mana") > 0 or consumables.get_needs("rejuv") > 0:
|
|
Logger.debug("Filling belt from inventory pots before stash")
|
|
belt.fill_up_belt_from_inventory(Config().char["num_loot_columns"])
|
|
belt.update_pot_needs()
|
|
|
|
# Stash stuff
|
|
if keep_items or stash_gold:
|
|
Logger.info("Stashing items")
|
|
prev_loc = self._curr_loc
|
|
self._curr_loc, result_items = self._town_manager.stash(self._curr_loc, items=items, game_stats=self._game_stats)
|
|
if not self._curr_loc:
|
|
Logger.warning("Stash failed, retrying from A5 town start")
|
|
wait(0.5, 0.6)
|
|
self._curr_loc, result_items = self._town_manager.stash(Location.A5_TOWN_START, items=items, game_stats=self._game_stats)
|
|
if not self._curr_loc:
|
|
Logger.warning("Stash failed after retry, skipping stash and keeping items")
|
|
self._curr_loc = prev_loc
|
|
return
|
|
sell_items = any([item.sell for item in result_items]) if result_items else None
|
|
#Acquire mutex to prevent controller from killing thread during transmutes
|
|
self._stash_mutex.acquire()
|
|
Logger.info("Running transmutes")
|
|
self._transmute.run_transmutes(force=False)
|
|
common.close()
|
|
self._stash_mutex.release()
|
|
self._picked_up_items = False
|
|
|
|
# Check if we are out of tps or need repairing
|
|
need_repair = is_visible(ScreenObjects.NeedRepair)
|
|
need_routine_repair = False if not Config().char["runs_per_repair"] else self._game_stats._run_counter % Config().char["runs_per_repair"] == 0
|
|
need_refill_teleport = self._char.capabilities.can_teleport_with_charges and (not self._char.select_tp() or self._char.is_low_on_teleport_charges())
|
|
if need_repair or need_routine_repair or need_refill_teleport or sell_items:
|
|
if need_repair:
|
|
Logger.info("Repair needed. Gear is about to break")
|
|
elif need_routine_repair:
|
|
Logger.info(f"Routine repair. Run count={self._game_stats._run_counter}, runs_per_repair={Config().char['runs_per_repair']}")
|
|
elif need_refill_teleport:
|
|
Logger.info("Teleport charges ran out. Need to repair")
|
|
elif sell_items:
|
|
Logger.info("Selling items at repair vendor")
|
|
self._curr_loc, result_items = self._town_manager.repair(self._curr_loc, items)
|
|
if not self._curr_loc:
|
|
Logger.warning("Repair failed, retrying from A4 town start")
|
|
wait(0.5, 0.6)
|
|
self._curr_loc, result_items = self._town_manager.repair(Location.A4_TOWN_START, items)
|
|
if self._curr_loc:
|
|
items = result_items
|
|
if not self._curr_loc:
|
|
# Keep maintenance best-effort for stability: avoid killing runs on flaky NPC/vendor detection.
|
|
if need_refill_teleport and self._char.capabilities.can_teleport_with_charges and not self._char.capabilities.can_teleport_natively:
|
|
Logger.warning("Repair failed and teleport charges are required, trying to buy TP from vendor")
|
|
consumables.set_needs("tp", 20)
|
|
self._curr_loc = Location.A5_TOWN_START
|
|
else:
|
|
Logger.warning("Repair/vendor interaction failed; skipping maintenance and continuing run.")
|
|
self._curr_loc = Location.A5_TOWN_START
|
|
|
|
# Check if merc needs to be revived
|
|
if Config().char["use_merc"]:
|
|
merc_visible = False
|
|
try:
|
|
merc_visible = is_visible(ScreenObjects.MercIcon)
|
|
except Exception:
|
|
Logger.debug("Merc icon detection error, skipping resurrect check")
|
|
merc_visible = True
|
|
# Confirm with 'O' key - opens merc panel if alive, does nothing if dead
|
|
if not merc_visible and not self._game_stats._merc_resurrect_failed:
|
|
Logger.debug("Merc icon not visible, confirming with merc panel check")
|
|
keyboard.send("o")
|
|
wait(0.3, 0.4)
|
|
merc_panel_open = is_visible(ScreenObjects.MercPanelText)
|
|
if merc_panel_open:
|
|
# Merc is alive - panel opened, close it
|
|
keyboard.send("o")
|
|
wait(0.2, 0.3)
|
|
merc_visible = True
|
|
if not merc_visible and not self._game_stats._merc_resurrect_failed:
|
|
Logger.info("Resurrect merc")
|
|
new_loc = self._town_manager.resurrect(self._curr_loc)
|
|
if new_loc is False:
|
|
# Retry once - sometimes resurrect fails due to timing
|
|
Logger.warning("Resurrect failed, retrying")
|
|
wait(0.5, 0.6)
|
|
new_loc = self._town_manager.resurrect(self._curr_loc)
|
|
if new_loc is False:
|
|
# Failed to resurrect (can't afford or other error) - don't log death, just continue
|
|
Logger.warning("Failed to resurrect merc after retry, continuing without merc")
|
|
self._game_stats._merc_resurrect_failed = True
|
|
elif new_loc is None:
|
|
Logger.warning("Resurrect returned None, continuing at current location")
|
|
pass
|
|
else:
|
|
self._game_stats.log_merc_death()
|
|
self._curr_loc = new_loc
|
|
|
|
# Gamble if needed
|
|
while vendor.get_gamble_status() and Config().char["gamble_items"]:
|
|
Logger.debug("Head to gamble")
|
|
self._curr_loc = self._town_manager.gamble(self._curr_loc)
|
|
if not self._curr_loc:
|
|
Logger.warning("Gamble failed, skipping gamble")
|
|
self._curr_loc = Location.A1_TOWN_START
|
|
break
|
|
items = vendor.gamble()
|
|
if items:
|
|
self._curr_loc, _ = self._town_manager.stash(self._curr_loc, items = items)
|
|
common.close()
|
|
if not self._curr_loc:
|
|
Logger.warning("Gamble stash failed")
|
|
self._curr_loc = Location.A1_TOWN_START
|
|
break
|
|
|
|
# Start a new run
|
|
started_run = False
|
|
self._previous_run_failed = False
|
|
for key in self._do_runs:
|
|
if self._do_runs[key]:
|
|
if should_skip_run():
|
|
self._do_runs[key] = False
|
|
continue
|
|
self.trigger_or_stop(key)
|
|
started_run = True
|
|
break
|
|
if not started_run:
|
|
self.trigger_or_stop("end_game")
|
|
|
|
def on_end_game(self, failed: bool = False):
|
|
if failed and not self._game_stats.get_failure_reason():
|
|
self._game_stats.set_failure_reason("Game ended without completing runs")
|
|
if Config().general["info_screenshots"] and failed:
|
|
safe_imwrite("./log/screenshots/info/info_failed_game_" + time.strftime("%Y%m%d_%H%M%S") + ".png", grab())
|
|
self._curr_loc = False
|
|
self._pre_buffed = False
|
|
if view.save_and_exit() == False:
|
|
Logger.error("Normal save_and_exit failed! Attempting fast_save_and_exit and resetting bot.")
|
|
img = grab()
|
|
if not self._game_stats.get_failure_reason():
|
|
self._game_stats.set_failure_reason("save_and_exit failed")
|
|
view.fast_save_and_exit()
|
|
self.stop()
|
|
if Config().general["info_screenshots"]:
|
|
self._last_chicken_screenshot = "./log/screenshots/info/info_failed_exit_" + time.strftime("%Y%m%d_%H%M%S") + ".png"
|
|
safe_imwrite(self._last_chicken_screenshot, img)
|
|
|
|
set_pause_state(True)
|
|
self._game_stats.log_end_game(failed=failed)
|
|
|
|
if Config().general["max_runtime_before_break_m"] and Config().general["break_length_m"]:
|
|
elapsed_time = time.time() - self._timer
|
|
Logger.debug(f'Session length = {math.ceil(elapsed_time/60)} minutes, max_runtime_before_break_m {Config().general["max_runtime_before_break_m"]}.')
|
|
|
|
if elapsed_time > (Config().general["max_runtime_before_break_m"]*60):
|
|
break_msg = f'Ran for {hms(elapsed_time)}, taking a break for {hms(Config().general["break_length_m"]*60)}.'
|
|
Logger.info(break_msg)
|
|
if self._messenger.enabled:
|
|
self._messenger.send_message(break_msg)
|
|
if not self._pausing:
|
|
self.toggle_pause()
|
|
|
|
wait(Config().general["break_length_m"]*60)
|
|
|
|
break_msg = f'Break over, will now run for {hms(Config().general["max_runtime_before_break_m"]*60)}.'
|
|
Logger.info(break_msg)
|
|
if self._messenger.enabled:
|
|
self._messenger.send_message(break_msg)
|
|
if self._pausing:
|
|
self.toggle_pause()
|
|
|
|
self._timer = time.time()
|
|
|
|
#We clear() and update() to avoid assignment on _do_runs as this list is referenced
|
|
#by other classes and we want updates persisted everywhere.
|
|
self._do_runs.clear()
|
|
self._do_runs.update(self._do_runs_reset)
|
|
if Config().general["randomize_runs"]:
|
|
self.shuffle_runs()
|
|
if Config().stealth.get("reshuffle_each_rotation"):
|
|
tmp = list(self._do_runs_reset.items())
|
|
random.shuffle(tmp)
|
|
self._do_runs_reset = OrderedDict(tmp)
|
|
self._do_runs.clear()
|
|
self._do_runs.update(self._do_runs_reset)
|
|
self.trigger_or_stop("init")
|
|
|
|
def on_end_run(self):
|
|
if not Config().char["pre_buff_every_run"]:
|
|
self._pre_buffed = True
|
|
# Try TP to town, with retries
|
|
success = self._char.tp_town()
|
|
if not success:
|
|
if not skills.has_tps():
|
|
consumables.set_needs("tp", 20)
|
|
Logger.warning("No TP charges left — ending game to restock on next game start")
|
|
set_pause_state(True)
|
|
self.trigger_or_stop("end_game")
|
|
return
|
|
# Retry TP once
|
|
Logger.warning("TP to town failed, retrying")
|
|
wait(0.5, 0.6)
|
|
success = self._char.tp_town()
|
|
if success:
|
|
tp_loc = self._curr_loc if self._curr_loc else Location.A5_TOWN_START
|
|
self._curr_loc = self._town_manager.wait_for_tp(tp_loc)
|
|
if self._curr_loc:
|
|
set_pause_state(True)
|
|
# Stealth: chance of unscheduled AFK break after returning to town
|
|
maybe_afk_break()
|
|
return self.trigger_or_stop("maintenance")
|
|
# wait_for_tp failed - try to recover
|
|
Logger.warning("wait_for_tp failed, attempting recovery")
|
|
set_pause_state(True)
|
|
self._curr_loc = Location.A5_TOWN_START
|
|
return self.trigger_or_stop("maintenance")
|
|
# TP failed twice
|
|
Logger.warning("TP to town failed twice, trying to walk back")
|
|
set_pause_state(True)
|
|
self._curr_loc = Location.A5_TOWN_START
|
|
self.trigger_or_stop("maintenance")
|
|
|
|
# All the runs go here
|
|
# ==================================
|
|
def _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)
|
|
self._curr_loc = run_obj.approach(self._curr_loc, *approach_args)
|
|
if not self._curr_loc:
|
|
# Approach failed — couldn't reach the boss
|
|
self._game_stats.set_failure_reason(f"Approach failed for {run_name}")
|
|
picked = None
|
|
self._game_stats.log_run_finished(run_name, True, picked)
|
|
self._ending_run_helper(res)
|
|
return
|
|
set_pause_state(False)
|
|
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)
|
|
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._game_stats.log_run_finished(run_name, not battle_succeeded, picked)
|
|
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,), ())
|
|
|
|
|
|
|