class_docstr: An environment runner to collect data from vectorized gymnasium environments.
class_docstr_tok: ["An","environment","runner","to","collect","data","from","vectorized","gymnasium","environments","."]
docstr: Initializes a DreamerV3EnvRunner instance.
  
  Args:
      config: The config to use to setup this EnvRunner.
code: def __init__(
      self,
      config: AlgorithmConfig,
      **kwargs,
  ):
      """Initializes a DreamerV3EnvRunner instance.
  
      Args:
          config: The config to use to setup this EnvRunner.
      """
      super().__init__(config=config)
  
      # Create the gym.vector.Env object.
      # Atari env.
      if self.config.env.startswith("ALE/"):
          # TODO (sven): This import currently causes a Tune test to fail. Either way,
          #  we need to figure out how to properly setup the CI environment with
          #  the correct versions of all gymnasium-related packages.
          from supersuit.generic_wrappers import resize_v1
  
          # [2]: "We down-scale the 84 × 84 grayscale images to 64 × 64 pixels so that
          # we can apply the convolutional architecture of DreamerV1."
          # ...
          # "We follow the evaluation protocol of Machado et al. (2018) with 200M
          # environment steps, action repeat of 4, a time limit of 108,000 steps per
          # episode that correspond to 30 minutes of game play, no access to life
          # information, full action space, and sticky actions. Because the world
          # model integrates information over time, DreamerV2 does not use frame
          # stacking."
          # However, in Danijar's repo, Atari100k experiments are configured as:
          # noop=30, 64x64x3 (no grayscaling), sticky actions=False,
          # full action space=False,
          wrappers = [
              partial(gym.wrappers.TimeLimit, max_episode_steps=108000),
              partial(
                  resize_v1, x_size=64, y_size=64
              ),  # resize to 64x64
              NormalizedImageEnv,
              NoopResetEnv,
              MaxAndSkipEnv,
          ]
  
          self.env = gym.vector.make(
              "GymV26Environment-v0",
              env_id=self.config.env,
              wrappers=wrappers,
              num_envs=self.config.num_envs_per_worker,
              asynchronous=self.config.remote_worker_envs,
              make_kwargs=dict(
                  self.config.env_config, **{"render_mode": "rgb_array"}
              ),
          )
      # DeepMind Control.
      elif self.config.env.startswith("DMC/"):
          parts = self.config.env.split("/")
          assert len(parts) == 3, (
              "ERROR: DMC env must be formatted as 'DMC/[task]/[domain]', e.g. "
              f"'DMC/cartpole/swingup'! You provided '{self.config.env}'."
          )
          gym.register(
              "dmc_env-v0",
              lambda from_pixels=True: DMCEnv(
                  parts[1],
                  parts[2],
                  from_pixels=from_pixels,
                  channels_first=False,
              ),
          )
          self.env = gym.vector.make(
              "dmc_env-v0",
              wrappers=[ActionClip],
              num_envs=self.config.num_envs_per_worker,
              asynchronous=self.config.remote_worker_envs,
              **dict(self.config.env_config),
          )
      # All other envs (gym or `tune.register_env()`'d by the user).
      else:
          # Register the env in this local context here.
          gym.register(
              "dreamerv3-custom-env-v0",
              partial(
                  _global_registry.get(ENV_CREATOR, self.config.env),
                  self.config.env_config,
              )
              if _global_registry.contains(ENV_CREATOR, self.config.env)
              else partial(
                  _gym_env_creator,
                  env_context=self.config.env_config,
                  env_descriptor=self.config.env,
              ),
          )
          # Create the vectorized gymnasium env.
          self.env = gym.vector.make(
              "dreamerv3-custom-env-v0",
              num_envs=self.config.num_envs_per_worker,
              asynchronous=False,  # self.config.remote_worker_envs,
          )
      self.num_envs = self.env.num_envs
      assert self.num_envs == self.config.num_envs_per_worker
  
      # Create our RLModule to compute actions with.
      if self.config.share_module_between_env_runner_and_learner:
          # DreamerV3 Algorithm will set this to the local Learner's module.
          self.module = None
      # Create our own instance of a DreamerV3RLModule (which then needs to be
      # weight-synched each iteration).
      else:
          policy_dict, _ = self.config.get_multi_agent_setup(
              env=self.env
          )
          module_spec = self.config.get_marl_module_spec(
              policy_dict=policy_dict
          )
          # TODO (sven): DreamerV3 is currently single-agent only.
          self.module = module_spec.build()[DEFAULT_POLICY_ID]
  
      self._needs_initial_reset = True
      self._episodes = [None for _ in range(self.num_envs)]
      self._states = [None for _ in range(self.num_envs)]
  
      # TODO (sven): Move metrics temp storage and collection out of EnvRunner
      #  and RolloutWorkers. These classes should not continue tracking some data
      #  that they have already returned (in a call to `sample()`). Instead, the
      #  episode data should be analyzed where it was sent to (the Algorithm itself
      #  via its replay buffer, etc..).
      self._done_episodes_for_metrics = []
      self._ongoing_episodes_for_metrics = defaultdict(list)
      self._ts_since_last_metrics = 0
  
imports: from collections import defaultdict
  from functools import partial
  from typing import List, Tuple
  import gymnasium
  import numpy
  import tree
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig
  from ray.rllib.core.models.base import STATE_IN, STATE_OUT
  from ray.rllib.env.env_runner import EnvRunner
  from ray.rllib.env.wrappers.atari_wrappers import NoopResetEnv, MaxAndSkipEnv
  from ray.rllib.env.wrappers.dm_control_wrapper import DMCEnv
  from ray.rllib.env.utils import _gym_env_creator
  from ray.rllib.evaluation.metrics import RolloutMetrics
  from ray.rllib.policy.sample_batch import DEFAULT_POLICY_ID, SampleBatch
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.framework import try_import_tf
  from ray.rllib.env.single_agent_episode import SingleAgentEpisode
  from ray.rllib.utils.numpy import one_hot
  from ray.tune.registry import ENV_CREATOR, _global_registry
label_desc: Use image node_id 1 to create a new DreamerV3EnvRunner object from inherited base classes: EnvRunner with example: obj = DreamerV3EnvRunner(config)