From 09bfd330e692377586746998e63122113d5043f4 Mon Sep 17 00:00:00 2001 From: alexpolo1 Date: Mon, 24 Aug 2026 22:30:04 +0200 Subject: [PATCH] feat(baal_xp): join public Baal games, hide, collect XP, leave MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/ui/game_browser.py: new module for game browser interaction (Play button → Join Game tab → OCR game list → click → loading) - src/config.py: [baal_xp] section with enabled, game_name_filter, max_wait_s, xp_threshold, min_hp_pct, hide_x/y, join_timeout_s - config/params.ini: [baal_xp] section + route doc - src/bot.py: baal_xp state, on_run_baal_xp handler (8-phase cycle: leave own game → hero select → join public game → wait in-game → corpse/nopickup/pre_buff → walk to hide spot → wait loop (XP/HP/timer) → leave → recover to own game), _recover_to_own_game helper Enable by adding run_baal_xp to [routes] order in params.ini. --- config/params.ini | 19 ++++ src/bot.py | 175 +++++++++++++++++++++++++++++++- src/config.py | 22 +++++ src/ui/game_browser.py | 220 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 433 insertions(+), 3 deletions(-) create mode 100644 src/ui/game_browser.py diff --git a/config/params.ini b/config/params.ini index fbe8ea8..45e6c01 100644 --- a/config/params.ini +++ b/config/params.ini @@ -216,6 +216,7 @@ skill_mistake_chance = 0.015 ; run_countess (Act 1 Forgotten Tower) ; run_mephisto (Act 3 Durance of Hate) ; run_baal (Act 5 Throne of Destruction) +; run_baal_xp (Join public Baal games, hide, collect XP, leave — see [baal_xp]) order=run_pindle, run_diablo [char] @@ -641,6 +642,24 @@ max_runtime_s=180 ; Engagements per step, so an unkillable/misdetected target cannot stall the run. max_engagements=8 +[baal_xp] +; Baal XP farm: join public Baal games, hide, collect XP, leave. Repeat. +; Enable by adding run_baal_xp to [routes] order above. +enabled=1 +; Substring to match against game names (case-insensitive). Empty = join first game. +game_name_filter= +; Max seconds to wait in a game before leaving (XP keeps ticking while you hide). +max_wait_s=900 +; Leave early if XP gained in this game reaches this value. +xp_threshold=50000000 +; Leave immediately if HP drops below this percentage (chicken). +min_hp_pct=40 +; Screen position (client coords) to walk to and stand still while hiding. +hide_x=640 +hide_y=360 +; Max seconds to wait for a matching game to appear in the browser. +join_timeout_s=60 + [advanced_options] ; startup hotkeys ; select_runs_key: open run selector UI diff --git a/src/bot.py b/src/bot.py index 4780ba0..74655da 100644 --- a/src/bot.py +++ b/src/bot.py @@ -38,8 +38,10 @@ from char.bone_necro import Bone_Necro from char.basic import Basic from char.basic_ranged import Basic_Ranged from ui_manager import wait_until_hidden, wait_until_visible, ScreenObjects, is_visible, detect_screen_object -from ui import meters, skills, view, character_select, main_menu +from ui import meters, skills, view, character_select, main_menu, game_browser +from ui import player_bar from inventory import personal, vendor, belt, common +from game_recovery import GameRecovery from run import Pindle, ShenkEld, Trav, Nihlathak, Arcane, Diablo, Vizier, Level from run import ColdPlains @@ -116,6 +118,7 @@ class Bot: self._do_runs = { "run_level": Config().routes.get("run_level"), "run_cold_plains": Config().routes.get("run_cold_plains"), + "run_baal_xp": Config().routes.get("run_baal_xp") and Config().baal_xp.get("enabled", False), "run_trav": Config().routes.get("run_trav"), "run_pindle": Config().routes.get("run_pindle"), "run_shenk": Config().routes.get("run_eldritch") or Config().routes.get("run_eldritch_shenk"), @@ -179,7 +182,7 @@ class Bot: self._do_runs_reset[_run_name] = False # Create State Machine - self._states=['initialization','hero_selection', 'town', 'level', 'cold_plains', 'pindle', 'shenk', 'trav', 'nihlathak', 'arcane', 'diablo', 'vizier', 'baal', 'mephisto', 'andariel', 'countess'] + self._states=['initialization','hero_selection', 'town', 'level', 'cold_plains', 'baal_xp', 'pindle', 'shenk', 'trav', 'nihlathak', 'arcane', 'diablo', 'vizier', 'baal', 'mephisto', 'andariel', 'countess'] self._transitions = [ { 'trigger': 'init', 'source': 'initialization', 'dest': '=','before': "on_init"}, { 'trigger': 'skip_to_level', 'source': 'initialization', 'dest': 'level', 'before': "on_skip_to_level"}, @@ -191,6 +194,7 @@ class Bot: # Different runs { 'trigger': 'run_level', 'source': 'town', 'dest': 'level', 'before': "on_run_level"}, { 'trigger': 'run_cold_plains', 'source': 'town', 'dest': 'cold_plains', 'before': "on_run_cold_plains"}, + { 'trigger': 'run_baal_xp', 'source': 'town', 'dest': 'baal_xp', 'before': "on_run_baal_xp"}, { 'trigger': 'run_pindle', 'source': 'town', 'dest': 'pindle', 'before': "on_run_pindle"}, { 'trigger': 'run_shenk', 'source': 'town', 'dest': 'shenk', 'before': "on_run_shenk"}, { 'trigger': 'run_trav', 'source': 'town', 'dest': 'trav', 'before': "on_run_trav"}, @@ -203,7 +207,7 @@ class Bot: { 'trigger': 'run_andariel', 'source': 'town', 'dest': 'andariel', 'before': "on_run_andariel" }, { 'trigger': 'run_countess', 'source': 'town', 'dest': 'countess', 'before': "on_run_countess" }, # End run / game - { 'trigger': 'end_run', 'source': ['level', 'cold_plains', 'shenk', 'pindle', 'nihlathak', 'trav', 'arcane', 'diablo','vizier', 'baal', 'mephisto', 'andariel', 'countess'], 'dest': 'town', 'before': "on_end_run"}, + { 'trigger': 'end_run', 'source': ['level', 'cold_plains', 'baal_xp', 'shenk', 'pindle', 'nihlathak', 'trav', 'arcane', 'diablo','vizier', 'baal', 'mephisto', 'andariel', 'countess'], 'dest': 'town', 'before': "on_end_run"}, { 'trigger': 'end_game', 'source': ['town', 'shenk', 'pindle', 'nihlathak', 'trav', 'arcane', 'diablo','vizier','end_run', 'baal', 'mephisto', 'andariel', 'countess'], 'dest': 'initialization', 'before': "on_end_game"}, ] self.machine = Machine(model=self, states=self._states, initial="initialization", transitions=self._transitions, queued=True) @@ -1124,6 +1128,171 @@ class Bot: self._game_stats.update_location("ColdPlains") self._run_wrapper("run_cold_plains", self._cold_plains, (not self._pre_buffed,), ()) + def on_run_baal_xp(self): + """Baal XP farm: leave own game, join a public game, hide, collect XP, leave. + This is a self-contained cycle — it does NOT use the run_wrapper/approach/battle + pattern because it exits and re-enters the game entirely.""" + self._game_stats.update_location("BaalXP") + self._do_runs["run_baal_xp"] = False + self._game_stats.log_run_started("run_baal_xp") + set_pause_state(False) + + cfg = Config().baal_xp + Logger.info(f"=== BAAL XP FARM === filter='{cfg['game_name_filter']}' max_wait={cfg['max_wait_s']}s xp_target={cfg['xp_threshold']}") + + # --- Phase 1: Leave current game, get to hero selection --- + if is_visible(ScreenObjects.InGame): + if not view.fast_save_and_exit(): + Logger.error("baal_xp: fast_save_and_exit failed") + self._game_stats.set_failure_reason("baal_xp: save_and_exit failed") + self._save_error_screenshot("baal_xp", "save_exit_failed") + self.trigger_or_stop("end_run") + return + # Wait for hero selection (main menu) + _recovery = GameRecovery(None) + if not _recovery.go_to_hero_selection(): + Logger.error("baal_xp: could not reach hero selection") + self._game_stats.set_failure_reason("baal_xp: hero selection unreachable") + self.trigger_or_stop("end_run") + return + + # --- Phase 2: Select character and join a public game --- + if not character_select.select_char(): + Logger.error("baal_xp: character selection failed") + self._game_stats.set_failure_reason("baal_xp: char select failed") + self.trigger_or_stop("end_run") + return + + if not game_browser.join_game( + name_filter=cfg["game_name_filter"], + max_wait_s=cfg["join_timeout_s"], + ): + Logger.error("baal_xp: failed to join a game") + self._game_stats.set_failure_reason("baal_xp: game join failed") + self._save_error_screenshot("baal_xp", "join_failed") + # Go back to our own game + self._recover_to_own_game() + self.trigger_or_stop("end_run") + return + + # --- Phase 3: Wait for in-game --- + if not game_browser.wait_for_in_game(timeout=60): + Logger.error("baal_xp: never entered the joined game") + self._game_stats.set_failure_reason("baal_xp: in-game timeout") + self._recover_to_own_game() + self.trigger_or_stop("end_run") + return + + # --- Phase 4: Enter game setup --- + # Handle corpse if we died in this game before + if is_visible(ScreenObjects.Corpse): + view.pickup_corpse() + wait_until_hidden(ScreenObjects.Corpse) + belt.fill_up_belt_from_inventory(Config().char["num_loot_columns"]) + self._char.discover_capabilities() + # /nopickup so we don't grab loot from other players + if Config().char["enable_no_pickup"] and not self._game_stats._nopickup_active: + if view.enable_no_pickup(): + self._game_stats._nopickup_active = True + Logger.info("baal_xp: /nopickup active") + # Pre-buff + self._char.pre_buff() + + # --- Phase 5: Walk to hide spot and stand still --- + hide_x, hide_y = cfg["hide_x"], cfg["hide_y"] + hide_pos = convert_screen_to_monitor((hide_x, hide_y)) + Logger.info(f"baal_xp: walking to hide spot ({hide_x}, {hide_y})") + from input_layer import mouse + mouse.move(*hide_pos) + wait(1.5, 2.5) # give the character time to walk there + + # Record starting XP + try: + start_exp, _ = player_bar.get_experience() + except Exception: + start_exp = 0 + Logger.info(f"baal_xp: starting XP = {start_exp}") + + # --- Phase 6: Hide and wait --- + max_wait = cfg["max_wait_s"] + xp_target = cfg["xp_threshold"] + min_hp = cfg["min_hp_pct"] + start_time = time.time() + last_log = 0.0 + + while True: + elapsed = time.time() - start_time + if elapsed > max_wait: + Logger.info(f"baal_xp: max wait {max_wait}s reached — leaving") + break + + # Check HP + img = grab() + hp_pct = 100.0 + if is_visible(ScreenObjects.InGame, img): + hp_pct = meters.get_health(img) * 100 + if hp_pct < min_hp: + Logger.warning(f"baal_xp: HP {hp_pct:.0f}% < {min_hp}% — chickening out") + break + + # Check XP (every 30s to avoid OCR spam) + if time.time() - last_log > 30: + last_log = time.time() + try: + cur_exp, req_exp = player_bar.get_experience() + gained = cur_exp - start_exp if cur_exp > start_exp else 0 + Logger.info(f"baal_xp: {elapsed:.0f}s in | XP {cur_exp} (+{gained}) | HP {hp_pct:.0f}%") + if gained >= xp_target: + Logger.info(f"baal_xp: XP target {xp_target} reached (+{gained}) — leaving") + break + except Exception as e: + Logger.debug(f"baal_xp: XP check failed: {e}") + + # Check if we got kicked (back at main menu) + if is_visible(ScreenObjects.MainMenu): + Logger.warning("baal_xp: kicked from game (main menu visible) — leaving") + break + + wait(3, 5) + + # --- Phase 7: Leave the game --- + Logger.info("baal_xp: leaving game") + if not view.fast_save_and_exit(): + Logger.error("baal_xp: save_and_exit on leave failed") + self._game_stats.set_failure_reason("baal_xp: leave failed") + self._save_error_screenshot("baal_xp", "leave_failed") + + # --- Phase 8: Recover to own game --- + self._recover_to_own_game() + + # Log run result + self._game_stats.log_run_finished("run_baal_xp", False, None, loot=[]) + self._game_stats.log_exp() + self.trigger_or_stop("end_run") + + def _recover_to_own_game(self): + """After leaving a public game, get back to hero selection and re-enter our own game.""" + # Wait for hero selection + if not is_visible(ScreenObjects.MainMenu): + _recovery = GameRecovery(None) + if not _recovery.go_to_hero_selection(): + Logger.error("baal_xp: could not reach hero selection after leaving public game") + return + # Select character + if not character_select.select_char(): + Logger.error("baal_xp: char select failed during recovery") + return + # Start our own game + if not main_menu.start_game(): + Logger.error("baal_xp: failed to start own game during recovery") + return + # Wait for town spawn + self._curr_loc = self._town_manager.wait_for_town_spawn() + if not self._curr_loc: + detected = self._town_manager.detect_current_act(timeout=4) + self._curr_loc = detected or Location.A5_TOWN_START + Logger.info(f"baal_xp: recovered to own game at {self._curr_loc}") + def on_run_level(self): self._game_stats.update_location("Level") self._run_wrapper("run_level", self._level, (not self._pre_buffed,), ()) diff --git a/src/config.py b/src/config.py index 104a0f4..9450f8f 100644 --- a/src/config.py +++ b/src/config.py @@ -27,6 +27,7 @@ class Config: routes = {} routes_order = [] cold_plains = {} + baal_xp = {} char = {} colors = {} shop = {} @@ -389,6 +390,27 @@ class Config: self.cold_plains.update(dict(self.configs["profile"]["parser"]["cold_plains"])) if "cold_plains" in self.configs["custom"]["parser"]: self.cold_plains.update(dict(self.configs["custom"]["parser"]["cold_plains"])) + # Baal XP farm (join public games, hide, collect XP, leave) + self.baal_xp = { + "enabled": bool(int(self._select_optional("baal_xp", "enabled", "0"))), + "game_name_filter": self._select_optional("baal_xp", "game_name_filter", ""), + "max_wait_s": float(self._select_optional("baal_xp", "max_wait_s", "900")), + "xp_threshold": int(self._select_optional("baal_xp", "xp_threshold", "50000000")), + "min_hp_pct": float(self._select_optional("baal_xp", "min_hp_pct", "40")), + "hide_x": int(self._select_optional("baal_xp", "hide_x", "640")), + "hide_y": int(self._select_optional("baal_xp", "hide_y", "360")), + "join_timeout_s": float(self._select_optional("baal_xp", "join_timeout_s", "60")), + } + if "baal_xp" in self.configs["profile"]["parser"]: + for k in list(self.baal_xp.keys()): + if k in self.configs["profile"]["parser"]["baal_xp"]: + raw = self.configs["profile"]["parser"]["baal_xp"][k] + self.baal_xp[k] = type(self.baal_xp[k])(raw) if not isinstance(self.baal_xp[k], str) else raw + if "baal_xp" in self.configs["custom"]["parser"]: + for k in list(self.baal_xp.keys()): + if k in self.configs["custom"]["parser"]["baal_xp"]: + raw = self.configs["custom"]["parser"]["baal_xp"][k] + self.baal_xp[k] = type(self.baal_xp[k])(raw) if not isinstance(self.baal_xp[k], str) else raw # Sorc base config sorc_base_cfg = dict(self.configs["config"]["parser"]["sorceress"]) if "sorceress" in self.configs["profile"]["parser"]: diff --git a/src/ui/game_browser.py b/src/ui/game_browser.py new file mode 100644 index 0000000..ee96f41 --- /dev/null +++ b/src/ui/game_browser.py @@ -0,0 +1,220 @@ +""" +Game browser: join a public game from the D2R main menu. + +Flow: Play button (hover) -> Join Game tab -> scan game list (OCR) -> click game -> loading screen. + +No dedicated templates exist for the browser UI, so this module relies on: +- ScreenObjects.PlayBtn for the Play button (existing template) +- Coordinate-based clicking for the "Join Game" tab (right of Play button) +- OCR for reading game names in the list +- ScreenObjects.Loading / InGame for confirming the join succeeded +""" +import time +from input_layer import keyboard, mouse +from config import Config +from utils.misc import wait, cut_roi, focus_d2r_window +from logger import Logger +from ui_manager import detect_screen_object, is_visible, select_screen_object_match, ScreenObjects +from screen import grab, convert_screen_to_monitor, find_and_set_window_position, stop_detecting_window, start_detecting_window +from d2r_image import ocr + +# Game list area (screen coords, 1280x720 client). The list occupies the center +# of the screen when the Join Game tab is active. Rows are ~28px tall. +_GAME_LIST_ROI = (150, 130, 980, 480) +_ROW_HEIGHT = 28 +_JOIN_TAB_OFFSET_X = 360 # "Join Game" tab is ~360px right of the Play button center + + +def _ocr_row(img, roi) -> str: + """OCR a single game-list row. Returns cleaned text or '' on failure.""" + try: + row_img = cut_roi(img, roi) + result = ocr.image_to_text( + images=row_img, + model="hover-eng_inconsolata_inv_th_fast", + psm=7, + scale=1.2, + crop_pad=False, + erode=False, + invert=False, + threshold=0, + digits_only=False, + fix_regexps=True, + check_known_errors=False, + correct_words=False, + ) + if result and result[0].text: + return result[0].text.strip() + except Exception as e: + Logger.debug(f"game_browser OCR row failed: {e}") + return "" + + +def _scan_game_list(img) -> list[dict]: + """Scan the game list area and return a list of {name, row, y_center} entries.""" + x, y, w, h = _GAME_LIST_ROI + games = [] + num_rows = max(1, h // _ROW_HEIGHT) + for row in range(num_rows): + row_y = y + row * _ROW_HEIGHT + row_h = min(_ROW_HEIGHT + 4, y + h - row_y) + if row_h < 5: + break + text = _ocr_row(img, [x, row_y, w, row_h]) + if text: + # Filter out obvious non-game text (headers, empty) + cleaned = text.replace(" ", "").lower() + if len(cleaned) >= 2 and not cleaned.isdigit(): + games.append({"name": text, "row": row, "y_center": row_y + row_h // 2}) + return games + + +def _click_game_row(y_center_screen: int) -> None: + """Click a game row at the given screen y-coordinate (center of list horizontally).""" + x, y, w, h = _GAME_LIST_ROI + click_x = x + w // 2 + pos = convert_screen_to_monitor((click_x, y_center_screen)) + mouse.move(*pos) + wait(0.15, 0.3) + mouse.click("left") + wait(0.2, 0.4) + + +def _open_join_tab() -> bool: + """Click the 'Join Game' tab next to the Play button. Returns True if the + game list area appears to have populated (OCR finds at least one row).""" + # Find the Play button to anchor our click + m = detect_screen_object(ScreenObjects.PlayBtn) + if not m.valid: + Logger.error("game_browser: Play button not found — cannot locate Join Game tab") + return False + # Join Game tab is to the right of the Play button + tab_x = m.center[0] + _JOIN_TAB_OFFSET_X + tab_y = m.center[1] + pos = convert_screen_to_monitor((tab_x, tab_y)) + Logger.debug(f"game_browser: clicking Join Game tab at screen ({tab_x}, {tab_y})") + mouse.move(*pos) + wait(0.2, 0.4) + mouse.click("left") + wait(1.0, 1.5) # give the list time to populate + + # Verify: try OCR on the list area + img = grab() + games = _scan_game_list(img) + if games: + Logger.debug(f"game_browser: game list populated, {len(games)} entries visible") + return True + # List might be empty (no games) — that's still a valid "tab opened" state. + # We can't distinguish "empty list" from "wrong tab" without a template, + # so assume success and let the caller handle the no-games case. + Logger.warning("game_browser: game list empty after opening tab (no games or wrong tab)") + return True + + +def join_game(name_filter: str = "", max_wait_s: float = 60.0) -> bool: + """ + Join a public game from the main menu. + + :param name_filter: substring to match against game names (case-insensitive). + Empty string = join the first game in the list. + :param max_wait_s: how long to wait for a matching game to appear. + :return: True if the join was initiated (loading screen or in-game detected). + """ + Logger.info(f"game_browser: joining game (filter='{name_filter or 'first'}')") + stop_detecting_window() + find_and_set_window_position(force=True) + + try: + # Step 1: Make sure we're at the main menu with the Play button visible + start = time.time() + while True: + if is_visible(ScreenObjects.InGame): + Logger.warning("game_browser: already in a game") + return True + if (m := detect_screen_object(ScreenObjects.PlayBtn)).valid: + break + if is_visible(ScreenObjects.MainMenu): + # Main menu visible but Play button not detected — wait for it to activate + wait(1, 2) + else: + Logger.error("game_browser: not at main menu") + return False + if time.time() - start > 30: + Logger.error("game_browser: Play button never appeared") + return False + + # Step 2: Open the Join Game tab + if not _open_join_tab(): + return False + + # Step 3: Find and click a matching game + start = time.time() + while time.time() - start < max_wait_s: + img = grab() + games = _scan_game_list(img) + if games: + target = None + if name_filter: + for g in games: + if name_filter.lower() in g["name"].lower(): + target = g + break + else: + target = games[0] + + if target: + Logger.info(f"game_browser: joining '{target['name']}' (row {target['row']})") + _click_game_row(target["y_center"]) + # Step 4: Wait for loading screen or in-game + return _wait_for_join() + + Logger.debug(f"game_browser: {len(games)} games visible, none match filter '{name_filter}'") + else: + Logger.debug("game_browser: no games visible, waiting...") + wait(2, 4) + + Logger.error(f"game_browser: no matching game found within {max_wait_s}s") + return False + finally: + start_detecting_window() + + +def _wait_for_join(timeout: float = 45.0) -> bool: + """Wait for the loading screen or in-game state after clicking a game.""" + start = time.time() + while time.time() - start < timeout: + if is_visible(ScreenObjects.Loading): + Logger.info("game_browser: loading screen detected — join in progress") + return True + if is_visible(ScreenObjects.InGame): + Logger.info("game_browser: already in game — join succeeded") + return True + # Server error or kicked back to menu? + if is_visible(ScreenObjects.ServerError): + Logger.error("game_browser: server error during join") + return False + # If we're back at the main menu with Play button, the join failed + if (m := detect_screen_object(ScreenObjects.PlayBtn)).valid: + # Could be a transient state — give it a moment + wait(1, 2) + if (m2 := detect_screen_object(ScreenObjects.PlayBtn)).valid: + Logger.error("game_browser: back at main menu — join failed (game full/left)") + return False + wait(0.5, 1.0) + Logger.error(f"game_browser: no loading/in-game screen within {timeout}s") + return False + + +def wait_for_in_game(timeout: float = 60.0) -> bool: + """Wait until we're actually in-game (InGame marker visible).""" + start = time.time() + while time.time() - start < timeout: + if is_visible(ScreenObjects.InGame): + Logger.info("game_browser: in-game confirmed") + return True + if is_visible(ScreenObjects.MainMenu): + Logger.error("game_browser: back at main menu — never entered game") + return False + wait(1, 2) + Logger.error(f"game_browser: InGame not detected within {timeout}s") + return False