my-botty/tools/testbed.py

"""
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).
  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).
  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 _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 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,
        "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()