gameboy_worlds.emulation.deja_vu.base_metrics

  1from typing import Optional
  2from abc import ABC
  3from gameboy_worlds.emulation.deja_vu.parsers import (
  4    AgentState,
  5    # MemoryBasedDejaVuStateParser,
  6    DejaVuStateParser,
  7    DejaVu1StateParser,
  8)
  9from gameboy_worlds.emulation.tracker import (
 10    MetricGroup,
 11    OCRegionMetric,
 12    # TerminationTruncationMetric,
 13)
 14
 15
 16import numpy as np
 17
 18
 19class CoreDejaVuMetrics(MetricGroup):
 20    """
 21    Deja Vu-specific core metrics.
 22
 23    Reports:
 24    - agent_state: The AgentState info. Is either Free Roam, In Dialogue or In Menu.
 25
 26    Final Reports:
 27    - None
 28    """
 29
 30    NAME = "dejavu_core"
 31    REQUIRED_PARSER = DejaVuStateParser
 32
 33    def start(self):
 34        self.n_battles_total = []
 35        super().start()
 36
 37    def reset(self, first=False):
 38        if not first:
 39            pass
 40        self.current_state: AgentState = (
 41            AgentState.IN_DIALOGUE
 42        )  # Start by default in dialogue because it has the least permissable actions.
 43        """ The current state of the agent in the game. """
 44        self._previous_state = self.current_state
 45
 46    def close(self):
 47        self.reset()
 48        return
 49
 50    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
 51        self._previous_state = self.current_state
 52        current_state = self.state_parser.get_agent_state(current_frame)
 53        self.current_state = current_state
 54
 55    def report(self) -> dict:
 56        """
 57        Reports the current Deja Vu core metrics:
 58        - Agent state
 59        Returns:
 60            dict: A dictionary containing the current agent state.
 61        """
 62        return {
 63            "agent_state": self.current_state,
 64        }
 65
 66    def report_final(self) -> dict:
 67        """
 68        Reports nothing:
 69        """
 70        return {}
 71
 72
 73class DejaVuOCRMetric(OCRegionMetric):
 74    REQUIRED_PARSER = DejaVuStateParser
 75
 76    def reset(self, first=False):
 77        super().reset(first)
 78        self.prev_was_in_dialogue = False
 79
 80    def start(self):
 81        self.kinds = {
 82            "dialogue": "dialogue_box_area",
 83            "menu": "menu_box_area",
 84        }
 85        super().start()
 86
 87    def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool:
 88        self.state_parser: DejaVuStateParser
 89        if kind == "dialogue":
 90            in_dialogue = self.state_parser.is_in_dialogue(current_screen=current_frame)
 91            # in_menu = self.state_parser.is_in_menu(current_screen=current_frame)
 92            return in_dialogue  # and not in_menu
 93        if kind == "menu":
 94            in_menu = self.state_parser.is_in_menu(current_screen=current_frame)
 95            return in_menu
 96        return False
 97
 98
 99class DejaVuTestMetric(MetricGroup):
100    """
101    Deja Vu-specific test metrics.
102
103    Reports:
104    - is_in_fight: Whether the agent is currently in a fight
105    - was_in_fight_last_step: Whether the agent was in a fight in the previous step
106    """
107
108    NAME = "dejavu_test"
109    REQUIRED_PARSER = DejaVuStateParser
110
111    def start(self):
112        super().start()
113
114    def reset(self, first=False):
115        if not first:
116            pass
117        self.is_in_dialogue = False
118        self.pre_was_dialogue = False
119
120    def close(self):
121        self.reset()
122        return
123
124    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
125        self.state_parser: DejaVuStateParser
126        is_dialogue = self.state_parser.is_in_dialogue(current_screen=current_frame)
127        self.pre_was_dialogue = self.is_in_dialogue
128        self.is_in_dialogue = is_dialogue
129
130    def report(self) -> dict:
131        return {
132            "is_in_dialogue": self.is_in_dialogue,
133            "was_in_dialogue_last_step": self.pre_was_dialogue,
134        }
135
136    def report_final(self) -> dict:
137        return {}