gameboy_worlds.interface.legend_of_zelda.actions
1from typing import Any, Dict, List, Optional, Tuple 2 3import numpy as np 4from gymnasium.spaces import Discrete 5 6from gameboy_worlds.emulation import LowLevelActions 7from gameboy_worlds.emulation.legend_of_zelda.parsers import BaseLegendOfZeldaParser 8from gameboy_worlds.emulation.legend_of_zelda.trackers import CoreLegendOfZeldaTracker 9from gameboy_worlds.interface.action import HighLevelAction, SingleHighLevelAction 10 11HARD_MAX_STEPS = 5 12 13STATE_METRIC_KEY = ("legend_of_zelda_core", "agent_state") 14FREE_ROAM = "free_roam" 15IN_DIALOGUE = "in_dialogue" 16IN_INVENTORY = "in_inventory" 17 18 19def frame_changed(previous: np.ndarray, current: np.ndarray, epsilon=0.01) -> bool: 20 return np.abs(previous - current).mean() > epsilon 21 22 23class MoveAction(HighLevelAction): 24 """ 25 Move in a cardinal direction for a number of steps. 26 """ 27 28 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 29 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 30 31 _DIRECTION_TO_ACTION = { 32 "up": LowLevelActions.PRESS_ARROW_UP, 33 "down": LowLevelActions.PRESS_ARROW_DOWN, 34 "left": LowLevelActions.PRESS_ARROW_LEFT, 35 "right": LowLevelActions.PRESS_ARROW_RIGHT, 36 } 37 38 def get_action_space(self): 39 return Discrete(4 * HARD_MAX_STEPS) 40 41 def space_to_parameters(self, space_action): 42 if space_action < 0 or space_action >= 4 * HARD_MAX_STEPS: 43 return None 44 if space_action < HARD_MAX_STEPS: 45 direction = "up" 46 steps = space_action 47 elif space_action < 2 * HARD_MAX_STEPS: 48 direction = "down" 49 steps = space_action - HARD_MAX_STEPS 50 elif space_action < 3 * HARD_MAX_STEPS: 51 direction = "left" 52 steps = space_action - 2 * HARD_MAX_STEPS 53 else: 54 direction = "right" 55 steps = space_action - 3 * HARD_MAX_STEPS 56 return {"direction": direction, "steps": steps + 1} 57 58 def parameters_to_space(self, direction: str, steps: int): 59 if ( 60 direction not in self._DIRECTION_TO_ACTION 61 or steps <= 0 62 or steps > HARD_MAX_STEPS 63 ): 64 return None 65 offset = {"up": 0, "down": 1, "left": 2, "right": 3}[direction] * HARD_MAX_STEPS 66 return offset + steps - 1 67 68 def is_valid(self, **kwargs): 69 direction = kwargs.get("direction", None) 70 steps = kwargs.get("steps", None) 71 if direction is not None and direction not in self._DIRECTION_TO_ACTION: 72 return False 73 if steps is not None and (not isinstance(steps, int) or steps <= 0): 74 return False 75 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM 76 77 def _execute(self, direction: str, steps: int): 78 action = self._DIRECTION_TO_ACTION[direction] 79 transition_state_dicts: List[Dict[str, Dict[str, Any]]] = [] 80 n_steps_taken = 0 81 previous_frame = self._emulator.get_current_frame() 82 action_success = -1 83 for _ in range(steps): 84 frames, done = self._emulator.step(action) 85 report = self._state_tracker.report() 86 transition_state_dicts.append(report) 87 current_frame = frames[-1] 88 if not frame_changed(previous_frame, current_frame): 89 action_success = 1 if n_steps_taken > 0 else -1 90 break 91 n_steps_taken += 1 92 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 93 if state != FREE_ROAM: 94 action_success = 2 95 break 96 if done: 97 action_success = 0 98 break 99 previous_frame = current_frame 100 else: 101 action_success = 0 102 if len(transition_state_dicts) > 0: 103 transition_state_dicts[-1]["core"]["action_return"] = { 104 "n_steps_taken": n_steps_taken 105 } 106 return transition_state_dicts, action_success 107 108 @staticmethod 109 def get_action_name(direction: str, steps: int) -> str: 110 return f"Move {direction} {steps}" 111 112 113class OpenInventoryAction(SingleHighLevelAction): 114 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 115 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 116 117 def is_valid(self, **kwargs): 118 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM 119 120 def _execute(self): 121 previous = self._emulator.get_current_frame() 122 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_START) 123 report = self._state_tracker.report() 124 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 125 if not frame_changed(previous, frames[-1]): 126 return [report], -1 127 return [report], 1 if state == IN_INVENTORY else 0 128 129 @staticmethod 130 def get_action_name() -> str: 131 return "OpenInventory" 132 133 134class CloseInventoryAction(SingleHighLevelAction): 135 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 136 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 137 138 def is_valid(self, **kwargs): 139 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == IN_INVENTORY 140 141 def _execute(self): 142 previous = self._emulator.get_current_frame() 143 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_START) 144 report = self._state_tracker.report() 145 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 146 if not frame_changed(previous, frames[-1]): 147 return [report], -1 148 return [report], 1 if state != IN_INVENTORY else 0 149 150 @staticmethod 151 def get_action_name() -> str: 152 return "CloseInventory" 153 154 155class SkipDialogueAction(SingleHighLevelAction): 156 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 157 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 158 159 def is_valid(self, **kwargs): 160 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == IN_DIALOGUE 161 162 def _execute(self): 163 previous = self._emulator.get_current_frame() 164 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_B) 165 report = self._state_tracker.report() 166 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 167 if not frame_changed(previous, frames[-1]): 168 return [report], -1 169 return [report], 0 if state != IN_DIALOGUE else 1 170 171 @staticmethod 172 def get_action_name() -> str: 173 return "SkipDialogue" 174 175 176class InteractAction(SingleHighLevelAction): 177 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 178 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 179 180 def is_valid(self, **kwargs): 181 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM 182 183 def _execute(self): 184 previous = self._emulator.get_current_frame() 185 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_A) 186 report = self._state_tracker.report() 187 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 188 if not frame_changed(previous, frames[-1]): 189 return [report], -1 190 return [report], 1 if state != FREE_ROAM else 0 191 192 @staticmethod 193 def get_action_name() -> str: 194 return "Interact" 195 196 197class UseOtherInventoryItemAction(SingleHighLevelAction): 198 """ 199 Uses the secondary equipped item (B button) while in free roam. 200 """ 201 202 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 203 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 204 205 def is_valid(self, **kwargs): 206 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM 207 208 def _execute(self): 209 previous = self._emulator.get_current_frame() 210 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_B) 211 report = self._state_tracker.report() 212 if not frame_changed(previous, frames[-1]): 213 return [report], -1 214 return [report], 0 215 216 @staticmethod 217 def get_action_name() -> str: 218 return "UseOtherInventoryItem"
24class MoveAction(HighLevelAction): 25 """ 26 Move in a cardinal direction for a number of steps. 27 """ 28 29 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 30 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 31 32 _DIRECTION_TO_ACTION = { 33 "up": LowLevelActions.PRESS_ARROW_UP, 34 "down": LowLevelActions.PRESS_ARROW_DOWN, 35 "left": LowLevelActions.PRESS_ARROW_LEFT, 36 "right": LowLevelActions.PRESS_ARROW_RIGHT, 37 } 38 39 def get_action_space(self): 40 return Discrete(4 * HARD_MAX_STEPS) 41 42 def space_to_parameters(self, space_action): 43 if space_action < 0 or space_action >= 4 * HARD_MAX_STEPS: 44 return None 45 if space_action < HARD_MAX_STEPS: 46 direction = "up" 47 steps = space_action 48 elif space_action < 2 * HARD_MAX_STEPS: 49 direction = "down" 50 steps = space_action - HARD_MAX_STEPS 51 elif space_action < 3 * HARD_MAX_STEPS: 52 direction = "left" 53 steps = space_action - 2 * HARD_MAX_STEPS 54 else: 55 direction = "right" 56 steps = space_action - 3 * HARD_MAX_STEPS 57 return {"direction": direction, "steps": steps + 1} 58 59 def parameters_to_space(self, direction: str, steps: int): 60 if ( 61 direction not in self._DIRECTION_TO_ACTION 62 or steps <= 0 63 or steps > HARD_MAX_STEPS 64 ): 65 return None 66 offset = {"up": 0, "down": 1, "left": 2, "right": 3}[direction] * HARD_MAX_STEPS 67 return offset + steps - 1 68 69 def is_valid(self, **kwargs): 70 direction = kwargs.get("direction", None) 71 steps = kwargs.get("steps", None) 72 if direction is not None and direction not in self._DIRECTION_TO_ACTION: 73 return False 74 if steps is not None and (not isinstance(steps, int) or steps <= 0): 75 return False 76 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM 77 78 def _execute(self, direction: str, steps: int): 79 action = self._DIRECTION_TO_ACTION[direction] 80 transition_state_dicts: List[Dict[str, Dict[str, Any]]] = [] 81 n_steps_taken = 0 82 previous_frame = self._emulator.get_current_frame() 83 action_success = -1 84 for _ in range(steps): 85 frames, done = self._emulator.step(action) 86 report = self._state_tracker.report() 87 transition_state_dicts.append(report) 88 current_frame = frames[-1] 89 if not frame_changed(previous_frame, current_frame): 90 action_success = 1 if n_steps_taken > 0 else -1 91 break 92 n_steps_taken += 1 93 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 94 if state != FREE_ROAM: 95 action_success = 2 96 break 97 if done: 98 action_success = 0 99 break 100 previous_frame = current_frame 101 else: 102 action_success = 0 103 if len(transition_state_dicts) > 0: 104 transition_state_dicts[-1]["core"]["action_return"] = { 105 "n_steps_taken": n_steps_taken 106 } 107 return transition_state_dicts, action_success 108 109 @staticmethod 110 def get_action_name(direction: str, steps: int) -> str: 111 return f"Move {direction} {steps}"
Move in a cardinal direction for a number of steps.
The state tracker that tracks the minimal state information required for the action to function.
The state parser that parses the minimal state information required for the action to function.
Returns the Gym defined Space that characterizes the high level action's parameter space.
You can use this API to get a Space for sampling high level actions of this type.
Returns:
Space: The Gym space that characterizes the high level action's parameter space.
42 def space_to_parameters(self, space_action): 43 if space_action < 0 or space_action >= 4 * HARD_MAX_STEPS: 44 return None 45 if space_action < HARD_MAX_STEPS: 46 direction = "up" 47 steps = space_action 48 elif space_action < 2 * HARD_MAX_STEPS: 49 direction = "down" 50 steps = space_action - HARD_MAX_STEPS 51 elif space_action < 3 * HARD_MAX_STEPS: 52 direction = "left" 53 steps = space_action - 2 * HARD_MAX_STEPS 54 else: 55 direction = "right" 56 steps = space_action - 3 * HARD_MAX_STEPS 57 return {"direction": direction, "steps": steps + 1}
Converts a Gym space action into high level action parameters. If the provided space action is invalid, return None.
Parameters
- space_action: The action in the high level action's parameter space.
Returns
The high level action's parameters corresponding to the space action.
59 def parameters_to_space(self, direction: str, steps: int): 60 if ( 61 direction not in self._DIRECTION_TO_ACTION 62 or steps <= 0 63 or steps > HARD_MAX_STEPS 64 ): 65 return None 66 offset = {"up": 0, "down": 1, "left": 2, "right": 3}[direction] * HARD_MAX_STEPS 67 return offset + steps - 1
Converts high level action parameters into a Gym space action. If the provided parameters are invalid, return None.
Parameters
- kwargs: The high level action's parameters.
Returns
The action in the high level action's parameter space.
69 def is_valid(self, **kwargs): 70 direction = kwargs.get("direction", None) 71 steps = kwargs.get("steps", None) 72 if direction is not None and direction not in self._DIRECTION_TO_ACTION: 73 return False 74 if steps is not None and (not isinstance(steps, int) or steps <= 0): 75 return False 76 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM
Checks if the high level action can be performed in the current state. If kwargs is empty, then must check whether there exists any valid way to perform the action.
Arguments:
- **kwargs: Additional arguments required for the specific high level action.
Returns:
bool: Whether the action is valid in the current state.
109 @staticmethod 110 def get_action_name(direction: str, steps: int) -> str: 111 return f"Move {direction} {steps}"
Returns a human readable name for the high level action with the given parameters.
Parameters
- kwargs: The high level action's parameters.
Returns
A human readable name for the high level action.
114class OpenInventoryAction(SingleHighLevelAction): 115 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 116 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 117 118 def is_valid(self, **kwargs): 119 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM 120 121 def _execute(self): 122 previous = self._emulator.get_current_frame() 123 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_START) 124 report = self._state_tracker.report() 125 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 126 if not frame_changed(previous, frames[-1]): 127 return [report], -1 128 return [report], 1 if state == IN_INVENTORY else 0 129 130 @staticmethod 131 def get_action_name() -> str: 132 return "OpenInventory"
An abstract class for a high level action that has only one possible parameterization.
The state tracker that tracks the minimal state information required for the action to function.
The state parser that parses the minimal state information required for the action to function.
118 def is_valid(self, **kwargs): 119 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM
Checks if the high level action can be performed in the current state. If kwargs is empty, then must check whether there exists any valid way to perform the action.
Arguments:
- **kwargs: Additional arguments required for the specific high level action.
Returns:
bool: Whether the action is valid in the current state.
135class CloseInventoryAction(SingleHighLevelAction): 136 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 137 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 138 139 def is_valid(self, **kwargs): 140 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == IN_INVENTORY 141 142 def _execute(self): 143 previous = self._emulator.get_current_frame() 144 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_START) 145 report = self._state_tracker.report() 146 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 147 if not frame_changed(previous, frames[-1]): 148 return [report], -1 149 return [report], 1 if state != IN_INVENTORY else 0 150 151 @staticmethod 152 def get_action_name() -> str: 153 return "CloseInventory"
An abstract class for a high level action that has only one possible parameterization.
The state tracker that tracks the minimal state information required for the action to function.
The state parser that parses the minimal state information required for the action to function.
139 def is_valid(self, **kwargs): 140 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == IN_INVENTORY
Checks if the high level action can be performed in the current state. If kwargs is empty, then must check whether there exists any valid way to perform the action.
Arguments:
- **kwargs: Additional arguments required for the specific high level action.
Returns:
bool: Whether the action is valid in the current state.
156class SkipDialogueAction(SingleHighLevelAction): 157 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 158 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 159 160 def is_valid(self, **kwargs): 161 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == IN_DIALOGUE 162 163 def _execute(self): 164 previous = self._emulator.get_current_frame() 165 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_B) 166 report = self._state_tracker.report() 167 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 168 if not frame_changed(previous, frames[-1]): 169 return [report], -1 170 return [report], 0 if state != IN_DIALOGUE else 1 171 172 @staticmethod 173 def get_action_name() -> str: 174 return "SkipDialogue"
An abstract class for a high level action that has only one possible parameterization.
The state tracker that tracks the minimal state information required for the action to function.
The state parser that parses the minimal state information required for the action to function.
160 def is_valid(self, **kwargs): 161 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == IN_DIALOGUE
Checks if the high level action can be performed in the current state. If kwargs is empty, then must check whether there exists any valid way to perform the action.
Arguments:
- **kwargs: Additional arguments required for the specific high level action.
Returns:
bool: Whether the action is valid in the current state.
177class InteractAction(SingleHighLevelAction): 178 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 179 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 180 181 def is_valid(self, **kwargs): 182 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM 183 184 def _execute(self): 185 previous = self._emulator.get_current_frame() 186 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_A) 187 report = self._state_tracker.report() 188 state = self._state_tracker.get_episode_metric(STATE_METRIC_KEY) 189 if not frame_changed(previous, frames[-1]): 190 return [report], -1 191 return [report], 1 if state != FREE_ROAM else 0 192 193 @staticmethod 194 def get_action_name() -> str: 195 return "Interact"
An abstract class for a high level action that has only one possible parameterization.
The state tracker that tracks the minimal state information required for the action to function.
The state parser that parses the minimal state information required for the action to function.
181 def is_valid(self, **kwargs): 182 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM
Checks if the high level action can be performed in the current state. If kwargs is empty, then must check whether there exists any valid way to perform the action.
Arguments:
- **kwargs: Additional arguments required for the specific high level action.
Returns:
bool: Whether the action is valid in the current state.
198class UseOtherInventoryItemAction(SingleHighLevelAction): 199 """ 200 Uses the secondary equipped item (B button) while in free roam. 201 """ 202 203 REQUIRED_STATE_TRACKER = CoreLegendOfZeldaTracker 204 REQUIRED_STATE_PARSER = BaseLegendOfZeldaParser 205 206 def is_valid(self, **kwargs): 207 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM 208 209 def _execute(self): 210 previous = self._emulator.get_current_frame() 211 frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_B) 212 report = self._state_tracker.report() 213 if not frame_changed(previous, frames[-1]): 214 return [report], -1 215 return [report], 0 216 217 @staticmethod 218 def get_action_name() -> str: 219 return "UseOtherInventoryItem"
Uses the secondary equipped item (B button) while in free roam.
The state tracker that tracks the minimal state information required for the action to function.
The state parser that parses the minimal state information required for the action to function.
206 def is_valid(self, **kwargs): 207 return self._state_tracker.get_episode_metric(STATE_METRIC_KEY) == FREE_ROAM
Checks if the high level action can be performed in the current state. If kwargs is empty, then must check whether there exists any valid way to perform the action.
Arguments:
- **kwargs: Additional arguments required for the specific high level action.
Returns:
bool: Whether the action is valid in the current state.