class_docstr: Collect experiences from user environment using Connectors.
docstr: Samples training episodes by stepping through environments.
docstr_tok: ["Samples","training","episodes","by","stepping","through","environments","."]
code: def step(self) -> List[SampleBatchType]:
      """Samples training episodes by stepping through environments."""
  
      self._perf_stats.incr("iters", 1)
  
      t0 = time.time()
      # Get observations from all ready agents.
      # types: MultiEnvDict, MultiEnvDict, MultiEnvDict, MultiEnvDict, ...
      (
          unfiltered_obs,
          rewards,
          terminateds,
          truncateds,
          infos,
          off_policy_actions,
      ) = self._base_env.poll()
      env_poll_time = time.time() - t0
  
      # Process observations and prepare for policy evaluation.
      t1 = time.time()
      # types: Set[EnvID], Dict[PolicyID, List[AgentConnectorDataType]],
      #       List[Union[RolloutMetrics, SampleBatchType]]
      active_envs, to_eval, outputs = self._process_observations(
          unfiltered_obs=unfiltered_obs,
          rewards=rewards,
          terminateds=terminateds,
          truncateds=truncateds,
          infos=infos,
      )
      self._perf_stats.incr("raw_obs_processing_time", time.time() - t1)
  
      # Do batched policy eval (accross vectorized envs).
      t2 = time.time()
      # types: Dict[PolicyID, Tuple[TensorStructType, StateBatch, dict]]
      eval_results = self._do_policy_eval(to_eval=to_eval)
      self._perf_stats.incr("inference_time", time.time() - t2)
  
      # Process results and update episode state.
      t3 = time.time()
      actions_to_send: Dict[
          EnvID, Dict[AgentID, EnvActionType]
      ] = self._process_policy_eval_results(
          active_envs=active_envs,
          to_eval=to_eval,
          eval_results=eval_results,
          off_policy_actions=off_policy_actions,
      )
      self._perf_stats.incr("action_processing_time", time.time() - t3)
  
      # Return computed actions to ready envs. We also send to envs that have
      # taken off-policy actions; those envs are free to ignore the action.
      t4 = time.time()
      self._base_env.send_actions(actions_to_send)
      self._perf_stats.incr(
          "env_wait_time", env_poll_time + time.time() - t4
      )
  
      self._maybe_render()
  
      return outputs
  
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 6 for calling the EnvRunnerV2 obj's underlying member method code with example usage: obj.step() and returns: outputs