alex/my-botty
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

feat(baal_xp): join public Baal games, hide, collect XP, leave

Browse Source 
- 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.
This commit is contained in:
alexpolo1
2026-08-24 22:30:04 +02:00
parent ce930f2a76
commit 09bfd330e6
4 changed files with 433 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
config/params.ini
View File

@@ -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

175
src/bot.py
View File

@@ -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,), ())

22
src/config.py
View File

@@ -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"]:

220
src/ui/game_browser.py Normal file
View File

@@ -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