diff --git a/agent_strategy.py b/agent_strategy.py index baaf3e0..ebe7213 100644 --- a/agent_strategy.py +++ b/agent_strategy.py @@ -4,6 +4,8 @@ from dataclasses import dataclass, field from enum import Enum +from memory_reader import ROUTE_1 + class GamePhase(Enum): """Different phases of the game.""" @@ -249,7 +251,7 @@ def suggest_action_for_goal( # early is a dead end -- runs reached the table row and # stalled there. Leave and go north until Route 1 has # been seen; only then approach the table. - if 0x0B not in self.visited_maps: + if ROUTE_1 not in self.visited_maps: return "DOWN" # Exit the lab (door is at the south) if y > 4: return "UP" @@ -272,7 +274,7 @@ def suggest_action_for_goal( return "RIGHT" # Try different direction return "UP" return "UP" - elif map_id == 0x0B: # Route 1 + elif map_id == ROUTE_1: # Route 1 # Continue north toward Viridian if pos and len(pos) >= 2: if self.is_at_boundary(pos, "UP", map_id): @@ -315,6 +317,28 @@ def suggest_action_for_goal( return None + def movement_target( + self, goal: Goal, game_state: str, memory_data: dict | None = None + ) -> tuple[int, int] | None: + """Map coordinate the pathfinder should walk toward, if any. + + Only overworld travel has a coordinate. Dialog, menus, and the + doorway-clear maneuver stay on ``suggest_action_for_goal``. + """ + if game_state != "overworld" or not memory_data or self._exit_maneuver_steps > 0: + return None + map_id = (memory_data.get("current_map") or {}).get("map_id") + pos = memory_data.get("player_position") or (0, 0) + x = pos[0] if len(pos) >= 1 else 0 + + if goal.name == "get_starter" and map_id == 0x00: + return self.OAK_TRIGGER + if goal.name == "reach_viridian" and map_id == 0x00: + return self.OAK_TRIGGER + if goal.name == "reach_viridian" and map_id == ROUTE_1: + return (x, 0) + return None + def add_event(self, event: GameEvent): """Add a game event to history. @@ -506,7 +530,7 @@ def check_goal_completion( if "explore_route_1" not in self.completed_goals: map_info = memory_data.get("current_map", {}) map_id = map_info.get("map_id", 0) - if map_id == 0x0B: # Route 1 + if map_id == ROUTE_1: # Route 1 self.mark_goal_complete("explore_route_1") def get_progress_summary(self) -> dict: diff --git a/game_state.py b/game_state.py index becb485..a789c41 100644 --- a/game_state.py +++ b/game_state.py @@ -12,6 +12,7 @@ from pyboy import PyBoy from memory_reader import MemoryReader, get_map_name +from navigation import downsample_collision from ocr_enhancer import OCREnhancer logger = logging.getLogger(__name__) @@ -90,6 +91,32 @@ def __init__( else: self.memory_reader = None + def read_walkable_grid(self) -> list[list[int]] | None: + """9 by 10 block grid of the visible map. 1 is walkable. + + Uses PyBoy's Pokemon collision map (tileset passable list plus the + screen tilemap). Returns None when that wrapper is not available, + so callers can fall back to the coarser direction heuristic. + """ + collision_fn = getattr(self.pyboy, "game_area_collision", None) + if collision_fn is None: + return None + try: + return downsample_collision(collision_fn()) + except Exception: + logger.debug("Walkability grid unavailable", exc_info=True) + return None + + def read_map_dimensions(self) -> tuple[int, int] | None: + """Current map size in blocks, or None if memory is not ready.""" + if not self.memory_enabled or not self.memory_reader: + return None + try: + return self.memory_reader.read_map_dimensions() + except Exception: + logger.debug("Map dimensions unavailable", exc_info=True) + return None + def get_screen_image(self) -> np.ndarray: """Get current screen as numpy array.""" screen = self.pyboy.screen.image @@ -706,8 +733,10 @@ def execute_action(self, action: str) -> bool: try: # WAIT N is in ticks of 15 frames (~0.25s of game time each); # raw single-frame waits were far too short to let loading - # screens and transitions play out + # screens and transitions play out. Cap N so a model reply + # of WAIT 999999 cannot stall the emulator. units = int(action.split()[1]) if len(action.split()) > 1 else 10 + units = max(0, min(units, 30)) for _ in range(units * 15): self.pyboy.tick() return True diff --git a/memory_reader.py b/memory_reader.py index 7a32b05..847c818 100644 --- a/memory_reader.py +++ b/memory_reader.py @@ -20,6 +20,9 @@ class MemoryAddresses: PLAYER_Y = 0xD361 # Player Y coordinate on current map CURRENT_MAP = 0xD35E # Current map ID MAP_HEADER_BANK = 0xD35F # Map header bank + # pret wram: wCurMapHeight / wCurMapWidth, in 2x2 blocks + MAP_HEIGHT = 0xD368 + MAP_WIDTH = 0xD369 # Player info PLAYER_NAME_START = 0xD158 # Player name (11 bytes, terminated by 0x50) @@ -30,15 +33,17 @@ class MemoryAddresses: PARTY_POKEMON_START = 0xD16B # First Pokemon data structure POKEMON_DATA_SIZE = 44 # Size of each Pokemon data structure - # Pokemon data structure offsets (relative to party Pokemon start) + # Pokemon data structure offsets (relative to party Pokemon start). + # pret pokered party_struct: species, current HP, box level, status, + # types, ... level at 33, max HP at 34. Offset 3 is box level, not max HP. POKEMON_SPECIES = 0 # Pokemon species ID POKEMON_HP_CURRENT = 1 # Current HP (2 bytes, little-endian) - POKEMON_HP_MAX = 3 # Max HP (2 bytes, little-endian) - POKEMON_STATUS = 5 # Status condition (0x00 = normal) - POKEMON_TYPE1 = 6 - POKEMON_TYPE2 = 7 + POKEMON_STATUS = 4 # Status condition (0x00 = normal) + POKEMON_TYPE1 = 5 + POKEMON_TYPE2 = 6 + POKEMON_EXP = 14 # Experience (3 bytes) POKEMON_LEVEL = 33 # Level - POKEMON_EXP = 34 # Experience (3 bytes) + POKEMON_HP_MAX = 34 # Max HP (2 bytes, little-endian) # Inventory INVENTORY_COUNT = 0xD31D # Number of items in inventory @@ -163,6 +168,18 @@ def read_current_map(self) -> dict[str, int]: "map_bank": map_bank, } + def read_map_dimensions(self) -> tuple[int, int] | None: + """Read the current map size in walkable blocks. + + Returns: + ``(width, height)``, or None when the header has not loaded. + """ + width = self.read_byte(MemoryAddresses.MAP_WIDTH) + height = self.read_byte(MemoryAddresses.MAP_HEIGHT) + if width == 0 or height == 0: + return None + return width, height + def read_player_name(self) -> str: """Read player name from memory. @@ -386,33 +403,38 @@ def read_full_game_state(self) -> dict[str, any]: 0x08: "Cinnabar Island", 0x09: "Indigo Plateau", 0x0A: "Saffron City", - 0x0B: "Route 1", - 0x0C: "Route 2", - 0x0D: "Route 3", - 0x0E: "Route 4", - 0x0F: "Route 5", - 0x10: "Route 6", - 0x11: "Route 7", - 0x12: "Route 8", - 0x13: "Route 9", - 0x14: "Route 10", - 0x15: "Route 11", - 0x16: "Route 12", - 0x17: "Route 13", - 0x18: "Route 14", - 0x19: "Route 15", - 0x1A: "Route 16", - 0x1B: "Route 17", - 0x1C: "Route 18", - 0x1D: "Route 19", - 0x1E: "Route 20", - 0x1F: "Route 21", - 0x20: "Route 22", - 0x21: "Route 23", - 0x22: "Route 24", - 0x23: "Route 25", + # 0x0B is UNUSED_MAP_0B in pret pokered map_constants.asm. + # Route 1 is 0x0C; Route 25 is 0x24. + 0x0C: "Route 1", + 0x0D: "Route 2", + 0x0E: "Route 3", + 0x0F: "Route 4", + 0x10: "Route 5", + 0x11: "Route 6", + 0x12: "Route 7", + 0x13: "Route 8", + 0x14: "Route 9", + 0x15: "Route 10", + 0x16: "Route 11", + 0x17: "Route 12", + 0x18: "Route 13", + 0x19: "Route 14", + 0x1A: "Route 15", + 0x1B: "Route 16", + 0x1C: "Route 17", + 0x1D: "Route 18", + 0x1E: "Route 19", + 0x1F: "Route 20", + 0x20: "Route 21", + 0x21: "Route 22", + 0x22: "Route 23", + 0x23: "Route 24", + 0x24: "Route 25", } +# pret pokered: Route 1 is map 0x0C (0x0B is unused). +ROUTE_1 = 0x0C + def get_map_name(map_id: int) -> str: """Get map name from map ID. diff --git a/navigation.py b/navigation.py new file mode 100644 index 0000000..548e61b --- /dev/null +++ b/navigation.py @@ -0,0 +1,207 @@ +"""Walkability grid and a one-step path toward a map coordinate. + +Pokemon Red's visible map is a 9 by 10 grid of walkable blocks (each block is +a 2x2 tile). PyBoy's ``game_area_collision`` reports the 18 by 20 tile view; +``downsample_collision`` folds that into the 9 by 10 block grid the player +actually steps on. The player is not a global "this direction is a wall": +a fence can block UP on one column and be open on the next. + +The pathfinder returns only the next button. Dialog, menus, and which +coordinate to head for stay with the policy chain and the strategy. +""" + +from collections import deque + +# Visible blocks. The camera keeps the player near (4, 4) on maps larger +# than the screen, matching the Claude / PyBoy collision overlay. +VIEW_ROWS = 9 +VIEW_COLS = 10 +CENTER_ROW = 4 +CENTER_COL = 4 + +_DELTAS = ( + (-1, 0, "UP"), + (1, 0, "DOWN"), + (0, -1, "LEFT"), + (0, 1, "RIGHT"), +) + + +def camera_origin( + player_x: int, + player_y: int, + map_width: int, + map_height: int, +) -> tuple[int, int]: + """Top-left map block currently on screen.""" + max_x = max(0, map_width - VIEW_COLS) + max_y = max(0, map_height - VIEW_ROWS) + origin_x = min(max(player_x - CENTER_COL, 0), max_x) + origin_y = min(max(player_y - CENTER_ROW, 0), max_y) + return origin_x, origin_y + + +def downsample_collision(collision) -> list[list[int]] | None: + """Fold an 18 by 20 tile collision map into a 9 by 10 block grid. + + A block is walkable when any of its four tiles is. Returns None when + the value is not a numeric grid (a missing game wrapper, for example). + """ + try: + if len(collision) < VIEW_ROWS * 2 or len(collision[0]) < VIEW_COLS * 2: + return None + int(collision[0][0]) + except (TypeError, ValueError, IndexError): + return None + + grid: list[list[int]] = [] + for row in range(VIEW_ROWS): + blocks: list[int] = [] + for col in range(VIEW_COLS): + tiles = ( + collision[row * 2][col * 2], + collision[row * 2][col * 2 + 1], + collision[row * 2 + 1][col * 2], + collision[row * 2 + 1][col * 2 + 1], + ) + blocks.append(1 if any(int(tile) != 0 for tile in tiles) else 0) + grid.append(blocks) + return grid + + +def next_step( + walkable: list[list[int]], + player: tuple[int, int], + target: tuple[int, int], + map_width: int, + map_height: int, +) -> str | None: + """Button that walks the player one block closer to ``target``. + + ``player`` and ``target`` are map coordinates ``(x, y)``. The search + runs on the visible walkability grid. When the target is off-screen, + the goal is the visible walkable block closest to it. + """ + if map_width <= 0 or map_height <= 0 or not walkable: + return None + origin_x, origin_y = camera_origin(player[0], player[1], map_width, map_height) + start = (player[1] - origin_y, player[0] - origin_x) + if not _inside(walkable, start) or walkable[start[0]][start[1]] == 0: + return None + + goal, came_from = _closest_reachable( + walkable, start, origin_x, origin_y, target + ) + if goal == start: + return _step_offscreen(walkable, start, player, target, origin_x, origin_y) + return _reconstruct(came_from, goal)[0] + + +def _world_distance( + cell: tuple[int, int], + origin_x: int, + origin_y: int, + target: tuple[int, int], +) -> int: + world_x = cell[1] + origin_x + world_y = cell[0] + origin_y + return abs(world_x - target[0]) + abs(world_y - target[1]) + + +def _closest_reachable( + walkable: list[list[int]], + start: tuple[int, int], + origin_x: int, + origin_y: int, + target: tuple[int, int], +) -> tuple[tuple[int, int], dict[tuple[int, int], tuple[int, int]]]: + """Reachable block nearest the target, and the BFS parent map. + + Breadth-first search makes the parent chain the shortest walk. A later + block replaces the goal only when it is strictly closer to the target, + so an equal-distance cell on the wrong side of a wall does not win. + """ + best = start + best_dist = _world_distance(start, origin_x, origin_y, target) + came_from: dict[tuple[int, int], tuple[int, int]] = {} + seen = {start} + queue: deque[tuple[int, int]] = deque([start]) + while queue: + current = queue.popleft() + distance = _world_distance(current, origin_x, origin_y, target) + if distance < best_dist: + best_dist = distance + best = current + for d_row, d_col, _direction in _DELTAS: + neighbor = (current[0] + d_row, current[1] + d_col) + if neighbor in seen or not _inside(walkable, neighbor): + continue + if walkable[neighbor[0]][neighbor[1]] == 0: + continue + seen.add(neighbor) + came_from[neighbor] = current + queue.append(neighbor) + return best, came_from + + +def _step_offscreen( + walkable: list[list[int]], + start: tuple[int, int], + player: tuple[int, int], + target: tuple[int, int], + origin_x: int, + origin_y: int, +) -> str | None: + """Step off the visible grid toward a target past the edge. + + Refuses the step when the next visible block is a wall. + """ + if (player[0], player[1]) == target: + return None + screen_target = (target[1] - origin_y, target[0] - origin_x) + row_delta = screen_target[0] - start[0] + col_delta = screen_target[1] - start[1] + if row_delta == 0 and col_delta == 0: + return None + if abs(row_delta) >= abs(col_delta) and row_delta != 0: + direction = "UP" if row_delta < 0 else "DOWN" + d_row, d_col = (-1 if row_delta < 0 else 1), 0 + elif col_delta < 0: + direction = "LEFT" + d_row, d_col = 0, -1 + elif col_delta > 0: + direction = "RIGHT" + d_row, d_col = 0, 1 + else: + return None + neighbor = (start[0] + d_row, start[1] + d_col) + if _inside(walkable, neighbor) and walkable[neighbor[0]][neighbor[1]] == 0: + return None + return direction + + +def _inside(walkable: list[list[int]], cell: tuple[int, int]) -> bool: + row, col = cell + return 0 <= row < len(walkable) and 0 <= col < len(walkable[0]) + + +def _reconstruct( + came_from: dict[tuple[int, int], tuple[int, int]], + current: tuple[int, int], +) -> list[str]: + steps: list[str] = [] + while current in came_from: + previous = came_from[current] + steps.append(_direction_between(previous, current)) + current = previous + steps.reverse() + return steps + + +def _direction_between(previous: tuple[int, int], current: tuple[int, int]) -> str: + d_row = current[0] - previous[0] + d_col = current[1] - previous[1] + for row, col, direction in _DELTAS: + if row == d_row and col == d_col: + return direction + return "UP" diff --git a/ocr_enhancer.py b/ocr_enhancer.py index 58e7341..fadd35c 100644 --- a/ocr_enhancer.py +++ b/ocr_enhancer.py @@ -256,24 +256,23 @@ def extract_text_enhanced( reverse=True, ) - # Extract text from each region - texts = [] + # Keep each string paired with the region that produced it. Indexing + # the original region list with the filtered text list mis-labels + # dialog after a short OCR result is dropped. + extracted: list[tuple[dict, str]] = [] for region in regions: text = self.extract_text_from_region(image, region) if text and len(text.strip()) > 2: - texts.append(text) - - # Combine texts, prioritizing dialog - if texts: - # If we have dialog text, use it primarily - dialog_texts = [ - t - for i, t in enumerate(texts) - if regions[i]["type"] == TextRegion.DIALOG_BOX - ] - if dialog_texts: - return " ".join(dialog_texts) - else: - return " ".join(texts) - - return "" + extracted.append((region, text)) + + if not extracted: + return "" + + dialog_texts = [ + text + for region, text in extracted + if region["type"] == TextRegion.DIALOG_BOX + ] + if dialog_texts: + return " ".join(dialog_texts) + return " ".join(text for _, text in extracted) diff --git a/pokemon_agent.py b/pokemon_agent.py index 9f0454c..2d91a17 100644 --- a/pokemon_agent.py +++ b/pokemon_agent.py @@ -3,9 +3,9 @@ Version: 0.0.7 """ -import contextlib import logging import random +import re import time from collections import Counter, deque from dataclasses import dataclass @@ -17,9 +17,17 @@ from llm_optimizer import ActionCache, PromptOptimizer, RepetitionDetector from llm_provider import LLMProvider from metrics import MetricsCollector +from navigation import next_step logger = logging.getLogger(__name__) +# Whole-word button tokens. Exact token match keeps START from collapsing +# into A ("A" is a substring of "START") and ignores prose like "GO EAST". +_BUTTON_TOKENS = frozenset({"START", "SELECT", "RIGHT", "LEFT", "DOWN", "UP", "A", "B"}) +# WAIT N is applied as N * 15 frames. Cap N so a model reply cannot stall the emulator. +_MAX_WAIT_UNITS = 30 +_BUTTON_WORD = re.compile(r"[A-Z]+") + @dataclass class Observation: @@ -631,6 +639,10 @@ def _route_policy(self, obs: "Observation") -> str | None: goal = self.strategy.get_current_goal() if goal is None: return None + path_step = self._path_step(obs, goal) + if path_step: + self._edge_scan_steps = 0 + return path_step action = self.strategy.suggest_action_for_goal( goal, obs.game_state, self._memory_data(obs.game_info) ) @@ -673,6 +685,29 @@ def _route_policy(self, obs: "Observation") -> str | None: self._edge_scan_steps = 0 return action + def _path_step(self, obs: "Observation", goal) -> str | None: + """One walkable step toward the goal, using the visible collision grid. + + ``blocked_directions`` is global, so a fence tile used to forbid UP + on every column. The grid is per block: the gap column stays open. + """ + memory = self._memory_data(obs.game_info) + target = self.strategy.movement_target(goal, obs.game_state, memory) + if target is None: + return None + grid = self.game_state.read_walkable_grid() + dimensions = self.game_state.read_map_dimensions() + pos = memory.get("player_position") + if not grid or not dimensions or not pos or len(pos) < 2: + return None + step = next_step(grid, (pos[0], pos[1]), target, dimensions[0], dimensions[1]) + if step: + logger.info( + f"[PATH] Step {obs.step_count}: {step} toward {target} " + f"from {pos}" + ) + return step + def _first_action_policy(self, obs: "Observation") -> str | None: """Cheap heuristics for the very first action (avoids an LLM call).""" if obs.step_count != 0 or self.action_history: @@ -864,7 +899,6 @@ def _dialog_shortcut_policy(self, obs: "Observation") -> str | None: def _llm_policy(self, obs: "Observation") -> str: """Ask the LLM for an action; falls back to heuristics on failure.""" - llm_start_time = None try: prompt = self.get_prompt() @@ -877,16 +911,11 @@ def _llm_policy(self, obs: "Observation") -> str: ) llm_duration = time.time() - llm_start_time - # Record LLM metrics + # Call counts live on the provider (it has token and error detail). + # Recording again here doubled total_calls and distorted latency. if self.metrics: self.metrics.performance.record_llm_time(llm_duration) - self.metrics.llm.record_call(llm_duration) except Exception as e: - llm_duration = ( - time.time() - llm_start_time if llm_start_time is not None else 0.0 - ) - if self.metrics: - self.metrics.llm.record_call(llm_duration, error=True) logger.warning(f"LLM call failed: {e}") logger.warning("Using fallback action based on game state") return self._llm_fallback(obs) @@ -913,50 +942,56 @@ def _llm_fallback(self, obs: "Observation") -> str: # during character creation) return "A" + def _match_button_token(self, text: str) -> str | None: + """First whole-word button in text. + + Substring matching treated START as A because ``"A" in "START"``. + Tokens are matched exactly, so START and SELECT stay intact and + prose like "GO EAST" does not become A. + """ + for token in _BUTTON_WORD.findall(text): + if token in _BUTTON_TOKENS: + return token + return None + + def _match_wait_units(self, text: str) -> int | None: + """WAIT count from a reply, clamped to ``_MAX_WAIT_UNITS``.""" + match = re.search(r"\bWAIT(?:\s+(\d+))?\b", text) + if match is None: + return None + if match.group(1) is None: + return 10 + return min(int(match.group(1)), _MAX_WAIT_UNITS) + + def _block_creation_b(self, obs: "Observation", button: str) -> str: + """B cancels a new game; substitute A while naming is in progress.""" + if button == "B" and self._in_creation_window(): + logger.info( + f"[CHARACTER_CREATION] Blocked B button, using A instead (step {obs.step_count})" + ) + return "A" + return button + def _parse_llm_response(self, obs: "Observation", response: str) -> str: """Extract a valid button action from the LLM's raw response.""" response_clean = response.strip().upper() - # CRITICAL: Final check - prevent B during character creation - if self._in_creation_window() and ( - response_clean == "B" or response_clean.startswith("B") - ): - logger.info( - f"[CHARACTER_CREATION] Blocked B press from LLM, using A instead (step {obs.step_count})" - ) - response_clean = "A" - - valid_buttons = ["UP", "DOWN", "LEFT", "RIGHT", "A", "B", "SELECT", "START"] - - # Check for direct match first (most common case) - for button in valid_buttons: - if button in response_clean: - # CRITICAL: Block B during character creation even if found in response - if button == "B" and self._in_creation_window(): - logger.info( - f"[CHARACTER_CREATION] Blocked B button, using A instead (step {obs.step_count})" - ) - return self._cache_action(obs, "A") - # Extract the button - if "," in response_clean: - # Handle comma-separated - parts = [p.strip() for p in response_clean.split(",")] - valid_parts = [p for p in parts if p in valid_buttons] - action = ( - ", ".join(valid_parts[:2]) if valid_parts else button - ) # Max 2 actions - else: - action = button - return self._cache_action(obs, action) - - # Check for WAIT command - if "WAIT" in response_clean: - parts = response_clean.split() - action = "WAIT 10" - if len(parts) >= 2: - with contextlib.suppress(ValueError): - action = f"WAIT {int(parts[1])}" - return self._cache_action(obs, action) + if "," in response_clean: + buttons = [] + for part in response_clean.split(","): + button = self._match_button_token(part) + if button is not None: + buttons.append(self._block_creation_b(obs, button)) + if buttons: + return self._cache_action(obs, ", ".join(buttons[:2])) + else: + button = self._match_button_token(response_clean) + if button is not None: + return self._cache_action(obs, self._block_creation_b(obs, button)) + + wait_units = self._match_wait_units(response_clean) + if wait_units is not None: + return self._cache_action(obs, f"WAIT {wait_units}") # Fallback: START while still orienting, A afterwards action = "START" if len(self.action_history) < 5 else "A" @@ -1008,29 +1043,12 @@ def step(self) -> dict: completed_goals=completed, ) - # Check if strategy suggests an action + # Do not apply a strategy suggestion here. suggest_action_for_goal + # returns raw A/DOWN/START for menus and dialog, and taking it + # skipped the policy chain (naming scripts, menu escape, dialog + # loops) on most steps. Overworld routing still runs inside + # get_action() via _route_policy. action = None - if self.strategy: - current_goal = self.strategy.get_current_goal() - if current_goal: - memory_data = { - "player_position": pre_state.get("player_position"), - "current_map": pre_state.get("current_map", {}), - "party": pre_state.get("party", []), - "health": pre_state.get("health", {}), - } - suggested_action = self.strategy.suggest_action_for_goal( - current_goal, pre_game_state, memory_data - ) - # Use strategy suggestion if available and not in pure exploration mode - # Calculate action diversity for adaptive exploration - action_diversity = ( - self.calculate_action_entropy(window=15) / 3.0 - ) # Normalize to 0-1 range - if suggested_action and not self.strategy.should_explore( - self.stuck_count, action_diversity - ): - action = suggested_action # Force exploration when stuck (before getting action). # Overworld only: random arrows inside a menu, dialog, or transition diff --git a/tests/conftest.py b/tests/conftest.py index cdc0bd8..d270835 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -74,11 +74,13 @@ def sample_memory_data(): 0xD163: 1, # 1 Pokemon in party # First Pokemon data (simplified) 0xD16B: 25, # Species (Pikachu) - 0xD16C: 0x20, # HP current low byte + 0xD16C: 0x20, # HP current low byte (offset 1) 0xD16D: 0x03, # HP current high byte (0x0320 = 800) - 0xD16E: 0x20, # HP max low byte - 0xD16F: 0x03, # HP max high byte (0x0320 = 800) - 0xD19C: 50, # Level (0xD16B + 33 = 0xD19C) + 0xD16E: 0x00, # Box level (offset 3), not max HP + 0xD16F: 0x00, # Status (offset 4) + 0xD18C: 50, # Level (0xD16B + 33) + 0xD18D: 0x20, # Max HP low byte (offset 34) + 0xD18E: 0x03, # Max HP high byte (0x0320 = 800) # Inventory 0xD31D: 2, # 2 items 0xD31E: 1, # Item 1 ID diff --git a/tests/test_agent_strategy_route.py b/tests/test_agent_strategy_route.py index e5b7828..7140033 100644 --- a/tests/test_agent_strategy_route.py +++ b/tests/test_agent_strategy_route.py @@ -1,6 +1,7 @@ """Route-to-Oak direction tests for the get_starter goal.""" from agent_strategy import AgentStrategy +from memory_reader import ROUTE_1 def suggest(map_id, pos, visited=()): @@ -32,9 +33,9 @@ def test_pallet_town_heads_north(): def test_oaks_lab_heads_to_ball_table_then_interacts(): - # Only after Route 1 (0x0B) has been seen — that is when Oak's cutscene + # Only after Route 1 (0x0C) has been seen — that is when Oak's cutscene # has fired and the balls become takeable. - r1 = (0x0B,) + r1 = (ROUTE_1,) assert suggest(0x28, (4, 11), r1) == "UP" # deep in the lab: go up assert suggest(0x28, (4, 4), r1) == "RIGHT" # at table row: go right assert suggest(0x28, (7, 4), r1) == "A" # at the table: interact @@ -50,7 +51,7 @@ def test_lab_is_a_dead_end_before_the_oak_cutscene(): def test_lab_approaches_table_after_route_1_seen(): s = AgentStrategy() - s.visited_maps.add(0x0B) # Route 1 visited -> Oak cutscene has fired + s.visited_maps.add(ROUTE_1) # Route 1 visited -> Oak cutscene has fired goal = next(g for g in s.goals if g.name == "get_starter") md = {"current_map": {"map_id": 0x28}, "player_position": (4, 11), "party": []} assert s.suggest_action_for_goal(goal, "overworld", md) == "UP" @@ -91,6 +92,23 @@ def _agent(self, mock_llm_provider, mock_pyboy, map_id, pos): agent.strategy.mark_goal_complete("start_game") return agent + def test_path_crosses_at_the_gap_even_if_up_is_globally_blocked( + self, mock_llm_provider, mock_pyboy + ): + from unittest.mock import Mock + + agent = self._agent(mock_llm_provider, mock_pyboy, 0x00, (2, 4)) + grid = [[1 for _ in range(10)] for _ in range(9)] + for col in range(10): + grid[3][col] = 0 + grid[3][6] = 1 + agent.game_state.read_walkable_grid = Mock(return_value=grid) + agent.game_state.read_map_dimensions = Mock(return_value=(10, 9)) + # The old global set would forbid UP everywhere after one bump. + # The path still steps toward the gap column. + agent.blocked_directions.add("UP") + assert agent.get_action() == "RIGHT" + def test_pallet_town_goes_north(self, mock_llm_provider, mock_pyboy): agent = self._agent(mock_llm_provider, mock_pyboy, 0x00, (10, 10)) assert agent.get_action() == "UP" @@ -165,12 +183,12 @@ def test_step_updates_map_tracking(mock_llm_provider, mock_pyboy): "game_state": "overworld", "party": [], "player_position": (5, 6), - "current_map": {"map_id": 0x0B, "map_name": "Route 1"}, + "current_map": {"map_id": ROUTE_1, "map_name": "Route 1"}, } ) gs.execute_action = Mock(return_value=True) agent.step() - assert 0x0B in agent.strategy.visited_maps + assert ROUTE_1 in agent.strategy.visited_maps class TestEdgeScan: diff --git a/tests/test_game_state.py b/tests/test_game_state.py index 62da811..da8b24b 100644 --- a/tests/test_game_state.py +++ b/tests/test_game_state.py @@ -145,6 +145,15 @@ def test_execute_action_wait(self, mock_pyboy): assert result is True assert mock_pyboy.tick.call_count == 5 * 15 # WAIT units are 15 frames each + def test_execute_action_wait_is_capped(self, mock_pyboy): + """Huge WAIT counts must not stall the emulator.""" + game_state = GameState(mock_pyboy, ocr_enabled=False) + + result = game_state.execute_action("WAIT 999999") + + assert result is True + assert mock_pyboy.tick.call_count == 30 * 15 + def test_execute_action_invalid(self, mock_pyboy): """Test executing invalid action.""" game_state = GameState(mock_pyboy, ocr_enabled=False) diff --git a/tests/test_memory_reader.py b/tests/test_memory_reader.py index cb623bc..05476a4 100644 --- a/tests/test_memory_reader.py +++ b/tests/test_memory_reader.py @@ -149,12 +149,6 @@ def getitem(address): mock_pyboy.memory.__getitem__ = Mock(side_effect=getitem) - # Level is at PARTY_POKEMON_START + POKEMON_LEVEL offset - # PARTY_POKEMON_START = 0xD16B, POKEMON_LEVEL = 33 - # So level address = 0xD16B + 33 = 0xD18C (not 0xD19C in sample data) - # Update sample data to have level at correct address - sample_memory_data[0xD18C] = 50 # Level at correct offset (0xD16B + 33) - reader = MemoryReader(mock_pyboy) party = reader.read_pokemon_party() @@ -162,9 +156,13 @@ def getitem(address): assert party[0]["species"] == 25 assert party[0]["level"] == 50 assert party[0]["hp_current"] == 800 + # Max HP is party_struct offset 34, not the box-level byte at offset 3. assert party[0]["hp_max"] == 800 + assert party[0]["status"] == 0 assert party[0]["hp_percent"] == 100.0 assert not party[0]["fainted"] + assert MemoryAddresses.POKEMON_HP_MAX == 34 + assert MemoryAddresses.POKEMON_STATUS == 4 def test_read_pokemon_party_empty(self, mock_pyboy): """Test reading empty party.""" @@ -308,7 +306,10 @@ def test_get_map_name(self): """Test getting map name from ID.""" assert get_map_name(0x00) == "Pallet Town" assert get_map_name(0x01) == "Viridian City" - assert get_map_name(0x0B) == "Route 1" + # pret map_constants: 0x0B is unused, Route 1 is 0x0C, Route 25 is 0x24 + assert get_map_name(0x0C) == "Route 1" + assert get_map_name(0x24) == "Route 25" + assert get_map_name(0x0B) == "Map 0B" def test_get_map_name_unknown(self): """Test getting name for unknown map ID.""" diff --git a/tests/test_metrics_integration.py b/tests/test_metrics_integration.py index 2a27c34..c13dd26 100644 --- a/tests/test_metrics_integration.py +++ b/tests/test_metrics_integration.py @@ -142,6 +142,40 @@ def mock_get_info(): assert llm_stats["total_calls"] >= 0 assert llm_stats["latency"]["avg"] >= 0 + def test_llm_call_recorded_once(self, mock_pyboy): + """One generate() records one LLM call, not one in the provider and one in the agent.""" + metrics = MetricsCollector() + with patch.object( + OllamaProvider, "_list_available_models", return_value=["llama3.2"] + ): + llm_provider = OllamaProvider(model="llama3.2", metrics=metrics) + llm_provider.client = Mock() + llm_provider.client.chat = Mock( + return_value={"message": {"content": "UP"}, "eval_count": 5} + ) + + game_state = GameState(mock_pyboy, ocr_enabled=False, metrics=metrics) + agent = PokemonAgent(llm_provider, game_state, metrics=metrics) + game_state.get_game_info = Mock( + return_value={ + "screen_text": "open grass", + "frame_count": 40, + "game_state": "overworld", + "party": [{"species": 1, "level": 6}], + "player_position": (3, 3), + "current_map": {}, + } + ) + game_state.execute_action = Mock(return_value=True) + agent.step_count = 10 + agent.new_game_started = True + agent._creation_over = True + + agent.get_action() + + assert metrics.get_all_stats()["llm"]["total_calls"] == 1 + assert llm_provider.client.chat.call_count == 1 + def test_metrics_summary_generation(self): """Test that metrics summary can be generated.""" metrics = MetricsCollector() diff --git a/tests/test_navigation.py b/tests/test_navigation.py new file mode 100644 index 0000000..ec416cc --- /dev/null +++ b/tests/test_navigation.py @@ -0,0 +1,91 @@ +"""Walkability grid and path step.""" + +from unittest.mock import Mock + +from agent_strategy import AgentStrategy +from navigation import camera_origin, downsample_collision, next_step + + +def _open_grid() -> list[list[int]]: + return [[1 for _ in range(10)] for _ in range(9)] + + +def _fence_with_gap(gap_col: int) -> list[list[int]]: + """A wall across row 3, open only at ``gap_col``.""" + grid = _open_grid() + for col in range(10): + grid[3][col] = 0 + grid[3][gap_col] = 1 + return grid + + +def test_camera_stays_at_origin_on_a_screen_sized_map(): + assert camera_origin(5, 6, 10, 9) == (0, 0) + + +def test_camera_centers_the_player_on_a_large_map(): + assert camera_origin(20, 15, 40, 30) == (16, 11) + + +def test_downsample_collision_folds_tiles_into_blocks(): + tiles = [[0 for _ in range(20)] for _ in range(18)] + # One walkable 2x2 block at block (1, 2) -> tiles rows 4-5, cols 2-3 + tiles[4][2] = 1 + tiles[4][3] = 1 + tiles[5][2] = 1 + tiles[5][3] = 1 + grid = downsample_collision(tiles) + assert grid is not None + assert len(grid) == 9 and len(grid[0]) == 10 + assert grid[2][1] == 1 + assert grid[0][0] == 0 + + +def test_downsample_rejects_a_non_grid(): + assert downsample_collision(Mock()) is None + + +def test_open_column_steps_toward_the_target(): + step = next_step(_open_grid(), (2, 4), (2, 1), 10, 9) + assert step == "UP" + + +def test_fence_gap_is_used_instead_of_the_blocked_column(): + # Straight UP from x=2 hits the wall. The only opening is column 6, + # and the target sits on the far side of that opening. + step = next_step(_fence_with_gap(6), (2, 4), (2, 1), 10, 9) + assert step == "RIGHT" + + +def test_does_not_step_into_a_wall_when_already_beside_it(): + grid = _fence_with_gap(6) + # Standing just south of the wall, off the gap column, with the + # target one step north through the wall. Nothing on this column is + # closer, and UP is a wall, so there is no legal step. + assert next_step(grid, (2, 4), (2, 3), 10, 9) is None + + +def test_already_on_the_target_does_not_move(): + assert next_step(_open_grid(), (4, 4), (4, 4), 10, 9) is None + + +def test_movement_target_is_the_oak_tile_on_pallet(): + strategy = AgentStrategy() + strategy.mark_goal_complete("start_game") + goal = strategy.get_current_goal() + target = strategy.movement_target( + goal, + "overworld", + {"current_map": {"map_id": 0x00}, "player_position": (3, 6)}, + ) + assert target == AgentStrategy.OAK_TRIGGER + + +def test_movement_target_skips_dialog_and_the_doorway_clear(): + strategy = AgentStrategy() + strategy.mark_goal_complete("start_game") + goal = strategy.get_current_goal() + memory = {"current_map": {"map_id": 0x00}, "player_position": (3, 6)} + assert strategy.movement_target(goal, "dialog", memory) is None + strategy._exit_maneuver_steps = 2 + assert strategy.movement_target(goal, "overworld", memory) is None diff --git a/tests/test_ocr_enhancer.py b/tests/test_ocr_enhancer.py new file mode 100644 index 0000000..22b12a9 --- /dev/null +++ b/tests/test_ocr_enhancer.py @@ -0,0 +1,26 @@ +"""Tests for OCR region association.""" + +from unittest.mock import Mock + +import numpy as np + +from ocr_enhancer import OCREnhancer, TextRegion + + +def test_dialog_text_stays_paired_after_short_reads_are_dropped(): + """A dropped short read must not shift later strings onto earlier regions.""" + enhancer = OCREnhancer() + image = np.zeros((144, 160, 3), dtype=np.uint8) + regions = [ + {"type": TextRegion.MENU, "confidence": 0.2, "bbox": (0, 0, 10, 10)}, + {"type": TextRegion.MENU, "confidence": 0.4, "bbox": (0, 20, 10, 10)}, + {"type": TextRegion.DIALOG_BOX, "confidence": 0.9, "bbox": (0, 104, 40, 20)}, + ] + enhancer.detect_text_regions = Mock(return_value=regions) + enhancer.extract_text_from_region = Mock( + side_effect=["HI", "OPEN MENU", "OAK SPEAKS"] + ) + + text = enhancer.extract_text_enhanced(image, prioritize_dialog=False) + + assert text == "OAK SPEAKS" diff --git a/tests/test_pokemon_agent.py b/tests/test_pokemon_agent.py index 7f9b25b..61bf3c7 100644 --- a/tests/test_pokemon_agent.py +++ b/tests/test_pokemon_agent.py @@ -3,7 +3,7 @@ from unittest.mock import Mock from game_state import GameState -from pokemon_agent import PokemonAgent +from pokemon_agent import Observation, PokemonAgent class TestPokemonAgent: @@ -406,3 +406,85 @@ def test_counter_resets_when_position_changes(self, mock_llm_provider, mock_pybo ) agent.get_action() assert agent._dialog_loop_steps == 1 + + +class TestPolicyChainNotSkipped: + """step() must run early-game and menu policies, not a raw strategy button.""" + + def test_step_uses_naming_script(self, mock_llm_provider, mock_pyboy): + game_state = GameState(mock_pyboy, ocr_enabled=False) + agent = PokemonAgent(mock_llm_provider, game_state) + game_state.get_game_info = Mock( + return_value={ + "screen_text": "NEW NAME RED ASH JACK", + "frame_count": 100, + "game_state": "menu", + "party": [], + "player_position": (0, 0), + } + ) + game_state.execute_action = Mock(return_value=True) + + result = agent.step() + + # Strategy's start_game menu suggestion is A. The naming script is DOWN. + assert result["action"] == "DOWN" + + def test_step_closes_lingering_menu(self, mock_llm_provider, mock_pyboy): + game_state = GameState(mock_pyboy, ocr_enabled=False) + agent = PokemonAgent(mock_llm_provider, game_state) + game_state.get_game_info = Mock( + return_value={ + "screen_text": "", + "frame_count": 100, + "game_state": "menu", + "party": [{"species": 1, "level": 6}], + "player_position": (5, 5), + } + ) + game_state.execute_action = Mock(return_value=True) + agent.new_game_started = True + agent.character_creation_steps = 60 + agent.step_count = 200 + + actions = [agent.step()["action"] for _ in range(3)] + + # Strategy would keep returning DOWN. Menu escape presses B. + assert actions[-1] == "B" + + +class TestParseLlmResponse: + """Button tokens are whole words, and WAIT is capped.""" + + def _agent(self, mock_llm_provider, mock_pyboy): + game_state = GameState(mock_pyboy, ocr_enabled=False) + return PokemonAgent(mock_llm_provider, game_state) + + def _obs(self) -> Observation: + return Observation( + game_info={"frame_count": 1}, + screen_text="", + game_state="title_screen", + step_count=10, + ) + + def test_start_is_not_a(self, mock_llm_provider, mock_pyboy): + agent = self._agent(mock_llm_provider, mock_pyboy) + obs = self._obs() + assert agent._parse_llm_response(obs, "START") == "START" + assert agent._parse_llm_response(obs, "PRESS START") == "START" + assert agent._parse_llm_response(obs, "GO EAST") == "START" + + def test_prose_without_a_button_falls_back(self, mock_llm_provider, mock_pyboy): + agent = self._agent(mock_llm_provider, mock_pyboy) + agent.action_history = ["A", "A", "A", "A", "A"] + assert agent._parse_llm_response(self._obs(), "GO EAST") == "A" + + def test_wait_is_capped(self, mock_llm_provider, mock_pyboy): + agent = self._agent(mock_llm_provider, mock_pyboy) + assert agent._parse_llm_response(self._obs(), "WAIT 999999") == "WAIT 30" + assert agent._parse_llm_response(self._obs(), "WAIT") == "WAIT 10" + + def test_comma_separated_keeps_two_buttons(self, mock_llm_provider, mock_pyboy): + agent = self._agent(mock_llm_provider, mock_pyboy) + assert agent._parse_llm_response(self._obs(), "UP, DOWN, LEFT") == "UP, DOWN"