class_docstr: Trainer class for OpenAI Evolution Strategies.
  
  :param algorithm_config: Algorithm parameters.
  :param torch_policy: Multi-step policy encapsulating the policy networks
  :param shared_noise: The noise table, with the same content for every worker and the master.
  :param normalization_stats: Normalization statistics as calculated by the NormalizeObservationWrapper.
class_docstr_tok: ["Trainer","class","for","OpenAI","Evolution","Strategies",".",":","param","algorithm_config",":","Algorithm","parameters",".",":","param","torch_policy",":","Multi-step","policy","encapsulating","the","policy","networks",":","param","shared_noise",":","The","noise","table",",","with","the","same","content","for","every","worker","and","the","master",".",":","param","normalization_stats",":","Normalization","statistics","as","calculated","by","the","NormalizeObservationWrapper","."]
code: def _update(self, distributed_rollouts: ESDistributedRollouts):
      # Pop off results for the current task
      n_train_episodes, n_timesteps_popped = 0, 0
  
      # aggregate all collected training rollouts for this episode
      epoch_results = ESRolloutResult(is_eval=False)
  
      # obtain a generator from the distribution interface
      rollouts_generator = distributed_rollouts.generate_rollouts(
          policy=self.policy,
          max_steps=self.algorithm_config.max_steps,
          noise_stddev=self.algorithm_config.noise_stddev,
          normalization_stats=self.normalization_stats,
      )
  
      # collect eval and training rollouts
      for result in rollouts_generator:
          if result.is_eval:
              # This was an eval job
              for e in result.episode_stats:
                  self.eval_stats.receive(e)
              continue
  
          # we received training experience from perturbed policy networks
          epoch_results.noise_indices.extend(result.noise_indices)
          epoch_results.episode_stats.extend(result.episode_stats)
  
          # update the training statistics
          for e in result.episode_stats:
              self.train_stats.receive(e)
  
              n_train_episodes += 1
              n_timesteps_popped += e[
                  (BaseEnvEvents.reward, "count", None)
              ]
  
          # continue until we collected enough episodes and timesteps
          if (
              n_train_episodes
              >= self.algorithm_config.n_rollouts_per_update
              and n_timesteps_popped
              >= self.algorithm_config.n_timesteps_per_update
          ):
              break
  
      # notify the model selection of the evaluation results
      if self.model_selection:
          eval_stats = self.eval_stats.reduce()
          train_stats = self.train_stats.reduce()
          if len(eval_stats):
              reward = eval_stats[(BaseEnvEvents.reward, "mean", None)]
          else:
              reward = train_stats[(BaseEnvEvents.reward, "mean", None)]
          self.model_selection.update(reward)
  
      # prepare returns, reshape the positive/negative antithetic estimation as (rollouts, 2)
      returns_n2 = np.array(
          [
              e[(BaseEnvEvents.reward, "sum", None)]
              for e in epoch_results.episode_stats
          ]
      ).reshape(-1, 2)
  
      # improve robustness: weight by rank, not by reward
      proc_returns_n2 = self._compute_centered_ranks(returns_n2)
  
      # compute the gradient
      g = self._batched_weighted_sum(
          proc_returns_n2[:, 0] - proc_returns_n2[:, 1],
          (
              self.shared_noise.get(idx, self.policy.num_params)
              for idx in epoch_results.noise_indices
          ),
          batch_size=500,
      )
  
      g /= n_train_episodes / 2.0
  
      # apply the weight update
      theta = get_flat_parameters(self.policy)
      update_ratio = self.optimizer.update(
          -g + self.algorithm_config.l2_penalty * theta.numpy()
      )
  
      # statistics logging
      self.es_events.update_ratio(update_ratio)
  
      for i in self.policy.state_dict().keys():
          self.es_events.policy_grad_norm(
              policy_id=i, value=np.square(g).sum() ** 0.5
          )
          self.es_events.policy_norm(
              policy_id=i, value=np.square(theta).sum() ** 0.5
          )
  
imports: import itertools
  import logging
  import time
  from typing import Optional, Iterable, Generator, Tuple, Union, Dict
  import numpy
  import torch
  from typing.io import BinaryIO
  from maze.core.agent.policy import Policy
  from maze.core.agent.torch_model import TorchModel
  from maze.core.agent.torch_policy import TorchPolicy
  from maze.core.annotations import override
  from maze.core.env.base_env_events import BaseEnvEvents
  from maze.core.log_stats.log_stats import LogStatsAggregator, LogStatsLevel, get_stats_logger, increment_log_step
  from maze.core.utils.factory import Factory
  from maze.train.trainers.common.model_selection.model_selection_base import ModelSelectionBase
  from maze.train.trainers.common.trainer import Trainer
  from maze.train.trainers.es.distributed.es_distributed_rollouts import ESDistributedRollouts, ESRolloutResult
  from maze.train.trainers.es.es_algorithm_config import ESAlgorithmConfig
  from maze.train.trainers.es.es_events import ESEvents
  from maze.train.trainers.es.es_shared_noise_table import SharedNoiseTable
  from maze.train.trainers.es.es_utils import get_flat_parameters
  from maze.train.trainers.es.optimizers.base_optimizer import Optimizer
label_desc: Use image node_id 6 for calling the ESTrainer obj's underlying member method code with example usage: obj._update(distributed_rollouts) without return types