Nik Afiq
a8ff95d6f5
- Added `wait_for_state` function in `navigation.py` for state monitoring and reaction handling. - Updated `mapping.md` to reflect changes in story sweep implementation and OCR usage. - Refactored `story_sweep.py` to utilize OCR for region and stage identification, replacing random selection with configured targets. - Enhanced modal handling and confirmation checks for AP usage in `story_sweep.py`. - Updated setup script to require `tesseract` for OCR functionality and included installation instructions. - Revised `plan.md` to document the transition from heuristic to OCR-based stage targeting and the associated findings from live testing.
356 lines
14 KiB
Python
356 lines
14 KiB
Python
"""Normal story AP sweep. Reference: baas-reference/module/explore_tasks/sweep_task.py
|
||
and task_utils.py.
|
||
|
||
Ported per CLAUDE.md's "OCR policy" and Handoff.md: sweeps a config-driven
|
||
list of exact (region, stage, count) targets (config.STORY_SWEEP_TARGETS),
|
||
navigating to each one deterministically instead of the previous "latest
|
||
unlocked region, then a random stage" heuristic -- see plan.md Phase 9's
|
||
retrospective for why that heuristic was a mistake.
|
||
|
||
- `_go_to_region` ports task_utils.py::to_region: OCR the current region
|
||
number, click the exact delta, re-check, bounded loop.
|
||
- `_find_stage_row` is a scoped-down port of core/image.py's
|
||
swipe_search_target_str: OCR each visible stage row's label and match it
|
||
against the configured target, rather than grabbing a random row.
|
||
- `_watch_sweep_result` is built on navigation.wait_for_state, this
|
||
project's scoped port of core/picture.py::co_detect, and returns a named
|
||
outcome ("swept", "inadequate_ap", "unrecognized_state", ...) the way the
|
||
reference's start_sweep does, instead of a single generic "Done".
|
||
|
||
The MAX-button click-then-verify and the modal's own X-button close (both
|
||
calibrated and confirmed live in Phase 9) are reused unchanged -- see
|
||
plan.md Phase 9/10.
|
||
"""
|
||
from ba_auto import detector, navigation
|
||
|
||
OPEN_RETRIES = 3
|
||
POST_SWEEP_DISMISS_ROUNDS = 6
|
||
STAGE_MODAL_DIM_MAX_CHANNEL = 150
|
||
MAX_BUTTON_RETRIES = 3
|
||
MODAL_CLOSE_RETRIES = 3
|
||
|
||
|
||
def _is_stage_modal_open(driver, config):
|
||
r, g, b = driver.color_at(*config.STAGE_MODAL_PROBE)
|
||
return r < STAGE_MODAL_DIM_MAX_CHANNEL and g < STAGE_MODAL_DIM_MAX_CHANNEL and b < STAGE_MODAL_DIM_MAX_CHANNEL
|
||
|
||
|
||
def _color_in_range(rgb, rgb_range):
|
||
lo, hi = rgb_range
|
||
r, g, b = rgb
|
||
return lo[0] <= r <= hi[0] and lo[1] <= g <= hi[1] and lo[2] <= b <= hi[2]
|
||
|
||
|
||
def _is_sweep_usage_confirm(driver, config):
|
||
# 掃討開始 always raises a "use N AP to sweep M times?" confirmation
|
||
# before actually sweeping. Its OK button is this bright cyan.
|
||
return _color_in_range(driver.color_at(*config.SWEEP_CONFIRM_BUTTON), config.SWEEP_CONFIRM_CYAN)
|
||
|
||
|
||
def _is_ap_purchase_prompt(driver, config):
|
||
# If AP is too low for even one sweep, a visually similar dialog appears
|
||
# at the *same* OK-button position but titled "AP購入" (spend real
|
||
# Pyroxene to buy more AP) with a gold/yellow OK instead of cyan --
|
||
# confirmed live by deliberately emptying the sweep count at low AP.
|
||
# Told apart from the safe confirm above by color, not position.
|
||
return _color_in_range(driver.color_at(*config.SWEEP_CONFIRM_BUTTON), config.SWEEP_CONFIRM_GOLD)
|
||
|
||
|
||
def _find_result_button(driver, config):
|
||
# The "掃討完了" (sweep complete) results screen shows a SKIP button
|
||
# (skips the reward-reveal animation) and then, once settled, a final
|
||
# OK button -- both the same cyan as the usage-confirm OK, but ~120px
|
||
# apart vertically. Finding whichever is showing by color avoids
|
||
# hardcoding both positions.
|
||
return detector.find_color_centroid(config.SWEEP_RESULT_BUTTON_REGION, *config.SWEEP_CONFIRM_CYAN)
|
||
|
||
|
||
def _count_raised_above_one(driver, config):
|
||
# The "-" stepper button is flat grey while count == 1 (its default,
|
||
# disabled at the minimum) and turns vivid orange once raised -- cheap
|
||
# way to confirm a MAX/"+" click actually registered without needing OCR
|
||
# on the count itself.
|
||
r, g, b = driver.color_at(*config.SWEEP_MINUS_BUTTON_PROBE)
|
||
return r > 200 and g < 180 and b < 100
|
||
|
||
|
||
def _open_task_screen(driver, config):
|
||
for attempt in range(1, OPEN_RETRIES + 1):
|
||
driver.click(*config.WORK_ICON)
|
||
driver.wait(2)
|
||
if navigation.is_on_subscreen(driver):
|
||
break
|
||
print(f"[story_sweep] work hub not detected after click (attempt {attempt}/{OPEN_RETRIES})")
|
||
else:
|
||
return False
|
||
|
||
for attempt in range(1, OPEN_RETRIES + 1):
|
||
driver.click(*config.TASK_CARD)
|
||
driver.wait(2)
|
||
if navigation.is_on_subscreen(driver):
|
||
return True
|
||
print(f"[story_sweep] task screen not detected after click (attempt {attempt}/{OPEN_RETRIES})")
|
||
return False
|
||
|
||
|
||
def _read_current_region(driver, config):
|
||
return detector.read_int(config.REGION_NUMBER_OCR_RECT)
|
||
|
||
|
||
def _region_arrow_visible(driver, config, center):
|
||
# Both arrows render as a solid navy-blue "<"/">" chevron on a
|
||
# light-blue backdrop when present. The last region in a given direction
|
||
# (or a locked one, per the reference's own "region-unavailable"
|
||
# template check) simply omits the arrow rather than greying it out --
|
||
# confirmed live: at region 30 (this account's current last region), the
|
||
# spot where the right arrow would be was plain background.
|
||
#
|
||
# Scans a small box around `center` rather than probing a single fixed
|
||
# point: a chevron is concave, and a centroid-derived single point
|
||
# landed in the notch between its two strokes -- reading as "absent"
|
||
# even while the glyph was clearly rendered a few pixels away. See
|
||
# plan.md Phase 10.
|
||
cx, cy = center
|
||
rect = (cx - 40, cy - 35, cx + 40, cy + 35)
|
||
return detector.region_contains_color(rect, (40, 70, 120), (100, 130, 190))
|
||
|
||
|
||
def _go_to_region(driver, config, target_region):
|
||
cur = _read_current_region(driver, config)
|
||
if cur is None:
|
||
print("[story_sweep] could not OCR the current region number")
|
||
return False
|
||
print(f"[story_sweep] current region {cur}, target region {target_region}")
|
||
|
||
for attempt in range(1, config.REGION_NAV_MAX_ATTEMPTS + 1):
|
||
if cur == target_region:
|
||
return True
|
||
going_left = cur > target_region
|
||
arrow_pos = config.REGION_LEFT_ARROW if going_left else config.REGION_RIGHT_ARROW
|
||
if not _region_arrow_visible(driver, config, arrow_pos):
|
||
direction = "left" if going_left else "right"
|
||
print(f"[story_sweep] region {target_region} unreachable -- no {direction} arrow at region {cur}")
|
||
return False
|
||
|
||
clicks = abs(cur - target_region)
|
||
for _ in range(clicks):
|
||
driver.click(*arrow_pos)
|
||
driver.wait(1)
|
||
|
||
new_cur = _read_current_region(driver, config)
|
||
if new_cur is None or new_cur == cur:
|
||
print(f"[story_sweep] region number unchanged after {clicks} click(s) (attempt {attempt}/{config.REGION_NAV_MAX_ATTEMPTS})")
|
||
return False
|
||
cur = new_cur
|
||
|
||
print(f"[story_sweep] gave up navigating to region {target_region} after {config.REGION_NAV_MAX_ATTEMPTS} attempts")
|
||
return False
|
||
|
||
|
||
def _normalize_label(text):
|
||
# OCR sometimes reads the row's dash as a different dash-like glyph, or
|
||
# picks up stray whitespace -- normalize before comparing.
|
||
return text.strip().upper().replace("—", "-").replace("–", "-").replace(" ", "")
|
||
|
||
|
||
def _label_suffix(label):
|
||
# Compare only the part after the dash (e.g. "2" in "30-2", "A" in
|
||
# "30-A"), not the full "{region}-{stage}" string. Confirmed live: this
|
||
# font's leading region-number digit is read unreliably by OCR (e.g. "3"
|
||
# misread as "2") even after threshold preprocessing, while the stage
|
||
# suffix after the dash reads correctly across every row tested -- and we
|
||
# don't need the region digit anyway, since _go_to_region has already
|
||
# independently confirmed we're in the right region. See plan.md Phase 10.
|
||
parts = [p for p in _normalize_label(label).split("-") if p]
|
||
return parts[-1] if parts else ""
|
||
|
||
|
||
def _row_label_rect(config, row_y):
|
||
x1, x2 = config.STAGE_LABEL_OCR_X
|
||
top_pad, bottom_pad = config.STAGE_LABEL_OCR_Y_PAD
|
||
return (x1, row_y - top_pad, x2, row_y - bottom_pad)
|
||
|
||
|
||
def _find_stage_row(driver, config, region, stage):
|
||
# Scoped-down swipe_search_target_str (see module docstring): this
|
||
# client's stage list only ever needs the two already-calibrated scroll
|
||
# extremes checked, not arbitrary swipe-and-retry.
|
||
target_suffix = str(stage)
|
||
x, y = config.STAGE_LIST_SCROLL_POINT
|
||
|
||
driver.scroll(x, y, "up", config.STAGE_LIST_SCROLL_CLICKS)
|
||
driver.wait(0.5)
|
||
for row_y in config.STAGE_ROWS_AT_TOP_Y:
|
||
label = detector.read_text(_row_label_rect(config, row_y), whitelist="0123456789-A")
|
||
print(f"[story_sweep] row @ {row_y} (scrolled up): read '{label}'")
|
||
if _label_suffix(label) == target_suffix:
|
||
return row_y
|
||
|
||
driver.scroll(x, y, "down", config.STAGE_LIST_SCROLL_CLICKS)
|
||
driver.wait(0.5)
|
||
for row_y in config.STAGE_ROWS_AT_BOTTOM_Y:
|
||
label = detector.read_text(_row_label_rect(config, row_y), whitelist="0123456789-A")
|
||
print(f"[story_sweep] row @ {row_y} (scrolled down): read '{label}'")
|
||
if _label_suffix(label) == target_suffix:
|
||
return row_y
|
||
|
||
return None
|
||
|
||
|
||
def _click_max_and_verify(driver, config):
|
||
for attempt in range(1, MAX_BUTTON_RETRIES + 1):
|
||
driver.click(*config.SWEEP_MAX_BUTTON)
|
||
driver.wait(0.8)
|
||
if _count_raised_above_one(driver, config):
|
||
return True
|
||
print(f"[story_sweep] MAX click not detected (attempt {attempt}/{MAX_BUTTON_RETRIES})")
|
||
return False
|
||
|
||
|
||
def _click_plus_and_verify(driver, config, count):
|
||
for attempt in range(1, MAX_BUTTON_RETRIES + 1):
|
||
for _ in range(count - 1):
|
||
driver.click(*config.SWEEP_PLUS_BUTTON)
|
||
driver.wait(0.8)
|
||
if _count_raised_above_one(driver, config):
|
||
return True
|
||
print(f"[story_sweep] count-raise via '+' not detected (attempt {attempt}/{MAX_BUTTON_RETRIES})")
|
||
return False
|
||
|
||
|
||
def _set_sweep_count(driver, config, count):
|
||
if count == "max":
|
||
return _click_max_and_verify(driver, config)
|
||
return _click_plus_and_verify(driver, config, count)
|
||
|
||
|
||
def _close_stage_modal(driver, config):
|
||
# Escape does not close this modal (confirmed live: it stayed open with
|
||
# focus on the live "任務開始" button after two Escape presses) -- the
|
||
# only reliable close path is clicking its own X button, verified.
|
||
for _ in range(MODAL_CLOSE_RETRIES):
|
||
if not _is_stage_modal_open(driver, config):
|
||
return True
|
||
driver.click(*config.STAGE_MODAL_CLOSE_BUTTON)
|
||
driver.wait(1)
|
||
return not _is_stage_modal_open(driver, config)
|
||
|
||
|
||
def _watch_sweep_result(driver, config):
|
||
# The reference's start_sweep returns one of "inadequate_ap",
|
||
# "charge_challenge_counts", or "sweep_complete" so its caller reacts
|
||
# appropriately -- this ports that same named-outcome contract via
|
||
# navigation.wait_for_state instead of the old single generic "Done".
|
||
#
|
||
# Called only after the usage-confirm dialog is already accepted (see
|
||
# _sweep_target), so from here it's purely "click through the
|
||
# 掃討完了 SKIP/OK screens until the bare stage-info modal reappears."
|
||
# Clicking the found button by color (not a keypress) means this never
|
||
# risks landing on the underlying "任務開始" button the way a blind
|
||
# Enter-press loop would.
|
||
def click_result_button(d):
|
||
pos = _find_result_button(d, config)
|
||
if pos:
|
||
d.click(*pos)
|
||
d.wait(1.5)
|
||
|
||
ends = {
|
||
(lambda d, c: _is_stage_modal_open(d, c) and _find_result_button(d, c) is None): "swept",
|
||
}
|
||
reactions = {
|
||
(lambda d, c: _find_result_button(d, c) is not None): click_result_button,
|
||
}
|
||
outcome = navigation.wait_for_state(
|
||
driver, config, reactions, ends,
|
||
max_iterations=POST_SWEEP_DISMISS_ROUNDS, poll_interval=1.5,
|
||
)
|
||
return outcome or "unrecognized_state"
|
||
|
||
|
||
def _sweep_target(driver, config, region, stage, count):
|
||
print(f"[story_sweep] --- target {region}-{stage} x {count} ---")
|
||
|
||
if not _go_to_region(driver, config, region):
|
||
return "region_unavailable"
|
||
|
||
row_y = _find_stage_row(driver, config, region, stage)
|
||
if row_y is None:
|
||
print(f"[story_sweep] stage {region}-{stage} not found in the visible stage list")
|
||
return "stage_not_found"
|
||
|
||
driver.click(config.STAGE_ENTER_X, row_y)
|
||
driver.wait(2)
|
||
|
||
if not _is_stage_modal_open(driver, config):
|
||
print("[story_sweep] stage info panel not detected, aborting")
|
||
if navigation.is_on_subscreen(driver):
|
||
driver.keypress("Escape")
|
||
driver.wait(1.5)
|
||
return "unrecognized_state"
|
||
|
||
if not _set_sweep_count(driver, config, count):
|
||
print("[story_sweep] could not confirm sweep count was raised, aborting without spending AP")
|
||
_close_stage_modal(driver, config)
|
||
if navigation.is_on_subscreen(driver):
|
||
driver.keypress("Escape")
|
||
driver.wait(1.5)
|
||
return "unrecognized_state"
|
||
|
||
driver.click(*config.SWEEP_START_BUTTON)
|
||
driver.wait(1.5)
|
||
|
||
if _is_ap_purchase_prompt(driver, config):
|
||
print("[story_sweep] insufficient AP for this sweep -- cancelling without purchasing")
|
||
driver.click(*config.SWEEP_CONFIRM_CANCEL_BUTTON)
|
||
driver.wait(1)
|
||
_close_stage_modal(driver, config)
|
||
if navigation.is_on_subscreen(driver):
|
||
driver.keypress("Escape")
|
||
driver.wait(1.5)
|
||
return "inadequate_ap"
|
||
|
||
if not _is_sweep_usage_confirm(driver, config):
|
||
print("[story_sweep] sweep-usage confirmation not detected, aborting without further input")
|
||
_close_stage_modal(driver, config)
|
||
if navigation.is_on_subscreen(driver):
|
||
driver.keypress("Escape")
|
||
driver.wait(1.5)
|
||
return "unrecognized_state"
|
||
|
||
driver.click(*config.SWEEP_CONFIRM_BUTTON)
|
||
driver.wait(1.5)
|
||
print("[story_sweep] sweep confirmed, waiting for results")
|
||
outcome = _watch_sweep_result(driver, config)
|
||
print(f"[story_sweep] result: {outcome}")
|
||
|
||
if not _close_stage_modal(driver, config):
|
||
print("[story_sweep] warning: could not confirm stage info modal closed -- leaving it open rather than pressing further keys blindly")
|
||
return outcome
|
||
|
||
if navigation.is_on_subscreen(driver):
|
||
driver.keypress("Escape")
|
||
driver.wait(1.5)
|
||
return outcome
|
||
|
||
|
||
def run(driver, config):
|
||
driver.focus_game()
|
||
|
||
if not config.STORY_SWEEP_TARGETS:
|
||
print("[story_sweep] no targets configured (config.STORY_SWEEP_TARGETS is empty), nothing to do")
|
||
return
|
||
|
||
if not _open_task_screen(driver, config):
|
||
print("[story_sweep] could not confirm task screen is open, aborting without pressing further keys")
|
||
return
|
||
|
||
for region, stage, count in config.STORY_SWEEP_TARGETS:
|
||
outcome = _sweep_target(driver, config, region, stage, count)
|
||
if outcome == "inadequate_ap":
|
||
print("[story_sweep] insufficient AP -- stopping, not attempting remaining targets")
|
||
break
|
||
if outcome != "swept":
|
||
print(f"[story_sweep] target {region}-{stage} ended in '{outcome}' -- skipping to next target")
|
||
|
||
print("[story_sweep] Done.")
|