class_docstr: Collect experiences from user environment using Connectors.
docstr: Args:
      worker: Reference to the current rollout worker.
      base_env: Env implementing BaseEnv.
      multiple_episodes_in_batch: Whether to pack multiple
          episodes into each batch. This guarantees batches will be exactly
          `rollout_fragment_length` in size.
      callbacks: User callbacks to run on episode events.
      perf_stats: Record perf stats into this object.
      rollout_fragment_length: The length of a fragment to collect
          before building a SampleBatch from the data and resetting
          the SampleBatchBuilder object.
      count_steps_by: One of "env_steps" (default) or "agent_steps".
          Use "agent_steps", if you want rollout lengths to be counted
          by individual agent steps. In a multi-agent env,
          a single env_step contains one or more agent_steps, depending
          on how many agents are present at any given time in the
          ongoing episode.
      render: Whether to try to render the environment after each
          step.
docstr_tok: ["Args",":","worker",":","Reference","to","the","current","rollout","worker",".","base_env",":","Env","implementing","BaseEnv",".","multiple_episodes_in_batch",":","Whether","to","pack","multiple","episodes","into","each","batch",".","This","guarantees","batches","will","be","exactly","`","rollout_fragment_length","`","in","size",".","callbacks",":","User","callbacks","to","run","on","episode","events",".","perf_stats",":","Record","perf","stats","into","this","object",".","rollout_fragment_length",":","The","length","of","a","fragment","to","collect","before","building","a","SampleBatch","from","the","data","and","resetting","the","SampleBatchBuilder","object",".","count_steps_by",":","One","of","``","env_steps","''","(","default",")","or","``","agent_steps","''",".","Use","``","agent_steps","''",",","if","you","want","rollout","lengths","to","be","counted","by","individual","agent","steps",".","In","a","multi-agent","env",",","a","single","env_step","contains","one","or","more","agent_steps",",","depending","on","how","many","agents","are","present","at","any","given","time","in","the","ongoing","episode",".","render",":","Whether","to","try","to","render","the","environment","after","each","step","."]
code: def __init__(
      self,
      worker: "RolloutWorker",
      base_env: BaseEnv,
      multiple_episodes_in_batch: bool,
      callbacks: "DefaultCallbacks",
      perf_stats: _PerfStats,
      rollout_fragment_length: int = 200,
      count_steps_by: str = "env_steps",
      render: bool = None,
  ):
      """
      Args:
          worker: Reference to the current rollout worker.
          base_env: Env implementing BaseEnv.
          multiple_episodes_in_batch: Whether to pack multiple
              episodes into each batch. This guarantees batches will be exactly
              `rollout_fragment_length` in size.
          callbacks: User callbacks to run on episode events.
          perf_stats: Record perf stats into this object.
          rollout_fragment_length: The length of a fragment to collect
              before building a SampleBatch from the data and resetting
              the SampleBatchBuilder object.
          count_steps_by: One of "env_steps" (default) or "agent_steps".
              Use "agent_steps", if you want rollout lengths to be counted
              by individual agent steps. In a multi-agent env,
              a single env_step contains one or more agent_steps, depending
              on how many agents are present at any given time in the
              ongoing episode.
          render: Whether to try to render the environment after each
              step.
      """
      self._worker = worker
      if isinstance(base_env, ExternalEnvWrapper):
          raise ValueError(
              "Policies using the new Connector API do not support ExternalEnv."
          )
      self._base_env = base_env
      self._multiple_episodes_in_batch = multiple_episodes_in_batch
      self._callbacks = callbacks
      self._perf_stats = perf_stats
      self._rollout_fragment_length = rollout_fragment_length
      self._count_steps_by = count_steps_by
      self._render = render
  
      # May be populated for image rendering.
      self._simple_image_viewer: Optional[
          "SimpleImageViewer"
      ] = self._get_simple_image_viewer()
  
      # Keeps track of active episodes.
      self._active_episodes: Dict[EnvID, EpisodeV2] = {}
      self._batch_builders: Dict[
          EnvID, _PolicyCollectorGroup
      ] = _NewDefaultDict(self._new_batch_builder)
  
      self._large_batch_threshold: int = (
          max(
              MIN_LARGE_BATCH_THRESHOLD,
              self._rollout_fragment_length * 10,
          )
          if self._rollout_fragment_length != float("inf")
          else DEFAULT_LARGE_BATCH_THRESHOLD
      )
  
imports: from collections import defaultdict
  import logging
  import time
  import tree
  from typing import TYPE_CHECKING, Dict, Iterator, List, Optional, Set, Tuple, Union
  import numpy
  from ray.rllib.env.base_env import ASYNC_RESET_RETURN, BaseEnv
  from ray.rllib.env.external_env import ExternalEnvWrapper
  from ray.rllib.env.wrappers.atari_wrappers import MonitorEnv, get_wrapper_by_cls
  from ray.rllib.evaluation.collectors.simple_list_collector import _PolicyCollectorGroup
  from ray.rllib.policy.rnn_sequencing import pad_batch_to_sequences_of_same_size
  from ray.rllib.evaluation.episode_v2 import EpisodeV2
  from ray.rllib.evaluation.metrics import RolloutMetrics
  from ray.rllib.models.preprocessors import Preprocessor
  from ray.rllib.policy.policy import Policy
  from ray.rllib.policy.sample_batch import MultiAgentBatch, SampleBatch, concat_samples
  from ray.rllib.utils.annotations import DeveloperAPI
  from ray.rllib.utils.filter import Filter
  from ray.rllib.utils.numpy import convert_to_numpy
  from ray.rllib.utils.spaces.space_utils import unbatch, get_original_space
  from ray.rllib.utils.typing import ActionConnectorDataType, AgentConnectorDataType, AgentID, EnvActionType, EnvID, EnvInfoDict, EnvObsType, MultiAgentDict, MultiEnvDict, PolicyID, PolicyOutputType, SampleBatchType, StateBatches, TensorStructType
  from ray.util.debug import log_once
label_desc: Use image node_id 1 to create a new EnvRunnerV2 object  with example: obj = EnvRunnerV2(worker, base_env, multiple_episodes_in_batch, callbacks, perf_stats, rollout_fragment_length, count_steps_by, render)