class maze.core.log_stats.log_stats.LogStatsAggregator(level: maze.core.log_stats.log_stats.LogStatsLevel, *consumers: maze.core.log_stats.log_stats.LogStatsConsumer)

Complements the event system by providing aggregation functionality. How the events are aggregated is specified by the event interface decorators (see event_decorators.py).

Note that the statistics calculation for episode aggregators will automatically be triggered on every increment_log_step() call.

add_event(event_record: maze.core.events.event_record.EventRecord)None

Add a recorded event to this aggregator.

The aggregator only keeps track of event/attributes with relevant statistics decoration, everything else is filtered out immediately.



add_value(event: Callable, value: Union[int, float, numpy.ndarray, dict], name: str = None, group: Tuple[Union[str, int], ] = None)None

Add a single value to this aggregator.

  • event – The event interface method the given value belongs to

  • value – The value to add

  • name – There may be multiple statistics per event, the name is used to refer to a specific statistics built from the event records

  • group – The group identifier if this event is grouped


Clear the input statistics (and start fresh).

create_event_topic(interface_class: Type[T]) → T

Provide an event topic proxy analogous to the event proxies provided by EventSystem/PubSub. But in contrast to the event system, this can be used to inject statistics also on the step, episode and epoch level.

Note that different LogStatsLevel result in different behaviour of the returned event topic proxies!


interface_class – The class object of an abstract interface that defines the events as methods.


A proxy object, dynamically derived from the passed interface_class, that can be used to trigger events.


keep track of all cumulative stats, required e.g. for an epoch where the previous epoch did not finish any episode

last_stats: Optional[LogStats]

keep track of the previous statistics, required e.g. for cumulative statistics

last_stats_step: Optional[int]

step number of the last statistics calculation

receive(stats: Dict[maze.core.log_stats.log_stats.LogStatsKey, Union[int, float, numpy.ndarray, dict]])None

Receive statistics from the subjacent aggregation level


stats – The statistics dictionary

reduce() → Dict[maze.core.log_stats.log_stats.LogStatsKey, Union[int, float, numpy.ndarray, dict]]

Consume the aggregated values by - calculating the statistics - sending the statistics to the consumers - resetting the values for the next aggregation step

:return Returns the statistics object. The same object has been sent to the consumers.

register_consumer(consumer: maze.core.log_stats.log_stats.LogStatsConsumer)None

Register a new consumer to receive the readily calculated statistics of this aggregator.


consumer – The consumer to add

remove_consumer(consumer: maze.core.log_stats.log_stats.LogStatsConsumer)None

Remove a registered consumer of this aggregator.


consumer – The consumer to remove