gameboy_worlds.interface.action

  1from abc import ABC, abstractmethod
  2from typing import Any, Dict, Tuple, List, Optional
  3from enum import Enum
  4from gameboy_worlds.emulation.emulator import LowLevelActions, Emulator
  5from gameboy_worlds.utils import (
  6    verify_parameters,
  7    log_info,
  8    log_warn,
  9    log_error,
 10    load_parameters,
 11    get_lowest_level_subclass,
 12)
 13from gameboy_worlds.emulation import StateTracker, StateParser
 14
 15import numpy as np
 16from gymnasium.spaces import Space, Discrete
 17
 18
 19class HighLevelAction(ABC):
 20    """
 21    Abstract base class for high level actions.
 22
 23    The execute() method performs the high level action and returns a list of state tracker reports after each low level action executed, along with an action success status.
 24
 25    The docstring of each subclass should specify the interpretation of the action success status.
 26
 27    The execute() method may also return some additional information in the final state tracker report under the "core" key, "action_return" subkey (usually as a dict itself)
 28    """
 29
 30    REQUIRED_STATE_TRACKER = StateTracker
 31    """ The state tracker that tracks the minimal state information required for the action to function. """
 32
 33    REQUIRED_STATE_PARSER = StateParser
 34    """ The state parser that parses the minimal state information required for the action to function. """
 35
 36    def __init__(self, parameters: dict, seed: Optional[int] = None):
 37        verify_parameters(parameters)
 38        self._parameters = parameters
 39        self._rng = np.random.default_rng(seed)
 40        self.unassign_emulator()
 41
 42    def seed(self, seed: Optional[int] = None):
 43        """
 44        Sets the random seed for the high level action.
 45        Args:
 46            seed (int): The random seed to set.
 47        """
 48        self._rng = np.random.default_rng(seed)
 49
 50    def assign_emulator(self, emulator: Emulator):
 51        """
 52        Sets a reference to the emulator instance.
 53        Args:
 54            emulator (Emulator): The emulator instance to be tracked.
 55        """
 56        if not isinstance(emulator, Emulator):
 57            log_error(
 58                f"HighLevelAction requires an Emulator instance, but got {type(emulator)}",
 59                self._parameters,
 60            )
 61        self._emulator = emulator
 62        self._state_tracker = emulator.state_tracker
 63        if not issubclass(type(self._state_tracker), self.REQUIRED_STATE_TRACKER):
 64            log_error(
 65                f"HighLevelAction requires a StateTracker of type {self.REQUIRED_STATE_TRACKER}, but got {type(self._state_tracker)}",
 66                self._parameters,
 67            )
 68        if not issubclass(type(emulator.state_parser), self.REQUIRED_STATE_PARSER):
 69            log_error(
 70                f"HighLevelAction requires a StateParser of type {self.REQUIRED_STATE_PARSER}, but got {type(emulator.state_parser)}",
 71                self._parameters,
 72            )
 73
 74    def unassign_emulator(self):
 75        """
 76        Clears the reference to the emulator instance.
 77        """
 78        self._emulator = None
 79        self._state_tracker = None
 80
 81    @abstractmethod
 82    def get_action_space(self) -> Space:
 83        """
 84        Returns the Gym defined Space that characterizes the high level action's parameter space.
 85
 86        You can use this API to get a Space for sampling high level actions of this type.
 87
 88        Returns:
 89            Space: The Gym space that characterizes the high level action's parameter space.
 90        """
 91        raise NotImplementedError
 92
 93    @abstractmethod
 94    def space_to_parameters(self, space_action: Space) -> Optional[Dict[str, Any]]:
 95        """
 96        Converts a Gym space action into high level action parameters.
 97        If the provided space action is invalid, return None.
 98
 99        :param space_action: The action in the high level action's parameter space.
100        :type space_action: Space
101        :return: The high level action's parameters corresponding to the space action.
102        :rtype: Optional[Dict[str, Any]]
103        """
104        raise NotImplementedError
105
106    @abstractmethod
107    def parameters_to_space(self, **kwargs) -> Optional[Space]:
108        """
109        Converts high level action parameters into a Gym space action.
110        If the provided parameters are invalid, return None.
111
112        :param kwargs: The high level action's parameters.
113        :type kwargs: Dict[str, Any]
114        :return: The action in the high level action's parameter space.
115        :rtype: Optional[Space]
116        """
117        raise NotImplementedError
118
119    @abstractmethod
120    def is_valid(self, **kwargs) -> bool:
121        """
122        Checks if the high level action can be performed in the current state.
123        If kwargs is empty, then must check whether there exists any valid way to perform the action.
124
125        Args:
126            **kwargs: Additional arguments required for the specific high level action.
127        Returns:
128            bool: Whether the action is valid in the current state.
129        """
130        raise NotImplementedError
131
132    @abstractmethod
133    def _execute(self, **kwargs) -> Tuple[List[Dict[str, Dict[str, Any]]], int]:
134        """
135        Executes the specified, valid high level action on the emulator.
136        Does not check for validity, assumes the action is valid.
137
138        :param self: Description
139        :param kwargs: Additional arguments required for the specific high level action.
140        :return:
141            - A list of state tracker reports after each low level action executed.
142
143            - Action success status.
144        :rtype: Tuple[List[Dict[str, Dict[str, Any]]], int]
145        """
146        raise NotImplementedError
147
148    def get_all_valid_parameters(self) -> List[Dict[str, Any]]:
149        """
150        Returns a list of all valid parameterizations for the high level action in the current state.
151
152        May not well defined for all high level actions, because some high level actions may have infinite parameterizations. (e.g. move to any (x, y) position.)
153
154        Use this to enumerate all valid ways to perform the action, and provide a way to sample over all valid parameterizations.
155
156
157        Returns:
158
159            List[Dict[str, Any]]: A list of valid parameterizations for the high level action.
160        """
161        raise ValueError(
162            "This high level action does not implement get_all_valid_parameters(). Most likely, it is not possible to enumerate an exhaustive list of all valid inputs. Use is_valid() instead. See documentation for more details. If you believe this is an error, please implement get_all_valid_parameters() in the high level action subclass."
163        )
164
165    def execute(
166        self, **kwargs
167    ) -> Tuple[Optional[List[Dict[str, Dict[str, Any]]]], Optional[int]]:
168        """
169        Executes the specified high level action on the emulator after checking for validity.
170
171        :param kwargs: Additional arguments required for the specific high level action.
172        :return: None, None if the action is not valid. Otherwise:
173
174            - A list of state tracker reports after each low level action executed.
175
176            - Action success status.
177        :rtype: Tuple[Optional[List[Dict[str, Dict[str, Any]]]], Optional[int]]
178        """
179        if self._emulator is None:
180            log_error(
181                f"Tried to execute action on HighLevelAction without an emulator",
182                self._parameters,
183            )
184        if not self.is_valid(**kwargs):
185            return None, None
186        return self._execute(**kwargs)
187
188    def execute_space_action(
189        self, space_action: Space
190    ) -> Tuple[Optional[List[Dict[str, Dict[str, Any]]]], Optional[int]]:
191        """
192        Executes the specified high level action on the emulator after checking for validity.
193
194        :param space_action: The action in the high level action's parameter space.
195        :type space_action: Space
196        :return: None, None if the action is not valid. Otherwise:
197
198            - A list of state tracker reports after each low level action executed.
199
200            - Action success status.
201        :rtype: Tuple[Optional[List[Dict[str, Dict[str, Any]]]], Optional[int]]
202        """
203        parameters = self.space_to_parameters(space_action)
204        if parameters is None:
205            return None, None
206        return self.execute(**parameters)
207
208    @staticmethod
209    @abstractmethod
210    def get_action_name(**kwargs) -> str:
211        """
212        Returns a human readable name for the high level action with the given parameters.
213
214        :param kwargs: The high level action's parameters.
215        :type kwargs: Dict[str, Any]
216        :return: A human readable name for the high level action.
217        :rtype: str
218        """
219        raise NotImplementedError
220
221
222class SingleHighLevelAction(HighLevelAction):
223    """
224    An abstract class for a high level action that has only one possible parameterization.
225    """
226
227    def space_to_parameters(self, space_action):
228        return {}
229
230    def parameters_to_space(self):
231        return 0
232
233    def get_action_space(self):
234        return Discrete(1)
235
236    def get_all_valid_parameters(self):
237        if self.is_valid():
238            return [{}]
239        else:
240            return []
241
242
243class LowLevelAction(HighLevelAction):
244    """A high level action that directly maps to a single low level action."""
245
246    def get_action_space(self):
247        """
248        Returns the Gym defined Space that characterizes the low level action's parameter space.
249        """
250        return Discrete(len(LowLevelActions))
251
252    def space_to_parameters(self, space_action: Space) -> Dict[str, Any]:
253        action = list(LowLevelActions)[space_action]
254        return {"low_level_action": action}
255
256    def parameters_to_space(self, low_level_action: LowLevelActions) -> Space:
257        if low_level_action is None or not isinstance(
258            low_level_action, LowLevelActions
259        ):
260            # log_warn(
261            #    "LowLevelAction requires a 'low_level_action' parameter of type LowLevelActions.",
262            #    self._parameters,
263            # )
264            return None
265        return low_level_action.value
266
267    def _execute(
268        self, low_level_action: LowLevelActions
269    ) -> Tuple[List[Dict[str, Dict[str, Any]]], int]:
270        self._emulator.step(low_level_action)
271        state_report = self._state_tracker.report()
272        return [
273            state_report
274        ], 0  # Low level actions are always successful in this context.
275
276    def is_valid(self, low_level_action: LowLevelActions) -> bool:
277        """
278        Checks if the low level action can be performed in the current state.
279
280        Args:
281            low_level_action (LowLevelActions): The low level action to check.
282        Returns:
283            bool: Whether the action is valid in the current state.
284        """
285        return True
286
287    def get_all_valid_parameters(self) -> List[Dict[str, Any]]:
288        """
289        Returns a list of all valid low level actions in the current state.
290
291        :return: A list of valid low level actions.
292        :rtype: List[Dict[str, Any]]
293        """
294        return [{"low_level_action": action} for action in LowLevelActions]
295
296    @staticmethod
297    def get_action_name(low_level_action: LowLevelActions) -> str:
298        action = (
299            str(low_level_action)
300            .replace("LowLevelActions.PRESS_ARROW_", "")
301            .replace("LowLevelActions.PRESS_BUTTON_", "")
302        )
303        return action
304
305
306class LowLevelPlayAction(HighLevelAction):
307    """A HighLevelAction subclass that directly maps to low level actions, except no menu button presses."""
308
309    def __init__(self, parameters: dict, seed: Optional[int] = None):
310        self.allowed_actions = [
311            LowLevelActions.PRESS_ARROW_UP,
312            LowLevelActions.PRESS_ARROW_DOWN,
313            LowLevelActions.PRESS_ARROW_RIGHT,
314            LowLevelActions.PRESS_ARROW_LEFT,
315            LowLevelActions.PRESS_BUTTON_A,
316            LowLevelActions.PRESS_BUTTON_B,
317        ]
318        super().__init__(parameters, seed=seed)
319
320    def get_action_space(self):
321        """
322        Returns the Gym defined Space that characterizes the low level play action's parameter space.
323        """
324        return Discrete(len(self.allowed_actions))
325
326    def space_to_parameters(self, space_action: Space) -> Dict[str, Any]:
327        action = self.allowed_actions[space_action]
328        return {"low_level_action": action}
329
330    def parameters_to_space(self, low_level_action: LowLevelActions) -> Space:
331        if low_level_action is None or low_level_action not in self.allowed_actions:
332            # log_warn(
333            #    "LowLevelPlayAction requires a 'low_level_action' parameter that is not a menu button press.",
334            #    self._parameters,
335            # )
336            return None
337        return self.allowed_actions.index(low_level_action)
338
339    def _execute(
340        self, low_level_action: LowLevelActions
341    ) -> Tuple[List[Dict[str, Dict[str, Any]]], int]:
342        self._emulator.step(low_level_action)
343        state_report = self._state_tracker.report()
344        return [
345            state_report
346        ], 0  # Low level actions are always successful in this context.
347
348    def is_valid(self, low_level_action: LowLevelActions) -> bool:
349        return low_level_action in self.allowed_actions
350
351    def get_all_valid_parameters(self) -> List[Dict[str, Any]]:
352        return [{"low_level_action": action} for action in self.allowed_actions]
353
354    @staticmethod
355    def get_action_name(low_level_action: LowLevelActions) -> str:
356        action = (
357            str(low_level_action)
358            .replace("LowLevelActions.PRESS_ARROW_", "")
359            .replace("LowLevelActions.PRESS_BUTTON_", "")
360        )
361        return action
362
363
364class RandomPlayAction(HighLevelAction):
365    """Execution either moves or presses A
366
367    Action Success Interpretation:
368
369    - 0: The frame changed after the action
370    - 1: The frame did not change after the action
371
372    """
373
374    def get_action_space(self):
375        """
376        Returns the Gym defined Space that characterizes the random play action's parameter space.
377        """
378        return Discrete(2)  # 0 for move, 1 for press A
379
380    def space_to_parameters(self, space_action: Space) -> Dict[str, Any]:
381        if space_action == 0:
382            return {"kind": "move"}
383        else:
384            return {"kind": "press"}
385
386    def parameters_to_space(self, kind: str) -> Space:
387        if kind == "move":
388            return 0
389        elif kind == "press":
390            return 1
391        else:
392            # log_warn(
393            #    "RandomPlayAction requires a 'kind' parameter of either 'move' or 'press'.",
394            #    self._parameters,
395            # )
396            return None
397
398    def _execute(self, kind: str):
399        if kind == "move":
400            actions = [
401                LowLevelActions.PRESS_ARROW_DOWN,
402                LowLevelActions.PRESS_ARROW_LEFT,
403                LowLevelActions.PRESS_ARROW_RIGHT,
404                LowLevelActions.PRESS_ARROW_UP,
405            ]
406
407        else:  # kind must be 'press'. Enforced in is_valid
408            actions = [
409                LowLevelActions.PRESS_BUTTON_A,
410            ]
411        action = self._rng.choice(actions)
412        self._emulator.step(action)
413        state_report = self._state_tracker.report()
414        not_success = not state_report["core"][
415            "frame_changed"
416        ]  # Whether the frame changed after the action
417        return [state_report], not_success
418
419    def is_valid(self, kind: str) -> bool:
420        if kind not in ["move", "press"]:
421            return False
422        return True
423
424    def get_all_valid_parameters(self) -> List[Dict[str, Any]]:
425        return [{"kind": "move"}, {"kind": "press"}]
426
427    @staticmethod
428    def get_action_name(kind: str) -> str:
429        return kind