gameboy_worlds.interface.bomberman.actions

  1from typing import Any, Dict, List, Type
  2
  3import numpy as np
  4from gymnasium.spaces import Discrete
  5
  6from gameboy_worlds.emulation import LowLevelActions
  7from gameboy_worlds.emulation.bomberman.parsers import (
  8    BombermanMaxParser,
  9    BombermanPocketParser,
 10    BombermanQuestParser,
 11)
 12from gameboy_worlds.emulation.bomberman.trackers import (
 13    BombermanMaxTracker,
 14    BombermanPocketTracker,
 15    BombermanQuestTracker,
 16)
 17from gameboy_worlds.interface.action import HighLevelAction, SingleHighLevelAction
 18
 19HARD_MAX_STEPS = 10
 20
 21MAX_MENU_METRIC = ("bomberman_max_core", "is_in_menu")
 22MAX_BATTLE_METRIC = ("bomberman_max_core", "is_in_battle")
 23POCKET_MENU_METRIC = ("bomberman_pocket_core", "is_in_menu")
 24QUEST_MENU_METRIC = ("bomberman_quest_core", "is_in_menu")
 25QUEST_BATTLE_METRIC = ("bomberman_quest_core", "is_in_battle")
 26
 27
 28def frame_changed(previous: np.ndarray, current: np.ndarray, epsilon: float = 0.01) -> bool:
 29    return np.abs(previous - current).mean() > epsilon
 30
 31
 32class _MoveAction(HighLevelAction):
 33    _DIRECTION_TO_ACTION = {}
 34    _BLOCKING_METRICS: List[tuple[str, str]] = []
 35
 36    def get_action_space(self):
 37        return Discrete(len(self._DIRECTION_TO_ACTION) * HARD_MAX_STEPS)
 38
 39    def space_to_parameters(self, space_action):
 40        directions = list(self._DIRECTION_TO_ACTION)
 41        if space_action < 0 or space_action >= len(directions) * HARD_MAX_STEPS:
 42            return None
 43        direction = directions[space_action // HARD_MAX_STEPS]
 44        steps = (space_action % HARD_MAX_STEPS) + 1
 45        return {"direction": direction, "steps": steps}
 46
 47    def parameters_to_space(self, direction: str, steps: int):
 48        directions = list(self._DIRECTION_TO_ACTION)
 49        if direction not in directions or steps <= 0 or steps > HARD_MAX_STEPS:
 50            return None
 51        return directions.index(direction) * HARD_MAX_STEPS + steps - 1
 52
 53    def is_valid(self, **kwargs):
 54        direction = kwargs.get("direction")
 55        steps = kwargs.get("steps")
 56        if direction is not None and direction not in self._DIRECTION_TO_ACTION:
 57            return False
 58        if steps is not None and (not isinstance(steps, int) or steps <= 0):
 59            return False
 60        return not any(
 61            self._state_tracker.get_episode_metric(metric_key)
 62            for metric_key in self._BLOCKING_METRICS
 63        )
 64
 65    def _execute(self, direction: str, steps: int):
 66        action = self._DIRECTION_TO_ACTION[direction]
 67        transition_state_dicts: List[Dict[str, Any]] = []
 68        n_steps_taken = 0
 69        previous_frame = self._emulator.get_current_frame()
 70        action_success = -1
 71        for _ in range(steps):
 72            frames, done = self._emulator.step(action)
 73            report = self._state_tracker.report()
 74            transition_state_dicts.append(report)
 75            current_frame = frames[-1]
 76            if not frame_changed(previous_frame, current_frame):
 77                action_success = 1 if n_steps_taken > 0 else -1
 78                break
 79            n_steps_taken += 1
 80            if any(
 81                self._state_tracker.get_episode_metric(metric_key)
 82                for metric_key in self._BLOCKING_METRICS
 83            ):
 84                action_success = 2
 85                break
 86            if done:
 87                action_success = 0
 88                break
 89            previous_frame = current_frame
 90        else:
 91            action_success = 0
 92        if transition_state_dicts:
 93            transition_state_dicts[-1]["core"]["action_return"] = {
 94                "n_steps_taken": n_steps_taken
 95            }
 96        return transition_state_dicts, action_success
 97
 98    @staticmethod
 99    def get_action_name(direction: str, steps: int) -> str:
100        return f"Move {direction} {steps}"
101
102
103class _MetricGatedSingleAction(SingleHighLevelAction):
104    _BUTTON = None
105    _REQUIRED_METRICS_FALSE: List[tuple[str, str]] = []
106    _REQUIRED_METRICS_TRUE: List[tuple[str, str]] = []
107
108    def is_valid(self, **kwargs):
109        return all(
110            not self._state_tracker.get_episode_metric(metric_key)
111            for metric_key in self._REQUIRED_METRICS_FALSE
112        ) and all(
113            self._state_tracker.get_episode_metric(metric_key)
114            for metric_key in self._REQUIRED_METRICS_TRUE
115        )
116
117    def _execute(self):
118        previous = self._emulator.get_current_frame()
119        frames, _ = self._emulator.step(self._BUTTON)
120        report = self._state_tracker.report()
121        return [report], 0 if frame_changed(previous, frames[-1]) else -1
122
123
124class BombermanMaxMoveAction(_MoveAction):
125    REQUIRED_STATE_TRACKER = BombermanMaxTracker
126    REQUIRED_STATE_PARSER = BombermanMaxParser
127    _DIRECTION_TO_ACTION = {
128        "up": LowLevelActions.PRESS_ARROW_UP,
129        "down": LowLevelActions.PRESS_ARROW_DOWN,
130        "left": LowLevelActions.PRESS_ARROW_LEFT,
131        "right": LowLevelActions.PRESS_ARROW_RIGHT,
132    }
133    _BLOCKING_METRICS = [MAX_MENU_METRIC, MAX_BATTLE_METRIC]
134
135
136class BombermanMaxPlaceBombAction(_MetricGatedSingleAction):
137    REQUIRED_STATE_TRACKER = BombermanMaxTracker
138    REQUIRED_STATE_PARSER = BombermanMaxParser
139    _BUTTON = LowLevelActions.PRESS_BUTTON_A
140    _REQUIRED_METRICS_FALSE = [MAX_MENU_METRIC, MAX_BATTLE_METRIC]
141
142    @staticmethod
143    def get_action_name() -> str:
144        return "PlaceBomb"
145
146
147class BombermanMaxKickBombAction(_MetricGatedSingleAction):
148    REQUIRED_STATE_TRACKER = BombermanMaxTracker
149    REQUIRED_STATE_PARSER = BombermanMaxParser
150    _BUTTON = LowLevelActions.PRESS_BUTTON_B
151    _REQUIRED_METRICS_FALSE = [MAX_MENU_METRIC, MAX_BATTLE_METRIC]
152
153    @staticmethod
154    def get_action_name() -> str:
155        return "KickBomb"
156
157
158class BombermanMaxOpenMenuAction(_MetricGatedSingleAction):
159    REQUIRED_STATE_TRACKER = BombermanMaxTracker
160    REQUIRED_STATE_PARSER = BombermanMaxParser
161    _BUTTON = LowLevelActions.PRESS_BUTTON_START
162    _REQUIRED_METRICS_FALSE = [MAX_MENU_METRIC, MAX_BATTLE_METRIC]
163
164    def _execute(self):
165        self._emulator.step(self._BUTTON)
166        report = self._state_tracker.report()
167        return [report], 0 if self._state_tracker.get_episode_metric(MAX_MENU_METRIC) else -1
168
169    @staticmethod
170    def get_action_name() -> str:
171        return "OpenMenu"
172
173
174class BombermanMaxCloseMenuAction(_MetricGatedSingleAction):
175    REQUIRED_STATE_TRACKER = BombermanMaxTracker
176    REQUIRED_STATE_PARSER = BombermanMaxParser
177    _BUTTON = LowLevelActions.PRESS_BUTTON_START
178    _REQUIRED_METRICS_TRUE = [MAX_MENU_METRIC]
179
180    def _execute(self):
181        self._emulator.step(self._BUTTON)
182        report = self._state_tracker.report()
183        return [report], 0 if not self._state_tracker.get_episode_metric(MAX_MENU_METRIC) else -1
184
185    @staticmethod
186    def get_action_name() -> str:
187        return "CloseMenu"
188
189
190class BombermanMaxNavigateMenuAction(SingleHighLevelAction):
191    REQUIRED_STATE_TRACKER = BombermanMaxTracker
192    REQUIRED_STATE_PARSER = BombermanMaxParser
193    _ACTION_MAP = {
194        "up": LowLevelActions.PRESS_ARROW_UP,
195        "down": LowLevelActions.PRESS_ARROW_DOWN,
196        "left": LowLevelActions.PRESS_ARROW_LEFT,
197        "right": LowLevelActions.PRESS_ARROW_RIGHT,
198        "confirm": LowLevelActions.PRESS_BUTTON_A,
199        "back": LowLevelActions.PRESS_BUTTON_B,
200    }
201
202    def get_action_space(self):
203        return Discrete(len(self._ACTION_MAP))
204
205    def space_to_parameters(self, space_action):
206        keys = list(self._ACTION_MAP)
207        if space_action < 0 or space_action >= len(keys):
208            return None
209        return {"menu_action": keys[space_action]}
210
211    def parameters_to_space(self, menu_action: str):
212        keys = list(self._ACTION_MAP)
213        return keys.index(menu_action) if menu_action in keys else None
214
215    def is_valid(self, **kwargs):
216        menu_action = kwargs.get("menu_action")
217        return (
218            menu_action in self._ACTION_MAP if menu_action is not None else True
219        ) and self._state_tracker.get_episode_metric(MAX_MENU_METRIC)
220
221    def _execute(self, menu_action: str):
222        previous = self._emulator.get_current_frame()
223        frames, _ = self._emulator.step(self._ACTION_MAP[menu_action])
224        report = self._state_tracker.report()
225        return [report], 0 if frame_changed(previous, frames[-1]) else -1
226
227    @staticmethod
228    def get_action_name(menu_action: str) -> str:
229        return f"NavigateMenu {menu_action}"
230
231
232class BombermanMaxBattleAction(SingleHighLevelAction):
233    REQUIRED_STATE_TRACKER = BombermanMaxTracker
234    REQUIRED_STATE_PARSER = BombermanMaxParser
235    _ACTION_MAP = {
236        "bomb": LowLevelActions.PRESS_BUTTON_A,
237        "up": LowLevelActions.PRESS_ARROW_UP,
238        "down": LowLevelActions.PRESS_ARROW_DOWN,
239        "left": LowLevelActions.PRESS_ARROW_LEFT,
240        "right": LowLevelActions.PRESS_ARROW_RIGHT,
241    }
242
243    def get_action_space(self):
244        return Discrete(len(self._ACTION_MAP))
245
246    def space_to_parameters(self, space_action):
247        keys = list(self._ACTION_MAP)
248        if space_action < 0 or space_action >= len(keys):
249            return None
250        return {"battle_action": keys[space_action]}
251
252    def parameters_to_space(self, battle_action: str):
253        keys = list(self._ACTION_MAP)
254        return keys.index(battle_action) if battle_action in keys else None
255
256    def is_valid(self, **kwargs):
257        battle_action = kwargs.get("battle_action")
258        return (
259            battle_action in self._ACTION_MAP if battle_action is not None else True
260        ) and self._state_tracker.get_episode_metric(MAX_BATTLE_METRIC)
261
262    def _execute(self, battle_action: str):
263        previous = self._emulator.get_current_frame()
264        frames, _ = self._emulator.step(self._ACTION_MAP[battle_action])
265        report = self._state_tracker.report()
266        return [report], 0 if frame_changed(previous, frames[-1]) else -1
267
268    @staticmethod
269    def get_action_name(battle_action: str) -> str:
270        return f"Battle {battle_action}"
271
272
273class BombermanPocketMoveAction(_MoveAction):
274    REQUIRED_STATE_TRACKER = BombermanPocketTracker
275    REQUIRED_STATE_PARSER = BombermanPocketParser
276    _DIRECTION_TO_ACTION = {
277        "left": LowLevelActions.PRESS_ARROW_LEFT,
278        "right": LowLevelActions.PRESS_ARROW_RIGHT,
279    }
280    _BLOCKING_METRICS = [POCKET_MENU_METRIC]
281
282
283class BombermanPocketJumpAction(_MetricGatedSingleAction):
284    REQUIRED_STATE_TRACKER = BombermanPocketTracker
285    REQUIRED_STATE_PARSER = BombermanPocketParser
286    _BUTTON = LowLevelActions.PRESS_BUTTON_B
287    _REQUIRED_METRICS_FALSE = [POCKET_MENU_METRIC]
288
289    @staticmethod
290    def get_action_name() -> str:
291        return "Jump"
292
293
294class BombermanPocketPlaceBombAction(_MetricGatedSingleAction):
295    REQUIRED_STATE_TRACKER = BombermanPocketTracker
296    REQUIRED_STATE_PARSER = BombermanPocketParser
297    _BUTTON = LowLevelActions.PRESS_BUTTON_A
298    _REQUIRED_METRICS_FALSE = [POCKET_MENU_METRIC]
299
300    @staticmethod
301    def get_action_name() -> str:
302        return "PlaceBomb"
303
304
305class BombermanPocketOpenPauseMenuAction(_MetricGatedSingleAction):
306    REQUIRED_STATE_TRACKER = BombermanPocketTracker
307    REQUIRED_STATE_PARSER = BombermanPocketParser
308    _BUTTON = LowLevelActions.PRESS_BUTTON_START
309    _REQUIRED_METRICS_FALSE = [POCKET_MENU_METRIC]
310
311    def _execute(self):
312        self._emulator.step(self._BUTTON)
313        report = self._state_tracker.report()
314        return [report], 0 if self._state_tracker.get_episode_metric(POCKET_MENU_METRIC) else -1
315
316    @staticmethod
317    def get_action_name() -> str:
318        return "OpenPauseMenu"
319
320
321class BombermanPocketClosePauseMenuAction(_MetricGatedSingleAction):
322    REQUIRED_STATE_TRACKER = BombermanPocketTracker
323    REQUIRED_STATE_PARSER = BombermanPocketParser
324    _BUTTON = LowLevelActions.PRESS_BUTTON_START
325    _REQUIRED_METRICS_TRUE = [POCKET_MENU_METRIC]
326
327    def _execute(self):
328        self._emulator.step(self._BUTTON)
329        report = self._state_tracker.report()
330        return [report], 0 if not self._state_tracker.get_episode_metric(POCKET_MENU_METRIC) else -1
331
332    @staticmethod
333    def get_action_name() -> str:
334        return "ClosePauseMenu"
335
336
337class BombermanQuestMoveAction(_MoveAction):
338    REQUIRED_STATE_TRACKER = BombermanQuestTracker
339    REQUIRED_STATE_PARSER = BombermanQuestParser
340    _DIRECTION_TO_ACTION = {
341        "up": LowLevelActions.PRESS_ARROW_UP,
342        "down": LowLevelActions.PRESS_ARROW_DOWN,
343        "left": LowLevelActions.PRESS_ARROW_LEFT,
344        "right": LowLevelActions.PRESS_ARROW_RIGHT,
345    }
346    _BLOCKING_METRICS = [QUEST_MENU_METRIC, QUEST_BATTLE_METRIC]
347
348
349class BombermanQuestPlaceBombAction(_MetricGatedSingleAction):
350    REQUIRED_STATE_TRACKER = BombermanQuestTracker
351    REQUIRED_STATE_PARSER = BombermanQuestParser
352    _BUTTON = LowLevelActions.PRESS_BUTTON_A
353    _REQUIRED_METRICS_FALSE = [QUEST_MENU_METRIC, QUEST_BATTLE_METRIC]
354
355    @staticmethod
356    def get_action_name() -> str:
357        return "PlaceBomb"
358
359
360class BombermanQuestUseBButtonItemAction(_MetricGatedSingleAction):
361    REQUIRED_STATE_TRACKER = BombermanQuestTracker
362    REQUIRED_STATE_PARSER = BombermanQuestParser
363    _BUTTON = LowLevelActions.PRESS_BUTTON_B
364    _REQUIRED_METRICS_FALSE = [QUEST_MENU_METRIC, QUEST_BATTLE_METRIC]
365
366    @staticmethod
367    def get_action_name() -> str:
368        return "UseBButtonItem"
369
370
371class BombermanQuestOpenMenuAction(_MetricGatedSingleAction):
372    REQUIRED_STATE_TRACKER = BombermanQuestTracker
373    REQUIRED_STATE_PARSER = BombermanQuestParser
374    _BUTTON = LowLevelActions.PRESS_BUTTON_START
375    _REQUIRED_METRICS_FALSE = [QUEST_MENU_METRIC, QUEST_BATTLE_METRIC]
376
377    def _execute(self):
378        self._emulator.step(self._BUTTON)
379        report = self._state_tracker.report()
380        return [report], 0 if self._state_tracker.get_episode_metric(QUEST_MENU_METRIC) else -1
381
382    @staticmethod
383    def get_action_name() -> str:
384        return "OpenMenu"
385
386
387class BombermanQuestCloseMenuAction(_MetricGatedSingleAction):
388    REQUIRED_STATE_TRACKER = BombermanQuestTracker
389    REQUIRED_STATE_PARSER = BombermanQuestParser
390    _BUTTON = LowLevelActions.PRESS_BUTTON_START
391    _REQUIRED_METRICS_TRUE = [QUEST_MENU_METRIC]
392
393    def _execute(self):
394        self._emulator.step(self._BUTTON)
395        report = self._state_tracker.report()
396        return [report], 0 if not self._state_tracker.get_episode_metric(QUEST_MENU_METRIC) else -1
397
398    @staticmethod
399    def get_action_name() -> str:
400        return "CloseMenu"
401
402
403class BombermanQuestNavigateMenuAction(BombermanMaxNavigateMenuAction):
404    REQUIRED_STATE_TRACKER = BombermanQuestTracker
405    REQUIRED_STATE_PARSER = BombermanQuestParser
406
407    def is_valid(self, **kwargs):
408        menu_action = kwargs.get("menu_action")
409        return (
410            menu_action in self._ACTION_MAP if menu_action is not None else True
411        ) and self._state_tracker.get_episode_metric(QUEST_MENU_METRIC)
412
413
414class BombermanQuestBattleAction(BombermanMaxBattleAction):
415    REQUIRED_STATE_TRACKER = BombermanQuestTracker
416    REQUIRED_STATE_PARSER = BombermanQuestParser
417    _ACTION_MAP = {
418        "bomb": LowLevelActions.PRESS_BUTTON_A,
419        "item": LowLevelActions.PRESS_BUTTON_B,
420        "up": LowLevelActions.PRESS_ARROW_UP,
421        "down": LowLevelActions.PRESS_ARROW_DOWN,
422        "left": LowLevelActions.PRESS_ARROW_LEFT,
423        "right": LowLevelActions.PRESS_ARROW_RIGHT,
424    }
425
426    def is_valid(self, **kwargs):
427        battle_action = kwargs.get("battle_action")
428        return (
429            battle_action in self._ACTION_MAP if battle_action is not None else True
430        ) and self._state_tracker.get_episode_metric(QUEST_BATTLE_METRIC)