alexandClaude Sonnet 4.6
35fa4ba442
Resolved 7 conflicts by keeping both local (Logger.info diagnostics in buy_consumables and repair) and remote (set_panel_check_paused health-check suppression during vendor/repair panels). All paths now log + pause correctly. Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
408 lines
19 KiB
Python
408 lines
19 KiB
Python
from item import consumables
|
|
from transmute.potion_transmute import convert_rejuv_potions
|
|
from input_layer import keyboard
|
|
import template_finder
|
|
from config import Config
|
|
from pather import Location
|
|
from logger import Logger
|
|
from town import IAct, A1, A2, A3, A4, A5
|
|
from utils.misc import wait
|
|
from screen import grab
|
|
from ui import waypoint, view
|
|
from inventory import personal, vendor, common
|
|
|
|
TOWN_MARKERS = [
|
|
"A5_TOWN_0", "A5_TOWN_1",
|
|
"A4_TOWN_4", "A4_TOWN_5",
|
|
"A3_TOWN_0", "A3_TOWN_1",
|
|
"A2_TOWN_0", "A2_TOWN_1", "A2_TOWN_10",
|
|
"A1_TOWN_1", "A1_TOWN_3"
|
|
]
|
|
|
|
class TownManager:
|
|
|
|
def __init__(self, a1: A1, a2: A2, a3: A3, a4: A4, a5: A5):
|
|
self._acts: dict[Location, IAct] = {
|
|
Location.A1_TOWN_START: a1,
|
|
Location.A2_TOWN_START: a2,
|
|
Location.A3_TOWN_START: a3,
|
|
Location.A4_TOWN_START: a4,
|
|
Location.A5_TOWN_START: a5
|
|
}
|
|
|
|
@staticmethod
|
|
def get_act_from_location(loc: Location) -> Location:
|
|
location = None
|
|
if not isinstance(loc, str):
|
|
return None
|
|
if loc.upper().startswith("A5_"):
|
|
location = Location.A5_TOWN_START
|
|
elif loc.upper().startswith("A4_"):
|
|
location = Location.A4_TOWN_START
|
|
elif loc.upper().startswith("A3_"):
|
|
location = Location.A3_TOWN_START
|
|
elif loc.upper().startswith("A2_"):
|
|
location = Location.A2_TOWN_START
|
|
elif loc.upper().startswith("A1_"):
|
|
location = Location.A1_TOWN_START
|
|
return location
|
|
|
|
def wait_for_town_spawn(self, timeout: float = 30) -> Location:
|
|
"""Wait for the char to spawn in town after starting a new game
|
|
:param timeout: Optional float value for time out in seconds, defaults to None
|
|
:return: Location of the town (e.g. Location.A4_TOWN_START) or None if nothing was found within timeout time
|
|
"""
|
|
template_match = template_finder.search_and_wait(TOWN_MARKERS, best_match=True, timeout=timeout)
|
|
if template_match.valid:
|
|
return TownManager.get_act_from_location(template_match.name)
|
|
return None
|
|
|
|
def wait_for_tp(self, curr_loc: Location):
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None: return False
|
|
return self._acts[curr_act].wait_for_tp()
|
|
|
|
def open_wp(self, curr_loc: Location):
|
|
from health_manager import set_panel_check_paused
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None: return False
|
|
set_panel_check_paused(True)
|
|
res = self._acts[curr_act].open_wp(curr_loc)
|
|
if not res:
|
|
set_panel_check_paused(False)
|
|
return res
|
|
|
|
def go_to_act(self, act_idx: int, curr_loc: Location) -> Location | bool:
|
|
from health_manager import set_panel_check_paused
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None: return False
|
|
# check if we already are in the desired act
|
|
if act_idx == 1: act = Location.A1_TOWN_START
|
|
elif act_idx == 2: act = Location.A2_TOWN_START
|
|
elif act_idx == 3: act = Location.A3_TOWN_START
|
|
elif act_idx == 4: act = Location.A4_TOWN_START
|
|
elif act_idx == 5: act = Location.A5_TOWN_START
|
|
else:
|
|
Logger.error(f"Act {act_idx} is not supported")
|
|
return False
|
|
if curr_act == act:
|
|
return curr_loc
|
|
# if not, move to the desired act via waypoint
|
|
set_panel_check_paused(True)
|
|
if not self._acts[curr_act].open_wp(curr_loc):
|
|
set_panel_check_paused(False)
|
|
return False
|
|
waypoint.use_wp(act = act_idx, idx = 0)
|
|
set_panel_check_paused(False)
|
|
return self._acts[act].get_wp_location()
|
|
|
|
def heal(self, curr_loc: Location) -> Location | bool:
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None: return False
|
|
# check if we can heal in current act
|
|
if self._acts[curr_act].can_heal():
|
|
return self._acts[curr_act].heal(curr_loc)
|
|
Logger.warning(f"Could not heal in {curr_act}. Continue without healing")
|
|
return curr_loc
|
|
|
|
def buy_consumables(self, curr_loc: Location, items: list = None):
|
|
from health_manager import set_panel_check_paused
|
|
Logger.info(f"TownManager buy_consumables: starting from {curr_loc}")
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None:
|
|
Logger.error("TownManager buy_consumables: could not determine act from location")
|
|
return False, items
|
|
if self._acts[curr_act].can_buy_pots():
|
|
set_panel_check_paused(True)
|
|
Logger.info(f"TownManager buy_consumables: calling open_trade_menu")
|
|
new_loc = self._acts[curr_act].open_trade_menu(curr_loc)
|
|
Logger.info(f"TownManager buy_consumables: open_trade_menu returned {new_loc}")
|
|
if not (new_loc and common.wait_for_left_inventory()):
|
|
Logger.error("TownManager buy_consumables: left inventory not detected after opening trade menu")
|
|
set_panel_check_paused(False)
|
|
return False, items
|
|
img=grab()
|
|
def buy_best_available_potion(potion_templates: list[str], quantity: int, shift_click: bool) -> bool:
|
|
for idx, template_name in enumerate(potion_templates):
|
|
# Intermediate fallback checks should not spam "not found" errors.
|
|
log_missing = idx == (len(potion_templates) - 1)
|
|
if vendor.buy_item(template_name=template_name, quantity=quantity, shift_click=shift_click, img=img, log_missing=log_missing):
|
|
if template_name not in ["SUPER_HEALING_POTION", "SUPER_MANA_POTION"]:
|
|
Logger.info(f"buy_consumables: Falling back to {template_name.replace('_', ' ')}")
|
|
return True
|
|
return False
|
|
# Buy HP pots
|
|
if consumables.get_needs("health") > 0:
|
|
can_shift_click = not sum([ x > 0 for x in [consumables.get_needs("health"), consumables.get_needs("mana"), consumables.get_needs("rejuv")]]) > 1
|
|
if buy_best_available_potion(
|
|
["SUPER_HEALING_POTION", "GREATER_HEALING_POTION"],
|
|
quantity=consumables.get_needs("health"),
|
|
shift_click=can_shift_click,
|
|
):
|
|
consumables.set_needs("health", 0)
|
|
else:
|
|
Logger.error("buy_consumables: Error purchasing health potions")
|
|
# Buy mana pots
|
|
if consumables.get_needs("mana") > 0:
|
|
can_shift_click = not sum([ x > 0 for x in [consumables.get_needs("health"), consumables.get_needs("mana"), consumables.get_needs("rejuv")]]) > 1
|
|
if buy_best_available_potion(
|
|
["SUPER_MANA_POTION", "GREATER_MANA_POTION"],
|
|
quantity=consumables.get_needs("mana"),
|
|
shift_click=can_shift_click,
|
|
):
|
|
consumables.set_needs("mana", 0)
|
|
else:
|
|
Logger.error("buy_consumables: Error purchasing mana potions")
|
|
# Buy TP scrolls
|
|
if consumables.get_needs("tp") > 0:
|
|
if vendor.buy_item(template_name="INV_SCROLL_TP", shift_click = True, img=img):
|
|
consumables.set_needs("tp", 0)
|
|
else:
|
|
Logger.error("buy_consumables: Error purchasing teleport scrolls")
|
|
# Buy ID scrolls
|
|
if consumables.get_needs("id") > 0:
|
|
if vendor.buy_item(template_name="INV_SCROLL_ID", shift_click = True, img=img):
|
|
consumables.set_needs("id", 0)
|
|
else:
|
|
Logger.error("buy_consumables: Error purchasing ID scrolls")
|
|
# Buy keys
|
|
if consumables.get_needs("key") > 0:
|
|
if vendor.buy_item(template_name="INV_KEY", shift_click = True, img=img):
|
|
consumables.set_needs("key", 0)
|
|
else:
|
|
Logger.error("buy_consumables: Error purchasing keys")
|
|
# Sell items, if any
|
|
if items:
|
|
Logger.info(f"TownManager buy_consumables: selling {len(items)} items")
|
|
items = personal.transfer_items(items, action = "sell")
|
|
Logger.info(f"TownManager buy_consumables: transfer_items returned {len(items) if items else 0} remaining items")
|
|
Logger.info("TownManager buy_consumables: closing vendor")
|
|
common.close()
|
|
return new_loc, items
|
|
Logger.warning(f"Could not buy consumables in {curr_act}. Continue.")
|
|
return curr_loc, items
|
|
|
|
def resurrect(self, curr_loc: Location) -> Location | bool:
|
|
from health_manager import set_panel_check_paused
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None: return False
|
|
# check if we can resurrect in current act
|
|
if self._acts[curr_act].can_resurrect():
|
|
set_panel_check_paused(True)
|
|
res = self._acts[curr_act].resurrect(curr_loc)
|
|
if not res:
|
|
set_panel_check_paused(False)
|
|
return res
|
|
new_loc = self.go_to_act(4, curr_loc)
|
|
if not new_loc: return False
|
|
set_panel_check_paused(True)
|
|
res = self._acts[Location.A4_TOWN_START].resurrect(new_loc)
|
|
if not res:
|
|
set_panel_check_paused(False)
|
|
return res
|
|
|
|
def identify(self, curr_loc: Location) -> Location | bool:
|
|
from health_manager import set_panel_check_paused
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None: return False
|
|
# try current act first
|
|
if self._acts[curr_act].can_identify():
|
|
set_panel_check_paused(True)
|
|
success = self._acts[curr_act].identify(curr_loc)
|
|
if success:
|
|
wait(0.2)
|
|
# close cain dialog so inventory key is not blocked
|
|
keyboard.send("esc")
|
|
set_panel_check_paused(False)
|
|
return True
|
|
else:
|
|
set_panel_check_paused(False)
|
|
view.return_to_play()
|
|
Logger.warning(f"Could not identify in act {curr_act}, trying A5 (Cain)")
|
|
# fallback to A5 Cain
|
|
new_loc = self.go_to_act(5, curr_loc)
|
|
if not new_loc: return False
|
|
set_panel_check_paused(True)
|
|
success = self._acts[Location.A5_TOWN_START].identify(new_loc)
|
|
if success:
|
|
wait(0.2)
|
|
# close cain dialog so inventory key is not blocked
|
|
keyboard.send("esc")
|
|
set_panel_check_paused(False)
|
|
else:
|
|
set_panel_check_paused(False)
|
|
view.return_to_play()
|
|
Logger.warning("Cain not available in A5 either, skipping identification")
|
|
return success
|
|
|
|
def open_stash(self, curr_loc: Location) -> Location | bool:
|
|
from health_manager import set_panel_check_paused
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
new_loc = curr_loc
|
|
|
|
if not self._acts[curr_act].can_stash():
|
|
new_loc = self.go_to_act(5, curr_loc)
|
|
if not new_loc: return False
|
|
curr_act = Location.A5_TOWN_START
|
|
|
|
set_panel_check_paused(True)
|
|
new_loc = self._acts[curr_act].open_stash(new_loc)
|
|
if not new_loc:
|
|
set_panel_check_paused(False)
|
|
return False
|
|
return new_loc
|
|
|
|
def gamble(self, curr_loc: Location) -> Location | bool:
|
|
from health_manager import set_panel_check_paused
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None: return False
|
|
# check if we can Identify in current act
|
|
if self._acts[curr_act].can_gamble():
|
|
set_panel_check_paused(True)
|
|
res = self._acts[curr_act].gamble(curr_loc)
|
|
if not res:
|
|
set_panel_check_paused(False)
|
|
return res
|
|
new_loc = self.go_to_act(4, curr_loc)
|
|
if not new_loc: return False
|
|
set_panel_check_paused(True)
|
|
res = self._acts[Location.A4_TOWN_START].gamble(new_loc)
|
|
if not res:
|
|
set_panel_check_paused(False)
|
|
return res
|
|
|
|
def stash(self, curr_loc: Location, items: list = None, game_stats: "GameStats" = None):
|
|
from health_manager import set_panel_check_paused
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None: return False, False
|
|
# check if we can stash in current act
|
|
if self._acts[curr_act].can_stash():
|
|
set_panel_check_paused(True)
|
|
new_loc = self._acts[curr_act].open_stash(curr_loc)
|
|
if not new_loc:
|
|
set_panel_check_paused(False)
|
|
return False, False
|
|
wait(1.0)
|
|
items = personal.stash_all_items(items, game_stats=game_stats)
|
|
# Convert regular Rejuv -> Full Rejuv while stash is open
|
|
convert_rejuv_potions()
|
|
return new_loc, items
|
|
new_loc = self.go_to_act(5, curr_loc)
|
|
if not new_loc: return False, False
|
|
set_panel_check_paused(True)
|
|
new_loc = self._acts[Location.A5_TOWN_START].open_stash(new_loc)
|
|
if not new_loc:
|
|
set_panel_check_paused(False)
|
|
return False, False
|
|
wait(1.0)
|
|
items = personal.stash_all_items(items, game_stats=game_stats)
|
|
# Convert regular Rejuv -> Full Rejuv while stash is open
|
|
convert_rejuv_potions()
|
|
return new_loc, items
|
|
|
|
def repair(self, curr_loc: Location, items: list = None):
|
|
from health_manager import set_panel_check_paused
|
|
Logger.info(f"TownManager repair: starting from {curr_loc}")
|
|
curr_act = TownManager.get_act_from_location(curr_loc)
|
|
if curr_act is None:
|
|
Logger.error("TownManager repair: could not determine act from location")
|
|
return False, False
|
|
repair_preference = (Config().char.get("repair_npc") or "a4_halbu").lower()
|
|
# check if we can rapair in current act
|
|
if curr_act == Location.A5_TOWN_START and repair_preference in ["a4_halbu", "halbu", "act4"]:
|
|
Logger.info("Repair preference is Act 4 Halbu. Skipping Larzuk flow for stability.")
|
|
new_loc = self.go_to_act(4, curr_loc)
|
|
if not new_loc:
|
|
return False, False
|
|
Logger.info("TownManager repair: opening Halbu trade/repair menu")
|
|
set_panel_check_paused(True)
|
|
new_loc = self._acts[Location.A4_TOWN_START].open_trade_and_repair_menu(new_loc)
|
|
Logger.info(f"TownManager repair: open_trade_and_repair_menu returned {new_loc}")
|
|
if not new_loc:
|
|
set_panel_check_paused(False)
|
|
return False, False
|
|
if items:
|
|
Logger.info(f"TownManager repair: selling {len(items)} items before repair")
|
|
items = personal.transfer_items(items, "sell")
|
|
Logger.info("TownManager repair: calling vendor.repair()")
|
|
vendor.repair()
|
|
wait(0.1, 0.2)
|
|
common.close()
|
|
return new_loc, items
|
|
|
|
if self._acts[curr_act].can_trade_and_repair():
|
|
Logger.info(f"TownManager repair: opening trade/repair menu at {curr_act}")
|
|
set_panel_check_paused(True)
|
|
new_loc = self._acts[curr_act].open_trade_and_repair_menu(curr_loc)
|
|
Logger.info(f"TownManager repair: open_trade_and_repair_menu returned {new_loc}")
|
|
if new_loc:
|
|
if items:
|
|
Logger.info(f"TownManager repair: selling {len(items)} items before repair")
|
|
items = personal.transfer_items(items, "sell")
|
|
Logger.info("TownManager repair: calling vendor.repair()")
|
|
vendor.repair()
|
|
wait(0.1, 0.2)
|
|
common.close()
|
|
return new_loc, items
|
|
set_panel_check_paused(False)
|
|
if curr_act == Location.A5_TOWN_START:
|
|
Logger.warning("A5 repair failed, attempting Act 4 Halbu fallback")
|
|
new_loc = self.go_to_act(4, Location.A5_LARZUK)
|
|
if not new_loc:
|
|
return False, False
|
|
Logger.info("TownManager repair: fallback - opening Halbu trade/repair menu")
|
|
set_panel_check_paused(True)
|
|
new_loc = self._acts[Location.A4_TOWN_START].open_trade_and_repair_menu(new_loc)
|
|
Logger.info(f"TownManager repair: fallback open_trade_and_repair_menu returned {new_loc}")
|
|
if not new_loc:
|
|
set_panel_check_paused(False)
|
|
return False, False
|
|
if items:
|
|
Logger.info(f"TownManager repair: selling {len(items)} items before repair (fallback)")
|
|
items = personal.transfer_items(items, "sell")
|
|
Logger.info("TownManager repair: calling vendor.repair() (fallback)")
|
|
vendor.repair()
|
|
wait(0.1, 0.2)
|
|
common.close()
|
|
return new_loc, items
|
|
return False, False
|
|
Logger.info("TownManager repair: going to A5 for repair")
|
|
new_loc = self.go_to_act(5, curr_loc)
|
|
if not new_loc: return False, False
|
|
Logger.info("TownManager repair: opening Larzuk trade/repair menu")
|
|
set_panel_check_paused(True)
|
|
new_loc = self._acts[Location.A5_TOWN_START].open_trade_and_repair_menu(new_loc)
|
|
Logger.info(f"TownManager repair: open_trade_and_repair_menu returned {new_loc}")
|
|
if not new_loc:
|
|
set_panel_check_paused(False)
|
|
return False, False
|
|
if items:
|
|
Logger.info(f"TownManager repair: selling {len(items)} items before repair")
|
|
items = personal.transfer_items(items, "sell")
|
|
Logger.info("TownManager repair: calling vendor.repair()")
|
|
vendor.repair()
|
|
wait(0.1, 0.2)
|
|
common.close()
|
|
return new_loc, items
|
|
|
|
# Test: Move to desired location in d2r and run any town action you want to test from there
|
|
if __name__ == "__main__":
|
|
from input_layer import keyboard
|
|
import os
|
|
from screen import start_detecting_window
|
|
start_detecting_window()
|
|
keyboard.add_hotkey('f12', lambda: Logger.info('Force Exit (f12)') or os._exit(1))
|
|
print("Move to d2r window and press f11")
|
|
keyboard.wait("f11")
|
|
from char.paladin.hammerdin import Hammerdin
|
|
from pather import Pather
|
|
pather = Pather()
|
|
char = Hammerdin(Config().hammerdin, Config().char, pather)
|
|
a5 = A5(pather, char)
|
|
a4 = A4(pather, char)
|
|
a3 = A3(pather, char)
|
|
a2 = A2(pather, char)
|
|
a1 = A1(pather, char)
|
|
town_manager = TownManager(a1, a2, a3, a4, a5)
|
|
print(town_manager.identify(Location.A3_TOWN_START))
|