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 build_iter(
      self,
      epoch: int,
      shuffle: Optional[bool] = None,
  ) -> Iterator[Tuple[List[str], Dict[str, torch.Tensor]]]:
      per_sample_loader = self.per_sample_iter_factory.build_iter(
          epoch, shuffle
      )
  
      if shuffle is None:
          shuffle = self.shuffle
      state = np.random.RandomState(epoch + self.seed)
  
      # NOTE(kamo):
      #   This iterator supports multiple chunk lengths and
      #   keep chunks for each lengths here until collecting specified numbers
      cache_chunks_dict = defaultdict(dict)
      cache_id_list_dict = defaultdict(dict)
      for ids, batch in per_sample_loader:
          # Must be per-sample-loader
          assert len(ids) == 1, f"Must be per-sample-loader: {len(ids)}"
          assert all(len(x) == 1 for x in batch.values())
  
          # Get keys of sequence data
          sequence_keys = []
          for key in batch:
              if key + "_lengths" in batch:
                  sequence_keys.append(key)
          # Remove lengths data and get the first sample
          batch = {
              k: v[0]
              for k, v in batch.items()
              if not k.endswith("_lengths")
          }
          id_ = ids[0]
  
          for key in sequence_keys:
              if (
                  self.excluded_key_pattern is not None
                  and re.fullmatch(self.excluded_key_pattern, key)
              ):
                  # ignore length inconsistency for `excluded_key_prefixes`
                  continue
              if len(batch[key]) != len(batch[sequence_keys[0]]):
                  raise RuntimeError(
                      f"All sequences must has same length: "
                      f"{len(batch[key])} != {len(batch[sequence_keys[0]])}"
                  )
  
          # Get sampling frequency of the batch to recalculate the chunk length
          fs = (
              batch.get("utt2fs", torch.LongTensor([16000]))
              .type(torch.int64)
              .item()
          )
          default_fs = (
              fs if self.default_fs is None else self.default_fs
          )
          assert fs % default_fs == 0 or default_fs % fs == 0
  
          L = len(batch[sequence_keys[0]])
          # Select chunk length
          chunk_lengths = [
              lg * fs // default_fs for lg in self.chunk_lengths
          ]
          chunk_lengths = [lg for lg in chunk_lengths if lg < L]
          if len(chunk_lengths) == 0:
              logging.warning(
                  f"The length of '{id_}' is {L}, but it is shorter than "
                  f"any candidates of chunk-length: {self.chunk_lengths}"
              )
              continue
  
          # Convert numpy array to number
          category = (
              batch.get("utt2category", torch.LongTensor([0]))
              .type(torch.int64)
              .item()
          )
  
          W = int(state.choice(chunk_lengths, 1))
          cache_id_list = cache_id_list_dict[category].setdefault(W, [])
          cache_chunks = cache_chunks_dict[category].setdefault(W, {})
  
          # Shift width to the next chunk
          S = int(W * self.chunk_shift_ratio)
          # Number of chunks
          N = (L - W) // S + 1
          if shuffle:
              Z = state.randint(0, (L - W) % S + 1)
          else:
              Z = 0
  
          # Split a sequence into chunks.
          # Note that the marginal frames divided by chunk length are discarded
          for k, v in batch.items():
              if k not in cache_chunks:
                  cache_chunks[k] = []
              if k in sequence_keys:
                  # Shift chunks with overlapped length for data augmentation
                  if (
                      self.excluded_key_pattern is not None
                      and re.fullmatch(self.excluded_key_pattern, k)
                  ):
                      for _ in range(N):
                          cache_chunks[k].append(v)
                  else:
                      cache_chunks[k] += [
                          v[Z + i * S : Z + i * S + W] for i in range(N)
                      ]
              else:
                  # If not sequence, use whole data instead of chunk
                  cache_chunks[k] += [v for _ in range(N)]
          cache_id_list += [id_ for _ in range(N)]
  
          if len(cache_id_list) > self.num_cache_chunks:
              (
                  cache_id_list,
                  cache_chunks,
              ) = yield from self._generate_mini_batches(
                  cache_id_list,
                  cache_chunks,
                  shuffle,
                  state,
              )
  
          cache_id_list_dict[category][W] = cache_id_list
          cache_chunks_dict[category][W] = cache_chunks
  
      else:
          for category in cache_id_list_dict.keys():
              for W in cache_id_list_dict[category]:
                  cache_id_list = cache_id_list_dict[
                      category
                  ].setdefault(W, [])
                  cache_chunks = cache_chunks_dict[category].setdefault(
                      W, {}
                  )
  
                  yield from self._generate_mini_batches(
                      cache_id_list,
                      cache_chunks,
                      shuffle,
                      state,
                  )
  
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 2 for calling the ChunkIterFactory obj's underlying member method code with example usage: obj.build_iter(epoch, shuffle) without return types