class_docstr: 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.
class_docstr_tok: ["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","."]
docstr: 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.
docstr_tok: ["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","."]
code: def reduce(self) -> LogStats:
      """
      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.
      """
      if self.level == LogStatsLevel.EPOCH:
          assert (
              self.last_stats_step != GlobalLogState.global_step
          ), "reduce() called twice on same global log step"
  
      all_values = defaultdict(list)
  
      # first stage: collect all values and handle group projection
      for (
          event,
          input_name,
          group_tuple,
      ), values in self.input.items():
          # retrieve the aggregation info for this level (created by the decorators)
          input_to_reducers = getattr(event, self.level.name, None)
  
          if (
              not input_to_reducers
              or input_name not in input_to_reducers
          ):
              continue
  
          all_grouping_attributes = getattr(event, "group_by", None)
  
          # iterate the aggregations
          reducers = input_to_reducers[input_name]
          for (
              reduce_function,
              output_name,
              grouping_attribute,
              cumulative,
          ) in reducers:
              projected_group = self._project_group(
                  group_tuple,
                  grouping_attribute,
                  all_grouping_attributes,
                  event,
              )
  
              all_values[
                  (
                      event,
                      output_name,
                      projected_group,
                      cumulative,
                      reduce_function,
                  )
              ].extend(values)
  
      # second stage: execute reducers
      aggregated_stats = dict()
  
      for (
          event,
          output_name,
          group_tuple,
          cumulative,
          reduce_function,
      ), values in all_values.items():
          try:
              stats_key = (event, output_name, group_tuple)
              reduced_value = self._reduce(
                  reduce_function, values, event
              )
  
              # handle cumulative statistics: add value from previous run
              if cumulative:
                  if stats_key not in self.cumulative_stats:
                      self.cumulative_stats[stats_key] = reduced_value
                  else:
                      self.cumulative_stats[stats_key] += reduced_value
                  continue
  
              aggregated_stats[stats_key] = reduced_value
          except AssertionError:
              # keep assertions as is
              raise
          except Exception as e:
              # wrap the exception to make it easier to trace the event that caused the exception
              raise ValueError(f"failed to reduce event {event}") from e
  
      aggregated_stats.update(self.cumulative_stats)
  
      # send the statistics to the consumers
      for consumer in self.consumers:
          consumer.receive(aggregated_stats)
  
      # reset the collected statistics
      self.input.clear()
      self.last_stats = aggregated_stats
  
      # keep track of the global step to detect unintended reduce steps
      self.last_stats_step = GlobalLogState.global_step
  
      return aggregated_stats
  
imports: from abc import ABC, abstractmethod
  from collections import defaultdict
  from enum import Enum
  from typing import Union, Callable, Dict, Optional, NamedTuple, List, Type, TypeVar, Tuple, Any
  import numpy
  from maze.core.events.event_record import EventRecord
  from maze.core.events.event_topic_factory import event_topic_factory
label_desc: Use image node_id 9 for calling the LogStatsAggregator obj's underlying member method code with example usage: obj.reduce() and returns: aggregated_stats