gameboy_worlds.emulation.harvest_moon.base_metrics
1from typing import Optional 2from abc import ABC 3from gameboy_worlds.emulation.harvest_moon.parsers import ( 4 AgentState, 5 HarvestMoon1Parser, 6 HarvestMoon2Parser, 7 HarvestMoon3Parser, 8 HarvestMoonStateParser, 9) 10from gameboy_worlds.emulation.tracker import ( 11 MetricGroup, 12 OCRegionMetric, 13 TerminationTruncationMetric, 14) 15 16import numpy as np 17 18 19class CoreHarvestMoonMetrics(MetricGroup): 20 """ 21 Harvest Moon-specific metrics. 22 23 Reports: 24 - `agent_state`: The AgentState info. Is either FREE_ROAM or IN_DIALOGUE. 25 26 Final Reports: 27 - None 28 """ 29 30 NAME = "harvest_moon_core" 31 REQUIRED_PARSER = HarvestMoonStateParser 32 33 def reset(self, first=False): 34 if not first: 35 pass 36 self.current_state: AgentState = ( 37 AgentState.IN_DIALOGUE 38 ) # Start by default in dialogue because it has the least permissable actions. 39 """ The current state of the agent in the game. """ 40 self._previous_state = self.current_state 41 42 def close(self): 43 self.reset() 44 return 45 46 def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]): 47 self._previous_state = self.current_state 48 current_state = self.state_parser.get_agent_state(current_frame) 49 self.current_state = current_state 50 51 def report(self) -> dict: 52 return { 53 "agent_state": self.current_state, 54 } 55 56 def report_final(self) -> dict: 57 return {} 58 59 60class HarvestMoonOCRMetric(OCRegionMetric): 61 REQUIRED_PARSER = HarvestMoonStateParser 62 63 def reset(self, first=False): 64 super().reset(first) 65 self.prev_was_in_menu = False 66 67 def start(self): 68 self.kinds = { 69 "dialogue": "dialogue_bottom_right", 70 "menu": "menu_top_right", 71 } 72 super().start() 73 74 def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool: 75 self.state_parser: PokemonStateParser 76 if kind == "dialogue": 77 in_dialogue = self.state_parser.dialogue_box_open( 78 current_screen=current_frame 79 ) 80 # dialogue_empty = self.state_parser.dialogue_box_empty( 81 # current_screen=current_frame 82 # ) 83 return ( 84 in_dialogue 85 # and not dialogue_empty 86 ) 87 if kind == "menu": 88 in_menu = self.state_parser.is_in_menu(current_screen=current_frame) 89 return in_menu 90 return False 91 92 93class HarvestMoon1OCRMetric(HarvestMoonOCRMetric): 94 REQUIRED_PARSER = HarvestMoon1Parser 95 96 def start(self): 97 self.kinds = { 98 "dialogue": "dialogue_bottom_right", 99 "menu": "menu_top_right", 100 } 101 OCRegionMetric.start(self) 102 103 104class HarvestMoon2OCRMetric(HarvestMoonOCRMetric): 105 REQUIRED_PARSER = HarvestMoon2Parser 106 107 def start(self): 108 self.kinds = { 109 "dialogue": "dialogue_bottom_right", 110 "menu": "menu_top_right", 111 } 112 OCRegionMetric.start(self) 113 114 115class HarvestMoon3OCRMetric(HarvestMoonOCRMetric): 116 REQUIRED_PARSER = HarvestMoon3Parser 117 118 def start(self): 119 self.kinds = { 120 "dialogue": "dialogue_bottom_right", 121 "menu": "menu_top_right", 122 } 123 OCRegionMetric.start(self) 124 125 126class HarvestMoonTestMetric(MetricGroup): 127 """ 128 Harvest Moon metrics for test environments. 129 130 Reports: 131 - `agent_state`: The current AgentState (FREE_ROAM or IN_DIALOGUE). 132 - `previous_agent_state`: The AgentState from the previous step. 133 134 Final Reports: 135 - None 136 """ 137 138 NAME = "harvest_moon_test" 139 REQUIRED_PARSER = HarvestMoonStateParser 140 141 def start(self): 142 super().start() 143 144 def reset(self, first=False): 145 self.agent_state: AgentState = AgentState.IN_DIALOGUE 146 self.previous_agent_state: AgentState = self.agent_state 147 148 def close(self): 149 self.reset() 150 return 151 152 def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]): 153 self.state_parser: HarvestMoonStateParser 154 self.previous_agent_state = self.agent_state 155 self.agent_state = self.state_parser.get_agent_state(current_frame) 156 157 def report(self) -> dict: 158 return { 159 "agent_state": self.agent_state, 160 "previous_agent_state": self.previous_agent_state, 161 } 162 163 def report_final(self) -> dict: 164 return {}
20class CoreHarvestMoonMetrics(MetricGroup): 21 """ 22 Harvest Moon-specific metrics. 23 24 Reports: 25 - `agent_state`: The AgentState info. Is either FREE_ROAM or IN_DIALOGUE. 26 27 Final Reports: 28 - None 29 """ 30 31 NAME = "harvest_moon_core" 32 REQUIRED_PARSER = HarvestMoonStateParser 33 34 def reset(self, first=False): 35 if not first: 36 pass 37 self.current_state: AgentState = ( 38 AgentState.IN_DIALOGUE 39 ) # Start by default in dialogue because it has the least permissable actions. 40 """ The current state of the agent in the game. """ 41 self._previous_state = self.current_state 42 43 def close(self): 44 self.reset() 45 return 46 47 def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]): 48 self._previous_state = self.current_state 49 current_state = self.state_parser.get_agent_state(current_frame) 50 self.current_state = current_state 51 52 def report(self) -> dict: 53 return { 54 "agent_state": self.current_state, 55 } 56 57 def report_final(self) -> dict: 58 return {}
Harvest Moon-specific metrics.
Reports:
agent_state: The AgentState info. Is either FREE_ROAM or IN_DIALOGUE.
Final Reports:
- None
The StateParser which implements the minimum required functionality for this MetricGroup to work.
34 def reset(self, first=False): 35 if not first: 36 pass 37 self.current_state: AgentState = ( 38 AgentState.IN_DIALOGUE 39 ) # Start by default in dialogue because it has the least permissable actions. 40 """ The current state of the agent in the game. """ 41 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.
Called when environment closes. Good for computing summary stats.
Step will not be called after this.
47 def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]): 48 self._previous_state = self.current_state 49 current_state = self.state_parser.get_agent_state(current_frame) 50 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.
Return metrics as dictionary for instantaneous variable tracking.
Returns
Dictionary of metrics
61class HarvestMoonOCRMetric(OCRegionMetric): 62 REQUIRED_PARSER = HarvestMoonStateParser 63 64 def reset(self, first=False): 65 super().reset(first) 66 self.prev_was_in_menu = False 67 68 def start(self): 69 self.kinds = { 70 "dialogue": "dialogue_bottom_right", 71 "menu": "menu_top_right", 72 } 73 super().start() 74 75 def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool: 76 self.state_parser: PokemonStateParser 77 if kind == "dialogue": 78 in_dialogue = self.state_parser.dialogue_box_open( 79 current_screen=current_frame 80 ) 81 # dialogue_empty = self.state_parser.dialogue_box_empty( 82 # current_screen=current_frame 83 # ) 84 return ( 85 in_dialogue 86 # and not dialogue_empty 87 ) 88 if kind == "menu": 89 in_menu = self.state_parser.is_in_menu(current_screen=current_frame) 90 return in_menu 91 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.
The StateParser which implements the minimum required functionality for this MetricGroup to work.
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.
68 def start(self): 69 self.kinds = { 70 "dialogue": "dialogue_bottom_right", 71 "menu": "menu_top_right", 72 } 73 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.
75 def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool: 76 self.state_parser: PokemonStateParser 77 if kind == "dialogue": 78 in_dialogue = self.state_parser.dialogue_box_open( 79 current_screen=current_frame 80 ) 81 # dialogue_empty = self.state_parser.dialogue_box_empty( 82 # current_screen=current_frame 83 # ) 84 return ( 85 in_dialogue 86 # and not dialogue_empty 87 ) 88 if kind == "menu": 89 in_menu = self.state_parser.is_in_menu(current_screen=current_frame) 90 return in_menu 91 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.
94class HarvestMoon1OCRMetric(HarvestMoonOCRMetric): 95 REQUIRED_PARSER = HarvestMoon1Parser 96 97 def start(self): 98 self.kinds = { 99 "dialogue": "dialogue_bottom_right", 100 "menu": "menu_top_right", 101 } 102 OCRegionMetric.start(self)
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.
The StateParser which implements the minimum required functionality for this MetricGroup to work.
97 def start(self): 98 self.kinds = { 99 "dialogue": "dialogue_bottom_right", 100 "menu": "menu_top_right", 101 } 102 OCRegionMetric.start(self)
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.
105class HarvestMoon2OCRMetric(HarvestMoonOCRMetric): 106 REQUIRED_PARSER = HarvestMoon2Parser 107 108 def start(self): 109 self.kinds = { 110 "dialogue": "dialogue_bottom_right", 111 "menu": "menu_top_right", 112 } 113 OCRegionMetric.start(self)
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.
The StateParser which implements the minimum required functionality for this MetricGroup to work.
108 def start(self): 109 self.kinds = { 110 "dialogue": "dialogue_bottom_right", 111 "menu": "menu_top_right", 112 } 113 OCRegionMetric.start(self)
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.
116class HarvestMoon3OCRMetric(HarvestMoonOCRMetric): 117 REQUIRED_PARSER = HarvestMoon3Parser 118 119 def start(self): 120 self.kinds = { 121 "dialogue": "dialogue_bottom_right", 122 "menu": "menu_top_right", 123 } 124 OCRegionMetric.start(self)
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.
The StateParser which implements the minimum required functionality for this MetricGroup to work.
119 def start(self): 120 self.kinds = { 121 "dialogue": "dialogue_bottom_right", 122 "menu": "menu_top_right", 123 } 124 OCRegionMetric.start(self)
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.
127class HarvestMoonTestMetric(MetricGroup): 128 """ 129 Harvest Moon metrics for test environments. 130 131 Reports: 132 - `agent_state`: The current AgentState (FREE_ROAM or IN_DIALOGUE). 133 - `previous_agent_state`: The AgentState from the previous step. 134 135 Final Reports: 136 - None 137 """ 138 139 NAME = "harvest_moon_test" 140 REQUIRED_PARSER = HarvestMoonStateParser 141 142 def start(self): 143 super().start() 144 145 def reset(self, first=False): 146 self.agent_state: AgentState = AgentState.IN_DIALOGUE 147 self.previous_agent_state: AgentState = self.agent_state 148 149 def close(self): 150 self.reset() 151 return 152 153 def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]): 154 self.state_parser: HarvestMoonStateParser 155 self.previous_agent_state = self.agent_state 156 self.agent_state = self.state_parser.get_agent_state(current_frame) 157 158 def report(self) -> dict: 159 return { 160 "agent_state": self.agent_state, 161 "previous_agent_state": self.previous_agent_state, 162 } 163 164 def report_final(self) -> dict: 165 return {}
Harvest Moon metrics for test environments.
Reports:
agent_state: The current AgentState (FREE_ROAM or IN_DIALOGUE).previous_agent_state: The AgentState from the previous step.
Final Reports:
- None
The StateParser which implements the minimum required functionality for this MetricGroup to work.
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.
145 def reset(self, first=False): 146 self.agent_state: AgentState = AgentState.IN_DIALOGUE 147 self.previous_agent_state: AgentState = self.agent_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.
Called when environment closes. Good for computing summary stats.
Step will not be called after this.
153 def step(self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]): 154 self.state_parser: HarvestMoonStateParser 155 self.previous_agent_state = self.agent_state 156 self.agent_state = self.state_parser.get_agent_state(current_frame)
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.
158 def report(self) -> dict: 159 return { 160 "agent_state": self.agent_state, 161 "previous_agent_state": self.previous_agent_state, 162 }
Return metrics as dictionary for instantaneous variable tracking.
Returns
Dictionary of metrics