class_docstr: Creates chunks from a sequence
  
  Examples:
      >>> batches = [["id1"], ["id2"], ...]
      >>> batch_size = 128
      >>> chunk_length = 1000
      >>> iter_factory = ChunkIterFactory(dataset, batches, batch_size, chunk_length)
      >>> it = iter_factory.build_iter(epoch)
      >>> for ids, batch in it:
      ...     ...
  
  - The number of mini-batches are varied in each epochs and
    we can't get the number in advance
    because IterFactory doesn't be given to the length information.
  - Since the first reason, "num_iters_per_epoch" can't be implemented
    for this iterator. Instead of it, "num_samples_per_epoch" is implemented.
class_docstr_tok: ["Creates","chunks","from","a","sequence","Examples",":",">",">",">","batches","=","[","[","``","id1","''","]",",","[","``","id2","''","]",",","...","]",">",">",">","batch_size","=","128",">",">",">","chunk_length","=","1000",">",">",">","iter_factory","=","ChunkIterFactory","(","dataset",",","batches",",","batch_size",",","chunk_length",")",">",">",">","it","=","iter_factory.build_iter","(","epoch",")",">",">",">","for","ids",",","batch","in","it",":","...","...","-","The","number","of","mini-batches","are","varied","in","each","epochs","and","we","ca","n't","get","the","number","in","advance","because","IterFactory","does","n't","be","given","to","the","length","information",".","-","Since","the","first","reason",",","``","num_iters_per_epoch","''","ca","n't","be","implemented","for","this","iterator",".","Instead","of","it",",","``","num_samples_per_epoch","''","is","implemented","."]
code: def __init__(
      self,
      dataset,
      batch_size: int,
      batches: Union[AbsSampler, Sequence[Sequence[Any]]],
      chunk_length: Union[int, str],
      chunk_shift_ratio: float = 0.5,
      num_cache_chunks: int = 1024,
      num_samples_per_epoch: Optional[int] = None,
      seed: int = 0,
      shuffle: bool = False,
      num_workers: int = 0,
      collate_fn=None,
      pin_memory: bool = False,
      excluded_key_prefixes: Optional[List[str]] = None,
      default_fs: Optional[int] = None,
  ):
      assert check_argument_types()
      assert all(len(x) == 1 for x in batches), "batch-size must be 1"
  
      self.per_sample_iter_factory = SequenceIterFactory(
          dataset=dataset,
          batches=batches,
          num_iters_per_epoch=num_samples_per_epoch,
          seed=seed,
          shuffle=shuffle,
          num_workers=num_workers,
          collate_fn=collate_fn,
          pin_memory=pin_memory,
      )
  
      self.num_cache_chunks = max(num_cache_chunks, batch_size)
      if isinstance(chunk_length, str):
          if len(chunk_length) == 0:
              raise ValueError("e.g. 5,8 or 3-5: but got empty string")
  
          self.chunk_lengths = []
          for x in chunk_length.split(","):
              try:
                  sps = list(map(int, x.split("-")))
              except ValueError:
                  raise ValueError(
                      f"e.g. 5,8 or 3-5: but got {chunk_length}"
                  )
  
              if len(sps) > 2:
                  raise ValueError(
                      f"e.g. 5,8 or 3-5: but got {chunk_length}"
                  )
              elif len(sps) == 2:
                  # Append all numbers between the range into the candidates
                  self.chunk_lengths += list(range(sps[0], sps[1] + 1))
              else:
                  self.chunk_lengths += [sps[0]]
      else:
          # Single candidates: Fixed chunk length
          self.chunk_lengths = [chunk_length]
  
      self.chunk_shift_ratio = chunk_shift_ratio
      self.batch_size = batch_size
      self.seed = seed
      self.shuffle = shuffle
      # Default sampling frequency used to decide the chunk length
      # in case that different batches have different sampling frequencies
      # (If None, the chunk length is always fixed)
      self.default_fs = default_fs
  
      # keys that satisfy either condition below will be excluded from the length
      # consistency check:
      #  - exactly match one of the prefixes in `excluded_key_prefixes`
      #  - have one of the prefixes in `excluded_key_prefixes` and end with numbers
      if excluded_key_prefixes is None:
          excluded_key_prefixes = DEFAULT_EXCLUDED_KEY_PREFIXES
      else:
          for k in DEFAULT_EXCLUDED_KEY_PREFIXES:
              if k not in excluded_key_prefixes:
                  excluded_key_prefixes.append(k)
      self.excluded_key_pattern = (
          "(" + "[0-9]*)|(".join(excluded_key_prefixes) + "[0-9]*)"
      )
      if self.excluded_key_pattern:
          logging.info(
              f"Data keys with the following patterns will be excluded from the "
              f"length consistency check:\n{self.excluded_key_pattern}"
          )
  
file_path: espnet/espnet2/iterators/chunk_iter_factory.py
imports: import logging
  import re
  from collections import defaultdict
  from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple, Union
  import numpy
  import torch
  from typeguard import check_argument_types
  from espnet2.iterators.abs_iter_factory import AbsIterFactory
  from espnet2.iterators.sequence_iter_factory import SequenceIterFactory
  from espnet2.samplers.abs_sampler import AbsSampler
label_desc: Use image node_id 1 to create a new ChunkIterFactory object from inherited base classes: AbsIterFactory with example: obj = ChunkIterFactory(dataset, batch_size, batches, chunk_length, chunk_shift_ratio, num_cache_chunks, num_samples_per_epoch, seed, shuffle, num_workers, collate_fn, pin_memory, excluded_key_prefixes, default_fs)