class_docstr: Collect experiences from user environment using Connectors.
docstr: Handle an all-finished episode.
  
  Add collected SampleBatch to batch builder. Reset corresponding env, etc.
  
  Args:
      env_id: Environment ID.
      env_obs_or_exception: Last per-environment observation or Exception.
      env_infos: Last per-environment infos.
      is_done: If all agents are done.
      active_envs: Set of active env ids.
      to_eval: Output container for policy eval data.
      outputs: Output container for collected sample batches.
docstr_tok: ["Handle","an","all-finished","episode",".","Add","collected","SampleBatch","to","batch","builder",".","Reset","corresponding","env",",","etc",".","Args",":","env_id",":","Environment","ID",".","env_obs_or_exception",":","Last","per-environment","observation","or","Exception",".","env_infos",":","Last","per-environment","infos",".","is_done",":","If","all","agents","are","done",".","active_envs",":","Set","of","active","env","ids",".","to_eval",":","Output","container","for","policy","eval","data",".","outputs",":","Output","container","for","collected","sample","batches","."]
code: def _handle_done_episode(
      self,
      env_id: EnvID,
      env_obs_or_exception: MultiAgentDict,
      is_done: bool,
      active_envs: Set[EnvID],
      to_eval: Dict[PolicyID, List[AgentConnectorDataType]],
      outputs: List[SampleBatchType],
  ) -> None:
      """Handle an all-finished episode.
  
      Add collected SampleBatch to batch builder. Reset corresponding env, etc.
  
      Args:
          env_id: Environment ID.
          env_obs_or_exception: Last per-environment observation or Exception.
          env_infos: Last per-environment infos.
          is_done: If all agents are done.
          active_envs: Set of active env ids.
          to_eval: Output container for policy eval data.
          outputs: Output container for collected sample batches.
      """
      if isinstance(env_obs_or_exception, Exception):
          episode_or_exception: Exception = env_obs_or_exception
          # Tell the sampler we have got a faulty episode.
          outputs.append(RolloutMetrics(episode_faulty=True))
      else:
          episode_or_exception: EpisodeV2 = self._active_episodes[
              env_id
          ]
          # Add rollout metrics.
          outputs.extend(
              self._get_rollout_metrics(
                  episode_or_exception,
                  policy_map=self._worker.policy_map,
              )
          )
          # Output the collected episode after adding rollout metrics so that we
          # always fetch metrics with RolloutWorker before we fetch samples.
          # This is because we need to behave like env_runner() for now.
          self._build_done_episode(env_id, is_done, outputs)
  
      # Clean up and deleted the post-processed episode now that we have collected
      # its data.
      self.end_episode(env_id, episode_or_exception)
      # Create a new episode instance (before we reset the sub-environment).
      new_episode: EpisodeV2 = self.create_episode(env_id)
  
      # The sub environment at index `env_id` might throw an exception
      # during the following `try_reset()` attempt. If configured with
      # `restart_failed_sub_environments=True`, the BaseEnv will restart
      # the affected sub environment (create a new one using its c'tor) and
      # must reset the recreated sub env right after that.
      # Should the sub environment fail indefinitely during these
      # repeated reset attempts, the entire worker will be blocked.
      # This would be ok, b/c the alternative would be the worker crashing
      # entirely.
      while True:
          resetted_obs, resetted_infos = self._base_env.try_reset(
              env_id
          )
  
          if (
              resetted_obs is None
              or resetted_obs == ASYNC_RESET_RETURN
              or not isinstance(resetted_obs[env_id], Exception)
          ):
              break
          else:
              # Report a faulty episode.
              outputs.append(RolloutMetrics(episode_faulty=True))
  
      # Reset connector state if this is a hard reset.
      for p in self._worker.policy_map.cache.values():
          p.agent_connectors.reset(env_id)
  
      # Creates a new episode if this is not async return.
      # If reset is async, we will get its result in some future poll.
      if (
          resetted_obs is not None
          and resetted_obs != ASYNC_RESET_RETURN
      ):
          self._active_episodes[env_id] = new_episode
          self._call_on_episode_start(new_episode, env_id)
  
          self.__process_resetted_obs_for_eval(
              env_id,
              resetted_obs,
              resetted_infos,
              new_episode,
              to_eval,
          )
  
          # Step after adding initial obs. This will give us 0 env and agent step.
          new_episode.step()
          active_envs.add(env_id)
  
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 11 for calling the EnvRunnerV2 obj's underlying member method code with example usage: obj._handle_done_episode(env_id, env_obs_or_exception, is_done, active_envs, to_eval, outputs) without return types