class_docstr: This buffer adds replayed samples to a stream of new experiences.
  
  - Any newly added batch (`add()`) is immediately returned upon
  the next `sample` call (close to on-policy) as well as being moved
  into the buffer.
  - Additionally, a certain number of old samples is mixed into the
  returned sample according to a given "replay ratio".
  - If >1 calls to `add()` are made without any `sample()` calls
  in between, all newly added batches are returned (plus some older samples
  according to the "replay ratio").
  
  .. testcode::
      :skipif: True
  
      # replay ratio 0.66 (2/3 replayed, 1/3 new samples):
      buffer = MultiAgentMixInReplayBuffer(capacity=100,
                                           replay_ratio=0.66)
      buffer.add(<A>)
      buffer.add(<B>)
      buffer.sample(1)
  
  .. testoutput::
  
      ..[<A>, <B>, <B>]
  
  .. testcode::
      :skipif: True
  
      buffer.add(<C>)
      buffer.sample(1)
  
  .. testoutput::
  
      [<C>, <A>, <B>]
      or: [<C>, <A>, <A>], [<C>, <B>, <A>] or [<C>, <B>, <B>],
      but always <C> as it is the newest sample
  
  .. testcode::
      :skipif: True
  
      buffer.add(<D>)
      buffer.sample(1)
  
  .. testoutput::
  
      [<D>, <A>, <C>]
      or [<D>, <A>, <A>], [<D>, <B>, <A>] or [<D>, <B>, <C>], etc..
      but always <D> as it is the newest sample
  
  .. testcode::
      :skipif: True
  
      # replay proportion 0.0 -> replay disabled:
      buffer = MixInReplay(capacity=100, replay_ratio=0.0)
      buffer.add(<A>)
      buffer.sample()
  
  .. testoutput::
  
      [<A>]
  
  .. testcode::
      :skipif: True
  
      buffer.add(<B>)
      buffer.sample()
  
  .. testoutput::
  
      [<B>]
class_docstr_tok: ["This","buffer","adds","replayed","samples","to","a","stream","of","new","experiences",".","-","Any","newly","added","batch","(","`","add","(",")","`",")","is","immediately","returned","upon","the","next","`","sample","`","call","(","close","to","on-policy",")","as","well","as","being","moved","into","the","buffer",".","-","Additionally",",","a","certain","number","of","old","samples","is","mixed","into","the","returned","sample","according","to","a","given","``","replay","ratio","''",".","-","If",">","1","calls","to","`","add","(",")","`","are","made","without","any","`","sample","(",")","`","calls","in","between",",","all","newly","added","batches","are","returned","(","plus","some","older","samples","according","to","the","``","replay","ratio","''",")",".","..","testcode",":",":",":","skipif",":","True","#","replay","ratio","0.66","(","2\/3","replayed",",","1\/3","new","samples",")",":","buffer","=","MultiAgentMixInReplayBuffer","(","capacity=100",",","replay_ratio=0.66",")","buffer.add","(","<","A",">",")","buffer.add","(","<","B",">",")","buffer.sample","(","1",")","..","testoutput",":",":","..","[","<","A",">",",","<","B",">",",","<","B",">","]","..","testcode",":",":",":","skipif",":","True","buffer.add","(","<","C",">",")","buffer.sample","(","1",")","..","testoutput",":",":","[","<","C",">",",","<","A",">",",","<","B",">","]","or",":","[","<","C",">",",","<","A",">",",","<","A",">","]",",","[","<","C",">",",","<","B",">",",","<","A",">","]","or","[","<","C",">",",","<","B",">",",","<","B",">","]",",","but","always","<","C",">","as","it","is","the","newest","sample","..","testcode",":",":",":","skipif",":","True","buffer.add","(","<","D",">",")","buffer.sample","(","1",")","..","testoutput",":",":","[","<","D",">",",","<","A",">",",","<","C",">","]","or","[","<","D",">",",","<","A",">",",","<","A",">","]",",","[","<","D",">",",","<","B",">",",","<","A",">","]","or","[","<","D",">",",","<","B",">",",","<","C",">","]",",","etc","..","but","always","<","D",">","as","it","is","the","newest","sample","..","testcode",":",":",":","skipif",":","True","#","replay","proportion","0.0","-",">","replay","disabled",":","buffer","=","MixInReplay","(","capacity=100",",","replay_ratio=0.0",")","buffer.add","(","<","A",">",")","buffer.sample","(",")","..","testoutput",":",":","[","<","A",">","]","..","testcode",":",":",":","skipif",":","True","buffer.add","(","<","B",">",")","buffer.sample","(",")","..","testoutput",":",":","[","<","B",">","]"]
docstr: Samples a batch of size `num_items` from a specified buffer.
  
  Concatenates old samples to new ones according to
  self.replay_ratio. If not enough new samples are available, mixes in
  less old samples to retain self.replay_ratio on average. Returns
  an empty batch if there are no items in the buffer.
  
  Args:
      num_items: Number of items to sample from this buffer.
      policy_id: ID of the policy that produced the experiences to be
      sampled.
      **kwargs: Forward compatibility kwargs.
  
  Returns:
      Concatenated MultiAgentBatch of items.
docstr_tok: ["Samples","a","batch","of","size","`","num_items","`","from","a","specified","buffer",".","Concatenates","old","samples","to","new","ones","according","to","self.replay_ratio",".","If","not","enough","new","samples","are","available",",","mixes","in","less","old","samples","to","retain","self.replay_ratio","on","average",".","Returns","an","empty","batch","if","there","are","no","items","in","the","buffer",".","Args",":","num_items",":","Number","of","items","to","sample","from","this","buffer",".","policy_id",":","ID","of","the","policy","that","produced","the","experiences","to","be","sampled",".","*","*","kwargs",":","Forward","compatibility","kwargs",".","Returns",":","Concatenated","MultiAgentBatch","of","items","."]
code: def sample(
      self,
      num_items: int,
      policy_id: PolicyID = DEFAULT_POLICY_ID,
      **kwargs,
  ) -> Optional[SampleBatchType]:
      """Samples a batch of size `num_items` from a specified buffer.
  
      Concatenates old samples to new ones according to
      self.replay_ratio. If not enough new samples are available, mixes in
      less old samples to retain self.replay_ratio on average. Returns
      an empty batch if there are no items in the buffer.
  
      Args:
          num_items: Number of items to sample from this buffer.
          policy_id: ID of the policy that produced the experiences to be
          sampled.
          **kwargs: Forward compatibility kwargs.
  
      Returns:
          Concatenated MultiAgentBatch of items.
      """
      # Merge kwargs, overwriting standard call arguments
      kwargs = merge_dicts_with_warning(
          self.underlying_buffer_call_args, kwargs
      )
  
      def mix_batches(_policy_id):
          """Mixes old with new samples.
  
          Tries to mix according to self.replay_ratio on average.
          If not enough new samples are available, mixes in less old samples
          to retain self.replay_ratio on average.
          """
  
          def round_up_or_down(value, ratio):
              """Returns an integer averaging to value*ratio."""
              product = value * ratio
              ceil_prob = product % 1
              if random.uniform(0, 1) < ceil_prob:
                  return int(np.ceil(product))
              else:
                  return int(np.floor(product))
  
          max_num_new = round_up_or_down(
              num_items, 1 - self.replay_ratio
          )
          # if num_samples * self.replay_ratio is not round,
          # we need one more sample with a probability of
          # (num_items*self.replay_ratio) % 1
  
          _buffer = self.replay_buffers[_policy_id]
          output_batches = self.last_added_batches[_policy_id][
              :max_num_new
          ]
          self.last_added_batches[_policy_id] = self.last_added_batches[
              _policy_id
          ][max_num_new:]
  
          # No replay desired
          if self.replay_ratio == 0.0:
              return concat_samples_into_ma_batch(output_batches)
          # Only replay desired
          elif self.replay_ratio == 1.0:
              return _buffer.sample(num_items, **kwargs)
  
          num_new = len(output_batches)
  
          if np.isclose(num_new, num_items * (1 - self.replay_ratio)):
              # The optimal case, we can mix in a round number of old
              # samples on average
              num_old = num_items - max_num_new
          else:
              # We never want to return more elements than num_items
              num_old = min(
                  num_items - max_num_new,
                  round_up_or_down(
                      num_new,
                      self.replay_ratio / (1 - self.replay_ratio),
                  ),
              )
  
          output_batches.append(_buffer.sample(num_old, **kwargs))
          # Depending on the implementation of underlying buffers, samples
          # might be SampleBatches
          output_batches = [
              batch.as_multi_agent() for batch in output_batches
          ]
          return concat_samples_into_ma_batch(output_batches)
  
      def check_buffer_is_ready(_policy_id):
          if (
              (len(self.replay_buffers[policy_id]) == 0)
              and self.replay_ratio > 0.0
          ) or (
              len(self.last_added_batches[_policy_id]) == 0
              and self.replay_ratio < 1.0
          ):
              return False
          return True
  
      with self.replay_timer:
          samples = []
  
          if self.replay_mode == ReplayMode.LOCKSTEP:
              assert (
                  policy_id is None
              ), "`policy_id` specifier not allowed in `lockstep` mode!"
              if check_buffer_is_ready(_ALL_POLICIES):
                  samples.append(
                      mix_batches(_ALL_POLICIES).as_multi_agent()
                  )
          elif policy_id is not None:
              if check_buffer_is_ready(policy_id):
                  samples.append(
                      mix_batches(policy_id).as_multi_agent()
                  )
          else:
              for (
                  policy_id,
                  replay_buffer,
              ) in self.replay_buffers.items():
                  if check_buffer_is_ready(policy_id):
                      samples.append(
                          mix_batches(policy_id).as_multi_agent()
                      )
  
          return concat_samples_into_ma_batch(samples)
  
imports: import collections
  import logging
  import random
  from typing import Any, Dict, Optional
  import numpy
  from ray.rllib.policy.rnn_sequencing import timeslice_along_seq_lens_with_overlap
  from ray.rllib.policy.sample_batch import DEFAULT_POLICY_ID, SampleBatch, concat_samples_into_ma_batch
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.replay_buffers.multi_agent_prioritized_replay_buffer import MultiAgentPrioritizedReplayBuffer
  from ray.rllib.utils.replay_buffers.multi_agent_replay_buffer import MultiAgentReplayBuffer, ReplayMode, merge_dicts_with_warning
  from ray.rllib.utils.replay_buffers.replay_buffer import _ALL_POLICIES, StorageUnit
  from ray.rllib.utils.typing import PolicyID, SampleBatchType
  from ray.util.annotations import DeveloperAPI
  from ray.util.debug import log_once
label_desc: Use image node_id 3 for calling the MultiAgentMixInReplayBuffer obj's underlying member method code with example usage: obj.sample(num_items, policy_id) and returns: concat_samples_into_ma_batch, True, concat_samples_into_ma_batch, concat_samples_into_ma_batch, False, int, int, _buffer