class_docstr: Buffer that stores (completed or truncated) episodes by their ID.
  
  Each "row" (a slot in a deque) in the buffer is occupied by one episode. If an
  incomplete episode is added to the buffer and then another chunk of that episode is
  added at a later time, the buffer will automatically concatenate the new fragment to
  the original episode. This way, episodes can be completed via subsequent `add`
  calls.
  
  Sampling returns batches of size B (number of "rows"), where each row is a
  trajectory of length T. Each trajectory contains consecutive timesteps from an
  episode, but might not start at the beginning of that episode. Should an episode end
  within such a trajectory, a random next episode (starting from its t0) will be
  concatenated to that "row". Example: `sample(B=2, T=4)` ->
  
     0 .. 1 .. 2 .. 3  <- T-axis
  0 e5   e6   e7   e8
  1 f2   f3   h0   h2
  ^ B-axis
  
  .. where e, f, and h are different (randomly picked) episodes, the 0-index (e.g. h0)
  indicates the start of an episode, and `f3` is an episode end (gym environment
  returned terminated=True or truncated=True).
  
  0-indexed returned timesteps contain the reset observation, a dummy 0.0 reward, as
  well as the first action taken in the episode (action picked after observing
  obs(0)).
  The last index in an episode (e.g. f3 in the example above) contains the final
  observation of the episode, the final reward received, a dummy action
  (repeat the previous action), as well as either terminated=True or truncated=True.
class_docstr_tok: ["Buffer","that","stores","(","completed","or","truncated",")","episodes","by","their","ID",".","Each","``","row","''","(","a","slot","in","a","deque",")","in","the","buffer","is","occupied","by","one","episode",".","If","an","incomplete","episode","is","added","to","the","buffer","and","then","another","chunk","of","that","episode","is","added","at","a","later","time",",","the","buffer","will","automatically","concatenate","the","new","fragment","to","the","original","episode",".","This","way",",","episodes","can","be","completed","via","subsequent","`","add","`","calls",".","Sampling","returns","batches","of","size","B","(","number","of","``","rows","''",")",",","where","each","row","is","a","trajectory","of","length","T.","Each","trajectory","contains","consecutive","timesteps","from","an","episode",",","but","might","not","start","at","the","beginning","of","that","episode",".","Should","an","episode","end","within","such","a","trajectory",",","a","random","next","episode","(","starting","from","its","t0",")","will","be","concatenated","to","that","``","row","''",".","Example",":","`","sample","(","B=2",",","T=4",")","`","-",">","0","..","1","..","2","..","3","<","-","T-axis","0","e5","e6","e7","e8","1","f2","f3","h0","h2","^","B-axis","..","where","e",",","f",",","and","h","are","different","(","randomly","picked",")","episodes",",","the","0-index","(","e.g",".","h0",")","indicates","the","start","of","an","episode",",","and","`","f3","`","is","an","episode","end","(","gym","environment","returned","terminated=True","or","truncated=True",")",".","0-indexed","returned","timesteps","contain","the","reset","observation",",","a","dummy","0.0","reward",",","as","well","as","the","first","action","taken","in","the","episode","(","action","picked","after","observing","obs","(","0",")",")",".","The","last","index","in","an","episode","(","e.g",".","f3","in","the","example","above",")","contains","the","final","observation","of","the","episode",",","the","final","reward","received",",","a","dummy","action","(","repeat","the","previous","action",")",",","as","well","as","either","terminated=True","or","truncated=True","."]
docstr: Converts the incoming SampleBatch into a number of SingleAgentEpisode objects.
  
  Then adds these episodes to the internal deque.
docstr_tok: ["Converts","the","incoming","SampleBatch","into","a","number","of","SingleAgentEpisode","objects",".","Then","adds","these","episodes","to","the","internal","deque","."]
code: def add(
      self,
      episodes: Union[List["SingleAgentEpisode"], "SingleAgentEpisode"],
  ):
      """Converts the incoming SampleBatch into a number of SingleAgentEpisode objects.
  
      Then adds these episodes to the internal deque.
      """
      if isinstance(episodes, SingleAgentEpisode):
          episodes = [episodes]
  
      for eps in episodes:
          # Make sure we don't change what's coming in from the user.
          # TODO (sven): It'd probably be better to make sure in the EnvRunner to not
          #  hold on to episodes (for metrics purposes only) that we are returning
          #  back to the user from `EnvRunner.sample()`. Then we wouldn't have to
          #  do any copying. Instead, either compile the metrics right away on the
          #  EnvRunner OR compile metrics entirely on the Algorithm side (this is
          #  actually preferred).
          eps = copy.deepcopy(eps)
  
          self._num_timesteps += len(eps)
          self._num_timesteps_added += len(eps)
  
          # Ongoing episode, concat to existing record.
          if eps.id_ in self.episode_id_to_index:
              eps_idx = self.episode_id_to_index[eps.id_]
              existing_eps = self.episodes[
                  eps_idx - self._num_episodes_evicted
              ]
              old_len = len(existing_eps)
              self._indices.extend(
                  [(eps_idx, old_len + i) for i in range(len(eps))]
              )
              existing_eps.concat_episode(eps)
          # New episode. Add to end of our episodes deque.
          else:
              self.episodes.append(eps)
              eps_idx = (
                  len(self.episodes) - 1 + self._num_episodes_evicted
              )
              self.episode_id_to_index[eps.id_] = eps_idx
              self._indices.extend(
                  [(eps_idx, i) for i in range(len(eps))]
              )
  
          # Eject old records from front of deque (only if we have more than 1 episode
          # in the buffer).
          while (
              self._num_timesteps > self.capacity
              and self.get_num_episodes() > 1
          ):
              # Eject oldest episode.
              evicted_eps = self.episodes.popleft()
              evicted_eps_len = len(evicted_eps)
              # Correct our size.
              self._num_timesteps -= evicted_eps_len
  
              # Erase episode from all our indices:
              # 1) Main episode index.
              evicted_idx = self.episode_id_to_index[evicted_eps.id_]
              del self.episode_id_to_index[evicted_eps.id_]
              # 2) All timestep indices that this episode owned.
              new_indices = (
                  []
              )  # New indices that will replace self._indices.
              idx_cursor = 0
              # Loop through all (eps_idx, ts_in_eps_idx)-tuples.
              for i, idx_tuple in enumerate(self._indices):
                  # This tuple is part of the evicted episode -> Add everything
                  # up until here to `new_indices` (excluding this very index, b/c
                  # it's already part of the evicted episode).
                  if (
                      idx_cursor is not None
                      and idx_tuple[0] == evicted_idx
                  ):
                      new_indices.extend(self._indices[idx_cursor:i])
                      # Set to None to indicate we are in the eviction zone.
                      idx_cursor = None
                  # We are/have been in the eviction zone (i pointing/pointed to the
                  # evicted episode) ..
                  elif idx_cursor is None:
                      # ... but are now not anymore (i is now an index into a
                      # non-evicted episode) -> Set cursor to valid int again.
                      if idx_tuple[0] != evicted_idx:
                          idx_cursor = i
                          # But early-out if evicted episode was only 1 single
                          # timestep long.
                          if evicted_eps_len == 1:
                              break
                      # Early-out: We reached the end of the to-be-evicted episode.
                      # We can stop searching further here (all following tuples
                      # will NOT be in the evicted episode).
                      elif idx_tuple[1] == evicted_eps_len - 1:
                          assert self._indices[i + 1][0] != idx_tuple[0]
                          idx_cursor = i + 1
                          break
  
              # Jump over (splice-out) the evicted episode if we are still in the
              # eviction zone.
              if idx_cursor is not None:
                  new_indices.extend(self._indices[idx_cursor:])
  
              # Reset our `self._indices` to the newly compiled list.
              self._indices = new_indices
  
              # Increase episode evicted counter.
              self._num_episodes_evicted += 1
  
imports: from collections import deque
  import copy
  from typing import Any, Dict, List, Optional, Union
  import numpy
  from ray.rllib.env.single_agent_episode import SingleAgentEpisode
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.replay_buffers.base import ReplayBufferInterface
  from ray.rllib.utils.typing import SampleBatchType
label_desc: Use image node_id 3 for calling the EpisodeReplayBuffer obj's underlying member method code with example usage: obj.add(episodes) without return types