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 {}
class CoreHarvestMoonMetrics(gameboy_worlds.emulation.tracker.MetricGroup):
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
NAME = 'harvest_moon_core'

Name of the MetricGroup.

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

def reset(self, first=False):
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.
def close(self):
43    def close(self):
44        self.reset()
45        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]):
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.
def report(self) -> dict:
52    def report(self) -> dict:
53        return {
54            "agent_state": self.current_state,
55        }

Return metrics as dictionary for instantaneous variable tracking.

Returns

Dictionary of metrics

def report_final(self) -> dict:
57    def report_final(self) -> dict:
58        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

class HarvestMoonOCRMetric(gameboy_worlds.emulation.tracker.OCRegionMetric):
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.

def reset(self, first=False):
64    def reset(self, first=False):
65        super().reset(first)
66        self.prev_was_in_menu = 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):
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.

def can_read_kind(self, current_frame: numpy.ndarray, kind: str) -> bool:
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.
class HarvestMoon1OCRMetric(HarvestMoonOCRMetric):
 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.

def start(self):
 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.

class HarvestMoon2OCRMetric(HarvestMoonOCRMetric):
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.

def start(self):
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.

class HarvestMoon3OCRMetric(HarvestMoonOCRMetric):
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.

def start(self):
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.

class HarvestMoonTestMetric(gameboy_worlds.emulation.tracker.MetricGroup):
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
NAME = 'harvest_moon_test'

Name of the MetricGroup.

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

def start(self):
142    def start(self):
143        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):
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.
def close(self):
149    def close(self):
150        self.reset()
151        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]):
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.
def report(self) -> dict:
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

def report_final(self) -> dict:
164    def report_final(self) -> dict:
165        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