class_docstr: The lowest-level replay buffer interface used by RLlib.
  
  This class implements a basic ring-type of buffer with random sampling.
  ReplayBuffer is the base class for advanced types that add functionality while
  retaining compatibility through inheritance.
  
  The following examples show how buffers behave with different storage_units
  and capacities. This behaviour is generally similar for other buffers, although
  they might not implement all storage_units.
  
  Examples:
  
  .. testcode::
  
      from ray.rllib.utils.replay_buffers.replay_buffer import ReplayBuffer
      from ray.rllib.utils.replay_buffers.replay_buffer import StorageUnit
      from ray.rllib.policy.sample_batch import SampleBatch
  
      # Store any batch as a whole
      buffer = ReplayBuffer(capacity=10, storage_unit=StorageUnit.FRAGMENTS)
      buffer.add(SampleBatch({"a": [1], "b": [2, 3, 4]}))
      buffer.sample(1)
  
      # Store only complete episodes
      buffer = ReplayBuffer(capacity=10,
                              storage_unit=StorageUnit.EPISODES)
      buffer.add(SampleBatch({"c": [1, 2, 3, 4],
                              SampleBatch.T: [0, 1, 0, 1],
                              SampleBatch.TERMINATEDS: [False, True, False, True],
                              SampleBatch.EPS_ID: [0, 0, 1, 1]}))
      buffer.sample(1)
  
      # Store single timesteps
      buffer = ReplayBuffer(capacity=2, storage_unit=StorageUnit.TIMESTEPS)
      buffer.add(SampleBatch({"a": [1, 2], SampleBatch.T: [0, 1]}))
      buffer.sample(1)
  
      buffer.add(SampleBatch({"a": [3], SampleBatch.T: [2]}))
      print(buffer._eviction_started)
      buffer.sample(1)
  
      buffer = ReplayBuffer(capacity=10, storage_unit=StorageUnit.SEQUENCES)
      buffer.add(SampleBatch({"c": [1, 2, 3], SampleBatch.SEQ_LENS: [1, 2]}))
      buffer.sample(1)
  
  .. testoutput::
  
      True
  
  `True` is not the output of the above testcode, but an artifact of unexpected
  behaviour of sphinx doctests.
  (see https://github.com/ray-project/ray/pull/32477#discussion_r1106776101)
docstr: Add a SampleBatch of experiences to self._storage.
  
  An item consists of either one or more timesteps, a sequence or an
  episode. Differs from add() in that it does not consider the storage
  unit or type of batch and simply stores it.
  
  Args:
      item: The batch to be added.
      ``**kwargs``: Forward compatibility kwargs.
docstr_tok: ["Add","a","SampleBatch","of","experiences","to","self._storage",".","An","item","consists","of","either","one","or","more","timesteps",",","a","sequence","or","an","episode",".","Differs","from","add","(",")","in","that","it","does","not","consider","the","storage","unit","or","type","of","batch","and","simply","stores","it",".","Args",":","item",":","The","batch","to","be","added",".","``","*","*","kwargs","``",":","Forward","compatibility","kwargs","."]
code: def _add_single_batch(self, item: SampleBatchType, **kwargs) -> None:
      """Add a SampleBatch of experiences to self._storage.
  
      An item consists of either one or more timesteps, a sequence or an
      episode. Differs from add() in that it does not consider the storage
      unit or type of batch and simply stores it.
  
      Args:
          item: The batch to be added.
          ``**kwargs``: Forward compatibility kwargs.
      """
      self._num_timesteps_added += item.count
      self._num_timesteps_added_wrap += item.count
  
      if self._next_idx >= len(self._storage):
          self._storage.append(item)
          self._est_size_bytes += item.size_bytes()
      else:
          item_to_be_removed = self._storage[self._next_idx]
          self._est_size_bytes -= item_to_be_removed.size_bytes()
          self._storage[self._next_idx] = item
          self._est_size_bytes += item.size_bytes()
  
      # Eviction of older samples has already started (buffer is "full").
      if self._eviction_started:
          self._evicted_hit_stats.push(self._hit_count[self._next_idx])
          self._hit_count[self._next_idx] = 0
  
      # Wrap around storage as a circular buffer once we hit capacity.
      if self._num_timesteps_added_wrap >= self.capacity:
          self._eviction_started = True
          self._num_timesteps_added_wrap = 0
          self._next_idx = 0
      else:
          self._next_idx += 1
  
imports: from enum import Enum
  import logging
  import numpy
  import random
  from typing import Any, Dict, List, Optional, Union
  import ray
  import psutil
  from ray.rllib.policy.sample_batch import SampleBatch, concat_samples
  from ray.rllib.utils.actor_manager import FaultAwareApply
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.metrics.window_stat import WindowStat
  from ray.rllib.utils.replay_buffers.base import ReplayBufferInterface
  from ray.rllib.utils.typing import SampleBatchType
  from ray.util.annotations import DeveloperAPI
  from ray.util.debug import log_once
label_desc: Use image node_id 4 for calling the ReplayBuffer obj's underlying member method code with example usage: obj._add_single_batch(item) without return types