class_docstr: Runner for evaluating using a fixed number of episodes rather than steps.
  
  Also restricts data collection to a strict cap,
  following conventions in data-efficient RL research.
class_docstr_tok: ["Runner","for","evaluating","using","a","fixed","number","of","episodes","rather","than","steps",".","Also","restricts","data","collection","to","a","strict","cap",",","following","conventions","in","data-efficient","RL","research","."]
docstr: Executes a full trajectory of the agent interacting with the environment.
  
  Args:
    envs: Environments to step in.
    episodes: Optional int, how many episodes to run. Unbounded if None.
    max_steps: Optional int, how many steps to run. Unbounded if None.
    one_to_one: Bool, whether to couple each episode to an environment.
    needs_reset: Bool, whether to reset environments before beginning.
    resume_state: State tuple to resume.
  
  Returns:
    The number of steps taken and the total reward.
docstr_tok: ["Executes","a","full","trajectory","of","the","agent","interacting","with","the","environment",".","Args",":","envs",":","Environments","to","step","in",".","episodes",":","Optional","int",",","how","many","episodes","to","run",".","Unbounded","if","None",".","max_steps",":","Optional","int",",","how","many","steps","to","run",".","Unbounded","if","None",".","one_to_one",":","Bool",",","whether","to","couple","each","episode","to","an","environment",".","needs_reset",":","Bool",",","whether","to","reset","environments","before","beginning",".","resume_state",":","State","tuple","to","resume",".","Returns",":","The","number","of","steps","taken","and","the","total","reward","."]
code: def _run_parallel(
      self,
      envs,
      episodes=None,
      max_steps=None,
      one_to_one=False,
      needs_reset=True,
      resume_state=None,
  ):
      """Executes a full trajectory of the agent interacting with the environment.
  
      Args:
        envs: Environments to step in.
        episodes: Optional int, how many episodes to run. Unbounded if None.
        max_steps: Optional int, how many steps to run. Unbounded if None.
        one_to_one: Bool, whether to couple each episode to an environment.
        needs_reset: Bool, whether to reset environments before beginning.
        resume_state: State tuple to resume.
  
      Returns:
        The number of steps taken and the total reward.
      """
      # You can't ask for 200 episodes run one-to-one on 100 envs
      if one_to_one:
          assert episodes is None or episodes == len(envs)
  
      # Create envs
      live_envs = list(range(len(envs)))
  
      if needs_reset:
          new_obs = self._initialize_episode(envs)
          new_obses = np.zeros(
              (2, len(envs), *self._agent.observation_shape, 1)
          )
          self._agent.reset_all(new_obs)
  
          rewards = np.zeros((len(envs),))
          terminals = np.zeros((len(envs),))
          episode_end = np.zeros((len(envs),))
  
          cum_rewards = []
          cum_lengths = []
      else:
          assert resume_state is not None
          (
              new_obses,
              rewards,
              terminals,
              episode_end,
              cum_rewards,
              cum_lengths,
          ) = resume_state
  
      total_steps = 0
      total_episodes = 0
      max_steps = np.inf if max_steps is None else max_steps
      step = 0
  
      # Keep interacting until we reach a terminal state.
      while True:
          live_env_index = 0
          step += 1
          episode_end.fill(0)
          total_steps += len(live_envs)
          actions = self._agent.step()
  
          # The agent may be hanging on to the previous new_obs, so we shouldn't
          # change it. By alternating, we can make sure we don't end up logging
          # with an offset while limiting memory usage.
          new_obs = new_obses[step % 2]
  
          # don't want to do a for-loop since live envs may change
          while live_env_index < len(live_envs):
              env_id = live_envs[live_env_index]
              obs, reward, done, _ = envs[env_id].step(
                  actions[live_env_index]
              )
              envs[env_id].cum_length += 1
              envs[env_id].cum_reward += reward
              new_obs[live_env_index] = obs
              rewards[live_env_index] = reward
              terminals[live_env_index] = done
  
              if (
                  envs[env_id].game_over
                  or envs[env_id].cum_length
                  == self._max_steps_per_episode
              ):
                  total_episodes += 1
                  cum_rewards.append(envs[env_id].cum_reward)
                  cum_lengths.append(envs[env_id].cum_length)
                  envs[env_id].cum_length = 0
                  envs[env_id].cum_reward = 0
  
                  log_str = (
                      "Steps executed: {} Num episodes: {} Episode length: {}"
                      " Return: {} ".format(
                          total_steps,
                          len(cum_rewards),
                          cum_lengths[-1],
                          cum_rewards[-1],
                      )
                  )
                  logging.info(log_str)
                  self._maybe_save_single_summary(
                      self.num_steps + total_steps,
                      cum_rewards[-1],
                      cum_lengths[-1],
                  )
  
                  if one_to_one:
                      new_obses = delete_ind_from_array(
                          new_obses, live_env_index, axis=1
                      )
                      new_obs = new_obses[step % 2]
                      actions = delete_ind_from_array(
                          actions, live_env_index
                      )
                      rewards = delete_ind_from_array(
                          rewards, live_env_index
                      )
                      terminals = delete_ind_from_array(
                          terminals, live_env_index
                      )
                      self._agent.delete_one(live_env_index)
                      del live_envs[live_env_index]
                      live_env_index -= 1  # Go back one to make up for deleting a value.
                  else:
                      episode_end[live_env_index] = 1
                      new_obs[
                          live_env_index
                      ] = self._initialize_episode([envs[env_id]])
                      self._agent.reset_one(env_id=live_env_index)
              elif done:
                  self._agent.reset_one(env_id=live_env_index)
  
              live_env_index += 1
  
          if self._clip_rewards:
              # Perform reward clipping.
              rewards = np.clip(rewards, -1, 1)
  
          self._agent.log_transition(
              new_obs, actions, rewards, terminals, episode_end
          )
  
          if (
              not live_envs
              or (max_steps is not None and total_steps > max_steps)
              or (episodes is not None and total_episodes > episodes)
          ):
              break
  
      state = (
          new_obses,
          rewards,
          terminals,
          episode_end,
          cum_rewards,
          cum_lengths,
      )
      return cum_lengths, cum_rewards, state, envs
  
imports: import functools
  import os
  import sys
  import time
  from absl import logging
  from dopamine.discrete_domains import atari_lib
  from dopamine.discrete_domains import iteration_statistics
  from dopamine.discrete_domains import run_experiment
  from dopamine.labs.atari_100k import normalization_utils
  from dopamine.metrics import collector_dispatcher
  from dopamine.metrics import statistics_instance
  import gin
  import jax
  import numpy
  import tensorflow
label_desc: Use image node_id 4 for calling the DataEfficientAtariRunner obj's underlying member method code with example usage: obj._run_parallel(envs, episodes, max_steps, one_to_one, needs_reset, resume_state) and returns: cum_lengths,  cum_rewards,  state,  envs