LogStatsWrapper

class maze.core.wrappers.log_stats_wrapper.LogStatsWrapper(env: MazeEnv, logging_prefix: str | None = None)

A statistics logging wrapper for BaseEnv.

Parameters:

env – The environment to wrap.

clear_epoch_stats() → None

(overrides LogStatsEnv)

Implementation of the LogStatsEnv interface, clear out episode statistics collected so far in this epoch.

clone_from(env: LogStatsWrapper) → None

(overrides SimulatedEnvMixin)

implementation of SimulatedEnvMixin.

close()

(overrides BaseEnv)

Close the stats rendering figure if needed.

get_last_step_events(query: Callable | Iterable[Callable] | None = None)

Convenience accessor to all events recorded during the last step.

Parameters:

query – Specify which events to return (one or more interface methods)

Returns:

Recorded events from the last step (all if no query is present)

get_observation_and_action_dicts(maze_state: Any | None, maze_action: Any | None, first_step_in_episode: bool) → Tuple[Dict[int | str, Any] | None, Dict[int | str, Any] | None]

(overrides Wrapper)

Keep both actions and observation the same.

get_stats(level: LogStatsLevel) → LogStatsAggregator

(overrides LogStatsEnv)

Implementation of the LogStatsEnv interface, return the statistics aggregator.

get_stats_value(event: Callable, level: LogStatsLevel, name: str | None = None) → int | float | numpy.ndarray | dict

(overrides LogStatsEnv)

Implementation of the LogStatsEnv interface, obtain the value from the cached aggregator statistics.

render_stats(event_name: str = 'BaseEnvEvents.reward', metric_name: str = 'value', aggregation_func: str | Callable | None = None, group_by: str | None = None, post_processing_func: str | Callable | None = 'cumsum')

Render statistics from the currently running episode.

Rendering is based on event logs. You can select arbitrary events from those dispatched by the currently running environment.

Parameters:

  • event_name – Name of the even the even log corresponds to

  • metric_name – Metric to use (one of the event attributes, e.g. “n_items” – depends on the event type)

  • aggregation_func – Optionally, specifies how to aggregate the metric on step level, i.e. when there are multiple same events dispatched during the same step.

  • group_by – Optionally, another of event attributes to group by on the step level (e.g. “product_id”)

  • post_processing_func – Optionally, a function to post-process the data (“cumsum” is often used)

reset() → Any

(overrides BaseEnv)

Reset the environment and trigger the episode statistics calculation of the previous run.

step(action: Any) → Tuple[Any, Any, bool, Dict[Any, Any]]

(overrides BaseEnv)

Collect the rewards for the logging statistics

classmethod wrap(env: T, logging_prefix: str | None = None) → T | LogStatsEnv

Creation method providing appropriate type hints. Preferred method to construct the wrapper compared to calling the class constructor directly.

Parameters:

  • env – The environment to be wrapped.

  • logging_prefix – The episode statistics is connected to the logging system with this tagging prefix. If None, no logging happens.

:return A newly created wrapper instance.

write_epoch_stats()

(overrides LogStatsEnv)

Implementation of the LogStatsEnv interface, call reduce on the episode aggregator.