class_docstr: Tests insertion and (weighted) sampling of the PrioritizedReplayBuffer.
class_docstr_tok: ["Tests","insertion","and","(","weighted",")","sampling","of","the","PrioritizedReplayBuffer","."]
code: def test_sequence_size(self):
      # Seq-len=1.
      buffer = PrioritizedReplayBuffer(
          capacity=100, alpha=0.1, storage_unit="fragments"
      )
      for _ in range(200):
          buffer.add(self._generate_data())
      assert len(buffer._storage) == 100, len(buffer._storage)
      assert buffer.stats()["added_count"] == 200, buffer.stats()
      # Test get_state/set_state.
      state = buffer.get_state()
      new_memory = PrioritizedReplayBuffer(capacity=100, alpha=0.1)
      new_memory.set_state(state)
      assert len(new_memory._storage) == 100, len(new_memory._storage)
      assert (
          new_memory.stats()["added_count"] == 200
      ), new_memory.stats()
  
      # Seq-len=5.
      buffer = PrioritizedReplayBuffer(
          capacity=100, alpha=0.1, storage_unit="fragments"
      )
      for _ in range(40):
          buffer.add(
              concat_samples([self._generate_data() for _ in range(5)])
          )
      assert len(buffer._storage) == 20, len(buffer._storage)
      assert buffer.stats()["added_count"] == 200, buffer.stats()
      # Test get_state/set_state.
      state = buffer.get_state()
      new_memory = PrioritizedReplayBuffer(capacity=100, alpha=0.1)
      new_memory.set_state(state)
      assert len(new_memory._storage) == 20, len(new_memory._storage)
      assert (
          new_memory.stats()["added_count"] == 200
      ), new_memory.stats()
  
file_path: ray/python/ray/rllib/utils/replay_buffers/tests/test_prioritized_replay_buffer_replay_buffer_api.py
imports: from collections import Counter
  import numpy
  import unittest
  from ray.rllib.utils.replay_buffers.prioritized_replay_buffer import PrioritizedReplayBuffer
  from ray.rllib.policy.sample_batch import SampleBatch, MultiAgentBatch, concat_samples
  from ray.rllib.utils.test_utils import check
label_desc: Use image node_id 3 for calling the TestPrioritizedReplayBuffer obj's underlying member method code with example usage: obj.test_sequence_size() without return types