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: Initializes a (FIFO) ReplayBuffer instance.
  
  Args:
      capacity: Max number of timesteps to store in this FIFO
          buffer. After reaching this number, older samples will be
          dropped to make space for new ones.
      storage_unit: If not a StorageUnit, either 'timesteps', 'sequences' or
          'episodes'. Specifies how experiences are stored.
      ``**kwargs``: Forward compatibility kwargs.
docstr_tok: ["Initializes","a","(","FIFO",")","ReplayBuffer","instance",".","Args",":","capacity",":","Max","number","of","timesteps","to","store","in","this","FIFO","buffer",".","After","reaching","this","number",",","older","samples","will","be","dropped","to","make","space","for","new","ones",".","storage_unit",":","If","not","a","StorageUnit",",","either","'timesteps","'",",","'sequences","'","or","'episodes","'",".","Specifies","how","experiences","are","stored",".","``","*","*","kwargs","``",":","Forward","compatibility","kwargs","."]
code: def __init__(
      self,
      capacity: int = 10000,
      storage_unit: Union[str, StorageUnit] = "timesteps",
      **kwargs,
  ):
      """Initializes a (FIFO) ReplayBuffer instance.
  
      Args:
          capacity: Max number of timesteps to store in this FIFO
              buffer. After reaching this number, older samples will be
              dropped to make space for new ones.
          storage_unit: If not a StorageUnit, either 'timesteps', 'sequences' or
              'episodes'. Specifies how experiences are stored.
          ``**kwargs``: Forward compatibility kwargs.
      """
  
      if storage_unit in ["timesteps", StorageUnit.TIMESTEPS]:
          self.storage_unit = StorageUnit.TIMESTEPS
      elif storage_unit in ["sequences", StorageUnit.SEQUENCES]:
          self.storage_unit = StorageUnit.SEQUENCES
      elif storage_unit in ["episodes", StorageUnit.EPISODES]:
          self.storage_unit = StorageUnit.EPISODES
      elif storage_unit in ["fragments", StorageUnit.FRAGMENTS]:
          self.storage_unit = StorageUnit.FRAGMENTS
      else:
          raise ValueError(
              f"storage_unit must be either '{StorageUnit.TIMESTEPS}', "
              f"'{StorageUnit.SEQUENCES}', '{StorageUnit.EPISODES}' "
              f"or '{StorageUnit.FRAGMENTS}', but is {storage_unit}"
          )
  
      # The actual storage (list of SampleBatches or MultiAgentBatches).
      self._storage = []
  
      # Caps the number of timesteps stored in this buffer
      if capacity <= 0:
          raise ValueError(
              "Capacity of replay buffer has to be greater than zero "
              "but was set to {}.".format(capacity)
          )
      self.capacity = capacity
      # The next index to override in the buffer.
      self._next_idx = 0
      # len(self._hit_count) must always be less than len(capacity)
      self._hit_count = np.zeros(self.capacity)
  
      # Whether we have already hit our capacity (and have therefore
      # started to evict older samples).
      self._eviction_started = False
  
      # Number of (single) timesteps that have been added to the buffer
      # over its lifetime. Note that each added item (batch) may contain
      # more than one timestep.
      self._num_timesteps_added = 0
      self._num_timesteps_added_wrap = 0
  
      # Number of (single) timesteps that have been sampled from the buffer
      # over its lifetime.
      self._num_timesteps_sampled = 0
  
      self._evicted_hit_stats = WindowStat("evicted_hit", 1000)
      self._est_size_bytes = 0
  
      self.batch_size = None
  
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 1 to create a new ReplayBuffer object from inherited base classes: ReplayBufferInterface, FaultAwareApply with example: obj = ReplayBuffer(capacity, storage_unit)