gameboy_worlds.interface.deja_vu.actions

  1from abc import ABC
  2from gameboy_worlds.utils import log_error, log_warn
  3from typing import List, Optional, Tuple, Dict
  4import numpy as np
  5from gymnasium.spaces import Box, Discrete
  6from gameboy_worlds.emulation import LowLevelActions
  7from gameboy_worlds.emulation.deja_vu.parsers import AgentState, DejaVuStateParser
  8from gameboy_worlds.emulation.deja_vu.trackers import CoreDejaVuTracker
  9from gameboy_worlds.interface.action import HighLevelAction, SingleHighLevelAction
 10
 11HARD_MAX_STEPS = 20
 12MENU_NAV_MAX_STEPS = 8
 13
 14
 15def frame_changed(past: np.ndarray, present: np.ndarray, epsilon: float = 0.01) -> bool:
 16    """Return True if the mean pixel difference between two frames is greater than epsilon."""
 17    return np.abs(past - present).mean() > epsilon
 18
 19
 20class TickUntilStable(SingleHighLevelAction):
 21    """Tick the emulator until the screen becomes stable (no change)."""
 22
 23    REQUIRED_STATE_PARSER = DejaVuStateParser
 24    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
 25
 26    def is_valid(self, **kwargs):
 27        return True
 28
 29    def _execute(self, max_ticks: int = 30):
 30        prev = self._emulator.get_current_frame()
 31        for _ in range(max_ticks):
 32            self._emulator.step(None)
 33            curr = self._emulator.get_current_frame()
 34            if not frame_changed(prev, curr):
 35                return [self._state_tracker.report()], 0
 36            prev = curr
 37        return [self._state_tracker.report()], -1
 38
 39    @staticmethod
 40    def get_action_name() -> str:
 41        return "TickUntilStable"
 42
 43
 44class MoveCursor(SingleHighLevelAction):
 45    """Move the menu cursor in a given direction (up, down, left, right)."""
 46
 47    REQUIRED_STATE_PARSER = DejaVuStateParser
 48    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
 49    _ACTION_MAP = {
 50        "up": LowLevelActions.PRESS_ARROW_UP,
 51        "down": LowLevelActions.PRESS_ARROW_DOWN,
 52        "left": LowLevelActions.PRESS_ARROW_LEFT,
 53        "right": LowLevelActions.PRESS_ARROW_RIGHT,
 54    }
 55
 56    def is_valid(self, direction=None, **kwargs):
 57        return direction in self._ACTION_MAP
 58
 59    def _execute(self, direction):
 60        action = self._ACTION_MAP[direction]
 61        prev = self._emulator.get_current_frame()
 62        self._emulator.step(action)
 63        curr = self._emulator.get_current_frame()
 64        return [self._state_tracker.report()], 0 if frame_changed(prev, curr) else -1
 65
 66    @staticmethod
 67    def get_action_name(**kwargs) -> str:
 68        return f"MoveCursor {kwargs.get('direction', '')}"
 69
 70
 71class MoveGrid(HighLevelAction):
 72    """Move the agent on the grid by (x_steps, y_steps). X is horizontal, Y is vertical."""
 73
 74    REQUIRED_STATE_PARSER = DejaVuStateParser
 75    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
 76
 77    def get_action_space(self):
 78        return Box(
 79            low=-HARD_MAX_STEPS // 2,
 80            high=HARD_MAX_STEPS // 2,
 81            shape=(2,),
 82            dtype=np.int8,
 83        )
 84
 85    def is_valid(self, x_steps=None, y_steps=None, **kwargs):
 86        state = self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
 87        return state == AgentState.FREE_ROAM and (x_steps or y_steps)
 88
 89    def _execute(self, x_steps, y_steps):
 90        reports = []
 91        for axis, steps, dir_pos, dir_neg in [
 92            ("x", x_steps, "right", "left"),
 93            ("y", y_steps, "down", "up"),
 94        ]:
 95            if steps:
 96                direction = dir_pos if steps > 0 else dir_neg
 97                for _ in range(abs(steps)):
 98                    prev = self._emulator.get_current_frame()
 99                    self._emulator.step(MoveStepsAction._ACTION_MAP[direction])
100                    curr = self._emulator.get_current_frame()
101                    reports.append(self._state_tracker.report())
102                    if not frame_changed(prev, curr):
103                        return reports, -1
104        return reports, 0
105
106    @staticmethod
107    def get_action_name(x_steps: int, y_steps: int) -> str:
108        return f"MoveGrid ({x_steps}, {y_steps})"
109
110
111class OpenButtonMenu(SingleHighLevelAction):
112    """Open the bottom button menu (usually START button)."""
113
114    REQUIRED_STATE_PARSER = DejaVuStateParser
115    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
116
117    def is_valid(self, **kwargs):
118        state = self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
119        return state == AgentState.FREE_ROAM
120
121    def _execute(self):
122        self._emulator.step(LowLevelActions.PRESS_BUTTON_START)
123        return [self._state_tracker.report()], 0
124
125    @staticmethod
126    def get_action_name() -> str:
127        return "OpenButtonMenu"
128
129
130class CloseButtonMenu(SingleHighLevelAction):
131    """Close the bottom button menu (usually B button)."""
132
133    REQUIRED_STATE_PARSER = DejaVuStateParser
134    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
135
136    def is_valid(self, **kwargs):
137        state = self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
138        return state == AgentState.IN_MENU
139
140    def _execute(self):
141        self._emulator.step(LowLevelActions.PRESS_BUTTON_B)
142        return [self._state_tracker.report()], 0
143
144    @staticmethod
145    def get_action_name() -> str:
146        return "CloseButtonMenu"
147
148
149class SelectMenuOption(SingleHighLevelAction):
150    """Select the current menu option (A button)."""
151
152    REQUIRED_STATE_PARSER = DejaVuStateParser
153    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
154
155    def is_valid(self, **kwargs):
156        state = self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
157        return state == AgentState.IN_MENU
158
159    def _execute(self):
160        self._emulator.step(LowLevelActions.PRESS_BUTTON_A)
161        return [self._state_tracker.report()], 0
162
163    @staticmethod
164    def get_action_name() -> str:
165        return "SelectMenuOption"
166
167
168class AdvanceDialogue(SingleHighLevelAction):
169    """Advance dialogue (B button)."""
170
171    REQUIRED_STATE_PARSER = DejaVuStateParser
172    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
173
174    def is_valid(self, **kwargs):
175        state = self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
176        return state == AgentState.IN_DIALOGUE
177
178    def _execute(self):
179        self._emulator.step(LowLevelActions.PRESS_BUTTON_B)
180        return [self._state_tracker.report()], 0
181
182    @staticmethod
183    def get_action_name() -> str:
184        return "AdvanceDialogue"
185
186
187# Helper for MoveGrid
188class MoveStepsAction:
189    _ACTION_MAP = {
190        "up": LowLevelActions.PRESS_ARROW_UP,
191        "down": LowLevelActions.PRESS_ARROW_DOWN,
192        "left": LowLevelActions.PRESS_ARROW_LEFT,
193        "right": LowLevelActions.PRESS_ARROW_RIGHT,
194    }
195
196
197class InteractAction(SingleHighLevelAction):
198    """
199    Presses the A button to interact with an object in front of the agent.
200
201    Is Valid When:
202    - In Free Roam State
203
204    Action Success Interpretation:
205    - -1: Frame did not change or agent still in free roam state
206    - 1: Agent not in free roam state
207    """
208
209    REQUIRED_STATE_PARSER = DejaVuStateParser
210    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
211
212    def is_valid(self, **kwargs):
213        return (
214            self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
215            == AgentState.FREE_ROAM
216        )
217
218    def _execute(self):
219        frames, done = self._emulator.step(LowLevelActions.PRESS_BUTTON_A)
220        action_success = 0
221        prev_frames = []
222        for frame in frames:
223            if (
224                self._emulator.state_parser.get_agent_state(frame)
225                != AgentState.FREE_ROAM
226            ):
227                action_success = 1
228                break
229            for past_frame in prev_frames:
230                if not frame_changed(past_frame, frame):
231                    action_success = -1
232                    break
233            if action_success != 0:
234                break
235            prev_frames.append(frame)
236        if action_success == 0:
237            action_success = -1
238        return [self._state_tracker.report()], action_success
239
240    @staticmethod
241    def get_action_name() -> str:
242        return "Interact"
243
244
245class BaseMovementAction(HighLevelAction, ABC):
246    """
247    Base class for movement actions in the Deja Vu environment.
248
249    Is Valid When:
250    - In Free Roam State
251
252    Action Success Interpretation:
253    - -1: Frame did not change, even on the first step
254    - 0: Finished all steps
255    - 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.
256    - 2: Took some steps, but agent state changed from free roam. This often means we entered a cutscene or battle.
257
258    Action Returns:
259    - n_steps_taken (int): Number of steps actually taken
260    - rotated (bool or None): True if the player has not moved, but has rotated.
261    """
262
263    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
264    REQUIRED_STATE_PARSER = DejaVuStateParser
265
266    def _is_uniform_quadrant(self, frame, quadrant_name) -> bool:
267        mapper = {
268            "screen_quadrant_1": "tr",
269            "screen_quadrant_2": "tl",
270            "screen_quadrant_3": "bl",
271            "screen_quadrant_4": "br",
272        }
273        quadrant_cells = self._emulator.state_parser.capture_grid_cells(
274            frame, quadrant=mapper[quadrant_name]
275        )
276        keys = list(quadrant_cells.keys())
277        x_min = 1e9
278        y_min = 1e9
279        x_max = -1e9
280        y_max = -1e9
281        for key in keys:
282            x_coord, y_coord = key
283            if x_coord * y_coord == 0:
284                quadrant_cells.pop(key)
285            else:
286                x_min = min(x_min, x_coord)
287                y_min = min(y_min, y_coord)
288                x_max = max(x_max, x_coord)
289                y_max = max(y_max, y_coord)
290        x_min = int(x_min)
291        y_min = int(y_min)
292        x_max = int(x_max)
293        y_max = int(y_max)
294        keys = list(quadrant_cells.keys())
295        vertical_uniform = True
296        horizontal_uniform = True
297        for y in range(y_min + 1, y_max):
298            first_cell = None
299            for x in range(x_min + 1, x_max):
300                cell = quadrant_cells[(x, y)]
301                if first_cell is None:
302                    first_cell = cell
303                else:
304                    if first_cell.shape == cell.shape:
305                        if frame_changed(first_cell, cell):
306                            horizontal_uniform = False
307                            break
308            if not horizontal_uniform:
309                break
310        if horizontal_uniform:
311            return True
312        for x in range(x_min + 1, x_max):
313            first_cell = None
314            for y in range(y_min + 1, y_max):
315                cell = quadrant_cells[(x, y)]
316                if first_cell is None:
317                    first_cell = cell
318                else:
319                    if first_cell.shape == cell.shape:
320                        if frame_changed(first_cell, cell):
321                            vertical_uniform = False
322                            break
323            if not vertical_uniform:
324                break
325        if vertical_uniform:
326            return True
327        return False
328
329    def judge_movement(
330        self, previous_frame: np.ndarray, current_frame: np.ndarray
331    ) -> Tuple[bool, Optional[bool]]:
332        if not frame_changed(previous_frame, current_frame):
333            return False, False
334        flag = False
335        for quadrant in [
336            "screen_quadrant_1",
337            "screen_quadrant_2",
338            "screen_quadrant_3",
339            "screen_quadrant_4",
340        ]:
341            prev_quad = self._emulator.state_parser.capture_named_region(
342                previous_frame, quadrant
343            )
344            curr_quad = self._emulator.state_parser.capture_named_region(
345                current_frame, quadrant
346            )
347            prev_uniform = (
348                prev_quad.max() == prev_quad.min()
349                or self._is_uniform_quadrant(previous_frame, quadrant)
350            )
351            curr_uniform = (
352                curr_quad.max() == curr_quad.min()
353                or self._is_uniform_quadrant(current_frame, quadrant)
354            )
355            if (
356                not frame_changed(prev_quad, curr_quad)
357                and not prev_uniform
358                and not curr_uniform
359            ):
360                flag = True
361                break
362        if flag:
363            prev_player_cell = self._emulator.state_parser.capture_grid_cells(
364                previous_frame
365            )[(0, 0)]
366            curr_player_cell = self._emulator.state_parser.capture_grid_cells(
367                current_frame
368            )[(0, 0)]
369            if frame_changed(prev_player_cell, curr_player_cell):
370                return False, True
371            return False, False
372        return True, None
373
374    def move(self, direction: str, steps: int) -> Tuple[List[Dict], int]:
375        action_dict = {
376            "right": LowLevelActions.PRESS_ARROW_RIGHT,
377            "down": LowLevelActions.PRESS_ARROW_DOWN,
378            "up": LowLevelActions.PRESS_ARROW_UP,
379            "left": LowLevelActions.PRESS_ARROW_LEFT,
380        }
381        if direction not in action_dict.keys():
382            log_error(f"Got invalid direction to move {direction}", self._parameters)
383        action = action_dict[direction]
384        action_success = -1
385        transition_state_dicts = []
386        transition_frames = []
387        previous_frame = self._emulator.get_current_frame()
388        n_step = 0
389        n_successful_steps = 0
390        has_rotated = None
391        agent_state = AgentState.FREE_ROAM
392        while n_step < steps and agent_state == AgentState.FREE_ROAM:
393            frames, done = self._emulator.step(action)
394            transition_state_dicts.append(self._state_tracker.report())
395            transition_frames.extend(frames)
396            current_frame = self._emulator.get_current_frame()
397            if done:
398                break
399            player_moved, player_rotated = self.judge_movement(
400                previous_frame, current_frame
401            )
402            if player_rotated is True:
403                has_rotated = True
404            if not player_moved and not player_rotated:
405                break
406            if player_moved:
407                n_successful_steps += 1
408            agent_state = self._emulator.state_parser.get_agent_state(
409                self._emulator.get_current_frame()
410            )
411            if agent_state != AgentState.FREE_ROAM:
412                break
413            n_step += 1
414            previous_frame = current_frame
415        if agent_state != AgentState.FREE_ROAM:
416            action_success = 2
417        else:
418            if n_step <= 0:
419                action_success = -1
420            elif n_step == steps:
421                action_success = 0
422            else:
423                action_success = 1
424        transition_state_dicts[-1]["core"]["action_return"] = {
425            "n_steps_taken": n_successful_steps,
426            "rotated": has_rotated,
427        }
428        return transition_state_dicts, action_success
429
430    def is_valid(self, **kwargs):
431        return (
432            self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
433            == AgentState.FREE_ROAM
434        )
435
436
437class MoveStepsAction(BaseMovementAction):
438    """
439    Moves the agent in a specified cardinal direction for a specified number of steps.
440    """
441
442    def get_action_space(self):
443        return Discrete(4 * HARD_MAX_STEPS)
444
445    def space_to_parameters(self, space_action):
446        direction = None
447        steps = None
448        if space_action < 0 or space_action >= 4 * HARD_MAX_STEPS:
449            return None
450        if space_action < HARD_MAX_STEPS:
451            direction = "up"
452            steps = space_action
453        elif space_action < 2 * HARD_MAX_STEPS:
454            direction = "down"
455            steps = space_action - HARD_MAX_STEPS
456        elif space_action < 3 * HARD_MAX_STEPS:
457            direction = "left"
458            steps = space_action - 2 * HARD_MAX_STEPS
459        else:
460            direction = "right"
461            steps = space_action - 3 * HARD_MAX_STEPS
462        return {"direction": direction, "steps": steps + 1}
463
464    def parameters_to_space(self, direction: str, steps: int):
465        if steps is None or steps <= 0 or steps > HARD_MAX_STEPS:
466            return None
467        if direction == "up":
468            return steps - 1
469        if direction == "down":
470            return HARD_MAX_STEPS + steps - 1
471        if direction == "left":
472            return 2 * HARD_MAX_STEPS + steps - 1
473        if direction == "right":
474            return 3 * HARD_MAX_STEPS + steps - 1
475        return None
476
477    def _execute(self, direction, steps):
478        transition_states, status = self.move(direction=direction, steps=steps)
479        return transition_states, status
480
481    def is_valid(self, **kwargs):
482        direction = kwargs.get("direction")
483        steps = kwargs.get("steps")
484        if direction is not None and direction not in ["up", "down", "left", "right"]:
485            return False
486        if steps is not None:
487            if not isinstance(steps, int):
488                return False
489            if steps <= 0:
490                return False
491        return super().is_valid(**kwargs)
492
493    @staticmethod
494    def get_action_name(direction: str, steps: int) -> str:
495        return f"Move {direction} {steps}"
496
497
498class MoveGridAction(BaseMovementAction):
499    """
500    Moves the agent on both axes. Will always try to move right/left first and then up/down.
501    """
502
503    def get_action_space(self):
504        return Box(
505            low=-HARD_MAX_STEPS // 2,
506            high=HARD_MAX_STEPS // 2,
507            shape=(2,),
508            dtype=np.int8,
509        )
510
511    def space_to_parameters(self, space_action):
512        right_action = space_action[0]
513        up_action = space_action[1]
514        return {"x_steps": right_action, "y_steps": up_action}
515
516    def parameters_to_space(self, x_steps, y_steps):
517        move_vec = np.zeros(2)
518        move_vec[0] = x_steps
519        move_vec[1] = y_steps
520        return move_vec
521
522    def _execute(self, x_steps, y_steps):
523        x_direction = "right" if x_steps >= 0 else "left"
524        y_direction = "up" if y_steps >= 0 else "down"
525        if x_steps != 0:
526            transition_states, status = self.move(
527                direction=x_direction, steps=abs(x_steps)
528            )
529            if status != 0:
530                return transition_states, status
531        else:
532            transition_states = []
533        if y_steps != 0:
534            more_transition_states, status = self.move(
535                direction=y_direction, steps=abs(y_steps)
536            )
537            transition_states.extend(more_transition_states)
538        try:
539            status is not None
540        except NameError:
541            log_warn(
542                "Weird case where both x_steps and y_steps are 0 in MoveGridAction or something.",
543                self._parameters,
544            )
545            transition_states = [self._state_tracker.report()]
546            status = -1
547        return transition_states, status
548
549    def is_valid(self, x_steps: int = None, y_steps: int = None):
550        if x_steps is not None and y_steps is not None:
551            if not isinstance(x_steps, int) or not isinstance(y_steps, int):
552                return False
553            if x_steps == 0 and y_steps == 0:
554                return False
555        return super().is_valid()
556
557    @staticmethod
558    def get_action_name(x_steps: int, y_steps: int) -> str:
559        return f"MoveGrid ({x_steps}, {y_steps})"
560
561
562class MenuAction(HighLevelAction):
563    """
564    Allows navigation and option selection within Deja Vu menus.
565
566    Is Valid When:
567    - In Menu State
568
569    Action Success Interpretation:
570    - -1: Frame did not change.
571    - 0: Frame changed.
572    """
573
574    REQUIRED_STATE_PARSER = DejaVuStateParser
575    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
576
577    _MENU_ACTION_MAP = {
578        "up": LowLevelActions.PRESS_ARROW_UP,
579        "down": LowLevelActions.PRESS_ARROW_DOWN,
580        "confirm": LowLevelActions.PRESS_BUTTON_A,
581        "left": LowLevelActions.PRESS_ARROW_LEFT,
582        "right": LowLevelActions.PRESS_ARROW_RIGHT,
583        "back": LowLevelActions.PRESS_BUTTON_B,
584    }
585
586    _MENU_ACTION_KEYS = list(_MENU_ACTION_MAP.keys())
587
588    def is_valid(self, **kwargs):
589        menu_action = kwargs.get("menu_action", None)
590        if menu_action is not None and menu_action not in self._MENU_ACTION_KEYS:
591            return False
592        state = self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
593        return state == AgentState.IN_MENU
594
595    def get_action_space(self):
596        return Discrete(len(self._MENU_ACTION_MAP))
597
598    def parameters_to_space(self, menu_action):
599        if menu_action not in self._MENU_ACTION_KEYS:
600            return None
601        return self._MENU_ACTION_KEYS.index(menu_action)
602
603    def space_to_parameters(self, space_action):
604        if space_action < 0 or space_action >= len(self._MENU_ACTION_MAP):
605            return None
606        menu_action = self._MENU_ACTION_KEYS[space_action]
607        return {"menu_action": menu_action}
608
609    def _execute(self, menu_action):
610        action = self._MENU_ACTION_MAP[menu_action]
611        current_frame = self._emulator.get_current_frame()
612        frames, done = self._emulator.step(action)
613        action_success = 0 if frame_changed(current_frame, frames[-1]) else -1
614        return [self._state_tracker.report()], action_success
615
616    @staticmethod
617    def get_action_name(menu_action: str) -> str:
618        return f"Menu {menu_action}"
619
620
621class OpenMenuAction(HighLevelAction):
622    """
623    Opens the investigation menu from free roam and optionally navigates to a section.
624
625    Is Valid When:
626    - In Free Roam State
627
628    Action Success Interpretation:
629    - -1: Could not open the menu or reach the requested section.
630    - 0: Menu opened (and section selected if specified).
631    """
632
633    OPTIONS = ["open", "case_notes", "evidence", "location"]
634
635    REQUIRED_STATE_PARSER = DejaVuStateParser
636    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
637
638    def get_action_space(self):
639        return Discrete(len(self.OPTIONS))
640
641    def space_to_parameters(self, space_action):
642        if space_action < 0 or space_action >= len(self.OPTIONS):
643            return None
644        return {"option": self.OPTIONS[space_action]}
645
646    def parameters_to_space(self, option: Optional[str] = None):
647        if option is None:
648            option = "open"
649        if option not in self.OPTIONS:
650            return None
651        return self.OPTIONS.index(option)
652
653    def is_valid(self, option: Optional[str] = None):
654        if option is not None and option not in self.OPTIONS:
655            return False
656        state = self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
657        return state == AgentState.FREE_ROAM
658
659    def _is_target_menu(self, option: str, frame: np.ndarray) -> bool:
660        parser = self._emulator.state_parser
661        if option == "case_notes":
662            return parser.is_in_case_notes(frame)
663        if option == "evidence":
664            return parser.is_in_evidence_menu(frame)
665        if option == "location":
666            return parser.is_location_menu_open(frame)
667        return False
668
669    def _execute(self, option: Optional[str] = None):
670        if option is None:
671            option = "open"
672        self._emulator.step(LowLevelActions.PRESS_BUTTON_START)
673        state_reports = [self._state_tracker.report()]
674        current_frame = self._emulator.get_current_frame()
675        if not self._emulator.state_parser.is_in_menu(current_frame):
676            return state_reports, -1
677        if option == "open":
678            return state_reports, 0
679        if self._is_target_menu(option, current_frame):
680            return state_reports, 0
681        for action in [
682            LowLevelActions.PRESS_ARROW_RIGHT,
683            LowLevelActions.PRESS_ARROW_LEFT,
684        ]:
685            for _ in range(MENU_NAV_MAX_STEPS):
686                self._emulator.step(action)
687                state_reports.append(self._state_tracker.report())
688                current_frame = self._emulator.get_current_frame()
689                if self._is_target_menu(option, current_frame):
690                    return state_reports, 0
691        return state_reports, -1
692
693    @staticmethod
694    def get_action_name(option: str) -> str:
695        return f"OpenMenu {option}"
696
697
698class PuzzleAction(HighLevelAction):
699    """
700    Allows navigation and option selection within Deja Vu puzzle/deduction states.
701
702    Is Valid When:
703    - In Puzzle State
704
705    Action Success Interpretation:
706    - -1: Frame did not change.
707    - 0: Frame changed.
708    """
709
710    REQUIRED_STATE_PARSER = DejaVuStateParser
711    REQUIRED_STATE_TRACKER = CoreDejaVuTracker
712
713    _PUZZLE_ACTION_MAP = {
714        "up": LowLevelActions.PRESS_ARROW_UP,
715        "down": LowLevelActions.PRESS_ARROW_DOWN,
716        "confirm": LowLevelActions.PRESS_BUTTON_A,
717        "left": LowLevelActions.PRESS_ARROW_LEFT,
718        "right": LowLevelActions.PRESS_ARROW_RIGHT,
719        "back": LowLevelActions.PRESS_BUTTON_B,
720    }
721
722    _PUZZLE_ACTION_KEYS = list(_PUZZLE_ACTION_MAP.keys())
723
724    def is_valid(self, **kwargs):
725        puzzle_action = kwargs.get("puzzle_action", None)
726        if puzzle_action is not None and puzzle_action not in self._PUZZLE_ACTION_KEYS:
727            return False
728        state = self._state_tracker.get_episode_metric(("dejavu_core", "agent_state"))
729        return state == AgentState.IN_PUZZLE
730
731    def get_action_space(self):
732        return Discrete(len(self._PUZZLE_ACTION_MAP))
733
734    def parameters_to_space(self, puzzle_action):
735        if puzzle_action not in self._PUZZLE_ACTION_KEYS:
736            return None
737        return self._PUZZLE_ACTION_KEYS.index(puzzle_action)
738
739    def space_to_parameters(self, space_action):
740        if space_action < 0 or space_action >= len(self._PUZZLE_ACTION_MAP):
741            return None
742        puzzle_action = self._PUZZLE_ACTION_KEYS[space_action]
743        return {"puzzle_action": puzzle_action}
744
745    def _execute(self, puzzle_action):
746        action = self._PUZZLE_ACTION_MAP[puzzle_action]
747        current_frame = self._emulator.get_current_frame()
748        frames, done = self._emulator.step(action)
749        action_success = 0 if frame_changed(current_frame, frames[-1]) else -1
750        return [self._state_tracker.report()], action_success
751
752    @staticmethod
753    def get_action_name(puzzle_action: str) -> str:
754        return f"Puzzle {puzzle_action}"