gameboy_worlds.emulation.pokemon.base_metrics

View Source

  1from typing import Optional
  2from abc import ABC
  3from gameboy_worlds.emulation.pokemon.parsers import (
  4    AgentState,
  5    MemoryBasedPokemonRedStateParser,
  6    PokemonRedStateParser,
  7    PokemonStateParser,
  8)
  9from gameboy_worlds.emulation.tracker import (
 10    MetricGroup,
 11    OCRegionMetric,
 12    TerminationTruncationMetric,
 13)
 14
 15
 16import numpy as np
 17
 18
 19class CorePokemonMetrics(MetricGroup):
 20    """
 21    Pokémon-specific metrics.
 22
 23    Reports:
 24    - agent_state: The AgentState info. Is either Free Roam, In Dialogue, In Menu or In Battle.
 25
 26    Final Reports:
 27    - None
 28
 29
 30    """
 31
 32    NAME = "pokemon_core"
 33    REQUIRED_PARSER = PokemonStateParser
 34
 35    def start(self):
 36        self.n_battles_total = []
 37        super().start()
 38
 39    def reset(self, first=False):
 40        if not first:
 41            pass
 42        self.current_state: AgentState = (
 43            AgentState.IN_DIALOGUE
 44        )  # Start by default in dialogue because it has the least permissable actions.
 45        """ The current state of the agent in the game. """
 46        self._previous_state = self.current_state
 47
 48    def close(self):
 49        self.reset()
 50        return
 51
 52    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
 53        self._previous_state = self.current_state
 54        current_state = self.state_parser.get_agent_state(current_frame)
 55        self.current_state = current_state
 56
 57    def report(self) -> dict:
 58        """
 59        Reports the current Pokémon core metrics:
 60        - Agent state
 61        Returns:
 62            dict: A dictionary containing the current agent state.
 63        """
 64        return {
 65            "agent_state": self.current_state
 66        }
 67
 68    def report_final(self) -> dict:
 69        """
 70        Reports nothing:
 71        """
 72        return {}
 73
 74
 75class PokemonOCRMetric(OCRegionMetric):
 76    REQUIRED_PARSER = PokemonStateParser
 77
 78    def reset(self, first=False):
 79        super().reset(first)
 80        self.prev_was_in_fight_options = False
 81
 82    def start(self):
 83        self.kinds = {
 84            "dialogue": "dialogue_box_full",
 85            "battle_attack_options": "screen_bottom_half",
 86        }
 87        super().start()
 88
 89    def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool:
 90        self.state_parser: PokemonStateParser
 91        if kind == "dialogue":
 92            in_dialogue = self.state_parser.dialogue_box_open(
 93                current_screen=current_frame
 94            )
 95            dialogue_empty = self.state_parser.dialogue_box_empty(
 96                current_screen=current_frame
 97            )
 98            in_battle_menu = self.state_parser.is_in_base_battle_menu(
 99                current_screen=current_frame
100            )
101            in_fight_options = self.state_parser.is_in_fight_options_menu(
102                current_screen=current_frame
103            )
104            in_bag = self.state_parser.is_in_fight_bag(current_screen=current_frame)
105            return (
106                in_dialogue
107                and not dialogue_empty
108                and not in_battle_menu
109                and not in_fight_options
110                and not in_bag
111            )
112        if kind == "battle_attack_options":
113            in_fight_options = self.state_parser.is_in_fight_options_menu(
114                current_screen=current_frame
115            )
116            if in_fight_options:
117                if self.prev_was_in_fight_options:
118                    self.prev_was_in_fight_options = True
119                    return False
120                else:
121                    self.prev_was_in_fight_options = True
122                    return True
123            else:
124                self.prev_was_in_fight_options = False
125                return False
126        return False
127
128
129class PokemonExitBattleTruncationMetric(TerminationTruncationMetric, ABC):
130    """
131    Truncates the episode if the agent exits a battle (enters into free roam)
132    Implement this class to test scenarios where the task can be completed entirely within a battle.
133    """
134
135    def determine_truncated(self, current_frame, recent_frames):
136        self.state_parser: PokemonStateParser
137        current_state = self.state_parser.get_agent_state(current_frame)
138        if current_state == AgentState.FREE_ROAM:
139            return True
140        return False
141
142
143class PokemonRedStarter(MetricGroup):
144    """
145    Specific tracking for choice of starter Pokémon in Pokémon Red.
146
147    Reports:
148    - current_starter: The starter Pokémon chosen in the current episode.
149
150    Final Reports:
151    - starter_choices: A dictionary with the total number of times each starter Pokémon was chosen across all episodes.
152    """
153
154    NAME = "pokemon_red_starter"
155    REQUIRED_PARSER = PokemonRedStateParser
156
157    def start(self):
158        self.starters_chosen = []
159        super().start()
160
161    def reset(self, first=False):
162        if not first:
163            self.starters_chosen.append(self.current_starter)
164        self.current_starter = None
165        """ The starter Pokémon chosen in the current episode. """
166
167    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
168        if self.current_starter is not None:
169            return
170        if recent_frames is not None:
171            all_frames = recent_frames
172        else:
173            all_frames = np.array([current_frame])
174        for frame in all_frames:
175            chose_charmander = self.state_parser.named_region_matches_multi_target(
176                frame, "dialogue_box_middle", "picked_charmander"
177            )
178            if chose_charmander:
179                self.current_starter = "charmander"
180                return
181            chose_bulbasaur = self.state_parser.named_region_matches_multi_target(
182                frame, "dialogue_box_middle", "picked_bulbasaur"
183            )
184            if chose_bulbasaur:
185                self.current_starter = "bulbasaur"
186                return
187            chose_squirtle = self.state_parser.named_region_matches_multi_target(
188                frame, "dialogue_box_middle", "picked_squirtle"
189            )
190            if chose_squirtle:
191                self.current_starter = "squirtle"
192                return
193        return
194
195    def report(self) -> dict:
196        """
197        Reports the current starter Pokémon chosen in the episode.
198
199        Returns:
200            dict: A dictionary containing the current starter Pokémon.
201        """
202        return {"current_starter": self.current_starter}
203
204    def close(self):
205        self.reset()
206        starter_choices = {"charmander": 0, "bulbasaur": 0, "squirtle": 0, None: 0}
207        for choice in self.starters_chosen:
208            starter_choices[choice] += 1
209        starter_choices["None"] = starter_choices.pop(None)
210        self.starter_choices = starter_choices
211
212    def report_final(self):
213        """
214        Reports the total number of times each starter Pokémon was chosen across all episodes.
215        """
216        return self.starter_choices
217
218
219class PokemonRedLocation(MetricGroup):
220    """
221    Reads from memory states to determine the player's current location in Pokemon Red.
222
223    Reports:
224    - direction: The direction the player is facing
225    - has_moved: Whether the player has moved since the last step
226    - current_global_location: (x, y)
227    - current_local_location: (x, y, map_name)
228    - n_walk_steps: Number of walk steps taken in the current episode
229    - unique_locations: List of unique locations visited in the current episode
230    - n_of_unique_locations: Number of unique locations visited in the current episode
231
232    Final Reports:
233    - mean_n_walk_steps_per_episode: Mean number of walk steps taken per episode
234    - mean_n_unique_locations_per_episode: Mean number of unique locations visited per episode
235    - std_n_walk_steps_per_episode: Standard deviation of walk steps taken per episode
236    - std_n_unique_locations_per_episode: Standard deviation of unique locations visited per episode
237    - max_n_walk_steps_per_episode: Maximum number of walk steps taken in a single episode
238    - max_n_unique_locations_per_episode: Maximum number of unique locations visited in a single episode
239
240
241    """
242
243    NAME = "pokemon_red_location"
244    REQUIRED_PARSER = MemoryBasedPokemonRedStateParser
245
246    def start(self):
247        self.total_n_walk_steps = []
248        self.total_n_of_unique_locations = []
249        super().start()
250
251    def reset(self, first=False):
252        if not first:
253            self.total_n_of_unique_locations.append(len(self.unique_locations))
254            self.total_n_walk_steps.append(self.n_walk_steps)
255        else:
256            self.direction = None
257            self.current_local_location = None
258            self.current_global_location = None
259            self.has_moved = False
260            self.n_walk_steps = 0
261            self.unique_locations = set()
262
263    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
264        self.state_parser: MemoryBasedPokemonRedStateParser
265        self.direction = self.state_parser.get_facing_direction()
266        current_local_position = self.state_parser.get_local_coords()
267        current_global_position = self.state_parser.get_global_coords()
268        x, y, map_number = current_local_position
269        map_name = self.state_parser.get_map_name(map_number)
270        if self.current_local_location is None:
271            self.current_local_location = (x, y, map_name)
272            self.current_global_location = current_global_position
273            self.unique_locations.add(map_name)
274        else:
275            if (
276                self.current_local_location[0] != x
277                or self.current_local_location[1] != y
278                or self.current_local_location[2] != map_name
279            ):
280                self.has_moved = True
281                if map_name == self.current_local_location[2]:
282                    initial_coord = np.array(self.current_local_location[0:2])
283                    new_coord = np.array(current_local_position[0:2])
284                    manhattan_distance = np.sum(np.abs(initial_coord - new_coord))
285                    self.n_walk_steps += manhattan_distance
286                else:
287                    # use global coords to estimate distance moved
288                    initial_coord = np.array(self.current_global_location)
289                    new_coord = np.array(current_global_position)
290                    manhattan_distance = np.sum(np.abs(initial_coord - new_coord))
291                    self.n_walk_steps += manhattan_distance
292                    self.unique_locations.add(map_name)
293            else:
294                self.has_moved = False
295            self.current_global_location = current_global_position
296            self.current_local_location = (x, y, map_number)
297
298    def report(self) -> dict:
299        """
300        Reports the current location metrics:
301        - direction: The direction the player is facing
302        - has_moved: Whether the player has moved since the last step
303        - current_global_location: (x, y)
304        - current_local_location: (x, y, map_name)
305        - n_walk_steps: Number of walk steps taken in the current episode
306        - unique_locations: List of unique locations visited in the current episode
307        - n_of_unique_locations: Number of unique locations visited in the current episode
308
309        Returns:
310            dict: A dictionary containing the current location metrics.
311        """
312        return {
313            "direction": self.direction,
314            "has_moved": self.has_moved,
315            "current_global_location": self.current_global_location,
316            "current_local_location": self.current_local_location,
317            "n_walk_steps": self.n_walk_steps,
318            "unique_locations": list(self.unique_locations),
319            "n_of_unique_locations": len(self.unique_locations),
320        }
321
322    def report_final(self):
323        return {
324            "mean_n_walk_steps_per_episode": float(np.mean(self.total_n_walk_steps)),
325            "mean_n_unique_locations_per_episode": float(
326                np.mean(self.total_n_of_unique_locations)
327            ),
328            "std_n_walk_steps_per_episode": float(np.std(self.total_n_walk_steps)),
329            "std_n_unique_locations_per_episode": float(
330                np.std(self.total_n_of_unique_locations)
331            ),
332            "max_n_walk_steps_per_episode": int(np.max(self.total_n_walk_steps)),
333            "max_n_unique_locations_per_episode": int(
334                np.max(self.total_n_of_unique_locations)
335            ),
336        }
337
338    def close(self):
339        pass
340
341
342class PokemonTestMetric(MetricGroup):
343    NAME = "pokemon_test"
344    REQUIRED_PARSER = PokemonStateParser
345
346    def start(self):
347        super().start()
348
349    def reset(self, first=False):
350        if not first:
351            pass
352        self.prev_was_fight = False
353        self.is_in_fight = False
354        self.is_got_away_safely = False
355
356    def close(self):
357        self.reset()
358        return
359
360    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
361        self.state_parser: PokemonStateParser
362        is_fight = False
363        self.is_got_away_safely = self.state_parser.named_region_matches_multi_target(
364            current_frame, "dialogue_box_middle", "got_away_safely"
365        )
366        # is_fight = self.state_parser.is_in_fight_options_menu(current_screen=current_frame)
367        self.prev_was_fight = self.is_in_fight
368        self.is_in_fight = is_fight
369
370    def report(self) -> dict:
371        return {
372            "is_in_fight": self.is_in_fight,
373            "is_got_away_safely": self.is_got_away_safely,
374            "was_in_fight_last_step": self.prev_was_fight,
375        }
376
377    def report_final(self) -> dict:
378        return {}

class CorePokemonMetrics(gameboy_worlds.emulation.tracker.MetricGroup): View Source

20class CorePokemonMetrics(MetricGroup):
21    """
22    Pokémon-specific metrics.
23
24    Reports:
25    - agent_state: The AgentState info. Is either Free Roam, In Dialogue, In Menu or In Battle.
26
27    Final Reports:
28    - None
29
30
31    """
32
33    NAME = "pokemon_core"
34    REQUIRED_PARSER = PokemonStateParser
35
36    def start(self):
37        self.n_battles_total = []
38        super().start()
39
40    def reset(self, first=False):
41        if not first:
42            pass
43        self.current_state: AgentState = (
44            AgentState.IN_DIALOGUE
45        )  # Start by default in dialogue because it has the least permissable actions.
46        """ The current state of the agent in the game. """
47        self._previous_state = self.current_state
48
49    def close(self):
50        self.reset()
51        return
52
53    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
54        self._previous_state = self.current_state
55        current_state = self.state_parser.get_agent_state(current_frame)
56        self.current_state = current_state
57
58    def report(self) -> dict:
59        """
60        Reports the current Pokémon core metrics:
61        - Agent state
62        Returns:
63            dict: A dictionary containing the current agent state.
64        """
65        return {
66            "agent_state": self.current_state
67        }
68
69    def report_final(self) -> dict:
70        """
71        Reports nothing:
72        """
73        return {}

Pokémon-specific metrics.

Reports:

agent_state: The AgentState info. Is either Free Roam, In Dialogue, In Menu or In Battle.

Final Reports:

None

NAME = 'pokemon_core'

Name of the MetricGroup.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.pokemon.parsers.PokemonStateParser'>

The StateParser which implements the minimum required functionality for this MetricGroup to work.

def start(self): View Source

36    def start(self):
37        self.n_battles_total = []
38        super().start()

Called once when environment starts. All subclasses should call super() AFTER initializing their own variables. Only variables that will persist across episodes should be initialized here.

def reset(self, first=False): View Source

40    def reset(self, first=False):
41        if not first:
42            pass
43        self.current_state: AgentState = (
44            AgentState.IN_DIALOGUE
45        )  # Start by default in dialogue because it has the least permissable actions.
46        """ The current state of the agent in the game. """
47        self._previous_state = self.current_state

Called when environment resets.

Arguments:

first (bool): Whether this is the first reset of the environment. If True, might need to aggregate metrics into running final totals.

def close(self): View Source

49    def close(self):
50        self.reset()
51        return

Called when environment closes. Good for computing summary stats.

Step will not be called after this.

def step( self, current_frame: numpy.ndarray, recent_frames: Optional[numpy.ndarray]): View Source

53    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
54        self._previous_state = self.current_state
55        current_state = self.state_parser.get_agent_state(current_frame)
56        self.current_state = current_state

Called each environment step to update metrics.

Arguments:

current_frame (np.ndarray): The current frame rendered by the emulator.
recent_frames (Optional[np.ndarray]): The stack of frames that were rendered during the last action. Shape is [n_frames, height, width, channels]. Can be None if rendering is disabled.

def report(self) -> dict: View Source

58    def report(self) -> dict:
59        """
60        Reports the current Pokémon core metrics:
61        - Agent state
62        Returns:
63            dict: A dictionary containing the current agent state.
64        """
65        return {
66            "agent_state": self.current_state
67        }

Reports the current Pokémon core metrics:

Agent state

Returns:

dict: A dictionary containing the current agent state.

def report_final(self) -> dict: View Source

69    def report_final(self) -> dict:
70        """
71        Reports nothing:
72        """
73        return {}

Reports nothing:

Inherited Members

gameboy_worlds.emulation.tracker.MetricGroup: MetricGroup; state_parser; final_metrics; log_info; log_warn; log_report

class PokemonOCRMetric(gameboy_worlds.emulation.tracker.OCRegionMetric): View Source

 76class PokemonOCRMetric(OCRegionMetric):
 77    REQUIRED_PARSER = PokemonStateParser
 78
 79    def reset(self, first=False):
 80        super().reset(first)
 81        self.prev_was_in_fight_options = False
 82
 83    def start(self):
 84        self.kinds = {
 85            "dialogue": "dialogue_box_full",
 86            "battle_attack_options": "screen_bottom_half",
 87        }
 88        super().start()
 89
 90    def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool:
 91        self.state_parser: PokemonStateParser
 92        if kind == "dialogue":
 93            in_dialogue = self.state_parser.dialogue_box_open(
 94                current_screen=current_frame
 95            )
 96            dialogue_empty = self.state_parser.dialogue_box_empty(
 97                current_screen=current_frame
 98            )
 99            in_battle_menu = self.state_parser.is_in_base_battle_menu(
100                current_screen=current_frame
101            )
102            in_fight_options = self.state_parser.is_in_fight_options_menu(
103                current_screen=current_frame
104            )
105            in_bag = self.state_parser.is_in_fight_bag(current_screen=current_frame)
106            return (
107                in_dialogue
108                and not dialogue_empty
109                and not in_battle_menu
110                and not in_fight_options
111                and not in_bag
112            )
113        if kind == "battle_attack_options":
114            in_fight_options = self.state_parser.is_in_fight_options_menu(
115                current_screen=current_frame
116            )
117            if in_fight_options:
118                if self.prev_was_in_fight_options:
119                    self.prev_was_in_fight_options = True
120                    return False
121                else:
122                    self.prev_was_in_fight_options = True
123                    return True
124            else:
125                self.prev_was_in_fight_options = False
126                return False
127        return False

Watch particular screen regions and capture subscreens for OCR when possible. Does not actually perform OCR itself, but makes it easy to capture the relevant regions. Children implementing this must define self.kinds in start() and then call on super().start().

Reports:

ocr_regions: A dictionary mapping kinds to captured regions that had OCR-eligible text detected in them. The keys are kinds of OCR regions, and the values are the stacks of captured screen regions as numpy arrays of shape (num_captures, height, width, channels).
step: The current step number. Useful for differentiating when multiple OCR texts were found in the same episode. You can typically safely ignore this.

Final Reports:

ocr_regions: A list of tuples for all steps where OCR was detected. Is in form: List[Tuple[int, Dict[str, np.ndarray]]] where the int is the step number and the Dict maps kinds to a stack of the captured screen region.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.pokemon.parsers.PokemonStateParser'>

The StateParser which implements the minimum required functionality for this MetricGroup to work.

def reset(self, first=False): View Source

79    def reset(self, first=False):
80        super().reset(first)
81        self.prev_was_in_fight_options = False

ocr_regions will track a list of the form List[Tuple[int, Dict[str, np.ndarray]]] which is a list of (step_number, {kind: ocr_region}) dictionaries.

def start(self): View Source

83    def start(self):
84        self.kinds = {
85            "dialogue": "dialogue_box_full",
86            "battle_attack_options": "screen_bottom_half",
87        }
88        super().start()

Assumes the child has initialized a dict called self.kinds which tracks the various kinds of OCR that could be done. self.kinds should be in the form: {kind: region_name} where region_name is the name of the region to OCR for that kind. Will track ocr captured region results in form of list of dictionaries where these kinds are keys.

def can_read_kind(self, current_frame: numpy.ndarray, kind: str) -> bool: View Source

 90    def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool:
 91        self.state_parser: PokemonStateParser
 92        if kind == "dialogue":
 93            in_dialogue = self.state_parser.dialogue_box_open(
 94                current_screen=current_frame
 95            )
 96            dialogue_empty = self.state_parser.dialogue_box_empty(
 97                current_screen=current_frame
 98            )
 99            in_battle_menu = self.state_parser.is_in_base_battle_menu(
100                current_screen=current_frame
101            )
102            in_fight_options = self.state_parser.is_in_fight_options_menu(
103                current_screen=current_frame
104            )
105            in_bag = self.state_parser.is_in_fight_bag(current_screen=current_frame)
106            return (
107                in_dialogue
108                and not dialogue_empty
109                and not in_battle_menu
110                and not in_fight_options
111                and not in_bag
112            )
113        if kind == "battle_attack_options":
114            in_fight_options = self.state_parser.is_in_fight_options_menu(
115                current_screen=current_frame
116            )
117            if in_fight_options:
118                if self.prev_was_in_fight_options:
119                    self.prev_was_in_fight_options = True
120                    return False
121                else:
122                    self.prev_was_in_fight_options = True
123                    return True
124            else:
125                self.prev_was_in_fight_options = False
126                return False
127        return False

Checks if the frame has text for the given kind.

Arguments:

frame (np.ndarray): The frame to check.
kind (str): The kind of text to check for.

Inherited Members

gameboy_worlds.emulation.tracker.MetricGroup: MetricGroup; state_parser; final_metrics; log_info; log_warn; log_report
gameboy_worlds.emulation.tracker.OCRegionMetric: NAME; step; report; report_final; close

class PokemonExitBattleTruncationMetric(gameboy_worlds.emulation.tracker.TerminationTruncationMetric, abc.ABC): View Source

130class PokemonExitBattleTruncationMetric(TerminationTruncationMetric, ABC):
131    """
132    Truncates the episode if the agent exits a battle (enters into free roam)
133    Implement this class to test scenarios where the task can be completed entirely within a battle.
134    """
135
136    def determine_truncated(self, current_frame, recent_frames):
137        self.state_parser: PokemonStateParser
138        current_state = self.state_parser.get_agent_state(current_frame)
139        if current_state == AgentState.FREE_ROAM:
140            return True
141        return False

Truncates the episode if the agent exits a battle (enters into free roam) Implement this class to test scenarios where the task can be completed entirely within a battle.

def determine_truncated(self, current_frame, recent_frames): View Source

136    def determine_truncated(self, current_frame, recent_frames):
137        self.state_parser: PokemonStateParser
138        current_state = self.state_parser.get_agent_state(current_frame)
139        if current_state == AgentState.FREE_ROAM:
140            return True
141        return False

Determines whether the environment was truncated.

Parameters

current_frame: The current frame rendered by the emulator.
recent_frames: The stack of frames that were rendered during the last action. Shape is [n_frames, height, width, channels]. Can be None if rendering is disabled.

Returns

True if the environment was truncated, False otherwise.

Inherited Members

gameboy_worlds.emulation.tracker.TerminationTruncationMetric: NAME; start; reset; close; determine_terminated; step; report; report_final
gameboy_worlds.emulation.tracker.MetricGroup: REQUIRED_PARSER; state_parser; final_metrics; log_info; log_warn; log_report

class PokemonRedStarter(gameboy_worlds.emulation.tracker.MetricGroup): View Source

144class PokemonRedStarter(MetricGroup):
145    """
146    Specific tracking for choice of starter Pokémon in Pokémon Red.
147
148    Reports:
149    - current_starter: The starter Pokémon chosen in the current episode.
150
151    Final Reports:
152    - starter_choices: A dictionary with the total number of times each starter Pokémon was chosen across all episodes.
153    """
154
155    NAME = "pokemon_red_starter"
156    REQUIRED_PARSER = PokemonRedStateParser
157
158    def start(self):
159        self.starters_chosen = []
160        super().start()
161
162    def reset(self, first=False):
163        if not first:
164            self.starters_chosen.append(self.current_starter)
165        self.current_starter = None
166        """ The starter Pokémon chosen in the current episode. """
167
168    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
169        if self.current_starter is not None:
170            return
171        if recent_frames is not None:
172            all_frames = recent_frames
173        else:
174            all_frames = np.array([current_frame])
175        for frame in all_frames:
176            chose_charmander = self.state_parser.named_region_matches_multi_target(
177                frame, "dialogue_box_middle", "picked_charmander"
178            )
179            if chose_charmander:
180                self.current_starter = "charmander"
181                return
182            chose_bulbasaur = self.state_parser.named_region_matches_multi_target(
183                frame, "dialogue_box_middle", "picked_bulbasaur"
184            )
185            if chose_bulbasaur:
186                self.current_starter = "bulbasaur"
187                return
188            chose_squirtle = self.state_parser.named_region_matches_multi_target(
189                frame, "dialogue_box_middle", "picked_squirtle"
190            )
191            if chose_squirtle:
192                self.current_starter = "squirtle"
193                return
194        return
195
196    def report(self) -> dict:
197        """
198        Reports the current starter Pokémon chosen in the episode.
199
200        Returns:
201            dict: A dictionary containing the current starter Pokémon.
202        """
203        return {"current_starter": self.current_starter}
204
205    def close(self):
206        self.reset()
207        starter_choices = {"charmander": 0, "bulbasaur": 0, "squirtle": 0, None: 0}
208        for choice in self.starters_chosen:
209            starter_choices[choice] += 1
210        starter_choices["None"] = starter_choices.pop(None)
211        self.starter_choices = starter_choices
212
213    def report_final(self):
214        """
215        Reports the total number of times each starter Pokémon was chosen across all episodes.
216        """
217        return self.starter_choices

Specific tracking for choice of starter Pokémon in Pokémon Red.

Reports:

current_starter: The starter Pokémon chosen in the current episode.

Final Reports:

starter_choices: A dictionary with the total number of times each starter Pokémon was chosen across all episodes.

NAME = 'pokemon_red_starter'

Name of the MetricGroup.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.pokemon.parsers.PokemonRedStateParser'>

The StateParser which implements the minimum required functionality for this MetricGroup to work.

def start(self): View Source

158    def start(self):
159        self.starters_chosen = []
160        super().start()

Called once when environment starts. All subclasses should call super() AFTER initializing their own variables. Only variables that will persist across episodes should be initialized here.

def reset(self, first=False): View Source

162    def reset(self, first=False):
163        if not first:
164            self.starters_chosen.append(self.current_starter)
165        self.current_starter = None
166        """ The starter Pokémon chosen in the current episode. """

Called when environment resets.

Arguments:

first (bool): Whether this is the first reset of the environment. If True, might need to aggregate metrics into running final totals.

def step( self, current_frame: numpy.ndarray, recent_frames: Optional[numpy.ndarray]): View Source

168    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
169        if self.current_starter is not None:
170            return
171        if recent_frames is not None:
172            all_frames = recent_frames
173        else:
174            all_frames = np.array([current_frame])
175        for frame in all_frames:
176            chose_charmander = self.state_parser.named_region_matches_multi_target(
177                frame, "dialogue_box_middle", "picked_charmander"
178            )
179            if chose_charmander:
180                self.current_starter = "charmander"
181                return
182            chose_bulbasaur = self.state_parser.named_region_matches_multi_target(
183                frame, "dialogue_box_middle", "picked_bulbasaur"
184            )
185            if chose_bulbasaur:
186                self.current_starter = "bulbasaur"
187                return
188            chose_squirtle = self.state_parser.named_region_matches_multi_target(
189                frame, "dialogue_box_middle", "picked_squirtle"
190            )
191            if chose_squirtle:
192                self.current_starter = "squirtle"
193                return
194        return

Called each environment step to update metrics.

Arguments:

current_frame (np.ndarray): The current frame rendered by the emulator.
recent_frames (Optional[np.ndarray]): The stack of frames that were rendered during the last action. Shape is [n_frames, height, width, channels]. Can be None if rendering is disabled.

def report(self) -> dict: View Source

196    def report(self) -> dict:
197        """
198        Reports the current starter Pokémon chosen in the episode.
199
200        Returns:
201            dict: A dictionary containing the current starter Pokémon.
202        """
203        return {"current_starter": self.current_starter}

Reports the current starter Pokémon chosen in the episode.

Returns:

dict: A dictionary containing the current starter Pokémon.

def close(self): View Source

205    def close(self):
206        self.reset()
207        starter_choices = {"charmander": 0, "bulbasaur": 0, "squirtle": 0, None: 0}
208        for choice in self.starters_chosen:
209            starter_choices[choice] += 1
210        starter_choices["None"] = starter_choices.pop(None)
211        self.starter_choices = starter_choices

Called when environment closes. Good for computing summary stats.

Step will not be called after this.

def report_final(self): View Source

213    def report_final(self):
214        """
215        Reports the total number of times each starter Pokémon was chosen across all episodes.
216        """
217        return self.starter_choices

Reports the total number of times each starter Pokémon was chosen across all episodes.

Inherited Members

gameboy_worlds.emulation.tracker.MetricGroup: MetricGroup; state_parser; final_metrics; log_info; log_warn; log_report

class PokemonRedLocation(gameboy_worlds.emulation.tracker.MetricGroup): View Source

220class PokemonRedLocation(MetricGroup):
221    """
222    Reads from memory states to determine the player's current location in Pokemon Red.
223
224    Reports:
225    - direction: The direction the player is facing
226    - has_moved: Whether the player has moved since the last step
227    - current_global_location: (x, y)
228    - current_local_location: (x, y, map_name)
229    - n_walk_steps: Number of walk steps taken in the current episode
230    - unique_locations: List of unique locations visited in the current episode
231    - n_of_unique_locations: Number of unique locations visited in the current episode
232
233    Final Reports:
234    - mean_n_walk_steps_per_episode: Mean number of walk steps taken per episode
235    - mean_n_unique_locations_per_episode: Mean number of unique locations visited per episode
236    - std_n_walk_steps_per_episode: Standard deviation of walk steps taken per episode
237    - std_n_unique_locations_per_episode: Standard deviation of unique locations visited per episode
238    - max_n_walk_steps_per_episode: Maximum number of walk steps taken in a single episode
239    - max_n_unique_locations_per_episode: Maximum number of unique locations visited in a single episode
240
241
242    """
243
244    NAME = "pokemon_red_location"
245    REQUIRED_PARSER = MemoryBasedPokemonRedStateParser
246
247    def start(self):
248        self.total_n_walk_steps = []
249        self.total_n_of_unique_locations = []
250        super().start()
251
252    def reset(self, first=False):
253        if not first:
254            self.total_n_of_unique_locations.append(len(self.unique_locations))
255            self.total_n_walk_steps.append(self.n_walk_steps)
256        else:
257            self.direction = None
258            self.current_local_location = None
259            self.current_global_location = None
260            self.has_moved = False
261            self.n_walk_steps = 0
262            self.unique_locations = set()
263
264    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
265        self.state_parser: MemoryBasedPokemonRedStateParser
266        self.direction = self.state_parser.get_facing_direction()
267        current_local_position = self.state_parser.get_local_coords()
268        current_global_position = self.state_parser.get_global_coords()
269        x, y, map_number = current_local_position
270        map_name = self.state_parser.get_map_name(map_number)
271        if self.current_local_location is None:
272            self.current_local_location = (x, y, map_name)
273            self.current_global_location = current_global_position
274            self.unique_locations.add(map_name)
275        else:
276            if (
277                self.current_local_location[0] != x
278                or self.current_local_location[1] != y
279                or self.current_local_location[2] != map_name
280            ):
281                self.has_moved = True
282                if map_name == self.current_local_location[2]:
283                    initial_coord = np.array(self.current_local_location[0:2])
284                    new_coord = np.array(current_local_position[0:2])
285                    manhattan_distance = np.sum(np.abs(initial_coord - new_coord))
286                    self.n_walk_steps += manhattan_distance
287                else:
288                    # use global coords to estimate distance moved
289                    initial_coord = np.array(self.current_global_location)
290                    new_coord = np.array(current_global_position)
291                    manhattan_distance = np.sum(np.abs(initial_coord - new_coord))
292                    self.n_walk_steps += manhattan_distance
293                    self.unique_locations.add(map_name)
294            else:
295                self.has_moved = False
296            self.current_global_location = current_global_position
297            self.current_local_location = (x, y, map_number)
298
299    def report(self) -> dict:
300        """
301        Reports the current location metrics:
302        - direction: The direction the player is facing
303        - has_moved: Whether the player has moved since the last step
304        - current_global_location: (x, y)
305        - current_local_location: (x, y, map_name)
306        - n_walk_steps: Number of walk steps taken in the current episode
307        - unique_locations: List of unique locations visited in the current episode
308        - n_of_unique_locations: Number of unique locations visited in the current episode
309
310        Returns:
311            dict: A dictionary containing the current location metrics.
312        """
313        return {
314            "direction": self.direction,
315            "has_moved": self.has_moved,
316            "current_global_location": self.current_global_location,
317            "current_local_location": self.current_local_location,
318            "n_walk_steps": self.n_walk_steps,
319            "unique_locations": list(self.unique_locations),
320            "n_of_unique_locations": len(self.unique_locations),
321        }
322
323    def report_final(self):
324        return {
325            "mean_n_walk_steps_per_episode": float(np.mean(self.total_n_walk_steps)),
326            "mean_n_unique_locations_per_episode": float(
327                np.mean(self.total_n_of_unique_locations)
328            ),
329            "std_n_walk_steps_per_episode": float(np.std(self.total_n_walk_steps)),
330            "std_n_unique_locations_per_episode": float(
331                np.std(self.total_n_of_unique_locations)
332            ),
333            "max_n_walk_steps_per_episode": int(np.max(self.total_n_walk_steps)),
334            "max_n_unique_locations_per_episode": int(
335                np.max(self.total_n_of_unique_locations)
336            ),
337        }
338
339    def close(self):
340        pass

Reads from memory states to determine the player's current location in Pokemon Red.

Reports:

direction: The direction the player is facing
has_moved: Whether the player has moved since the last step
current_global_location: (x, y)
current_local_location: (x, y, map_name)
n_walk_steps: Number of walk steps taken in the current episode
unique_locations: List of unique locations visited in the current episode
n_of_unique_locations: Number of unique locations visited in the current episode

Final Reports:

mean_n_walk_steps_per_episode: Mean number of walk steps taken per episode
mean_n_unique_locations_per_episode: Mean number of unique locations visited per episode
std_n_walk_steps_per_episode: Standard deviation of walk steps taken per episode
std_n_unique_locations_per_episode: Standard deviation of unique locations visited per episode
max_n_walk_steps_per_episode: Maximum number of walk steps taken in a single episode
max_n_unique_locations_per_episode: Maximum number of unique locations visited in a single episode

NAME = 'pokemon_red_location'

Name of the MetricGroup.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.pokemon.parsers.MemoryBasedPokemonRedStateParser'>

The StateParser which implements the minimum required functionality for this MetricGroup to work.

def start(self): View Source

247    def start(self):
248        self.total_n_walk_steps = []
249        self.total_n_of_unique_locations = []
250        super().start()

Called once when environment starts. All subclasses should call super() AFTER initializing their own variables. Only variables that will persist across episodes should be initialized here.

def reset(self, first=False): View Source

252    def reset(self, first=False):
253        if not first:
254            self.total_n_of_unique_locations.append(len(self.unique_locations))
255            self.total_n_walk_steps.append(self.n_walk_steps)
256        else:
257            self.direction = None
258            self.current_local_location = None
259            self.current_global_location = None
260            self.has_moved = False
261            self.n_walk_steps = 0
262            self.unique_locations = set()

Called when environment resets.

Arguments:

first (bool): Whether this is the first reset of the environment. If True, might need to aggregate metrics into running final totals.

def step( self, current_frame: numpy.ndarray, recent_frames: Optional[numpy.ndarray]): View Source

264    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
265        self.state_parser: MemoryBasedPokemonRedStateParser
266        self.direction = self.state_parser.get_facing_direction()
267        current_local_position = self.state_parser.get_local_coords()
268        current_global_position = self.state_parser.get_global_coords()
269        x, y, map_number = current_local_position
270        map_name = self.state_parser.get_map_name(map_number)
271        if self.current_local_location is None:
272            self.current_local_location = (x, y, map_name)
273            self.current_global_location = current_global_position
274            self.unique_locations.add(map_name)
275        else:
276            if (
277                self.current_local_location[0] != x
278                or self.current_local_location[1] != y
279                or self.current_local_location[2] != map_name
280            ):
281                self.has_moved = True
282                if map_name == self.current_local_location[2]:
283                    initial_coord = np.array(self.current_local_location[0:2])
284                    new_coord = np.array(current_local_position[0:2])
285                    manhattan_distance = np.sum(np.abs(initial_coord - new_coord))
286                    self.n_walk_steps += manhattan_distance
287                else:
288                    # use global coords to estimate distance moved
289                    initial_coord = np.array(self.current_global_location)
290                    new_coord = np.array(current_global_position)
291                    manhattan_distance = np.sum(np.abs(initial_coord - new_coord))
292                    self.n_walk_steps += manhattan_distance
293                    self.unique_locations.add(map_name)
294            else:
295                self.has_moved = False
296            self.current_global_location = current_global_position
297            self.current_local_location = (x, y, map_number)

Called each environment step to update metrics.

Arguments:

current_frame (np.ndarray): The current frame rendered by the emulator.
recent_frames (Optional[np.ndarray]): The stack of frames that were rendered during the last action. Shape is [n_frames, height, width, channels]. Can be None if rendering is disabled.

def report(self) -> dict: View Source

299    def report(self) -> dict:
300        """
301        Reports the current location metrics:
302        - direction: The direction the player is facing
303        - has_moved: Whether the player has moved since the last step
304        - current_global_location: (x, y)
305        - current_local_location: (x, y, map_name)
306        - n_walk_steps: Number of walk steps taken in the current episode
307        - unique_locations: List of unique locations visited in the current episode
308        - n_of_unique_locations: Number of unique locations visited in the current episode
309
310        Returns:
311            dict: A dictionary containing the current location metrics.
312        """
313        return {
314            "direction": self.direction,
315            "has_moved": self.has_moved,
316            "current_global_location": self.current_global_location,
317            "current_local_location": self.current_local_location,
318            "n_walk_steps": self.n_walk_steps,
319            "unique_locations": list(self.unique_locations),
320            "n_of_unique_locations": len(self.unique_locations),
321        }

Reports the current location metrics:

direction: The direction the player is facing
has_moved: Whether the player has moved since the last step
current_global_location: (x, y)
current_local_location: (x, y, map_name)
n_walk_steps: Number of walk steps taken in the current episode
unique_locations: List of unique locations visited in the current episode
n_of_unique_locations: Number of unique locations visited in the current episode

Returns:

dict: A dictionary containing the current location metrics.

def report_final(self): View Source

323    def report_final(self):
324        return {
325            "mean_n_walk_steps_per_episode": float(np.mean(self.total_n_walk_steps)),
326            "mean_n_unique_locations_per_episode": float(
327                np.mean(self.total_n_of_unique_locations)
328            ),
329            "std_n_walk_steps_per_episode": float(np.std(self.total_n_walk_steps)),
330            "std_n_unique_locations_per_episode": float(
331                np.std(self.total_n_of_unique_locations)
332            ),
333            "max_n_walk_steps_per_episode": int(np.max(self.total_n_walk_steps)),
334            "max_n_unique_locations_per_episode": int(
335                np.max(self.total_n_of_unique_locations)
336            ),
337        }

Return metrics as dictionary for logging. Called at end of environment (before close). Will never be called before self.close.

Returns

Dictionary of metrics

def close(self): View Source

339    def close(self):
340        pass

Called when environment closes. Good for computing summary stats.

Step will not be called after this.

Inherited Members

gameboy_worlds.emulation.tracker.MetricGroup: MetricGroup; state_parser; final_metrics; log_info; log_warn; log_report

class PokemonTestMetric(gameboy_worlds.emulation.tracker.MetricGroup): View Source

343class PokemonTestMetric(MetricGroup):
344    NAME = "pokemon_test"
345    REQUIRED_PARSER = PokemonStateParser
346
347    def start(self):
348        super().start()
349
350    def reset(self, first=False):
351        if not first:
352            pass
353        self.prev_was_fight = False
354        self.is_in_fight = False
355        self.is_got_away_safely = False
356
357    def close(self):
358        self.reset()
359        return
360
361    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
362        self.state_parser: PokemonStateParser
363        is_fight = False
364        self.is_got_away_safely = self.state_parser.named_region_matches_multi_target(
365            current_frame, "dialogue_box_middle", "got_away_safely"
366        )
367        # is_fight = self.state_parser.is_in_fight_options_menu(current_screen=current_frame)
368        self.prev_was_fight = self.is_in_fight
369        self.is_in_fight = is_fight
370
371    def report(self) -> dict:
372        return {
373            "is_in_fight": self.is_in_fight,
374            "is_got_away_safely": self.is_got_away_safely,
375            "was_in_fight_last_step": self.prev_was_fight,
376        }
377
378    def report_final(self) -> dict:
379        return {}

Abstract Base class for organizing related metrics.

Documentation Guidlines:

Every subchild should document the following in their class docstrings:

Reports (List of keys that are present in the return dict of report)
Final Reports (List of keys that are present in the return dict of report_final)

NAME = 'pokemon_test'

Name of the MetricGroup.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.pokemon.parsers.PokemonStateParser'>

The StateParser which implements the minimum required functionality for this MetricGroup to work.

def start(self): View Source

347    def start(self):
348        super().start()

Called once when environment starts. All subclasses should call super() AFTER initializing their own variables. Only variables that will persist across episodes should be initialized here.

def reset(self, first=False): View Source

350    def reset(self, first=False):
351        if not first:
352            pass
353        self.prev_was_fight = False
354        self.is_in_fight = False
355        self.is_got_away_safely = False

Called when environment resets.

Arguments:

first (bool): Whether this is the first reset of the environment. If True, might need to aggregate metrics into running final totals.

def close(self): View Source

357    def close(self):
358        self.reset()
359        return

Called when environment closes. Good for computing summary stats.

Step will not be called after this.

def step( self, current_frame: numpy.ndarray, recent_frames: Optional[numpy.ndarray]): View Source

361    def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]):
362        self.state_parser: PokemonStateParser
363        is_fight = False
364        self.is_got_away_safely = self.state_parser.named_region_matches_multi_target(
365            current_frame, "dialogue_box_middle", "got_away_safely"
366        )
367        # is_fight = self.state_parser.is_in_fight_options_menu(current_screen=current_frame)
368        self.prev_was_fight = self.is_in_fight
369        self.is_in_fight = is_fight

Called each environment step to update metrics.

Arguments:

current_frame (np.ndarray): The current frame rendered by the emulator.
recent_frames (Optional[np.ndarray]): The stack of frames that were rendered during the last action. Shape is [n_frames, height, width, channels]. Can be None if rendering is disabled.

def report(self) -> dict: View Source

371    def report(self) -> dict:
372        return {
373            "is_in_fight": self.is_in_fight,
374            "is_got_away_safely": self.is_got_away_safely,
375            "was_in_fight_last_step": self.prev_was_fight,
376        }

Return metrics as dictionary for instantaneous variable tracking.

Returns

Dictionary of metrics

def report_final(self) -> dict: View Source

378    def report_final(self) -> dict:
379        return {}

Return metrics as dictionary for logging. Called at end of environment (before close). Will never be called before self.close.

Returns

Dictionary of metrics

Inherited Members

gameboy_worlds.emulation.tracker.MetricGroup: MetricGroup; state_parser; final_metrics; log_info; log_warn; log_report