Gem transmute v3: OCR counting, GEMS-tab cube flow, cube safety helpers
New production flow for convert_all_gems_to_perfect: - Count gem stacks via OCR badge on each GEMS-tab slot (Hermes as fallback) - Open cube from PERSONAL stash per-transmute, return to GEMS tab while cube stays open - ctrl+shift+right-click x3 loads gems directly into cube from GEMS tab - ctrl+shift+left-click moves result from cube back to GEMS tab - Pre-flight: clear any leftover items in cube before planning - Stash reopen guard at every iteration in case ESC closed the stash - _ensure_cube_available / _locate_cube / _ensure_cube_in_stash safety chain - _empty_cube_to_gems_tab: blindly ctrl+shift+left-click all 12 slots to clear Also adds tier_filter, gem_filter, max_transmutes params to convert_all_gems_to_perfect and wires the new flow into run_transmutes when stash_tabs > 4. Includes 8 pytest unit tests covering OCR plan, gem/tier filters, cube-location logic and _open_available_cube_for_gems_tab edge cases. Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
alexpolo1
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
23f9488853
commit
4bb460db2a
+485
-100
@@ -231,6 +231,33 @@ class Transmute:
|
||||
Logger.error("Cube not found in PERSONAL stash tab")
|
||||
return False
|
||||
|
||||
def _open_cube_from_gems_tab(self) -> bool:
|
||||
"""Open the Horadric Cube from PERSONAL stash, then return to GEMS tab."""
|
||||
if not self._open_cube_from_stash():
|
||||
return False
|
||||
self._switch_to_gems_tab()
|
||||
wait(0.3, 0.4)
|
||||
return True
|
||||
|
||||
def _open_available_cube_for_gems_tab(self) -> bool:
|
||||
"""Open the cube from the PERSONAL stash, then return to GEMS tab."""
|
||||
if detect_screen_object(ScreenObjects.CubeOpened).valid:
|
||||
self._switch_to_gems_tab()
|
||||
wait(0.3, 0.4)
|
||||
return True
|
||||
self._switch_to_personal_tab()
|
||||
if detect_screen_object(ScreenObjects.CubeStash).valid:
|
||||
return self._open_cube_from_gems_tab()
|
||||
if detect_screen_object(ScreenObjects.CubeInventory).valid:
|
||||
Logger.error(
|
||||
"_open_available_cube_for_gems_tab: cube is in character inventory; "
|
||||
"move it to the PERSONAL stash tab before running gems_all"
|
||||
)
|
||||
return False
|
||||
Logger.error("_open_available_cube_for_gems_tab: cube not found in PERSONAL stash")
|
||||
return False
|
||||
|
||||
|
||||
def _open_cube_from_inventory(self) -> bool:
|
||||
if (match := wait_until_visible(ScreenObjects.CubeInventory, 3.0)).valid:
|
||||
mouse.move(*match.center_monitor)
|
||||
@@ -252,16 +279,31 @@ class Transmute:
|
||||
self._wait()
|
||||
|
||||
def _ctrl_shift_click_monitor(self, x: int, y: int, label: str = "") -> None:
|
||||
"""Ctrl+Shift+Click — moves item from current stash tab into the open cube."""
|
||||
Logger.info(f" ctrl+shift+click {label} @ monitor ({x},{y})")
|
||||
"""Ctrl+Shift+Right-Click — moves item from current stash tab into the Horadric Cube."""
|
||||
from input_layer.win_input import key_down, key_up, _get_vk
|
||||
Logger.info(f" ctrl+shift+rclick {label} @ monitor ({x},{y})")
|
||||
mouse.move(x, y)
|
||||
self._wait()
|
||||
keyboard.send('shift', do_release=False)
|
||||
keyboard.send('ctrl', do_release=False)
|
||||
key_down(_get_vk('ctrl'))
|
||||
key_down(_get_vk('shift'))
|
||||
self._wait()
|
||||
mouse.click("right")
|
||||
key_up(_get_vk('shift'))
|
||||
key_up(_get_vk('ctrl'))
|
||||
self._wait()
|
||||
|
||||
def _ctrl_shift_left_click_monitor(self, x: int, y: int, label: str = "") -> None:
|
||||
"""Ctrl+Shift+Left-Click — moves item from Horadric Cube back to current stash tab."""
|
||||
from input_layer.win_input import key_down, key_up, _get_vk
|
||||
Logger.info(f" ctrl+shift+lclick {label} @ monitor ({x},{y})")
|
||||
mouse.move(x, y)
|
||||
self._wait()
|
||||
key_down(_get_vk('ctrl'))
|
||||
key_down(_get_vk('shift'))
|
||||
self._wait()
|
||||
mouse.click("left")
|
||||
keyboard.release('ctrl')
|
||||
keyboard.release('shift')
|
||||
key_up(_get_vk('shift'))
|
||||
key_up(_get_vk('ctrl'))
|
||||
self._wait()
|
||||
|
||||
def _gems_tab_pull_3(self, gems_in: list) -> int:
|
||||
@@ -292,6 +334,95 @@ class Transmute:
|
||||
Logger.warning(" result gem not found in cube after transmute")
|
||||
return False
|
||||
|
||||
def _locate_cube(self) -> str | None:
|
||||
"""Return 'stash' or 'inventory' depending on where the cube is, or None if not found."""
|
||||
self._switch_to_personal_tab()
|
||||
if detect_screen_object(ScreenObjects.CubeStash).valid:
|
||||
return 'stash'
|
||||
if detect_screen_object(ScreenObjects.CubeInventory).valid:
|
||||
return 'inventory'
|
||||
Logger.error("_locate_cube: Horadric Cube not found in stash or character inventory")
|
||||
return None
|
||||
|
||||
def _ensure_cube_in_stash(self) -> bool:
|
||||
"""Move the Horadric Cube to the stash PERSONAL tab if it's currently in character inventory.
|
||||
Only works if the cube is in the loot columns (0 to num_loot_columns-1); cols 4+ are
|
||||
restricted by the equipped-area safety guard and require manual relocation to the stash."""
|
||||
self._switch_to_personal_tab()
|
||||
if detect_screen_object(ScreenObjects.CubeStash).valid:
|
||||
Logger.info(" cube already in stash PERSONAL tab")
|
||||
return True
|
||||
match = detect_screen_object(ScreenObjects.CubeInventory)
|
||||
if match.valid:
|
||||
from screen import convert_monitor_to_screen
|
||||
screen_x = convert_monitor_to_screen(match.center_monitor)[0]
|
||||
loot_cols = Config().char["num_loot_columns"]
|
||||
slot_w = Config().ui_pos["slot_width"]
|
||||
inv_x0 = Config().ui_roi["right_inventory"][0]
|
||||
loot_end_x = inv_x0 + loot_cols * slot_w
|
||||
if screen_x < loot_end_x:
|
||||
Logger.info(" moving cube from inventory loot cols to stash PERSONAL tab")
|
||||
self._ctrl_click_monitor(*match.center_monitor, label="cube")
|
||||
wait(0.5, 0.6)
|
||||
if detect_screen_object(ScreenObjects.CubeStash).valid:
|
||||
return True
|
||||
Logger.error("_ensure_cube_in_stash: ctrl+click did not land cube in stash")
|
||||
return False
|
||||
else:
|
||||
Logger.error(
|
||||
"_ensure_cube_in_stash: Horadric Cube is in restricted inventory cols "
|
||||
f"(screen_x={screen_x}, loot ends at {loot_end_x}). "
|
||||
"Move the Cube to the stash PERSONAL tab manually before running gems_all."
|
||||
)
|
||||
return False
|
||||
Logger.error("_ensure_cube_in_stash: cube not found in stash or character inventory")
|
||||
return False
|
||||
|
||||
def _ensure_cube_available(self) -> bool:
|
||||
"""Verify the cube is in PERSONAL stash for the GEMS-tab flow."""
|
||||
if detect_screen_object(ScreenObjects.CubeOpened).valid:
|
||||
Logger.info(" cube UI already open")
|
||||
return True
|
||||
location = self._locate_cube()
|
||||
if location == "stash":
|
||||
Logger.info(" cube available in stash")
|
||||
return True
|
||||
if location == "inventory":
|
||||
Logger.error(
|
||||
" cube is in character inventory; move it to the PERSONAL stash tab "
|
||||
"before running gems_all"
|
||||
)
|
||||
return False
|
||||
return False
|
||||
|
||||
def _reopen_stash(self) -> bool:
|
||||
"""Click the A5 stash chest to reopen the stash after cube was closed."""
|
||||
from inventory.common import left_inventory_ready
|
||||
from ui_manager import is_visible, ScreenObjects as SO
|
||||
|
||||
def _stash_open():
|
||||
img = grab()
|
||||
return (is_visible(SO.GoldBtnInventory, img)
|
||||
or is_visible(SO.GoldBtnStash, img)
|
||||
or left_inventory_ready(img))
|
||||
|
||||
for threshold in (0.60, 0.45):
|
||||
img = grab()
|
||||
match = template_finder.search(
|
||||
["A5_STASH", "A5_STASH_2"], img,
|
||||
threshold=threshold, use_grayscale=True, best_match=True,
|
||||
)
|
||||
if match.valid:
|
||||
mouse.move(*match.center_monitor)
|
||||
self._wait()
|
||||
mouse.click("left")
|
||||
for _ in range(20):
|
||||
wait(0.2, 0.2)
|
||||
if _stash_open():
|
||||
return True
|
||||
Logger.error("_reopen_stash: A5_STASH template not found")
|
||||
return False
|
||||
|
||||
def _run_gem_transmutes_new_ui(self, gems_in: list, gems_out: list, tier_name: str) -> None:
|
||||
"""Transmute one tier of gems using the new named-tab stash UI.
|
||||
|
||||
@@ -393,57 +524,145 @@ class Transmute:
|
||||
"DIAMOND": "diamond", "RUBY": "ruby", "EMERALD": "emerald", "SKULL": "skull",
|
||||
}
|
||||
|
||||
def _count_gems_with_hermes(self) -> dict:
|
||||
"""Screenshot the GEMS tab, ask Hermes to count each gem type/tier.
|
||||
Returns dict keyed like 'topaz_flawless': N.
|
||||
Falls back to template search_all counts if LLM is unreachable."""
|
||||
try:
|
||||
from PIL import Image
|
||||
img_np = grab(True)
|
||||
img_pil = Image.fromarray(img_np[..., ::-1])
|
||||
buf = io.BytesIO()
|
||||
img_pil.save(buf, format="JPEG", quality=85)
|
||||
b64 = base64.b64encode(buf.getvalue()).decode()
|
||||
_HERMES_PROMPT = (
|
||||
"You are looking at the GEMS tab of a Diablo II Resurrected stash. "
|
||||
"The grid has 7 columns (gem types) and 5 rows (tiers). "
|
||||
"Scan left to right, top to bottom. Every cell has a stacked gem icon with a count number. "
|
||||
"There are exactly 35 slots total: "
|
||||
"columns topaz, amethyst, sapphire, diamond, ruby, emerald, skull; "
|
||||
"rows chipped, flawed, normal, flawless, perfect. "
|
||||
"Read the number on EVERY slot — do not skip any. "
|
||||
"Reply with ONLY a JSON object, no explanation, no code fences. "
|
||||
"Keys: <type>_<tier> e.g. ruby_flawless, diamond_perfect. "
|
||||
"Values: integer counts. "
|
||||
"Example: {\"topaz_chipped\": 7, \"diamond_perfect\": 13, \"skull_flawless\": 6}"
|
||||
)
|
||||
|
||||
prompt = (
|
||||
"You are looking at the GEMS tab of a Diablo II Resurrected stash. "
|
||||
"Each slot shows a stacked gem icon with a count number. "
|
||||
"Count every gem type and tier visible. "
|
||||
"Reply with ONLY a JSON object, no explanation. Keys are like 'ruby_flawless', "
|
||||
"'topaz_chipped', 'emerald_perfect' etc. Values are integer counts. "
|
||||
"Tiers: chipped, flawed, normal, flawless, perfect. "
|
||||
"Types: topaz, amethyst, sapphire, diamond, ruby, emerald, skull. "
|
||||
"Example: {\"ruby_flawless\": 9, \"topaz_chipped\": 3}"
|
||||
)
|
||||
payload = json.dumps({
|
||||
"model": self._HERMES_MODEL,
|
||||
"max_tokens": 300,
|
||||
"temperature": 0.1,
|
||||
"repetition_penalty": 1.1,
|
||||
"chat_template_kwargs": {"enable_thinking": False},
|
||||
"messages": [{"role": "user", "content": [
|
||||
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
|
||||
{"type": "text", "text": prompt},
|
||||
]}],
|
||||
}).encode()
|
||||
req = urllib.request.Request(
|
||||
self._HERMES_URL, data=payload,
|
||||
headers={"Content-Type": "application/json"}, method="POST"
|
||||
)
|
||||
resp = urllib.request.urlopen(req, timeout=30)
|
||||
text = json.loads(resp.read())["choices"][0]["message"]["content"].strip()
|
||||
# Strip markdown code fences if present
|
||||
if text.startswith("```"):
|
||||
text = text.split("```")[1]
|
||||
if text.startswith("json"):
|
||||
text = text[4:]
|
||||
def _hermes_parse(self, text: str) -> dict:
|
||||
text = text.strip()
|
||||
start = text.find("{")
|
||||
end = text.rfind("}") + 1
|
||||
if start != -1 and end > start:
|
||||
text = text[start:end]
|
||||
try:
|
||||
counts = json.loads(text)
|
||||
Logger.info(f"Hermes gem counts: {counts}")
|
||||
return {k.lower(): int(v) for k, v in counts.items()}
|
||||
except json.JSONDecodeError:
|
||||
import ast
|
||||
counts = ast.literal_eval(text)
|
||||
return {k.lower(): int(v) for k, v in counts.items()}
|
||||
|
||||
def _hermes_call(self, prompt: str, b64: str) -> dict:
|
||||
"""Send one image+prompt to Hermes and return parsed count dict."""
|
||||
payload = json.dumps({
|
||||
"model": self._HERMES_MODEL,
|
||||
"max_tokens": 600,
|
||||
"temperature": 0.1,
|
||||
"repetition_penalty": 1.1,
|
||||
"chat_template_kwargs": {"enable_thinking": False},
|
||||
"messages": [{"role": "user", "content": [
|
||||
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
|
||||
{"type": "text", "text": prompt},
|
||||
]}],
|
||||
}).encode()
|
||||
req = urllib.request.Request(
|
||||
self._HERMES_URL, data=payload,
|
||||
headers={"Content-Type": "application/json"}, method="POST"
|
||||
)
|
||||
resp = urllib.request.urlopen(req, timeout=30)
|
||||
text = json.loads(resp.read())["choices"][0]["message"]["content"]
|
||||
return self._hermes_parse(text)
|
||||
|
||||
def _hermes_single_call(self) -> dict:
|
||||
"""Grab screen, send to Hermes with standard count prompt."""
|
||||
from PIL import Image
|
||||
img_np = grab(True)
|
||||
img_pil = Image.fromarray(img_np[..., ::-1])
|
||||
buf = io.BytesIO()
|
||||
img_pil.save(buf, format="JPEG", quality=85)
|
||||
b64 = base64.b64encode(buf.getvalue()).decode()
|
||||
return self._hermes_call(self._HERMES_PROMPT, b64)
|
||||
|
||||
|
||||
def _count_gems_by_ocr(self) -> dict:
|
||||
"""Count gem stacks in the GEMS tab using template matching + OCR of the count badge.
|
||||
Each stacked slot has a small digit in the bottom-right corner of the icon.
|
||||
Returns dict keyed like 'inventory_ruby_flawless': N (or 999 if OCR failed)."""
|
||||
import re as _re
|
||||
from d2r_image import ocr as d2r_ocr
|
||||
img = grab()
|
||||
roi = Config().ui_roi["left_inventory"]
|
||||
slot_w = Config().ui_pos["slot_width"]
|
||||
slot_h = Config().ui_pos["slot_height"]
|
||||
img_h, img_w = img.shape[:2]
|
||||
all_gems = CHIPPED_GEMS + FLAWED_GEMS + STANDARD_GEMS + FLAWLESS_GEMS + PERFECT_GEMS
|
||||
counts = {}
|
||||
for gem in all_gems:
|
||||
matches = template_finder.search_all(gem, img, threshold=0.60, roi=roi)
|
||||
if not matches:
|
||||
continue
|
||||
best = max(matches, key=lambda m: m.score)
|
||||
cx, cy = int(best.center[0]), int(best.center[1])
|
||||
# Count badge sits in the bottom-right quadrant of the slot.
|
||||
x1 = max(0, cx)
|
||||
y1 = max(0, cy)
|
||||
x2 = min(img_w, cx + slot_w // 2)
|
||||
y2 = min(img_h, cy + slot_h // 2)
|
||||
crop = img[y1:y2, x1:x2]
|
||||
if crop.size == 0:
|
||||
counts[gem.lower()] = 999
|
||||
continue
|
||||
# Scale up 4x so tesseract can read the small badge text
|
||||
crop_up = cv2.resize(crop, None, fx=4, fy=4, interpolation=cv2.INTER_CUBIC)
|
||||
text = "?"
|
||||
n = 999
|
||||
try:
|
||||
ocr_result = d2r_ocr.image_to_text(
|
||||
[crop_up],
|
||||
digits_only=True, psm=8,
|
||||
crop_pad=False, invert=True, threshold=30,
|
||||
fix_regexps=False, check_known_errors=False, correct_words=False,
|
||||
)
|
||||
text = ocr_result[0].text.strip().replace(" ", "")
|
||||
# Accept clean digit strings; also extract first digit run from noisy output
|
||||
m = _re.search(r'\d+', text)
|
||||
if m:
|
||||
val = int(m.group())
|
||||
n = val if val > 0 else 999
|
||||
except Exception:
|
||||
pass
|
||||
counts[gem.lower()] = n
|
||||
Logger.debug(f" OCR {gem}: '{text}' -> {n}")
|
||||
Logger.info(f"OCR gem counts ({len(counts)} types): {counts}")
|
||||
return counts
|
||||
|
||||
def _count_gems_with_hermes(self) -> dict:
|
||||
"""Single Hermes call to count all 35 gem stacks on the GEMS tab.
|
||||
Used for diagnostics/comparison; production flow uses _count_gems_by_ocr."""
|
||||
try:
|
||||
result = self._hermes_single_call()
|
||||
Logger.info(f"Hermes counts ({len(result)} types): {result}")
|
||||
return result
|
||||
except Exception as e:
|
||||
Logger.warning(f"Hermes gem count failed ({e}), falling back to template scan")
|
||||
return self._count_gems_by_template()
|
||||
|
||||
def _count_gems_for_plan(self) -> dict:
|
||||
"""Read gem counts, preferring OCR but falling back when OCR cannot read badge digits."""
|
||||
counts = self._count_gems_by_ocr()
|
||||
if not counts:
|
||||
return counts
|
||||
|
||||
unknown = sum(1 for n in counts.values() if n == 999)
|
||||
if unknown >= max(4, len(counts) // 2):
|
||||
Logger.warning(
|
||||
f"OCR failed to read {unknown}/{len(counts)} gem counts; trying Hermes vision fallback"
|
||||
)
|
||||
hermes_counts = self._count_gems_with_hermes()
|
||||
if hermes_counts and any(n != 999 for n in hermes_counts.values()):
|
||||
return hermes_counts
|
||||
Logger.warning("Hermes fallback did not return usable numeric counts; continuing with OCR/template depletion mode")
|
||||
return counts
|
||||
|
||||
def _count_gems_by_template(self) -> dict:
|
||||
"""Fallback: detect which gem types exist in the GEMS tab via template search_all.
|
||||
Stacked slots show one icon per type, so len(matches) = number of stack slots, not gem count.
|
||||
@@ -460,7 +679,12 @@ class Transmute:
|
||||
return counts
|
||||
|
||||
def _plan_transmutes(self, counts: dict) -> list:
|
||||
"""Given gem counts, return list of (n_transmutes, gem_template, gems_out_list) to execute."""
|
||||
"""Given gem counts, return list of (n_transmutes, gem_template, matching_gem_out, tier) to execute.
|
||||
|
||||
A count of 999 means OCR saw the gem stack but could not read the
|
||||
number. The execution loop will attempt that stack until the transmute
|
||||
button no longer appears, then stop that gem type.
|
||||
"""
|
||||
tier_map = [
|
||||
("chipped", CHIPPED_GEMS, FLAWED_GEMS),
|
||||
("flawed", FLAWED_GEMS, STANDARD_GEMS),
|
||||
@@ -469,90 +693,235 @@ class Transmute:
|
||||
]
|
||||
plan = []
|
||||
for tier_name, gems_in, gems_out in tier_map:
|
||||
for gem in gems_in:
|
||||
for gem, gem_out in zip(gems_in, gems_out):
|
||||
full_key = gem.lower() # "inventory_ruby_flawless"
|
||||
short_key = full_key.replace("inventory_", "") # "ruby_flawless" (Hermes style)
|
||||
n = counts.get(full_key, counts.get(short_key, 0))
|
||||
if tier_name == "standard" and n == 0:
|
||||
n = counts.get(full_key.replace("_standard", "_normal"), counts.get(short_key.replace("_standard", "_normal"), 0))
|
||||
if n >= 3:
|
||||
plan.append((n // 3, gem, gems_out, tier_name))
|
||||
Logger.info(f" plan: {n // 3}x {gem} ({tier_name} up)")
|
||||
n_transmutes = 999 if n == 999 else n // 3
|
||||
plan.append((n_transmutes, gem, [gem_out], tier_name))
|
||||
planned = "until depleted" if n_transmutes == 999 else f"{n_transmutes}x"
|
||||
Logger.info(f" plan: {planned} {gem} -> {gem_out} ({tier_name} up)")
|
||||
return plan
|
||||
|
||||
def _filter_transmute_plan(
|
||||
self,
|
||||
plan: list,
|
||||
tier_filter: list[str] | None = None,
|
||||
gem_filter: list[str] | None = None,
|
||||
) -> list:
|
||||
if tier_filter:
|
||||
tier_filter = [tier.strip().lower() for tier in tier_filter if tier and tier.strip()]
|
||||
plan = [entry for entry in plan if entry[3] in tier_filter]
|
||||
if gem_filter:
|
||||
gem_filter = [gem.strip().lower() for gem in gem_filter if gem and gem.strip()]
|
||||
plan = [
|
||||
entry for entry in plan
|
||||
if any(f"_{gem.upper()}_" in entry[1] for gem in gem_filter)
|
||||
]
|
||||
return plan
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def convert_all_gems_to_perfect(self) -> None:
|
||||
"""Screenshot GEMS tab, plan transmutes via Hermes vision, then execute."""
|
||||
def _cube_slot_monitors(self) -> list:
|
||||
"""Return monitor-coords for the 12 cube slots (3 cols × 4 rows) in cube_area_roi."""
|
||||
roi = Config().ui_roi["cube_area_roi"] # screen x,y,w,h
|
||||
slot_w = roi[2] / 3
|
||||
slot_h = roi[3] / 4
|
||||
from screen import convert_screen_to_monitor
|
||||
slots = []
|
||||
for row in range(4):
|
||||
for col in range(3):
|
||||
sx = roi[0] + col * slot_w + slot_w / 2
|
||||
sy = roi[1] + row * slot_h + slot_h / 2
|
||||
slots.append(convert_screen_to_monitor((int(sx), int(sy))))
|
||||
return slots
|
||||
|
||||
def _empty_cube_to_gems_tab(self) -> int:
|
||||
"""ctrl+shift+left-click all 12 cube slots → any items land on the current stash tab.
|
||||
Returns number of slots that had something (any item that moved makes a sound but we
|
||||
can't detect that here — we just blindly click all 12 and let D2R sort it out)."""
|
||||
slots = self._cube_slot_monitors()
|
||||
moved = 0
|
||||
for mx, my in slots:
|
||||
self._ctrl_shift_left_click_monitor(mx, my, label="cube-slot")
|
||||
wait(0.12, 0.15)
|
||||
moved += 1
|
||||
return moved
|
||||
|
||||
def convert_all_gems_to_perfect(
|
||||
self,
|
||||
tier_filter: list[str] | None = None,
|
||||
gem_filter: list[str] | None = None,
|
||||
max_transmutes: int | None = None,
|
||||
) -> None:
|
||||
"""Convert all gem tiers to perfect using the GEMS-tab stash flow.
|
||||
|
||||
Pre-conditions:
|
||||
- Stash must be open
|
||||
- Horadric Cube must be in the PERSONAL stash tab
|
||||
|
||||
Per-transmute flow:
|
||||
1. Open the cube from PERSONAL stash
|
||||
2. Switch to GEMS while cube stays open
|
||||
3. Ctrl+shift+right-click gem x3 -> 3 gems load into cube
|
||||
4. Click Transmute
|
||||
5. Ctrl+shift+left-click result from cube -> result lands on GEMS tab
|
||||
|
||||
tier_filter: optional list of tier names to process (e.g. ["chipped"]).
|
||||
None means all tiers.
|
||||
gem_filter: optional list of gem types to process (e.g. ["diamond"]).
|
||||
None means all gem types.
|
||||
max_transmutes: optional total transmute cap for one-off runs.
|
||||
"""
|
||||
from inventory.common import left_inventory_ready
|
||||
if not wait_until_visible(ScreenObjects.GoldBtnStash, timeout=4).valid:
|
||||
if not left_inventory_ready(grab()):
|
||||
Logger.error("convert_all_gems: stash not open")
|
||||
return
|
||||
Logger.info("convert_all_gems: GoldBtn not matched but left inventory ready — proceeding")
|
||||
|
||||
# Step 1: switch to GEMS tab, screenshot, count via Hermes
|
||||
if not self._ensure_cube_available():
|
||||
Logger.error("convert_all_gems: cube must be available in PERSONAL stash")
|
||||
return
|
||||
|
||||
# Open cube before clicking cube slots; otherwise these coordinates hit the stash panel.
|
||||
if self._open_available_cube_for_gems_tab():
|
||||
self._switch_to_gems_tab()
|
||||
wait(0.4, 0.5)
|
||||
Logger.info("convert_all_gems: pre-flight clear (ctrl+shift+left-click 12 cube slots)")
|
||||
self._empty_cube_to_gems_tab()
|
||||
self.close_cube()
|
||||
wait(0.4, 0.5)
|
||||
else:
|
||||
Logger.warning("convert_all_gems: skipping pre-flight cube clear because cube UI could not be opened")
|
||||
|
||||
from ui_manager import is_visible
|
||||
img = grab()
|
||||
if not (is_visible(ScreenObjects.GoldBtnStash, img) or left_inventory_ready(img)):
|
||||
if not self._reopen_stash():
|
||||
Logger.error("convert_all_gems: could not reopen stash after pre-flight cube clear")
|
||||
return
|
||||
|
||||
# Switch to GEMS tab before counting and planning.
|
||||
self._switch_to_gems_tab()
|
||||
wait(0.5, 0.6)
|
||||
Logger.info("convert_all_gems: reading gem counts via Hermes vision")
|
||||
counts = self._count_gems_with_hermes()
|
||||
wait(0.4, 0.5)
|
||||
|
||||
Logger.info("convert_all_gems: reading gem counts via OCR")
|
||||
counts = self._count_gems_for_plan()
|
||||
|
||||
if not counts:
|
||||
Logger.info("convert_all_gems: no gems found")
|
||||
Logger.info("convert_all_gems: no gems found in GEMS tab")
|
||||
return
|
||||
|
||||
# Step 2: plan
|
||||
plan = self._plan_transmutes(counts)
|
||||
if not plan:
|
||||
Logger.info("convert_all_gems: nothing to transmute (no gem type has 3+)")
|
||||
return
|
||||
total = sum(n for n, *_ in plan)
|
||||
Logger.info(f"convert_all_gems: {len(plan)} gem types to process, {total} total transmutes planned")
|
||||
|
||||
# Step 3: execute.
|
||||
# Flow per gem type:
|
||||
# Open cube from CHARACTER INVENTORY (right-click)
|
||||
# Switch to GEMS tab — cube UI stays open alongside stash
|
||||
# For each transmute:
|
||||
# ctrl+shift+click gem stack 3x -> gems go directly into open cube
|
||||
# transmute
|
||||
# ctrl+click result from cube -> goes back to active GEMS tab
|
||||
# Close cube when done with this gem type.
|
||||
plan = self._filter_transmute_plan(plan, tier_filter=tier_filter, gem_filter=gem_filter)
|
||||
if not plan:
|
||||
Logger.info(
|
||||
"convert_all_gems: nothing to transmute for configured filters "
|
||||
f"(tiers={tier_filter}, gems={gem_filter})"
|
||||
)
|
||||
return
|
||||
|
||||
known_total = sum(n for n, *_ in plan if n != 999)
|
||||
unknown_count = sum(1 for n, *_ in plan if n == 999)
|
||||
total_label = f"{known_total}+depleted-stacks({unknown_count})" if unknown_count else str(known_total)
|
||||
Logger.info(f"convert_all_gems: {len(plan)} gem types, {total_label} transmutes planned")
|
||||
|
||||
left_roi = Config().ui_roi["left_inventory"]
|
||||
done = 0
|
||||
for n_transmutes, gem_template, gems_out, tier_name in plan:
|
||||
Logger.info(f"convert_all_gems: {n_transmutes}x {gem_template}")
|
||||
|
||||
# Open cube from character inventory, then switch to GEMS tab
|
||||
if not self._open_cube_from_inventory():
|
||||
Logger.error(" cube not found in inventory — aborting")
|
||||
return
|
||||
self._switch_to_gems_tab()
|
||||
for n_transmutes, gem_template, gems_out, tier_name in plan:
|
||||
Logger.info(f"convert_all_gems: {n_transmutes}x {gem_template} ({tier_name})")
|
||||
|
||||
for i in range(n_transmutes):
|
||||
# Ensure stash is open and on GEMS tab
|
||||
from ui_manager import is_visible
|
||||
from inventory.common import left_inventory_ready
|
||||
img = grab()
|
||||
roi = Config().ui_roi["left_inventory"]
|
||||
matches = template_finder.search_all(gem_template, img, threshold=0.85, roi=roi)
|
||||
if not (is_visible(ScreenObjects.GoldBtnStash, img) or left_inventory_ready(img)):
|
||||
Logger.info(f" [{i+1}] stash closed -- reopening")
|
||||
if not self._reopen_stash():
|
||||
Logger.error(" could not reopen stash -- aborting")
|
||||
return
|
||||
self._switch_to_gems_tab()
|
||||
wait(0.3, 0.4)
|
||||
|
||||
img = grab()
|
||||
matches = template_finder.search_all(gem_template, img, threshold=0.85, roi=left_roi)
|
||||
if not matches:
|
||||
Logger.info(f" no {gem_template} left in GEMS tab, stopping")
|
||||
Logger.info(f" no {gem_template} left in GEMS tab -- stopping {tier_name}")
|
||||
break
|
||||
Logger.info(f" [{i+1}/{n_transmutes}] ctrl+shift {gem_template} x3 -> cube @ {matches[0].center_monitor}")
|
||||
|
||||
# Step 1: open the cube, then return to GEMS.
|
||||
if not self._open_available_cube_for_gems_tab():
|
||||
Logger.error(f" [{i+1}/{n_transmutes}] cube not found -- aborting")
|
||||
return
|
||||
wait(0.3, 0.4)
|
||||
|
||||
# Step 2: ctrl+shift+right-click x3 -- 3 gems move from GEMS tab into the open cube
|
||||
pos = matches[0].center_monitor
|
||||
Logger.info(f" [{i+1}/{n_transmutes}] loading 3x {gem_template} into cube @ {pos}")
|
||||
for _ in range(3):
|
||||
self._ctrl_shift_click_monitor(*matches[0].center_monitor, label=gem_template)
|
||||
wait(0.15, 0.2)
|
||||
self._ctrl_shift_click_monitor(*pos, label=gem_template)
|
||||
wait(0.25, 0.3)
|
||||
|
||||
# Step 3: transmute button visible in cube UI
|
||||
wait(0.3, 0.4)
|
||||
t_match = wait_until_visible(ScreenObjects.CubeOpened, timeout=3.0)
|
||||
if not t_match.valid:
|
||||
Logger.warning(f" [{i+1}/{n_transmutes}] transmute btn not visible -- clearing cube and stopping this gem type")
|
||||
self._empty_cube_to_gems_tab()
|
||||
self.close_cube()
|
||||
wait(0.4, 0.5)
|
||||
break
|
||||
|
||||
# Step 4: click transmute
|
||||
Logger.info(f" [{i+1}/{n_transmutes}] transmuting")
|
||||
self.transmute()
|
||||
wait(0.5, 0.5)
|
||||
select_screen_object_match(t_match)
|
||||
wait(0.6, 0.7)
|
||||
|
||||
# Step 5: move result directly from cube to the active GEMS tab.
|
||||
cube_roi = Config().ui_roi["cube_area_roi"]
|
||||
img = grab()
|
||||
result_match = template_finder.search(gems_out, img, threshold=0.72, roi=cube_roi)
|
||||
if result_match.valid:
|
||||
Logger.info(f" [{i+1}/{n_transmutes}] ctrl+shift+lclick {result_match.name} from cube to GEMS @ {result_match.center_monitor}")
|
||||
self._ctrl_shift_left_click_monitor(*result_match.center_monitor, label=result_match.name)
|
||||
else:
|
||||
Logger.warning(
|
||||
f" [{i+1}/{n_transmutes}] expected result {gems_out} not found in cube -- "
|
||||
"clearing cube and stopping this gem type"
|
||||
)
|
||||
self._empty_cube_to_gems_tab()
|
||||
self.close_cube()
|
||||
wait(0.4, 0.5)
|
||||
break
|
||||
wait(0.3, 0.4)
|
||||
|
||||
# Step 6: close cube, keep/reopen stash on GEMS tab
|
||||
self.close_cube()
|
||||
wait(0.4, 0.5)
|
||||
img = grab()
|
||||
if not (is_visible(ScreenObjects.GoldBtnStash, img) or left_inventory_ready(img)):
|
||||
if not self._reopen_stash():
|
||||
Logger.error(f" [{i+1}/{n_transmutes}] could not reopen stash after cube close -- aborting")
|
||||
return
|
||||
self._switch_to_gems_tab()
|
||||
wait(0.3, 0.4)
|
||||
|
||||
# ctrl+click result from cube -> goes to active GEMS tab
|
||||
if not self._pick_result_from_cube(gems_out):
|
||||
Logger.warning(f" [{i+1}/{n_transmutes}] result not found in cube")
|
||||
done += 1
|
||||
if max_transmutes is not None and done >= max_transmutes:
|
||||
Logger.info(f"convert_all_gems: reached max_transmutes={max_transmutes}")
|
||||
Logger.info(f"convert_all_gems: done -- {done}/{total_label} transmutes completed")
|
||||
return
|
||||
|
||||
self.close_cube()
|
||||
wait(0.3, 0.4)
|
||||
# Reopen GEMS tab for next gem type
|
||||
self._switch_to_gems_tab()
|
||||
|
||||
Logger.info(f"convert_all_gems: done — {done}/{total} transmutes completed")
|
||||
Logger.info(f"convert_all_gems: done -- {done}/{total_label} transmutes completed")
|
||||
|
||||
def should_transmute(self) -> bool:
|
||||
every_x_game = Config().configs["transmute"]["parser"]["transmute_every_x_game"]
|
||||
@@ -569,7 +938,23 @@ class Transmute:
|
||||
Logger.info(f"Skipping transmutes. Force: {force}, Game#: {self._game_stats._game_counter}")
|
||||
return None
|
||||
|
||||
transmute_gems=Config().configs["transmute"]["parser"]["transmute"]
|
||||
transmute_gems = [
|
||||
gem.strip().lower()
|
||||
for gem in Config().configs["transmute"]["parser"]["transmute"]
|
||||
if gem and gem.strip()
|
||||
]
|
||||
if not transmute_gems:
|
||||
Logger.info("Skipping transmutes. No gem tiers configured.")
|
||||
return None
|
||||
|
||||
# ROTW/named-tab stashes expose a dedicated GEMS tab; use the newer
|
||||
# stacked-gem flow there. Keep the original fixed-slot routine for
|
||||
# classic 4-tab stashes.
|
||||
if Config().configs["transmute"]["parser"]["stash_tabs"] > 4:
|
||||
self._last_game = self._game_stats._game_counter
|
||||
self.convert_all_gems_to_perfect(tier_filter=transmute_gems)
|
||||
return None
|
||||
|
||||
gemsToTransmute=[]
|
||||
gemsToPutBack=[]
|
||||
gemLoggerName=""
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
import transmute.transmute as transmute_module
|
||||
from transmute import Transmute
|
||||
|
||||
|
||||
class _Match:
|
||||
def __init__(self, valid):
|
||||
self.valid = valid
|
||||
|
||||
|
||||
class TestTransmutePlanning:
|
||||
|
||||
def test_unknown_ocr_count_runs_until_depleted(self):
|
||||
subject = object.__new__(Transmute)
|
||||
|
||||
plan = subject._plan_transmutes({
|
||||
"inventory_ruby_flawless": 999,
|
||||
})
|
||||
|
||||
assert plan == [(
|
||||
999,
|
||||
"INVENTORY_RUBY_FLAWLESS",
|
||||
["INVENTORY_RUBY_PERFECT"],
|
||||
"flawless",
|
||||
)]
|
||||
|
||||
def test_plan_uses_matching_output_gem_type(self):
|
||||
subject = object.__new__(Transmute)
|
||||
|
||||
plan = subject._plan_transmutes({
|
||||
"diamond_standard": 3,
|
||||
})
|
||||
|
||||
assert plan == [(
|
||||
1,
|
||||
"INVENTORY_DIAMOND_STANDARD",
|
||||
["INVENTORY_DIAMOND_FLAWLESS"],
|
||||
"standard",
|
||||
)]
|
||||
|
||||
def test_standard_tier_accepts_hermes_normal_key(self):
|
||||
subject = object.__new__(Transmute)
|
||||
|
||||
plan = subject._plan_transmutes({
|
||||
"diamond_normal": 3,
|
||||
})
|
||||
|
||||
assert plan == [(
|
||||
1,
|
||||
"INVENTORY_DIAMOND_STANDARD",
|
||||
["INVENTORY_DIAMOND_FLAWLESS"],
|
||||
"standard",
|
||||
)]
|
||||
|
||||
def test_known_ocr_count_uses_groups_of_three(self):
|
||||
subject = object.__new__(Transmute)
|
||||
|
||||
plan = subject._plan_transmutes({
|
||||
"ruby_flawless": 8,
|
||||
})
|
||||
|
||||
assert plan[0][0] == 2
|
||||
|
||||
def test_gem_type_filter_keeps_only_diamonds(self):
|
||||
subject = object.__new__(Transmute)
|
||||
|
||||
plan = subject._plan_transmutes({
|
||||
"diamond_chipped": 6,
|
||||
"ruby_chipped": 6,
|
||||
})
|
||||
diamonds = subject._filter_transmute_plan(plan, gem_filter=["diamond"])
|
||||
|
||||
assert len(diamonds) == 1
|
||||
assert diamonds[0][1] == "INVENTORY_DIAMOND_CHIPPED"
|
||||
|
||||
def test_cube_available_rejects_inventory_cube(self, monkeypatch):
|
||||
subject = object.__new__(Transmute)
|
||||
monkeypatch.setattr(subject, "_locate_cube", lambda: "inventory")
|
||||
|
||||
assert subject._ensure_cube_available() is False
|
||||
|
||||
def test_open_available_cube_rejects_inventory_when_stash_missing(self, monkeypatch):
|
||||
subject = object.__new__(Transmute)
|
||||
calls = []
|
||||
|
||||
def detect(screen_object):
|
||||
calls.append(screen_object)
|
||||
if screen_object is transmute_module.ScreenObjects.CubeStash:
|
||||
return _Match(False)
|
||||
if screen_object is transmute_module.ScreenObjects.CubeInventory:
|
||||
return _Match(True)
|
||||
return _Match(False)
|
||||
|
||||
monkeypatch.setattr(transmute_module, "detect_screen_object", detect)
|
||||
monkeypatch.setattr(subject, "_switch_to_personal_tab", lambda: calls.append("personal"))
|
||||
monkeypatch.setattr(subject, "_switch_to_gems_tab", lambda: calls.append("gems"))
|
||||
monkeypatch.setattr(subject, "_open_cube_from_inventory", lambda: calls.append("open_inventory") or True)
|
||||
monkeypatch.setattr(transmute_module, "wait", lambda *args, **kwargs: None)
|
||||
|
||||
assert subject._open_available_cube_for_gems_tab() is False
|
||||
assert "open_inventory" not in calls
|
||||
assert "gems" not in calls
|
||||
|
||||
def test_open_available_cube_accepts_already_open_cube_ui(self, monkeypatch):
|
||||
subject = object.__new__(Transmute)
|
||||
calls = []
|
||||
|
||||
def detect(screen_object):
|
||||
calls.append(screen_object)
|
||||
if screen_object is transmute_module.ScreenObjects.CubeOpened:
|
||||
return _Match(True)
|
||||
return _Match(False)
|
||||
|
||||
monkeypatch.setattr(transmute_module, "detect_screen_object", detect)
|
||||
monkeypatch.setattr(subject, "_switch_to_personal_tab", lambda: calls.append("personal"))
|
||||
monkeypatch.setattr(subject, "_switch_to_gems_tab", lambda: calls.append("gems"))
|
||||
monkeypatch.setattr(subject, "_open_cube_from_inventory", lambda: calls.append("open_inventory") or True)
|
||||
monkeypatch.setattr(transmute_module, "wait", lambda *args, **kwargs: None)
|
||||
|
||||
assert subject._open_available_cube_for_gems_tab() is True
|
||||
assert "gems" in calls
|
||||
assert "personal" not in calls
|
||||
assert "open_inventory" not in calls
|
||||
Reference in New Issue
Block a user