gameboy_worlds.interface.harvest_moon.actions

View Source

  1from gameboy_worlds.utils import log_error, log_warn
  2from gameboy_worlds.interface.action import HighLevelAction, SingleHighLevelAction
  3from gameboy_worlds.emulation.harvest_moon.parsers import (
  4    AgentState,
  5    HarvestMoonStateParser,
  6    BaseHarvestMoonStateParser,
  7)
  8from gameboy_worlds.emulation.harvest_moon.trackers import CoreHarvestMoonTracker
  9from gameboy_worlds.emulation import LowLevelActions
 10from abc import ABC, abstractmethod
 11from typing import List, Tuple, Dict
 12from gameboy_worlds.utils import show_frames
 13import numpy as np
 14
 15from gymnasium.spaces import Box, Discrete, Text, OneOf
 16import matplotlib.pyplot as plt
 17from PIL import Image
 18
 19HARD_MAX_STEPS = 5
 20""" The hard maximum number of steps we'll let agents take in a sequence """
 21
 22
 23def frame_changed(past: np.ndarray, preset: np.ndarray, epsilon=0.01):
 24    return np.abs(past - preset).mean() > epsilon
 25
 26
 27def _plot(past: np.ndarray, current: np.ndarray):
 28    # plot both side by side for debug
 29    fig, axs = plt.subplots(1, 2)
 30    axs[0].imshow(past)
 31    axs[1].imshow(current)
 32    plt.show()
 33
 34
 35class PassDialogueAction(SingleHighLevelAction):
 36    """
 37    Skips dialogue by pressing the B button.
 38
 39    Is Valid When:
 40    - In Dialogue State
 41
 42    Action Success Interpretation:
 43    - -1: Frame did not change
 44    - 0: Frame changed and no longer in dialogue state
 45    - 1: Frame changed but still in dialogue state
 46    """
 47
 48    REQUIRED_STATE_PARSER = HarvestMoonStateParser
 49    REQUIRED_STATE_TRACKER = CoreHarvestMoonTracker
 50
 51    def is_valid(self, **kwargs):
 52        """
 53        Just checks if the agent is in dialogue state.
 54        """
 55        return (
 56            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
 57            == AgentState.IN_DIALOGUE
 58        )
 59
 60    def _execute(self):
 61        frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_B)
 62        report = self._state_tracker.report()
 63        if not report["core"]["frame_changed"]:
 64            action_success = -1
 65        else:
 66            action_success = (
 67                0
 68                if self._emulator.state_parser.get_agent_state(frames[-1])
 69                != AgentState.IN_DIALOGUE
 70                else 1
 71            )
 72        return [report], action_success
 73
 74    @staticmethod
 75    def get_action_name() -> str:
 76        return "PassDialogue"
 77
 78
 79class InteractAction(SingleHighLevelAction):
 80    """
 81    Presses the A button to interact with an object in front of the agent.
 82
 83    Is Valid When:
 84    - In Free Roam State
 85
 86    Action Success Interpretation:
 87    - -1: Frame did not change or agent still in free roam state
 88    - 1: Agent not in free roam state
 89    """
 90
 91    REQUIRED_STATE_PARSER = HarvestMoonStateParser
 92    REQUIRED_STATE_TRACKER = CoreHarvestMoonTracker
 93
 94    def is_valid(self, **kwargs):
 95        """
 96        Just checks if the agent is in free roam state.
 97        """
 98        return (
 99            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
100            == AgentState.FREE_ROAM
101        )
102
103    def _execute(self):
104        current_frame = self._emulator.get_current_frame()
105        frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_A)
106        action_success = 0
107        # Check if the frames have changed. Be strict and require all to not permit jittering screens.
108        prev_frames = []
109        for frame in frames:
110            if (
111                self._emulator.state_parser.get_agent_state(frame)
112                != AgentState.FREE_ROAM
113            ):  # something happened lol
114                action_success = 1
115                break
116            for past_frame in prev_frames:
117                if not frame_changed(past_frame, frame):
118                    action_success = -1
119                    break
120            if action_success != 0:
121                break
122            prev_frames.append(frame)
123        if action_success == 0:
124            action_success = (
125                -1
126            )  # I guess? For some reason the previous thing doesn't catch same frames
127        return [
128            self._state_tracker.report()
129        ], action_success  # 0 means something maybe happened. 1 means def happened.
130
131    @staticmethod
132    def get_action_name() -> str:
133        return "Interact"
134
135
136class BaseMovementAction(HighLevelAction, ABC):
137    """
138    Base class for movement actions in the Harvest Moon environment.
139    Has utility methods for moving in directions.
140
141    Is Valid When:
142    - In Free Roam State
143
144    Action Success Interpretation:
145    - -1: Frame did not change, even on the first step
146    - 0: Finished all steps
147    - 1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
148    - 2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.
149
150    Action Returns:
151    - `n_steps_taken` (`int`): Number of steps actually taken
152    - `rotated` (`bool` or `None`): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.
153    """
154
155    REQUIRED_STATE_TRACKER = CoreHarvestMoonTracker
156    REQUIRED_STATE_PARSER = HarvestMoonStateParser
157
158    def judge_movement(
159        self, previous_frame: np.ndarray, current_frame: np.ndarray
160    ) -> Tuple[bool, bool]:
161        """
162        Judges whether movement has occurred between two frames.
163
164        Args:
165            previous_frame (np.ndarray): The previous frame.
166            current_frame (np.ndarray): The current frame.
167        Returns:
168            Tuple[bool, bool]: A tuple containing:
169            - bool: True if movement has occurred, False otherwise.
170            - bool: True if the player has not moved, but has rotated.
171        """
172        if not frame_changed(previous_frame, current_frame):
173            return False, False
174        return True, None
175
176    def move(self, direction: str, steps: int) -> Tuple[np.ndarray, int]:
177        """
178        Move in a given direction for a number of steps.
179
180        :param direction: One of "up", "down", "left", "right"
181        :type direction: str
182        :param steps: Number of steps to move in that direction
183        :type steps: int
184        :return:  A tuple containing:
185
186                - A list of state tracker reports after each low level action executed. Length is equal to the number of low level actions executed.
187
188                - An integer action success status
189        :rtype: Tuple[ndarray[_AnyShape, dtype[Any]], int]
190        """
191        action_dict = {
192            "right": LowLevelActions.PRESS_ARROW_RIGHT,
193            "down": LowLevelActions.PRESS_ARROW_DOWN,
194            "up": LowLevelActions.PRESS_ARROW_UP,
195            "left": LowLevelActions.PRESS_ARROW_LEFT,
196        }
197        if direction not in action_dict.keys():
198            log_error(f"Got invalid direction to move {direction}", self._parameters)
199        action = action_dict[direction]
200        # keep trying the action.
201        # exit status 0 -> finished steps
202        # 1 -> took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter)
203        # 2 -> took some steps, but agent state changed from free roam
204        # -1 -> frame didn't change, even on the first step
205        action_success = -1
206        transition_state_dicts = []
207        transition_frames = []
208        previous_frame = (
209            self._emulator.get_current_frame()
210        )  # Do NOT get the state tracker frame, as it may have a grid on it.
211        n_step = 0
212        n_successful_steps = 0
213        has_rotated = None
214        agent_state = AgentState.FREE_ROAM
215        while n_step < steps and agent_state == AgentState.FREE_ROAM:
216            frames, done = self._emulator.step(action)
217            transition_state_dicts.append(self._state_tracker.report())
218            transition_frames.extend(frames)
219            current_frame = (
220                self._emulator.get_current_frame()
221            )  # Do NOT use the emulator frame, as it may have a grid on it.
222            if done:
223                break
224            # check if frames changed. If not, break out.
225            player_moved, player_rotated = self.judge_movement(
226                previous_frame, current_frame
227            )
228            if player_rotated == True:
229                has_rotated = True
230            if not player_moved and not player_rotated:
231                break
232            if player_moved:
233                n_successful_steps += 1  # don't count rotation as a step
234            agent_state = self._emulator.state_parser.get_agent_state(
235                self._emulator.get_current_frame()
236            )
237            if agent_state != AgentState.FREE_ROAM:
238                break
239            n_step += 1
240            previous_frame = current_frame
241        if agent_state != AgentState.FREE_ROAM:
242            action_success = 2
243        else:
244            if n_step <= 0:
245                action_success = -1
246            elif n_step == steps:
247                action_success = 0
248            else:
249                action_success = 1
250        transition_state_dicts[-1]["core"]["action_return"] = {
251            "n_steps_taken": n_successful_steps,
252            "rotated": has_rotated,
253        }
254        return transition_state_dicts, action_success
255
256    def is_valid(self, **kwargs):
257        """
258        Just checks if the agent is in free roam state.
259        """
260        return (
261            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
262            == AgentState.FREE_ROAM
263        )
264
265
266class MoveStepsAction(BaseMovementAction):
267    """
268    Moves the agent in a specified cardinal direction for a specified number of steps.
269
270    Is Valid When:
271    - In Free Roam State
272    Action Success Interpretation:
273    - -1: Frame did not change, even on the first step
274    - 0: Finished all steps
275    - 1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
276    - 2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.
277
278    Action Returns:
279    - `n_steps_taken` (`int`): Number of steps actually taken
280    - `rotated` (`bool` or `None`): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.
281    """
282
283    def get_action_space(self):
284        """
285        Returns a Box space representing movement in 2D.
286        The first dimension represents vertical movement (positive is up, negative is down).
287        The second dimension represents horizontal movement (positive is right, negative is left).
288
289        Returns:
290            Box: A Box space with shape (2,) and values ranging from -HARD_MAX_STEPS//2 to HARD_MAX_STEPS//2.
291
292        """
293        return Discrete(4 * HARD_MAX_STEPS)
294
295    def space_to_parameters(self, space_action):
296        direction = None
297        steps = None
298        if space_action < 0 or space_action >= 4 * HARD_MAX_STEPS:
299            # log_warn(f"Invalid space action {space_action}", self._parameters)
300            return None
301        if space_action < HARD_MAX_STEPS:
302            direction = "up"
303            steps = space_action
304        elif space_action < 2 * HARD_MAX_STEPS:
305            direction = "down"
306            steps = space_action - HARD_MAX_STEPS
307        elif space_action < 3 * HARD_MAX_STEPS:
308            direction = "left"
309            steps = space_action - 2 * HARD_MAX_STEPS
310        else:
311            direction = "right"
312            steps = space_action - 3 * HARD_MAX_STEPS
313        return {"direction": direction, "steps": steps + 1}
314
315    def parameters_to_space(self, direction: str, steps: int):
316        if steps <= 0 or steps > HARD_MAX_STEPS:
317            return None
318        if direction == "up":
319            return steps - 1
320        elif direction == "down":
321            return HARD_MAX_STEPS + steps - 1
322        elif direction == "left":
323            return 2 * HARD_MAX_STEPS + steps - 1
324        elif direction == "right":
325            return 3 * HARD_MAX_STEPS + steps - 1
326        else:
327            # log_warn(f"Unrecognized direction {direction}", self._parameters)
328            return None
329
330    def _execute(self, direction, steps):
331        transition_states, status = self.move(direction=direction, steps=steps)
332        return transition_states, status
333
334    def is_valid(self, **kwargs):
335        direction = kwargs.get("direction")
336        step = kwargs.get("step")
337        if direction is not None and direction not in ["up", "down", "left", "right"]:
338            return False
339        if step is not None:
340            if not isinstance(step, str):
341                return False
342            if step <= 0:
343                return False
344        return super().is_valid(**kwargs)
345
346    @staticmethod
347    def get_action_name(direction: str, steps: int) -> str:
348        return f"Move {direction} {steps}"
349
350
351class MoveGridAction(BaseMovementAction):
352    """
353    Moves the agent on both axes. Will always try to move right first and then up.
354
355    Is Valid When:
356    - In Free Roam State
357    Action Success Interpretation:
358    - -1: Frame did not change, even on the first step
359    - 0: Finished all steps
360    - 1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
361    - 2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.
362
363    Action Returns:
364    - `n_steps_taken` (`int`): Number of steps actually taken
365    - `rotated` (`bool` or `None`): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.
366    """
367
368    def get_action_space(self):
369        """
370        Returns a Box space representing movement in 2D.
371        The first dimension represents vertical movement (positive is up, negative is down).
372        The second dimension represents horizontal movement (positive is right, negative is left).
373
374        Returns:
375            Box: A Box space with shape (2,) and values ranging from -HARD_MAX_STEPS//2 to HARD_MAX_STEPS//2.
376
377        """
378        return Box(
379            low=-HARD_MAX_STEPS // 2,
380            high=HARD_MAX_STEPS // 2,
381            shape=(2,),
382            dtype=np.int8,
383        )
384
385    def space_to_parameters(self, space_action):
386        right_action = space_action[0]
387        up_action = space_action[1]
388        return {"x_steps": right_action, "y_steps": up_action}
389
390    def parameters_to_space(self, x_steps, y_steps):
391        move_vec = np.zeros(2)  # x, y
392        move_vec[0] = x_steps
393        move_vec[1] = y_steps
394        return move_vec
395
396    def _execute(self, x_steps, y_steps):
397        x_direction = "right" if x_steps >= 0 else "left"
398        y_direction = "up" if y_steps >= 0 else "down"
399        if x_steps != 0:
400            transition_states, status = self.move(
401                direction=x_direction, steps=abs(x_steps)
402            )
403            if status != 0:
404                return transition_states, status
405        else:
406            transition_states = []
407        if y_steps != 0:
408            more_transition_states, status = self.move(
409                direction=y_direction, steps=abs(y_steps)
410            )
411            transition_states.extend(more_transition_states)
412        try:
413            status is not None
414        except NameError:
415            log_warn(
416                f"Weird case where both x_steps and y_steps are 0 in MoveGridAction or something. {x_steps}, {y_steps}",
417                self._parameters,
418            )
419            transition_states = [self._state_tracker.report()]
420            status = -1
421        return transition_states, status
422
423    def is_valid(self, x_steps: int = None, y_steps: int = None):
424        if x_steps is not None and y_steps is not None:
425            if not isinstance(x_steps, int) or not isinstance(y_steps, int):
426                return False
427            if x_steps == 0 and y_steps == 0:
428                return False
429        return super().is_valid()
430
431    @staticmethod
432    def get_action_name(x_steps: int, y_steps: int) -> str:
433        return f"MoveGrid ({x_steps}, {y_steps})"

HARD_MAX_STEPS = 5

The hard maximum number of steps we'll let agents take in a sequence

def frame_changed(past: numpy.ndarray, preset: numpy.ndarray, epsilon=0.01): View Source

24def frame_changed(past: np.ndarray, preset: np.ndarray, epsilon=0.01):
25    return np.abs(past - preset).mean() > epsilon

class PassDialogueAction(gameboy_worlds.interface.action.SingleHighLevelAction): View Source

36class PassDialogueAction(SingleHighLevelAction):
37    """
38    Skips dialogue by pressing the B button.
39
40    Is Valid When:
41    - In Dialogue State
42
43    Action Success Interpretation:
44    - -1: Frame did not change
45    - 0: Frame changed and no longer in dialogue state
46    - 1: Frame changed but still in dialogue state
47    """
48
49    REQUIRED_STATE_PARSER = HarvestMoonStateParser
50    REQUIRED_STATE_TRACKER = CoreHarvestMoonTracker
51
52    def is_valid(self, **kwargs):
53        """
54        Just checks if the agent is in dialogue state.
55        """
56        return (
57            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
58            == AgentState.IN_DIALOGUE
59        )
60
61    def _execute(self):
62        frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_B)
63        report = self._state_tracker.report()
64        if not report["core"]["frame_changed"]:
65            action_success = -1
66        else:
67            action_success = (
68                0
69                if self._emulator.state_parser.get_agent_state(frames[-1])
70                != AgentState.IN_DIALOGUE
71                else 1
72            )
73        return [report], action_success
74
75    @staticmethod
76    def get_action_name() -> str:
77        return "PassDialogue"

Skips dialogue by pressing the B button.

Is Valid When:

In Dialogue State

Action Success Interpretation:

-1: Frame did not change
0: Frame changed and no longer in dialogue state
1: Frame changed but still in dialogue state

REQUIRED_STATE_PARSER = <class 'gameboy_worlds.emulation.harvest_moon.parsers.HarvestMoonStateParser'>

The state parser that parses the minimal state information required for the action to function.

REQUIRED_STATE_TRACKER = <class 'gameboy_worlds.emulation.harvest_moon.trackers.CoreHarvestMoonTracker'>

The state tracker that tracks the minimal state information required for the action to function.

def is_valid(self, **kwargs): View Source

52    def is_valid(self, **kwargs):
53        """
54        Just checks if the agent is in dialogue state.
55        """
56        return (
57            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
58            == AgentState.IN_DIALOGUE
59        )

Just checks if the agent is in dialogue state.

@staticmethod

def get_action_name() -> str: View Source

75    @staticmethod
76    def get_action_name() -> str:
77        return "PassDialogue"

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.

Inherited Members

gameboy_worlds.interface.action.HighLevelAction: HighLevelAction; seed; assign_emulator; unassign_emulator; execute; execute_space_action
gameboy_worlds.interface.action.SingleHighLevelAction: space_to_parameters; parameters_to_space; get_action_space; get_all_valid_parameters

class InteractAction(gameboy_worlds.interface.action.SingleHighLevelAction): View Source

 80class InteractAction(SingleHighLevelAction):
 81    """
 82    Presses the A button to interact with an object in front of the agent.
 83
 84    Is Valid When:
 85    - In Free Roam State
 86
 87    Action Success Interpretation:
 88    - -1: Frame did not change or agent still in free roam state
 89    - 1: Agent not in free roam state
 90    """
 91
 92    REQUIRED_STATE_PARSER = HarvestMoonStateParser
 93    REQUIRED_STATE_TRACKER = CoreHarvestMoonTracker
 94
 95    def is_valid(self, **kwargs):
 96        """
 97        Just checks if the agent is in free roam state.
 98        """
 99        return (
100            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
101            == AgentState.FREE_ROAM
102        )
103
104    def _execute(self):
105        current_frame = self._emulator.get_current_frame()
106        frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_A)
107        action_success = 0
108        # Check if the frames have changed. Be strict and require all to not permit jittering screens.
109        prev_frames = []
110        for frame in frames:
111            if (
112                self._emulator.state_parser.get_agent_state(frame)
113                != AgentState.FREE_ROAM
114            ):  # something happened lol
115                action_success = 1
116                break
117            for past_frame in prev_frames:
118                if not frame_changed(past_frame, frame):
119                    action_success = -1
120                    break
121            if action_success != 0:
122                break
123            prev_frames.append(frame)
124        if action_success == 0:
125            action_success = (
126                -1
127            )  # I guess? For some reason the previous thing doesn't catch same frames
128        return [
129            self._state_tracker.report()
130        ], action_success  # 0 means something maybe happened. 1 means def happened.
131
132    @staticmethod
133    def get_action_name() -> str:
134        return "Interact"

Presses the A button to interact with an object in front of the agent.

Is Valid When:

In Free Roam State

Action Success Interpretation:

-1: Frame did not change or agent still in free roam state
1: Agent not in free roam state

REQUIRED_STATE_PARSER = <class 'gameboy_worlds.emulation.harvest_moon.parsers.HarvestMoonStateParser'>

The state parser that parses the minimal state information required for the action to function.

REQUIRED_STATE_TRACKER = <class 'gameboy_worlds.emulation.harvest_moon.trackers.CoreHarvestMoonTracker'>

The state tracker that tracks the minimal state information required for the action to function.

def is_valid(self, **kwargs): View Source

 95    def is_valid(self, **kwargs):
 96        """
 97        Just checks if the agent is in free roam state.
 98        """
 99        return (
100            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
101            == AgentState.FREE_ROAM
102        )

Just checks if the agent is in free roam state.

@staticmethod

def get_action_name() -> str: View Source

132    @staticmethod
133    def get_action_name() -> str:
134        return "Interact"

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.

Inherited Members

gameboy_worlds.interface.action.HighLevelAction: HighLevelAction; seed; assign_emulator; unassign_emulator; execute; execute_space_action
gameboy_worlds.interface.action.SingleHighLevelAction: space_to_parameters; parameters_to_space; get_action_space; get_all_valid_parameters

class BaseMovementAction(gameboy_worlds.interface.action.HighLevelAction, abc.ABC): View Source

137class BaseMovementAction(HighLevelAction, ABC):
138    """
139    Base class for movement actions in the Harvest Moon environment.
140    Has utility methods for moving in directions.
141
142    Is Valid When:
143    - In Free Roam State
144
145    Action Success Interpretation:
146    - -1: Frame did not change, even on the first step
147    - 0: Finished all steps
148    - 1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
149    - 2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.
150
151    Action Returns:
152    - `n_steps_taken` (`int`): Number of steps actually taken
153    - `rotated` (`bool` or `None`): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.
154    """
155
156    REQUIRED_STATE_TRACKER = CoreHarvestMoonTracker
157    REQUIRED_STATE_PARSER = HarvestMoonStateParser
158
159    def judge_movement(
160        self, previous_frame: np.ndarray, current_frame: np.ndarray
161    ) -> Tuple[bool, bool]:
162        """
163        Judges whether movement has occurred between two frames.
164
165        Args:
166            previous_frame (np.ndarray): The previous frame.
167            current_frame (np.ndarray): The current frame.
168        Returns:
169            Tuple[bool, bool]: A tuple containing:
170            - bool: True if movement has occurred, False otherwise.
171            - bool: True if the player has not moved, but has rotated.
172        """
173        if not frame_changed(previous_frame, current_frame):
174            return False, False
175        return True, None
176
177    def move(self, direction: str, steps: int) -> Tuple[np.ndarray, int]:
178        """
179        Move in a given direction for a number of steps.
180
181        :param direction: One of "up", "down", "left", "right"
182        :type direction: str
183        :param steps: Number of steps to move in that direction
184        :type steps: int
185        :return:  A tuple containing:
186
187                - A list of state tracker reports after each low level action executed. Length is equal to the number of low level actions executed.
188
189                - An integer action success status
190        :rtype: Tuple[ndarray[_AnyShape, dtype[Any]], int]
191        """
192        action_dict = {
193            "right": LowLevelActions.PRESS_ARROW_RIGHT,
194            "down": LowLevelActions.PRESS_ARROW_DOWN,
195            "up": LowLevelActions.PRESS_ARROW_UP,
196            "left": LowLevelActions.PRESS_ARROW_LEFT,
197        }
198        if direction not in action_dict.keys():
199            log_error(f"Got invalid direction to move {direction}", self._parameters)
200        action = action_dict[direction]
201        # keep trying the action.
202        # exit status 0 -> finished steps
203        # 1 -> took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter)
204        # 2 -> took some steps, but agent state changed from free roam
205        # -1 -> frame didn't change, even on the first step
206        action_success = -1
207        transition_state_dicts = []
208        transition_frames = []
209        previous_frame = (
210            self._emulator.get_current_frame()
211        )  # Do NOT get the state tracker frame, as it may have a grid on it.
212        n_step = 0
213        n_successful_steps = 0
214        has_rotated = None
215        agent_state = AgentState.FREE_ROAM
216        while n_step < steps and agent_state == AgentState.FREE_ROAM:
217            frames, done = self._emulator.step(action)
218            transition_state_dicts.append(self._state_tracker.report())
219            transition_frames.extend(frames)
220            current_frame = (
221                self._emulator.get_current_frame()
222            )  # Do NOT use the emulator frame, as it may have a grid on it.
223            if done:
224                break
225            # check if frames changed. If not, break out.
226            player_moved, player_rotated = self.judge_movement(
227                previous_frame, current_frame
228            )
229            if player_rotated == True:
230                has_rotated = True
231            if not player_moved and not player_rotated:
232                break
233            if player_moved:
234                n_successful_steps += 1  # don't count rotation as a step
235            agent_state = self._emulator.state_parser.get_agent_state(
236                self._emulator.get_current_frame()
237            )
238            if agent_state != AgentState.FREE_ROAM:
239                break
240            n_step += 1
241            previous_frame = current_frame
242        if agent_state != AgentState.FREE_ROAM:
243            action_success = 2
244        else:
245            if n_step <= 0:
246                action_success = -1
247            elif n_step == steps:
248                action_success = 0
249            else:
250                action_success = 1
251        transition_state_dicts[-1]["core"]["action_return"] = {
252            "n_steps_taken": n_successful_steps,
253            "rotated": has_rotated,
254        }
255        return transition_state_dicts, action_success
256
257    def is_valid(self, **kwargs):
258        """
259        Just checks if the agent is in free roam state.
260        """
261        return (
262            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
263            == AgentState.FREE_ROAM
264        )

Base class for movement actions in the Harvest Moon environment. Has utility methods for moving in directions.

Is Valid When:

In Free Roam State

Action Success Interpretation:

-1: Frame did not change, even on the first step
0: Finished all steps
1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.

Action Returns:

n_steps_taken (int): Number of steps actually taken
rotated (bool or None): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.

REQUIRED_STATE_TRACKER = <class 'gameboy_worlds.emulation.harvest_moon.trackers.CoreHarvestMoonTracker'>

The state tracker that tracks the minimal state information required for the action to function.

REQUIRED_STATE_PARSER = <class 'gameboy_worlds.emulation.harvest_moon.parsers.HarvestMoonStateParser'>

The state parser that parses the minimal state information required for the action to function.

def judge_movement( self, previous_frame: numpy.ndarray, current_frame: numpy.ndarray) -> Tuple[bool, bool]: View Source

159    def judge_movement(
160        self, previous_frame: np.ndarray, current_frame: np.ndarray
161    ) -> Tuple[bool, bool]:
162        """
163        Judges whether movement has occurred between two frames.
164
165        Args:
166            previous_frame (np.ndarray): The previous frame.
167            current_frame (np.ndarray): The current frame.
168        Returns:
169            Tuple[bool, bool]: A tuple containing:
170            - bool: True if movement has occurred, False otherwise.
171            - bool: True if the player has not moved, but has rotated.
172        """
173        if not frame_changed(previous_frame, current_frame):
174            return False, False
175        return True, None

Judges whether movement has occurred between two frames.

Arguments:

previous_frame (np.ndarray): The previous frame.
current_frame (np.ndarray): The current frame.

Returns:

Tuple[bool, bool]: A tuple containing:

bool: True if movement has occurred, False otherwise.

bool: True if the player has not moved, but has rotated.

def move(self, direction: str, steps: int) -> Tuple[numpy.ndarray, int]: View Source

177    def move(self, direction: str, steps: int) -> Tuple[np.ndarray, int]:
178        """
179        Move in a given direction for a number of steps.
180
181        :param direction: One of "up", "down", "left", "right"
182        :type direction: str
183        :param steps: Number of steps to move in that direction
184        :type steps: int
185        :return:  A tuple containing:
186
187                - A list of state tracker reports after each low level action executed. Length is equal to the number of low level actions executed.
188
189                - An integer action success status
190        :rtype: Tuple[ndarray[_AnyShape, dtype[Any]], int]
191        """
192        action_dict = {
193            "right": LowLevelActions.PRESS_ARROW_RIGHT,
194            "down": LowLevelActions.PRESS_ARROW_DOWN,
195            "up": LowLevelActions.PRESS_ARROW_UP,
196            "left": LowLevelActions.PRESS_ARROW_LEFT,
197        }
198        if direction not in action_dict.keys():
199            log_error(f"Got invalid direction to move {direction}", self._parameters)
200        action = action_dict[direction]
201        # keep trying the action.
202        # exit status 0 -> finished steps
203        # 1 -> took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter)
204        # 2 -> took some steps, but agent state changed from free roam
205        # -1 -> frame didn't change, even on the first step
206        action_success = -1
207        transition_state_dicts = []
208        transition_frames = []
209        previous_frame = (
210            self._emulator.get_current_frame()
211        )  # Do NOT get the state tracker frame, as it may have a grid on it.
212        n_step = 0
213        n_successful_steps = 0
214        has_rotated = None
215        agent_state = AgentState.FREE_ROAM
216        while n_step < steps and agent_state == AgentState.FREE_ROAM:
217            frames, done = self._emulator.step(action)
218            transition_state_dicts.append(self._state_tracker.report())
219            transition_frames.extend(frames)
220            current_frame = (
221                self._emulator.get_current_frame()
222            )  # Do NOT use the emulator frame, as it may have a grid on it.
223            if done:
224                break
225            # check if frames changed. If not, break out.
226            player_moved, player_rotated = self.judge_movement(
227                previous_frame, current_frame
228            )
229            if player_rotated == True:
230                has_rotated = True
231            if not player_moved and not player_rotated:
232                break
233            if player_moved:
234                n_successful_steps += 1  # don't count rotation as a step
235            agent_state = self._emulator.state_parser.get_agent_state(
236                self._emulator.get_current_frame()
237            )
238            if agent_state != AgentState.FREE_ROAM:
239                break
240            n_step += 1
241            previous_frame = current_frame
242        if agent_state != AgentState.FREE_ROAM:
243            action_success = 2
244        else:
245            if n_step <= 0:
246                action_success = -1
247            elif n_step == steps:
248                action_success = 0
249            else:
250                action_success = 1
251        transition_state_dicts[-1]["core"]["action_return"] = {
252            "n_steps_taken": n_successful_steps,
253            "rotated": has_rotated,
254        }
255        return transition_state_dicts, action_success

Move in a given direction for a number of steps.

Parameters

direction: One of "up", "down", "left", "right"
steps: Number of steps to move in that direction

Returns

A tuple containing:

    - A list of state tracker reports after each low level action executed. Length is equal to the number of low level actions executed.

    - An integer action success status

def is_valid(self, **kwargs): View Source

257    def is_valid(self, **kwargs):
258        """
259        Just checks if the agent is in free roam state.
260        """
261        return (
262            self._state_tracker.get_episode_metric(("harvest_moon_core", "agent_state"))
263            == AgentState.FREE_ROAM
264        )

Just checks if the agent is in free roam state.

Inherited Members

gameboy_worlds.interface.action.HighLevelAction: seed; assign_emulator; unassign_emulator; get_action_space; space_to_parameters; parameters_to_space; get_all_valid_parameters; execute; execute_space_action; get_action_name

class MoveStepsAction(BaseMovementAction): View Source

267class MoveStepsAction(BaseMovementAction):
268    """
269    Moves the agent in a specified cardinal direction for a specified number of steps.
270
271    Is Valid When:
272    - In Free Roam State
273    Action Success Interpretation:
274    - -1: Frame did not change, even on the first step
275    - 0: Finished all steps
276    - 1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
277    - 2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.
278
279    Action Returns:
280    - `n_steps_taken` (`int`): Number of steps actually taken
281    - `rotated` (`bool` or `None`): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.
282    """
283
284    def get_action_space(self):
285        """
286        Returns a Box space representing movement in 2D.
287        The first dimension represents vertical movement (positive is up, negative is down).
288        The second dimension represents horizontal movement (positive is right, negative is left).
289
290        Returns:
291            Box: A Box space with shape (2,) and values ranging from -HARD_MAX_STEPS//2 to HARD_MAX_STEPS//2.
292
293        """
294        return Discrete(4 * HARD_MAX_STEPS)
295
296    def space_to_parameters(self, space_action):
297        direction = None
298        steps = None
299        if space_action < 0 or space_action >= 4 * HARD_MAX_STEPS:
300            # log_warn(f"Invalid space action {space_action}", self._parameters)
301            return None
302        if space_action < HARD_MAX_STEPS:
303            direction = "up"
304            steps = space_action
305        elif space_action < 2 * HARD_MAX_STEPS:
306            direction = "down"
307            steps = space_action - HARD_MAX_STEPS
308        elif space_action < 3 * HARD_MAX_STEPS:
309            direction = "left"
310            steps = space_action - 2 * HARD_MAX_STEPS
311        else:
312            direction = "right"
313            steps = space_action - 3 * HARD_MAX_STEPS
314        return {"direction": direction, "steps": steps + 1}
315
316    def parameters_to_space(self, direction: str, steps: int):
317        if steps <= 0 or steps > HARD_MAX_STEPS:
318            return None
319        if direction == "up":
320            return steps - 1
321        elif direction == "down":
322            return HARD_MAX_STEPS + steps - 1
323        elif direction == "left":
324            return 2 * HARD_MAX_STEPS + steps - 1
325        elif direction == "right":
326            return 3 * HARD_MAX_STEPS + steps - 1
327        else:
328            # log_warn(f"Unrecognized direction {direction}", self._parameters)
329            return None
330
331    def _execute(self, direction, steps):
332        transition_states, status = self.move(direction=direction, steps=steps)
333        return transition_states, status
334
335    def is_valid(self, **kwargs):
336        direction = kwargs.get("direction")
337        step = kwargs.get("step")
338        if direction is not None and direction not in ["up", "down", "left", "right"]:
339            return False
340        if step is not None:
341            if not isinstance(step, str):
342                return False
343            if step <= 0:
344                return False
345        return super().is_valid(**kwargs)
346
347    @staticmethod
348    def get_action_name(direction: str, steps: int) -> str:
349        return f"Move {direction} {steps}"

Moves the agent in a specified cardinal direction for a specified number of steps.

Is Valid When:

In Free Roam State Action Success Interpretation:
-1: Frame did not change, even on the first step
0: Finished all steps
1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.

Action Returns:

n_steps_taken (int): Number of steps actually taken
rotated (bool or None): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.

def get_action_space(self): View Source

284    def get_action_space(self):
285        """
286        Returns a Box space representing movement in 2D.
287        The first dimension represents vertical movement (positive is up, negative is down).
288        The second dimension represents horizontal movement (positive is right, negative is left).
289
290        Returns:
291            Box: A Box space with shape (2,) and values ranging from -HARD_MAX_STEPS//2 to HARD_MAX_STEPS//2.
292
293        """
294        return Discrete(4 * HARD_MAX_STEPS)

Returns a Box space representing movement in 2D. The first dimension represents vertical movement (positive is up, negative is down). The second dimension represents horizontal movement (positive is right, negative is left).

Returns:

Box: A Box space with shape (2,) and values ranging from -HARD_MAX_STEPS//2 to HARD_MAX_STEPS//2.

def space_to_parameters(self, space_action): View Source

296    def space_to_parameters(self, space_action):
297        direction = None
298        steps = None
299        if space_action < 0 or space_action >= 4 * HARD_MAX_STEPS:
300            # log_warn(f"Invalid space action {space_action}", self._parameters)
301            return None
302        if space_action < HARD_MAX_STEPS:
303            direction = "up"
304            steps = space_action
305        elif space_action < 2 * HARD_MAX_STEPS:
306            direction = "down"
307            steps = space_action - HARD_MAX_STEPS
308        elif space_action < 3 * HARD_MAX_STEPS:
309            direction = "left"
310            steps = space_action - 2 * HARD_MAX_STEPS
311        else:
312            direction = "right"
313            steps = space_action - 3 * HARD_MAX_STEPS
314        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.

def parameters_to_space(self, direction: str, steps: int): View Source

316    def parameters_to_space(self, direction: str, steps: int):
317        if steps <= 0 or steps > HARD_MAX_STEPS:
318            return None
319        if direction == "up":
320            return steps - 1
321        elif direction == "down":
322            return HARD_MAX_STEPS + steps - 1
323        elif direction == "left":
324            return 2 * HARD_MAX_STEPS + steps - 1
325        elif direction == "right":
326            return 3 * HARD_MAX_STEPS + steps - 1
327        else:
328            # log_warn(f"Unrecognized direction {direction}", self._parameters)
329            return None

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.

def is_valid(self, **kwargs): View Source

335    def is_valid(self, **kwargs):
336        direction = kwargs.get("direction")
337        step = kwargs.get("step")
338        if direction is not None and direction not in ["up", "down", "left", "right"]:
339            return False
340        if step is not None:
341            if not isinstance(step, str):
342                return False
343            if step <= 0:
344                return False
345        return super().is_valid(**kwargs)

Just checks if the agent is in free roam state.

@staticmethod

def get_action_name(direction: str, steps: int) -> str: View Source

347    @staticmethod
348    def get_action_name(direction: str, steps: int) -> str:
349        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.

Inherited Members

gameboy_worlds.interface.action.HighLevelAction: HighLevelAction; seed; assign_emulator; unassign_emulator; get_all_valid_parameters; execute; execute_space_action
BaseMovementAction: REQUIRED_STATE_TRACKER; REQUIRED_STATE_PARSER; judge_movement; move

class MoveGridAction(BaseMovementAction): View Source

352class MoveGridAction(BaseMovementAction):
353    """
354    Moves the agent on both axes. Will always try to move right first and then up.
355
356    Is Valid When:
357    - In Free Roam State
358    Action Success Interpretation:
359    - -1: Frame did not change, even on the first step
360    - 0: Finished all steps
361    - 1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
362    - 2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.
363
364    Action Returns:
365    - `n_steps_taken` (`int`): Number of steps actually taken
366    - `rotated` (`bool` or `None`): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.
367    """
368
369    def get_action_space(self):
370        """
371        Returns a Box space representing movement in 2D.
372        The first dimension represents vertical movement (positive is up, negative is down).
373        The second dimension represents horizontal movement (positive is right, negative is left).
374
375        Returns:
376            Box: A Box space with shape (2,) and values ranging from -HARD_MAX_STEPS//2 to HARD_MAX_STEPS//2.
377
378        """
379        return Box(
380            low=-HARD_MAX_STEPS // 2,
381            high=HARD_MAX_STEPS // 2,
382            shape=(2,),
383            dtype=np.int8,
384        )
385
386    def space_to_parameters(self, space_action):
387        right_action = space_action[0]
388        up_action = space_action[1]
389        return {"x_steps": right_action, "y_steps": up_action}
390
391    def parameters_to_space(self, x_steps, y_steps):
392        move_vec = np.zeros(2)  # x, y
393        move_vec[0] = x_steps
394        move_vec[1] = y_steps
395        return move_vec
396
397    def _execute(self, x_steps, y_steps):
398        x_direction = "right" if x_steps >= 0 else "left"
399        y_direction = "up" if y_steps >= 0 else "down"
400        if x_steps != 0:
401            transition_states, status = self.move(
402                direction=x_direction, steps=abs(x_steps)
403            )
404            if status != 0:
405                return transition_states, status
406        else:
407            transition_states = []
408        if y_steps != 0:
409            more_transition_states, status = self.move(
410                direction=y_direction, steps=abs(y_steps)
411            )
412            transition_states.extend(more_transition_states)
413        try:
414            status is not None
415        except NameError:
416            log_warn(
417                f"Weird case where both x_steps and y_steps are 0 in MoveGridAction or something. {x_steps}, {y_steps}",
418                self._parameters,
419            )
420            transition_states = [self._state_tracker.report()]
421            status = -1
422        return transition_states, status
423
424    def is_valid(self, x_steps: int = None, y_steps: int = None):
425        if x_steps is not None and y_steps is not None:
426            if not isinstance(x_steps, int) or not isinstance(y_steps, int):
427                return False
428            if x_steps == 0 and y_steps == 0:
429                return False
430        return super().is_valid()
431
432    @staticmethod
433    def get_action_name(x_steps: int, y_steps: int) -> str:
434        return f"MoveGrid ({x_steps}, {y_steps})"

Moves the agent on both axes. Will always try to move right first and then up.

Is Valid When:

In Free Roam State Action Success Interpretation:
-1: Frame did not change, even on the first step
0: Finished all steps
1: Took some steps, but not all, and then frame stopped changing OR the frame starts oscillating (trying to check for jitter). This usually means we ran into an obstacle.
2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene.

Action Returns:

n_steps_taken (int): Number of steps actually taken
rotated (bool or None): True if the player has not moved, but has rotated. If the player has moved, this will be None. If it is False, it means the player tried to walk straight into an obstacle.

def get_action_space(self): View Source

369    def get_action_space(self):
370        """
371        Returns a Box space representing movement in 2D.
372        The first dimension represents vertical movement (positive is up, negative is down).
373        The second dimension represents horizontal movement (positive is right, negative is left).
374
375        Returns:
376            Box: A Box space with shape (2,) and values ranging from -HARD_MAX_STEPS//2 to HARD_MAX_STEPS//2.
377
378        """
379        return Box(
380            low=-HARD_MAX_STEPS // 2,
381            high=HARD_MAX_STEPS // 2,
382            shape=(2,),
383            dtype=np.int8,
384        )

Returns:

Box: A Box space with shape (2,) and values ranging from -HARD_MAX_STEPS//2 to HARD_MAX_STEPS//2.

def space_to_parameters(self, space_action): View Source

386    def space_to_parameters(self, space_action):
387        right_action = space_action[0]
388        up_action = space_action[1]
389        return {"x_steps": right_action, "y_steps": up_action}

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.

def parameters_to_space(self, x_steps, y_steps): View Source

391    def parameters_to_space(self, x_steps, y_steps):
392        move_vec = np.zeros(2)  # x, y
393        move_vec[0] = x_steps
394        move_vec[1] = y_steps
395        return move_vec

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.

def is_valid(self, x_steps: int = None, y_steps: int = None): View Source

424    def is_valid(self, x_steps: int = None, y_steps: int = None):
425        if x_steps is not None and y_steps is not None:
426            if not isinstance(x_steps, int) or not isinstance(y_steps, int):
427                return False
428            if x_steps == 0 and y_steps == 0:
429                return False
430        return super().is_valid()

Just checks if the agent is in free roam state.

@staticmethod

def get_action_name(x_steps: int, y_steps: int) -> str: View Source

432    @staticmethod
433    def get_action_name(x_steps: int, y_steps: int) -> str:
434        return f"MoveGrid ({x_steps}, {y_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.

Inherited Members

gameboy_worlds.interface.action.HighLevelAction: HighLevelAction; seed; assign_emulator; unassign_emulator; get_all_valid_parameters; execute; execute_space_action
BaseMovementAction: REQUIRED_STATE_TRACKER; REQUIRED_STATE_PARSER; judge_movement; move