class_docstr: Stress tests for the replay buffer, especially for new flags.
class_docstr_tok: ["Stress","tests","for","the","replay","buffer",",","especially","for","new","flags","."]
code: def test_sparse_input(self):
      replay_capacity = 100
      num_transitions = replay_capacity // 2
      memory = ReplayBuffer(
          stack_size=1,
          replay_capacity=replay_capacity,
          update_horizon=1,
      )
  
      def trans(i):
          sparse_feat1 = list(range(0, i % 4))
          sparse_feat2 = list(range(i % 4, 4))
          id_list = {
              "sparse_feat1": sparse_feat1,
              "sparse_feat2": sparse_feat2,
          }
          sparse_feat3 = (
              list(range(0, i % 7)),
              [k + 0.5 for k in range(0, i % 7)],
          )
          sparse_feat4 = (
              list(range(i % 7, 7)),
              [k + 0.5 for k in range(i % 7, 7)],
          )
          id_score_list = {
              "sparse_feat3": sparse_feat3,
              "sparse_feat4": sparse_feat4,
          }
          return {
              "observation": np.ones(OBS_SHAPE, dtype=OBS_TYPE),
              "action": int(2 * i),
              "reward": float(3 * i),
              "terminal": i % 4,
              "id_list": id_list,
              "id_score_list": id_score_list,
          }
  
      for i in range(num_transitions):
          memory.add(**trans(i))
  
      indices = list(range(num_transitions - 1))
      batch = memory.sample_transition_batch(
          len(indices), torch.tensor(indices)
      )
  
      # calculate expected
      res = {
          "id_list": {
              "sparse_feat1": ([], []),
              "sparse_feat2": ([], []),
          },
          "id_score_list": {
              "sparse_feat3": ([], [], []),
              "sparse_feat4": ([], [], []),
          },
          "next_id_list": {
              "sparse_feat1": ([], []),
              "sparse_feat2": ([], []),
          },
          "next_id_score_list": {
              "sparse_feat3": ([], [], []),
              "sparse_feat4": ([], [], []),
          },
      }
      for i in range(num_transitions - 1):
          feats_i = trans(i)
          feats_next = trans(i + 1)
          for k in ["id_list", "id_score_list"]:
              for feat_id in res[k]:
                  res[k][feat_id][0].append(len(res[k][feat_id][1]))
                  if k == "id_list":
                      res[k][feat_id][1].extend(feats_i[k][feat_id])
                  else:
                      res[k][feat_id][1].extend(feats_i[k][feat_id][0])
                      res[k][feat_id][2].extend(feats_i[k][feat_id][1])
  
          for k in ["next_id_list", "next_id_score_list"]:
              for feat_id in res[k]:
                  res[k][feat_id][0].append(len(res[k][feat_id][1]))
                  orig_k = k[len("next_") :]
                  if k == "next_id_list":
                      res[k][feat_id][1].extend(
                          feats_next[orig_k][feat_id]
                      )
                  else:
                      res[k][feat_id][1].extend(
                          feats_next[orig_k][feat_id][0]
                      )
                      res[k][feat_id][2].extend(
                          feats_next[orig_k][feat_id][1]
                      )
  
      for k in [
          "id_list",
          "id_score_list",
          "next_id_list",
          "next_id_score_list",
      ]:
          for feat_id in res[k]:
              if k in ["id_list", "next_id_list"]:
                  npt.assert_array_equal(
                      res[k][feat_id][0], getattr(batch, k)[feat_id][0]
                  )
                  npt.assert_array_equal(
                      res[k][feat_id][1], getattr(batch, k)[feat_id][1]
                  )
              else:
                  npt.assert_array_equal(
                      res[k][feat_id][0], getattr(batch, k)[feat_id][0]
                  )
                  npt.assert_array_equal(
                      res[k][feat_id][1], getattr(batch, k)[feat_id][1]
                  )
                  npt.assert_array_equal(
                      res[k][feat_id][2], getattr(batch, k)[feat_id][2]
                  )
  
      # sample random
      _ = memory.sample_transition_batch(10)
  
file_path: ReAgent/reagent/test/replay_memory/extra_replay_buffer_test.py
imports: import logging
  import numpy
  import numpy.testing
  import torch
  from reagent.replay_memory.circular_replay_buffer import ReplayBuffer
  from reagent.test.base.horizon_test_base import HorizonTestBase
label_desc: Use image node_id 4 for calling the ExtraReplayBufferTest obj's underlying member method code with example usage: obj.test_sparse_input() and returns: dict