code: def dynamic_bucketed_batch(
      data,
      len_key=None,
      len_fn=len,
      min_sample_len=None,
      max_sample_len=None,
      buffersize=1024,
      collate_fn=PaddedBatch,
      sampler_fn=indices_around_random_pivot,
      sampler_kwargs={},
      drop_end=False,
  ):
      """Produce batches from a sorted buffer
  
      This function keeps a sorted buffer of the incoming samples.
      The samples can be filtered for min/max length.
      An external sampler is used to choose samples for each batch,
      which allows different dynamic batching algorithms to be used.
  
      Arguments
      ---------
      data : iterable
          An iterable source of samples, such as an IterableDataset.
      len_key : str, None
          The key in the sample dict to use to fetch the length of the sample, or
          None if no key should be used.
      len_fn : callable
          Called with sample[len_key] if len_key is not None, else sample. Needs
          to return the sample length as an integer.
      min_sample_len : int, None
          Discard samples with length lower than this. If None, no minimum is
          applied.
      max_sample_len : int, None
          Discard samples with length larger than this. If None, no maximum is
          applied.
      buffersize : int
          The size of the internal sorted buffer. The buffer is always filled up
          before yielding a batch of samples.
      collate_fn : callable
          Called with a list of samples. This should return a batch. By default, using
          the SpeechBrain PaddedBatch class, which works for dict-like samples, and
          pads any tensors.
      sampler_fn : callable
          Called with the sorted data buffer. Needs to return a list of indices, which
          make up the next batch. By default using ``indices_around_random_pivot``
      sampler_kwargs : dict
          Keyword arguments, passed to sampler_fn.
      drop_end : bool
          After the data stream is exhausted, should batches be made until the data
          buffer is exhausted, or should the rest of the buffer be discarded. Without
          new samples, the last batches might not be efficient to process.
          Note: you can use ``.repeat`` on `webdataset` IterableDatasets to never
          run out of new samples, and then use
          `speechbrain.dataio.dataloader.LoopedLoader` to set a nominal epoch length.
      """
      databuffer = []
      if sampler_kwargs:
          sampler_fn = partial(sampler_fn, **sampler_kwargs)
      for sample in data:
          # Length fetching interface has multiple valid call signatures:
          if len_key is not None and len_fn is not None:
              length = len_fn(sample[len_key])
          elif len_key is not None:
              length = sample[len_key]
          elif len_fn is not None:
              length = len_fn(sample)
          else:
              raise ValueError(
                  "Must specify at least one of len_key or len_fn"
              )
          # Possibly filter by length:
          if (
              min_sample_len is not None and length < min_sample_len
          ) or (max_sample_len is not None and length > max_sample_len):
              # Drop sample
              continue
          item = LengthItem(length, sample)
          # bisect.insort inserts in sorted order.
          # This should be a good way to maintain a sorted list,
          # but perhaps simply filling up the buffer and calling .sort()
          # could be good as well (Python's sort leverages already sorted segments)
          bisect.insort(databuffer, item)
          if len(databuffer) == buffersize:
              indices = sampler_fn(databuffer)
              batch_list = []
              # popping from highest to lowest is safe
              for i in sorted(indices, reverse=True):
                  item = databuffer.pop(i)
                  batch_list.append(item.data)
              yield collate_fn(batch_list)
      # Data stream was exhausted. Data buffer is relatively full at first,
      # but cannot be replenished, so batches might not be efficiently produced.
      # Either stop, or exhaust buffer.
      if drop_end:
          return
      while databuffer:
          indices = sampler_fn(databuffer)
          batch_list = []
          for i in sorted(indices, reverse=True):
              item = databuffer.pop(i)
              batch_list.append(item.data)
          yield collate_fn(batch_list)
  
file_path: speechbrain/speechbrain/dataio/iterators.py
imports: import bisect
  import random
  from dataclasses import dataclass, field
  from functools import partial
  from typing import Any
  from speechbrain.dataio.batch import PaddedBatch
label_desc: Use image node_id 4 for calling a global function with example usage: dynamic_bucketed_batch(data, len_key, len_fn, min_sample_len, max_sample_len, buffersize, collate_fn, sampler_fn, sampler_kwargs, drop_end) without return types