378 lines
16 KiB
Python
378 lines
16 KiB
Python
"""
|
|
Testbed: named in-game test runs for individual bot subsystems.
|
|
|
|
Each scenario assumes D2R is in the required state (noted below), runs ONE
|
|
subsystem in isolation, and reports — so features can be tested and improved
|
|
without farming runs around them.
|
|
|
|
<botty-env-python> tools/testbed.py <scenario>
|
|
|
|
Scenarios:
|
|
binds Verify + auto-set skill hotkeys from params/profile (in a game).
|
|
capture Capture skill-slot icon templates for the current binds (in a game).
|
|
audit Audit all stash pages against the current pickit; list FREE-able
|
|
items (in a game, stash OPEN).
|
|
gems Force-run the gem transmute routine: cube flawless -> perfect from
|
|
stash, restock per stash_destination (in a game, stash OPEN).
|
|
gems_all Convert gem tiers to perfect in one pass: chipped->flawed->
|
|
standard->flawless->perfect (in a game, stash OPEN).
|
|
Optional filters: a tier and/or gem type, e.g. "gems_all diamond"
|
|
or "gems_all flawless diamond".
|
|
For chipped-only conversion with before/after analysis, use
|
|
run_chipped_gems.bat instead.
|
|
stash Force-run the stashing routine: deposit gold + stash keep-flagged
|
|
inventory items, with page rotation (in a game, stash OPEN).
|
|
keyo Verify Controls-page key bindings (.keyo) match params (anywhere;
|
|
--fix requires D2R closed).
|
|
count Read gem counts from the GEMS stash tab via OCR and print the
|
|
transmute plan (in a game, stash OPEN on GEMS tab).
|
|
loot Unit-check pickup priority order + dive-mode flag (no game needed).
|
|
hermes Interactive LLM vision agent: grabs live D2R screen and sends it to
|
|
the local Qwen model for game-state analysis / feature development.
|
|
Optional: pass an initial prompt as extra args (no game state needed
|
|
for the LLM call itself, but D2R must be visible to grab).
|
|
|
|
State tip: to get 'stash OPEN with the bot paused', start the bot, and the
|
|
moment the log prints 'Stashing items', press F11 — or run a watcher like
|
|
log/_stash_pause_watcher.ps1.
|
|
"""
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
TOOLS = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.path.dirname(TOOLS)
|
|
PY = sys.executable
|
|
|
|
|
|
def _run_tool(script: str, *args) -> int:
|
|
return subprocess.call([PY, os.path.join(TOOLS, script), *args], cwd=ROOT)
|
|
|
|
|
|
def _patch_ssl():
|
|
# The conda env's python chokes on a malformed cert in the Windows store
|
|
# (ssl.SSLError: ASN1 NOT_ENOUGH_DATA) when aiohttp builds its default
|
|
# context at import time. Fall back to certifi's bundle.
|
|
import ssl
|
|
_orig = ssl.SSLContext.load_default_certs
|
|
|
|
def _safe(self, purpose=ssl.Purpose.CLIENT_AUTH):
|
|
try:
|
|
_orig(self, purpose)
|
|
except ssl.SSLError:
|
|
try:
|
|
import certifi
|
|
self.load_verify_locations(certifi.where())
|
|
except Exception:
|
|
pass
|
|
ssl.SSLContext.load_default_certs = _safe
|
|
|
|
|
|
def _gems() -> int:
|
|
sys.path.insert(0, os.path.join(ROOT, "src"))
|
|
_patch_ssl()
|
|
import ctypes
|
|
ctypes.windll.user32.SetProcessDPIAware()
|
|
import time
|
|
print("Stash must be OPEN in-game. Forcing gem transmute in 5s...")
|
|
time.sleep(5)
|
|
from screen import start_detecting_window, stop_detecting_window
|
|
from game_stats import GameStats
|
|
from transmute import Transmute
|
|
start_detecting_window()
|
|
try:
|
|
Transmute(GameStats()).run_transmutes(force=True)
|
|
finally:
|
|
stop_detecting_window()
|
|
return 0
|
|
|
|
|
|
def _gems_all() -> int:
|
|
sys.path.insert(0, os.path.join(ROOT, "src"))
|
|
_patch_ssl()
|
|
import ctypes
|
|
ctypes.windll.user32.SetProcessDPIAware()
|
|
import time
|
|
print("Converting ALL gems to perfect. Opening stash if needed in 3s...")
|
|
time.sleep(3)
|
|
from screen import start_detecting_window, stop_detecting_window, grab
|
|
from game_stats import GameStats
|
|
from transmute import Transmute
|
|
from ui_manager import is_visible, ScreenObjects
|
|
from inventory import common
|
|
start_detecting_window()
|
|
try:
|
|
# Fast-path: ESC any open panels, then try clicking the stash chest directly
|
|
from input_layer import keyboard
|
|
from utils.misc import wait
|
|
for _ in range(3):
|
|
keyboard.send("esc")
|
|
wait(0.3, 0.3)
|
|
img = grab()
|
|
stash_open = is_visible(ScreenObjects.GoldBtnStash, img) or common.left_inventory_ready(img)
|
|
if not stash_open:
|
|
import template_finder as _tf
|
|
_m = _tf.search(["A5_STASH", "A5_STASH_2"], grab(), threshold=0.45,
|
|
use_grayscale=True, best_match=True)
|
|
if _m.valid:
|
|
from input_layer import mouse
|
|
print(f"gems_all: clicking stash chest directly @ {_m.center_monitor}")
|
|
mouse.move(*_m.center_monitor)
|
|
wait(0.2, 0.2)
|
|
mouse.click("left")
|
|
wait(1.5, 1.5)
|
|
img = grab()
|
|
stash_open = is_visible(ScreenObjects.GoldBtnStash, img) or common.left_inventory_ready(img)
|
|
if not stash_open:
|
|
print("gems_all: stash not open — navigating to stash...")
|
|
import template_finder
|
|
from ui import main_menu
|
|
from config import Config
|
|
from pather import Pather
|
|
from item.pickit import PickIt
|
|
from town import TownManager, A1, A2, A3, A4, A5
|
|
pather = Pather()
|
|
if Config().char["type"] in ("hammerdin", "paladin"):
|
|
from char.paladin.hammerdin import Hammerdin
|
|
char = Hammerdin(Config().hammerdin, pather, PickIt())
|
|
else:
|
|
from char.basic import Basic
|
|
char = Basic(Config().basic, pather)
|
|
char.discover_capabilities()
|
|
tm = TownManager(A1(pather, char), A2(pather, char), A3(pather, char),
|
|
A4(pather, char), A5(pather, char))
|
|
if template_finder.search(main_menu.MAIN_MENU_MARKERS, grab(), best_match=True).valid:
|
|
print("gems_all: on main menu — creating a game first...")
|
|
if not main_menu.start_game():
|
|
print("gems_all: FAIL — could not create game")
|
|
return 1
|
|
act = tm.wait_for_town_spawn()
|
|
else:
|
|
act = tm.detect_current_act()
|
|
if act is None:
|
|
from pather import Location
|
|
act = Location.A5_TOWN_START
|
|
print("gems_all: act detection failed, assuming A5")
|
|
if not tm.open_stash(act):
|
|
print("gems_all: FAIL — could not open stash")
|
|
return 1
|
|
if not common.wait_for_left_inventory():
|
|
print("gems_all: FAIL — stash UI not ready")
|
|
return 1
|
|
tier_names = {"chipped", "flawed", "standard", "flawless"}
|
|
gem_names = {"topaz", "amethyst", "sapphire", "diamond", "ruby", "emerald", "skull"}
|
|
args = [arg.strip().lower() for arg in sys.argv[2:] if arg.strip()]
|
|
max_transmutes = None
|
|
if "--max" in args:
|
|
max_idx = args.index("--max")
|
|
try:
|
|
max_transmutes = int(args[max_idx + 1])
|
|
del args[max_idx:max_idx + 2]
|
|
except (IndexError, ValueError):
|
|
print("gems_all: --max requires an integer")
|
|
return 1
|
|
tiers = [arg for arg in args if arg in tier_names] or None
|
|
gems = [arg for arg in args if arg in gem_names] or None
|
|
unknown = [arg for arg in args if arg not in tier_names and arg not in gem_names]
|
|
if unknown:
|
|
print(f"gems_all: ignoring unknown filter(s): {', '.join(unknown)}")
|
|
Transmute(GameStats()).convert_all_gems_to_perfect(
|
|
tier_filter=tiers,
|
|
gem_filter=gems,
|
|
max_transmutes=max_transmutes,
|
|
)
|
|
print("gems_all: DONE")
|
|
# Leave stash open so re-runs can skip navigation
|
|
finally:
|
|
stop_detecting_window()
|
|
return 0
|
|
|
|
|
|
def _stash() -> int:
|
|
sys.path.insert(0, os.path.join(ROOT, "src"))
|
|
_patch_ssl()
|
|
import ctypes
|
|
ctypes.windll.user32.SetProcessDPIAware()
|
|
import time
|
|
print("Requires being in a game, in town. Running stash routine in 5s...")
|
|
time.sleep(5)
|
|
from screen import start_detecting_window, stop_detecting_window, grab
|
|
start_detecting_window()
|
|
try:
|
|
from config import Config
|
|
from inventory import personal, common
|
|
from inventory import stash as stash_state
|
|
from ui_manager import is_visible, ScreenObjects
|
|
|
|
img = grab()
|
|
stash_open = (is_visible(ScreenObjects.GoldBtnStash, img)
|
|
or is_visible(ScreenObjects.GoldBtnInventory, img))
|
|
if not stash_open:
|
|
print("stash: stash UI not open — walking to the stash chest...")
|
|
import template_finder
|
|
from ui import main_menu
|
|
from pather import Pather
|
|
from item.pickit import PickIt
|
|
from town import TownManager, A1, A2, A3, A4, A5
|
|
pather = Pather()
|
|
if Config().char["type"] in ("hammerdin", "paladin"):
|
|
from char.paladin.hammerdin import Hammerdin
|
|
char = Hammerdin(Config().hammerdin, pather, PickIt())
|
|
else:
|
|
from char.basic import Basic
|
|
char = Basic(Config().basic, pather)
|
|
char.discover_capabilities()
|
|
tm = TownManager(A1(pather, char), A2(pather, char), A3(pather, char),
|
|
A4(pather, char), A5(pather, char))
|
|
if template_finder.search(main_menu.MAIN_MENU_MARKERS, grab(), best_match=True).valid:
|
|
print("stash: on main menu — creating a game first...")
|
|
if not main_menu.start_game():
|
|
print("stash: FAIL — could not create a game from the main menu")
|
|
return 1
|
|
act = tm.wait_for_town_spawn()
|
|
else:
|
|
act = tm.detect_current_act()
|
|
if act is None:
|
|
print("stash: FAIL — no town marker found (is the char in town?)")
|
|
return 1
|
|
if not tm.open_stash(act):
|
|
print("stash: FAIL — could not open the stash chest")
|
|
return 1
|
|
if not common.wait_for_left_inventory():
|
|
print("stash: FAIL — stash UI not detected after opening")
|
|
return 1
|
|
# "all" widens the scan to all 10 inventory columns so existing items
|
|
# (charms etc. in cols 4-9) get inspected — used to exercise the
|
|
# keep-item transfer branch when the loot columns (0-3) are empty.
|
|
if "all" in sys.argv[2:]:
|
|
prev_cols = Config().char["num_loot_columns"]
|
|
Config().char["num_loot_columns"] = 10
|
|
print(f"stash: scanning all 10 columns (was {prev_cols}) to test keep-item path")
|
|
items = personal.inspect_items(grab(), close_window=False, ignore_sell=True) or []
|
|
keep = [i for i in items if i.keep]
|
|
print(f"stash: inventory has {len(items)} item(s), {len(keep)} flagged keep"
|
|
+ (f" ({', '.join(i.name for i in keep)})" if keep else ""))
|
|
pages = stash_state.get_curr_stash()
|
|
print(f"stash: starting pages — items={pages['items']}, gold={pages['gold']}")
|
|
personal.set_inventory_gold_full(True) # force the gold-deposit branch
|
|
left = personal.stash_all_items(items if keep else None)
|
|
leftover = [i for i in (left or []) if i.keep]
|
|
gold_done = is_visible(ScreenObjects.GoldNone)
|
|
print(f"stash: gold deposited: {'yes' if gold_done else 'NO — gold still in inventory'}")
|
|
if leftover:
|
|
if "all" in sys.argv[2:]:
|
|
# In `all` mode cols 4-9 include the charm/belt area. Ctrl+click on those
|
|
# positions is blocked by D2R (equipped-area guard) so the items can't be
|
|
# stashed — they're meant to STAY in inventory. The correct behavior after
|
|
# the Bug-19 fix is: detect the transfer failure, leave them in inventory,
|
|
# and never call stash_full(). That is a PASS, not a FAIL.
|
|
unstashable = [i for i in leftover if i.column is not None and i.column >= 4]
|
|
real_fail = [i for i in leftover if i.column is None or i.column < 4]
|
|
if unstashable:
|
|
print(f"stash: OK — {len(unstashable)} item(s) in equipped-area cols "
|
|
f"correctly left in inventory (not stash-full): "
|
|
f"{', '.join(i.name for i in unstashable)}")
|
|
if real_fail:
|
|
print(f"stash: FAIL — {len(real_fail)} keep item(s) in loot cols "
|
|
f"left in inventory: {', '.join(i.name for i in real_fail)}")
|
|
return 1
|
|
print("stash: PASS — gold branch ran; equipped-area items correctly retained")
|
|
return 0
|
|
print(f"stash: FAIL — {len(leftover)} keep item(s) left in inventory: "
|
|
f"{', '.join(i.name for i in leftover)}")
|
|
return 1
|
|
print("stash: PASS — gold branch ran and no keep items left in inventory")
|
|
return 0
|
|
finally:
|
|
stop_detecting_window()
|
|
|
|
|
|
def _loot() -> int:
|
|
sys.path.insert(0, os.path.join(ROOT, "src"))
|
|
_patch_ssl()
|
|
import time
|
|
import health_manager as hm
|
|
from item.pickit import PickIt
|
|
|
|
class FakeItem:
|
|
def __init__(self, name, dist):
|
|
self.Name, self.Distance = name, dist
|
|
|
|
items = [FakeItem("SUPER MANA POTION", 50), FakeItem("JAH RUNE", 400), FakeItem("700 GOLD", 20)]
|
|
ordered = sorted(items, key=lambda i: (0 if PickIt._is_high_value(i) else 1, i.Distance))
|
|
assert ordered[0].Name == "JAH RUNE", [i.Name for i in ordered]
|
|
hm.set_loot_priority(True, duration=0.2)
|
|
assert hm.loot_priority_active()
|
|
time.sleep(0.3)
|
|
assert not hm.loot_priority_active()
|
|
print("loot: PASS (Jah-first ordering, dive flag + expiry)")
|
|
return 0
|
|
|
|
|
|
def _count() -> int:
|
|
"""Read gem counts from GEMS tab via OCR (primary) and print plan."""
|
|
sys.path.insert(0, os.path.join(ROOT, "src"))
|
|
_patch_ssl()
|
|
import ctypes
|
|
ctypes.windll.user32.SetProcessDPIAware()
|
|
import time
|
|
print("Stash must be OPEN in-game. Reading gem counts in 3s...")
|
|
time.sleep(3)
|
|
from screen import start_detecting_window, stop_detecting_window, grab
|
|
from game_stats import GameStats
|
|
from transmute import Transmute
|
|
from inventory.common import left_inventory_ready
|
|
from ui_manager import is_visible, ScreenObjects
|
|
start_detecting_window()
|
|
try:
|
|
img = grab()
|
|
if not (is_visible(ScreenObjects.GoldBtnStash, img) or left_inventory_ready(img)):
|
|
print("count: stash not open")
|
|
return 1
|
|
t = Transmute(GameStats())
|
|
t._switch_to_gems_tab()
|
|
time.sleep(0.6)
|
|
|
|
print("\n--- OCR counts (primary) ---")
|
|
ocr = t._count_gems_by_ocr()
|
|
if ocr:
|
|
for k, v in sorted(ocr.items()):
|
|
display = str(v) if v != 999 else "999 (OCR failed)"
|
|
print(f" {k}: {display}")
|
|
else:
|
|
print(" (none)")
|
|
|
|
print("\n--- Plan (3+ of same type) ---")
|
|
plan = t._plan_transmutes(ocr)
|
|
if plan:
|
|
for n, gem, _, tier in plan:
|
|
print(f" {n}x {gem} [{tier}]")
|
|
else:
|
|
print(" nothing to transmute")
|
|
finally:
|
|
stop_detecting_window()
|
|
return 0
|
|
|
|
|
|
def main():
|
|
scenarios = {
|
|
"binds": lambda: _run_tool("set_binds_from_params.py"),
|
|
"capture": lambda: _run_tool("capture_skill_hotkeys.py"),
|
|
"audit": lambda: _run_tool("audit_stash_pickit.py"),
|
|
"keyo": lambda: _run_tool("set_controls_keyo.py", *sys.argv[2:]),
|
|
"gems": _gems,
|
|
"gems_all": _gems_all,
|
|
"count": _count,
|
|
"stash": _stash,
|
|
"loot": _loot,
|
|
"hermes": lambda: _run_tool("hermes.py", *sys.argv[2:]),
|
|
}
|
|
if len(sys.argv) < 2 or sys.argv[1] not in scenarios:
|
|
print(__doc__)
|
|
raise SystemExit(1)
|
|
raise SystemExit(scenarios[sys.argv[1]]())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|