gameboy_worlds.emulation.survival_kids.base_metrics

Base metric groups for Survival Kids.

View Source

  1"""Base metric groups for Survival Kids."""
  2
  3from typing import Any, Dict, Optional, Set
  4
  5import numpy as np
  6
  7from gameboy_worlds.emulation.tracker import MetricGroup, OCRegionMetric
  8from gameboy_worlds.emulation.survival_kids.parsers import (
  9    AgentState,
 10    SurvivalKidsParser,
 11)
 12
 13
 14class CoreSurvivalKidsMetrics(MetricGroup):
 15    NAME = "survival_kids_core"
 16    REQUIRED_PARSER = SurvivalKidsParser
 17
 18    def start(self):
 19        self.total_steps_all: int = 0
 20        self.episode_count: int = 0
 21        super().start()
 22
 23    def reset(self, first: bool = False):
 24        if not first:
 25            self.total_steps_all += self.step_count
 26            self.episode_count += 1
 27        self.step_count: int = 0
 28        self.agent_state: AgentState = AgentState.FREE_ROAM
 29        self.in_dialogue: bool = False
 30        self.in_menu: bool = False
 31
 32    def close(self):
 33        self.final_metrics: Dict[str, Any] = {
 34            "total_steps_all_episodes": self.total_steps_all + self.step_count,
 35            "total_episodes": self.episode_count,
 36        }
 37
 38    def step(
 39        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
 40    ):  # noqa: ARG002
 41        self.step_count += 1
 42        self.agent_state = self.state_parser.get_agent_state(current_frame)
 43        self.in_dialogue = self.agent_state == AgentState.IN_DIALOGUE
 44        self.in_menu = self.agent_state == AgentState.IN_MENU
 45
 46    def report(self) -> Dict[str, Any]:
 47        return {
 48            "agent_state": self.agent_state,
 49            "in_dialogue": self.in_dialogue,
 50            "in_menu": self.in_menu,
 51            "step_count": self.step_count,
 52        }
 53
 54    def report_final(self) -> Dict[str, Any]:
 55        return {
 56            "total_steps_all_episodes": self.total_steps_all + self.step_count,
 57            "total_episodes": self.episode_count,
 58        }
 59
 60
 61class SurvivalKidsExploreMetrics(MetricGroup):
 62    NAME = "survival_kids_explore"
 63    REQUIRED_PARSER = SurvivalKidsParser
 64
 65    _HASH_W = 20
 66    _HASH_H = 18
 67
 68    def start(self):
 69        self._all_seen_hashes: Set[int] = set()
 70        self.total_episodes: int = 0
 71        super().start()
 72
 73    def reset(self, first: bool = False):
 74        if not first:
 75            self.total_episodes += 1
 76        self._episode_hashes: Set[int] = set()
 77        self.steps_exploring: int = 0
 78        self.step_count: int = 0
 79
 80    def close(self):
 81        self.final_metrics: Dict[str, Any] = {
 82            "unique_frames_total": len(self._all_seen_hashes),
 83            "total_episodes": self.total_episodes,
 84        }
 85
 86    @staticmethod
 87    def _hash_frame(frame: np.ndarray, w: int, h: int) -> int:
 88        small = frame[::144 // h, ::160 // w, 0]
 89        return hash(small.tobytes())
 90
 91    def step(
 92        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
 93    ):  # noqa: ARG002
 94        self.step_count += 1
 95        h = self._hash_frame(current_frame, self._HASH_W, self._HASH_H)
 96        if h not in self._episode_hashes:
 97            self._episode_hashes.add(h)
 98            self._all_seen_hashes.add(h)
 99            self.steps_exploring += 1
100
101    def report(self) -> Dict[str, Any]:
102        unique = len(self._episode_hashes)
103        ratio = unique / self.step_count if self.step_count > 0 else 0.0
104        return {
105            "unique_frames_episode": unique,
106            "total_steps_exploring": self.steps_exploring,
107            "exploration_ratio": round(ratio, 4),
108        }
109
110    def report_final(self) -> Dict[str, Any]:
111        return {
112            "unique_frames_total": len(self._all_seen_hashes),
113            "total_episodes": self.total_episodes,
114        }
115
116
117class SurvivalKidsVitalMetrics(MetricGroup):
118    NAME = "survival_kids_vitals"
119    REQUIRED_PARSER = SurvivalKidsParser
120
121    _ADDR_HP: Optional[int] = None
122    _ADDR_HUNGER: Optional[int] = None
123    _ADDR_THIRST: Optional[int] = None
124    _ADDR_STAMINA: Optional[int] = None
125    _CRITICAL_THRESHOLD: int = 200
126
127    def _read(self, address: Optional[int]) -> Optional[int]:
128        if address is None:
129            return None
130        return self.state_parser.read_memory_byte(address)
131
132    def start(self):
133        super().start()
134
135    def reset(self, first: bool = False):
136        self.hp: Optional[int] = None
137        self.hunger: Optional[int] = None
138        self.thirst: Optional[int] = None
139        self.stamina: Optional[int] = None
140        self._min_hp: Optional[int] = None
141        self._times_starving: int = 0
142        self._times_dehydrated: int = 0
143
144    def close(self):
145        self.final_metrics: Dict[str, Any] = {
146            "min_hp_seen": self._min_hp,
147            "times_starving": self._times_starving,
148            "times_dehydrated": self._times_dehydrated,
149        }
150
151    def step(
152        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
153    ):  # noqa: ARG002
154        self.hp = self._read(self._ADDR_HP)
155        self.hunger = self._read(self._ADDR_HUNGER)
156        self.thirst = self._read(self._ADDR_THIRST)
157        self.stamina = self._read(self._ADDR_STAMINA)
158        if self.hp is not None:
159            if self._min_hp is None or self.hp < self._min_hp:
160                self._min_hp = self.hp
161        if self.hunger is not None and self.hunger >= self._CRITICAL_THRESHOLD:
162            self._times_starving += 1
163        if self.thirst is not None and self.thirst >= self._CRITICAL_THRESHOLD:
164            self._times_dehydrated += 1
165
166    def report(self) -> Dict[str, Any]:
167        return {
168            "hp": self.hp,
169            "hunger": self.hunger,
170            "thirst": self.thirst,
171            "stamina": self.stamina,
172        }
173
174    def report_final(self) -> Dict[str, Any]:
175        return {
176            "min_hp_seen": self._min_hp,
177            "times_starving": self._times_starving,
178            "times_dehydrated": self._times_dehydrated,
179        }
180
181
182class SurvivalKidsHudMetrics(MetricGroup):
183    NAME = "survival_kids_hud"
184    REQUIRED_PARSER = SurvivalKidsParser
185
186    _REGIONS = [
187        "status_bar",
188        "hp_area",
189        "hunger_area",
190        "thirst_area",
191        "stamina_area",
192        "equipped_items_area",
193        "equipped_item_area",
194        "pack_icon_area",
195    ]
196    _CHANGE_MAE_THRESHOLD = 10
197
198    def reset(self, first: bool = False):  # noqa: ARG002
199        self._baselines: Dict[str, Optional[np.ndarray]] = {
200            region: None for region in self._REGIONS
201        }
202        self._mae: Dict[str, float] = {region: 0.0 for region in self._REGIONS}
203        self._changed: Dict[str, bool] = {region: False for region in self._REGIONS}
204
205    def close(self):
206        self.final_metrics = {
207            f"{region}_changed": self._changed[region] for region in self._REGIONS
208        }
209
210    def step(
211        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
212    ):  # noqa: ARG002
213        for region in self._REGIONS:
214            cropped = self.state_parser.capture_named_region(current_frame, region)
215            if self._baselines[region] is None:
216                self._baselines[region] = cropped.copy()
217                self._mae[region] = 0.0
218                self._changed[region] = False
219                continue
220            mae = np.abs(
221                cropped.astype(float) - self._baselines[region].astype(float)
222            ).mean()
223            self._mae[region] = float(mae)
224            self._changed[region] = mae > self._CHANGE_MAE_THRESHOLD
225
226    def report(self) -> Dict[str, Any]:
227        report: Dict[str, Any] = {}
228        for region in self._REGIONS:
229            report[f"{region}_mae"] = round(self._mae[region], 4)
230            report[f"{region}_changed"] = self._changed[region]
231        return report
232
233    def report_final(self) -> Dict[str, Any]:
234        return self.final_metrics
235
236
237class SurvivalKidsOCRMetric(OCRegionMetric):
238    """Expose OCR regions only when Survival Kids shows readable UI text."""
239
240    REQUIRED_PARSER = SurvivalKidsParser
241
242    def start(self):
243        self.kinds = {
244            "dialogue": "dialogue_area",
245            "dialogue_bottom": "screen_bottom",
246            "menu": "menu_area",
247        }
248        super().start()
249
250    def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool:
251        if kind in {"dialogue", "dialogue_bottom"}:
252            return self.state_parser.is_in_dialogue(current_frame)
253        if kind == "menu":
254            return self.state_parser.is_in_menu(current_frame)
255        return False

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

15class CoreSurvivalKidsMetrics(MetricGroup):
16    NAME = "survival_kids_core"
17    REQUIRED_PARSER = SurvivalKidsParser
18
19    def start(self):
20        self.total_steps_all: int = 0
21        self.episode_count: int = 0
22        super().start()
23
24    def reset(self, first: bool = False):
25        if not first:
26            self.total_steps_all += self.step_count
27            self.episode_count += 1
28        self.step_count: int = 0
29        self.agent_state: AgentState = AgentState.FREE_ROAM
30        self.in_dialogue: bool = False
31        self.in_menu: bool = False
32
33    def close(self):
34        self.final_metrics: Dict[str, Any] = {
35            "total_steps_all_episodes": self.total_steps_all + self.step_count,
36            "total_episodes": self.episode_count,
37        }
38
39    def step(
40        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
41    ):  # noqa: ARG002
42        self.step_count += 1
43        self.agent_state = self.state_parser.get_agent_state(current_frame)
44        self.in_dialogue = self.agent_state == AgentState.IN_DIALOGUE
45        self.in_menu = self.agent_state == AgentState.IN_MENU
46
47    def report(self) -> Dict[str, Any]:
48        return {
49            "agent_state": self.agent_state,
50            "in_dialogue": self.in_dialogue,
51            "in_menu": self.in_menu,
52            "step_count": self.step_count,
53        }
54
55    def report_final(self) -> Dict[str, Any]:
56        return {
57            "total_steps_all_episodes": self.total_steps_all + self.step_count,
58            "total_episodes": self.episode_count,
59        }

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 = 'survival_kids_core'

Name of the MetricGroup.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.survival_kids.parsers.SurvivalKidsParser'>

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

def start(self): View Source

19    def start(self):
20        self.total_steps_all: int = 0
21        self.episode_count: int = 0
22        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: bool = False): View Source

24    def reset(self, first: bool = False):
25        if not first:
26            self.total_steps_all += self.step_count
27            self.episode_count += 1
28        self.step_count: int = 0
29        self.agent_state: AgentState = AgentState.FREE_ROAM
30        self.in_dialogue: bool = False
31        self.in_menu: bool = 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

33    def close(self):
34        self.final_metrics: Dict[str, Any] = {
35            "total_steps_all_episodes": self.total_steps_all + self.step_count,
36            "total_episodes": self.episode_count,
37        }

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

39    def step(
40        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
41    ):  # noqa: ARG002
42        self.step_count += 1
43        self.agent_state = self.state_parser.get_agent_state(current_frame)
44        self.in_dialogue = self.agent_state == AgentState.IN_DIALOGUE
45        self.in_menu = self.agent_state == AgentState.IN_MENU

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[str, Any]: View Source

47    def report(self) -> Dict[str, Any]:
48        return {
49            "agent_state": self.agent_state,
50            "in_dialogue": self.in_dialogue,
51            "in_menu": self.in_menu,
52            "step_count": self.step_count,
53        }

Return metrics as dictionary for instantaneous variable tracking.

Returns

Dictionary of metrics

def report_final(self) -> Dict[str, Any]: View Source

55    def report_final(self) -> Dict[str, Any]:
56        return {
57            "total_steps_all_episodes": self.total_steps_all + self.step_count,
58            "total_episodes": self.episode_count,
59        }

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

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

 62class SurvivalKidsExploreMetrics(MetricGroup):
 63    NAME = "survival_kids_explore"
 64    REQUIRED_PARSER = SurvivalKidsParser
 65
 66    _HASH_W = 20
 67    _HASH_H = 18
 68
 69    def start(self):
 70        self._all_seen_hashes: Set[int] = set()
 71        self.total_episodes: int = 0
 72        super().start()
 73
 74    def reset(self, first: bool = False):
 75        if not first:
 76            self.total_episodes += 1
 77        self._episode_hashes: Set[int] = set()
 78        self.steps_exploring: int = 0
 79        self.step_count: int = 0
 80
 81    def close(self):
 82        self.final_metrics: Dict[str, Any] = {
 83            "unique_frames_total": len(self._all_seen_hashes),
 84            "total_episodes": self.total_episodes,
 85        }
 86
 87    @staticmethod
 88    def _hash_frame(frame: np.ndarray, w: int, h: int) -> int:
 89        small = frame[::144 // h, ::160 // w, 0]
 90        return hash(small.tobytes())
 91
 92    def step(
 93        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
 94    ):  # noqa: ARG002
 95        self.step_count += 1
 96        h = self._hash_frame(current_frame, self._HASH_W, self._HASH_H)
 97        if h not in self._episode_hashes:
 98            self._episode_hashes.add(h)
 99            self._all_seen_hashes.add(h)
100            self.steps_exploring += 1
101
102    def report(self) -> Dict[str, Any]:
103        unique = len(self._episode_hashes)
104        ratio = unique / self.step_count if self.step_count > 0 else 0.0
105        return {
106            "unique_frames_episode": unique,
107            "total_steps_exploring": self.steps_exploring,
108            "exploration_ratio": round(ratio, 4),
109        }
110
111    def report_final(self) -> Dict[str, Any]:
112        return {
113            "unique_frames_total": len(self._all_seen_hashes),
114            "total_episodes": self.total_episodes,
115        }

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 = 'survival_kids_explore'

Name of the MetricGroup.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.survival_kids.parsers.SurvivalKidsParser'>

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

def start(self): View Source

69    def start(self):
70        self._all_seen_hashes: Set[int] = set()
71        self.total_episodes: int = 0
72        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: bool = False): View Source

74    def reset(self, first: bool = False):
75        if not first:
76            self.total_episodes += 1
77        self._episode_hashes: Set[int] = set()
78        self.steps_exploring: int = 0
79        self.step_count: int = 0

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

81    def close(self):
82        self.final_metrics: Dict[str, Any] = {
83            "unique_frames_total": len(self._all_seen_hashes),
84            "total_episodes": self.total_episodes,
85        }

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

 92    def step(
 93        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
 94    ):  # noqa: ARG002
 95        self.step_count += 1
 96        h = self._hash_frame(current_frame, self._HASH_W, self._HASH_H)
 97        if h not in self._episode_hashes:
 98            self._episode_hashes.add(h)
 99            self._all_seen_hashes.add(h)
100            self.steps_exploring += 1

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[str, Any]: View Source

102    def report(self) -> Dict[str, Any]:
103        unique = len(self._episode_hashes)
104        ratio = unique / self.step_count if self.step_count > 0 else 0.0
105        return {
106            "unique_frames_episode": unique,
107            "total_steps_exploring": self.steps_exploring,
108            "exploration_ratio": round(ratio, 4),
109        }

Return metrics as dictionary for instantaneous variable tracking.

Returns

Dictionary of metrics

def report_final(self) -> Dict[str, Any]: View Source

111    def report_final(self) -> Dict[str, Any]:
112        return {
113            "unique_frames_total": len(self._all_seen_hashes),
114            "total_episodes": self.total_episodes,
115        }

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

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

118class SurvivalKidsVitalMetrics(MetricGroup):
119    NAME = "survival_kids_vitals"
120    REQUIRED_PARSER = SurvivalKidsParser
121
122    _ADDR_HP: Optional[int] = None
123    _ADDR_HUNGER: Optional[int] = None
124    _ADDR_THIRST: Optional[int] = None
125    _ADDR_STAMINA: Optional[int] = None
126    _CRITICAL_THRESHOLD: int = 200
127
128    def _read(self, address: Optional[int]) -> Optional[int]:
129        if address is None:
130            return None
131        return self.state_parser.read_memory_byte(address)
132
133    def start(self):
134        super().start()
135
136    def reset(self, first: bool = False):
137        self.hp: Optional[int] = None
138        self.hunger: Optional[int] = None
139        self.thirst: Optional[int] = None
140        self.stamina: Optional[int] = None
141        self._min_hp: Optional[int] = None
142        self._times_starving: int = 0
143        self._times_dehydrated: int = 0
144
145    def close(self):
146        self.final_metrics: Dict[str, Any] = {
147            "min_hp_seen": self._min_hp,
148            "times_starving": self._times_starving,
149            "times_dehydrated": self._times_dehydrated,
150        }
151
152    def step(
153        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
154    ):  # noqa: ARG002
155        self.hp = self._read(self._ADDR_HP)
156        self.hunger = self._read(self._ADDR_HUNGER)
157        self.thirst = self._read(self._ADDR_THIRST)
158        self.stamina = self._read(self._ADDR_STAMINA)
159        if self.hp is not None:
160            if self._min_hp is None or self.hp < self._min_hp:
161                self._min_hp = self.hp
162        if self.hunger is not None and self.hunger >= self._CRITICAL_THRESHOLD:
163            self._times_starving += 1
164        if self.thirst is not None and self.thirst >= self._CRITICAL_THRESHOLD:
165            self._times_dehydrated += 1
166
167    def report(self) -> Dict[str, Any]:
168        return {
169            "hp": self.hp,
170            "hunger": self.hunger,
171            "thirst": self.thirst,
172            "stamina": self.stamina,
173        }
174
175    def report_final(self) -> Dict[str, Any]:
176        return {
177            "min_hp_seen": self._min_hp,
178            "times_starving": self._times_starving,
179            "times_dehydrated": self._times_dehydrated,
180        }

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 = 'survival_kids_vitals'

Name of the MetricGroup.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.survival_kids.parsers.SurvivalKidsParser'>

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

def start(self): View Source

133    def start(self):
134        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: bool = False): View Source

136    def reset(self, first: bool = False):
137        self.hp: Optional[int] = None
138        self.hunger: Optional[int] = None
139        self.thirst: Optional[int] = None
140        self.stamina: Optional[int] = None
141        self._min_hp: Optional[int] = None
142        self._times_starving: int = 0
143        self._times_dehydrated: int = 0

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

145    def close(self):
146        self.final_metrics: Dict[str, Any] = {
147            "min_hp_seen": self._min_hp,
148            "times_starving": self._times_starving,
149            "times_dehydrated": self._times_dehydrated,
150        }

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

152    def step(
153        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
154    ):  # noqa: ARG002
155        self.hp = self._read(self._ADDR_HP)
156        self.hunger = self._read(self._ADDR_HUNGER)
157        self.thirst = self._read(self._ADDR_THIRST)
158        self.stamina = self._read(self._ADDR_STAMINA)
159        if self.hp is not None:
160            if self._min_hp is None or self.hp < self._min_hp:
161                self._min_hp = self.hp
162        if self.hunger is not None and self.hunger >= self._CRITICAL_THRESHOLD:
163            self._times_starving += 1
164        if self.thirst is not None and self.thirst >= self._CRITICAL_THRESHOLD:
165            self._times_dehydrated += 1

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[str, Any]: View Source

167    def report(self) -> Dict[str, Any]:
168        return {
169            "hp": self.hp,
170            "hunger": self.hunger,
171            "thirst": self.thirst,
172            "stamina": self.stamina,
173        }

Return metrics as dictionary for instantaneous variable tracking.

Returns

Dictionary of metrics

def report_final(self) -> Dict[str, Any]: View Source

175    def report_final(self) -> Dict[str, Any]:
176        return {
177            "min_hp_seen": self._min_hp,
178            "times_starving": self._times_starving,
179            "times_dehydrated": self._times_dehydrated,
180        }

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

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

183class SurvivalKidsHudMetrics(MetricGroup):
184    NAME = "survival_kids_hud"
185    REQUIRED_PARSER = SurvivalKidsParser
186
187    _REGIONS = [
188        "status_bar",
189        "hp_area",
190        "hunger_area",
191        "thirst_area",
192        "stamina_area",
193        "equipped_items_area",
194        "equipped_item_area",
195        "pack_icon_area",
196    ]
197    _CHANGE_MAE_THRESHOLD = 10
198
199    def reset(self, first: bool = False):  # noqa: ARG002
200        self._baselines: Dict[str, Optional[np.ndarray]] = {
201            region: None for region in self._REGIONS
202        }
203        self._mae: Dict[str, float] = {region: 0.0 for region in self._REGIONS}
204        self._changed: Dict[str, bool] = {region: False for region in self._REGIONS}
205
206    def close(self):
207        self.final_metrics = {
208            f"{region}_changed": self._changed[region] for region in self._REGIONS
209        }
210
211    def step(
212        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
213    ):  # noqa: ARG002
214        for region in self._REGIONS:
215            cropped = self.state_parser.capture_named_region(current_frame, region)
216            if self._baselines[region] is None:
217                self._baselines[region] = cropped.copy()
218                self._mae[region] = 0.0
219                self._changed[region] = False
220                continue
221            mae = np.abs(
222                cropped.astype(float) - self._baselines[region].astype(float)
223            ).mean()
224            self._mae[region] = float(mae)
225            self._changed[region] = mae > self._CHANGE_MAE_THRESHOLD
226
227    def report(self) -> Dict[str, Any]:
228        report: Dict[str, Any] = {}
229        for region in self._REGIONS:
230            report[f"{region}_mae"] = round(self._mae[region], 4)
231            report[f"{region}_changed"] = self._changed[region]
232        return report
233
234    def report_final(self) -> Dict[str, Any]:
235        return self.final_metrics

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 = 'survival_kids_hud'

Name of the MetricGroup.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.survival_kids.parsers.SurvivalKidsParser'>

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

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

199    def reset(self, first: bool = False):  # noqa: ARG002
200        self._baselines: Dict[str, Optional[np.ndarray]] = {
201            region: None for region in self._REGIONS
202        }
203        self._mae: Dict[str, float] = {region: 0.0 for region in self._REGIONS}
204        self._changed: Dict[str, bool] = {region: False for region in self._REGIONS}

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

206    def close(self):
207        self.final_metrics = {
208            f"{region}_changed": self._changed[region] for region in self._REGIONS
209        }

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

211    def step(
212        self, current_frame: np.ndarray, recent_frames: Optional[np.ndarray]
213    ):  # noqa: ARG002
214        for region in self._REGIONS:
215            cropped = self.state_parser.capture_named_region(current_frame, region)
216            if self._baselines[region] is None:
217                self._baselines[region] = cropped.copy()
218                self._mae[region] = 0.0
219                self._changed[region] = False
220                continue
221            mae = np.abs(
222                cropped.astype(float) - self._baselines[region].astype(float)
223            ).mean()
224            self._mae[region] = float(mae)
225            self._changed[region] = mae > self._CHANGE_MAE_THRESHOLD

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[str, Any]: View Source

227    def report(self) -> Dict[str, Any]:
228        report: Dict[str, Any] = {}
229        for region in self._REGIONS:
230            report[f"{region}_mae"] = round(self._mae[region], 4)
231            report[f"{region}_changed"] = self._changed[region]
232        return report

Return metrics as dictionary for instantaneous variable tracking.

Returns

Dictionary of metrics

def report_final(self) -> Dict[str, Any]: View Source

234    def report_final(self) -> Dict[str, Any]:
235        return self.final_metrics

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; start; log_info; log_warn; log_report

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

238class SurvivalKidsOCRMetric(OCRegionMetric):
239    """Expose OCR regions only when Survival Kids shows readable UI text."""
240
241    REQUIRED_PARSER = SurvivalKidsParser
242
243    def start(self):
244        self.kinds = {
245            "dialogue": "dialogue_area",
246            "dialogue_bottom": "screen_bottom",
247            "menu": "menu_area",
248        }
249        super().start()
250
251    def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool:
252        if kind in {"dialogue", "dialogue_bottom"}:
253            return self.state_parser.is_in_dialogue(current_frame)
254        if kind == "menu":
255            return self.state_parser.is_in_menu(current_frame)
256        return False

Expose OCR regions only when Survival Kids shows readable UI text.

REQUIRED_PARSER = <class 'gameboy_worlds.emulation.survival_kids.parsers.SurvivalKidsParser'>

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

def start(self): View Source

243    def start(self):
244        self.kinds = {
245            "dialogue": "dialogue_area",
246            "dialogue_bottom": "screen_bottom",
247            "menu": "menu_area",
248        }
249        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

251    def can_read_kind(self, current_frame: np.ndarray, kind: str) -> bool:
252        if kind in {"dialogue", "dialogue_bottom"}:
253            return self.state_parser.is_in_dialogue(current_frame)
254        if kind == "menu":
255            return self.state_parser.is_in_menu(current_frame)
256        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; reset; step; report; report_final; close