Ported from an unpushed local main commit (477c693) that the PR #41 merge could not
include: the merge was built on origin/main, where these commits never landed.
Code only - the CLAUDE.md write-up is skipped, its bug numbers were renumbered
by the merge.
Co-Authored-By: Claude Opus 5 <[email protected]>
218 lines
8.5 KiB
Python
218 lines
8.5 KiB
Python
"""Inventory scanner — read the character's inventory and report what is in it.
|
|
|
|
Hovers every occupied slot, OCRs the item, runs the SAME pickit rules the bot
|
|
uses (bnip.actions.should_keep), and prints a grid plus a per-item verdict. Use
|
|
it to answer "why is my inventory full?" without guessing from the log.
|
|
|
|
Read-only: it moves the mouse to hover and opens the inventory panel. It never
|
|
clicks an item, so nothing is picked up, sold, dropped or stashed.
|
|
|
|
Run from the project root with D2R in a game (town is fine):
|
|
<botty-env-python> scripts/inventory_scan.py
|
|
<botty-env-python> scripts/inventory_scan.py --json log/inventory_scan.json
|
|
|
|
Verdict column:
|
|
KEEP a pickit rule matched -> will be stashed
|
|
ID needs identifying first
|
|
JUNK no rule matched -> will be sold, or stashed if protected
|
|
STUCK no rule matched AND protected from sale (charms/shields) -> it can
|
|
only leave via the stash; if that is not happening it accumulates
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
_script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
_root = os.path.dirname(_script_dir)
|
|
_src = os.path.join(_root, "src")
|
|
if _src not in sys.path:
|
|
sys.path.insert(0, _src)
|
|
os.chdir(_root)
|
|
|
|
import ctypes
|
|
ctypes.windll.user32.SetProcessDPIAware()
|
|
|
|
from config import Config
|
|
from screen import grab, convert_screen_to_monitor, start_detecting_window, stop_detecting_window
|
|
from ui_manager import is_visible, ScreenObjects
|
|
from input_layer import mouse
|
|
from inventory import common, personal
|
|
from d2r_image import processing as d2r_image
|
|
from bnip.actions import should_keep
|
|
from utils.misc import wait
|
|
|
|
ROWS = 4
|
|
COLS = 10
|
|
|
|
|
|
def _center_mouse():
|
|
mouse.move(*convert_screen_to_monitor((640, 360)), randomize=6, delay_factor=[0.2, 0.4])
|
|
wait(0.1, 0.2)
|
|
|
|
|
|
def _is_protected(name: str) -> str | None:
|
|
"""Mirror of personal.transfer_items._is_protected — why an item cannot be sold."""
|
|
n = (name or "").lower()
|
|
if "tome" in n or n.startswith("key of") or n == "key":
|
|
return "tome/key"
|
|
if "torch" in n:
|
|
return "torch"
|
|
if Config().char.get("protect_charms_from_sell", True) and "charm" in n:
|
|
return "charm"
|
|
if Config().char.get("protect_shields_from_sell", True) and "shield" in n:
|
|
return "shield"
|
|
for kw in str(Config().char.get("never_sell_keywords", "")).split(","):
|
|
if kw.strip() and kw.strip().lower() in n:
|
|
return f"keyword:{kw.strip()}"
|
|
return None
|
|
|
|
|
|
def scan() -> list[dict]:
|
|
img = personal.open_inventory()
|
|
if img is None:
|
|
print("Could not open the inventory panel.")
|
|
return []
|
|
|
|
found = []
|
|
for row in range(ROWS):
|
|
for col in range(COLS):
|
|
_, slot_img = common.get_slot_pos_and_img(img, col, row)
|
|
if not common.slot_has_item(slot_img):
|
|
continue
|
|
pos, _ = common.get_slot_pos_and_img(img, col, row)
|
|
x_m, y_m = convert_screen_to_monitor(pos)
|
|
|
|
_center_mouse()
|
|
mouse.move(x_m, y_m, randomize=4, delay_factor=[0.2, 0.3])
|
|
wait(0.25, 0.35)
|
|
hovered = grab(True)
|
|
try:
|
|
props, box = d2r_image.get_hovered_item(hovered)
|
|
except Exception as e:
|
|
found.append({"col": col, "row": row, "name": "<read failed>",
|
|
"verdict": "?", "detail": str(e)[:60], "text": ""})
|
|
continue
|
|
if box is None:
|
|
found.append({"col": col, "row": row, "name": "<no tooltip>",
|
|
"verdict": "?", "detail": "", "text": ""})
|
|
continue
|
|
|
|
text = (box.ocr_result.text or "").strip()
|
|
name = text.splitlines()[0] if text else "<unnamed>"
|
|
unidentified = is_visible(ScreenObjects.Unidentified, box.img)
|
|
try:
|
|
keep, expr = should_keep(props.as_dict())
|
|
except Exception as e:
|
|
keep, expr = False, f"rule error: {e}"
|
|
|
|
protected = _is_protected(name)
|
|
if keep:
|
|
verdict, detail = "KEEP", (expr or "").strip()
|
|
elif unidentified:
|
|
verdict, detail = "ID", "unidentified"
|
|
elif protected:
|
|
verdict, detail = "STUCK", f"no rule matched; protected ({protected})"
|
|
else:
|
|
verdict, detail = "JUNK", "no rule matched"
|
|
|
|
found.append({"col": col, "row": row, "name": name, "verdict": verdict,
|
|
"detail": detail, "text": text.replace("\n", " | "),
|
|
"slots": 1})
|
|
_center_mouse()
|
|
return _merge_multislot(found)
|
|
|
|
|
|
def _merge_multislot(found):
|
|
"""Collapse the slots of one multi-slot item into a single entry.
|
|
|
|
Armour is 2x3 and a Large Charm 1x2, so hovering every occupied slot reports
|
|
the same item once per slot it covers - a first scan listed one Breast Plate
|
|
three times. Slots that touch orthogonally and carry an identical tooltip are
|
|
the same item. Two identical items sitting adjacent would merge, which is why
|
|
`slots` is reported: a count that is not a plausible footprint is the tell.
|
|
"""
|
|
by_pos = {(i["col"], i["row"]): i for i in found}
|
|
seen, merged = set(), []
|
|
for item in found:
|
|
key = (item["col"], item["row"])
|
|
if key in seen:
|
|
continue
|
|
group, stack = [], [key]
|
|
while stack:
|
|
pos = stack.pop()
|
|
if pos in seen or pos not in by_pos or by_pos[pos]["text"] != item["text"]:
|
|
continue
|
|
seen.add(pos)
|
|
group.append(pos)
|
|
c, r = pos
|
|
stack += [(c + 1, r), (c - 1, r), (c, r + 1), (c, r - 1)]
|
|
top = min(group, key=lambda q: (q[1], q[0]))
|
|
entry = dict(by_pos[top])
|
|
entry["slots"] = len(group)
|
|
merged.append(entry)
|
|
return merged
|
|
|
|
|
|
def report(items: list[dict]) -> None:
|
|
occupied = {(i["col"], i["row"]): i for i in items}
|
|
total = ROWS * COLS
|
|
print()
|
|
used = sum(i.get("slots", 1) for i in items)
|
|
loot_cols = Config().char["num_loot_columns"]
|
|
print(f"Inventory: {used}/{total} slots used by {len(items)} items, {total - used} free")
|
|
print(f"Loot zone: columns 1-{loot_cols} Safe zone (bot never touches): {loot_cols + 1}-{COLS}")
|
|
print()
|
|
sym = {"KEEP": "K", "JUNK": "j", "STUCK": "X", "ID": "?", "?": "."}
|
|
print(" " + " ".join(f"{c:>2}" for c in range(COLS)))
|
|
for r in range(ROWS):
|
|
cells = []
|
|
for c in range(COLS):
|
|
it = occupied.get((c, r))
|
|
cells.append(f"{sym.get(it['verdict'], '.') if it else '.':>2}")
|
|
print(f" r{r} " + " ".join(cells))
|
|
print("\n K=keep j=junk(sellable) X=STUCK(junk but unsellable) ?=needs ID .=empty")
|
|
|
|
counts = {}
|
|
for i in items:
|
|
counts[i["verdict"]] = counts.get(i["verdict"], 0) + 1
|
|
print("\n " + " ".join(f"{k}={v}" for k, v in sorted(counts.items())))
|
|
|
|
print(f"\n{'slot':6} {'verdict':8} {'item':34} why")
|
|
print("-" * 100)
|
|
for i in sorted(items, key=lambda x: (x["verdict"] != "STUCK", x["row"], x["col"])):
|
|
zone = "loot" if i["col"] < Config().char["num_loot_columns"] else "SAFE"
|
|
print(f"c{i['col']+1}r{i['row']+1:<2} {i.get('slots',1):>2} {zone:5}{i['verdict']:8} "
|
|
f"{i['name'][:32]:32} {i['detail'][:40]}")
|
|
|
|
stuck = [i for i in items if i["verdict"] == "STUCK"]
|
|
if stuck:
|
|
print(f"\n{len(stuck)} STUCK item(s): no pickit rule matched, and they cannot be sold or")
|
|
print("dropped. They can only leave via the stash. If they are still here after a")
|
|
print("stash cycle, that is the bug — grep the log for 'marking for stash'.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--json", metavar="PATH", help="also write the result as JSON")
|
|
args = ap.parse_args()
|
|
|
|
start_detecting_window()
|
|
time.sleep(1.5)
|
|
if not is_visible(ScreenObjects.InGame, grab(force_new=True)):
|
|
print("Not in a game — enter a game (town is fine) and rerun. Nothing was touched.")
|
|
stop_detecting_window()
|
|
sys.exit(2)
|
|
try:
|
|
items = scan()
|
|
report(items)
|
|
if args.json:
|
|
os.makedirs(os.path.dirname(os.path.abspath(args.json)), exist_ok=True)
|
|
with open(args.json, "w", encoding="utf-8") as f:
|
|
json.dump(items, f, indent=2)
|
|
print(f"\nwrote {args.json}")
|
|
finally:
|
|
stop_detecting_window()
|