class_docstr: Base class for loading text-to-text manifests
  Map-style and Iterable datasets should inherit this class
class_docstr_tok: ["Base","class","for","loading","text-to-text","manifests","Map-style","and","Iterable","datasets","should","inherit","this","class"]
code: def __init__(
      self,
      manifest_filepath: Union[AnyPath, List[AnyPath]],
      speakers_filepath: Union[AnyPath, List[AnyPath]],
      asr_tokenizer: TokenizerSpec,
      asr_use_start_end_token: bool,
      tts_parser: Callable,
      tts_text_pad_id: int,
      tts_text_normalizer: "Normalizer",
      tts_text_normalizer_call_kwargs: Dict,
      min_words: int = 1,
      max_words: int = 1_000_000,
      tokenizer_workers: int = 1,
      num_parts: int = 1,
      current_part_index: int = 0,
  ):
      super().__init__()
      # ASR tokenizer setup
      if asr_use_start_end_token and hasattr(
          asr_tokenizer, "bos_token"
      ):
          self.asr_bos_id = asr_tokenizer.bos_id
  
      if asr_use_start_end_token and hasattr(
          asr_tokenizer, "eos_token"
      ):
          self.asr_eos_id = asr_tokenizer.eos_id
  
      if hasattr(asr_tokenizer, "pad_token"):
          self.asr_pad_id = asr_tokenizer.pad_id
      else:
          self.asr_pad_id = 0
  
      self.asr_tokenizer = asr_tokenizer
  
      # TTS tokenizer setup
      self.tts_parser = tts_parser
      self.tts_normalizer = tts_text_normalizer
      self.tts_normalizer_kwargs = tts_text_normalizer_call_kwargs
      self.tts_text_pad_id = tts_text_pad_id
  
      # Load speakers
      if isinstance(speakers_filepath, str):
          speakers_filepath = speakers_filepath.split(",")
      elif isinstance(speakers_filepath, Path):
          speakers_filepath = [speakers_filepath]
      speakers: Set[int] = set()
      for filepath in speakers_filepath:
          with open(Path(filepath).expanduser(), "r") as f:
              speakers.update(map(int, f.read().split()))
      self.speakers = np.asarray(sorted(speakers))
      logging.info(f"Loaded {len(self.speakers)} speakers")
  
      # Load manifest
      if isinstance(manifest_filepath, str):
          manifest_filepath = manifest_filepath.split(",")
      elif isinstance(manifest_filepath, Path):
          manifest_filepath = [manifest_filepath]
      self.manifest_paths = [
          Path(filepath) for filepath in manifest_filepath
      ]
  
      num_skipped_words = 0
      num_skipped_utterances = 0
      asr_texts = []
      tts_texts = []
      need_normalization = False
  
      for manifest_path in self.manifest_paths:
          for tmp_item in tqdm(_iterate_manifest(manifest_path)):
              text = tmp_item["text"]
              num_words = len(text.split())
              # skip if number of works not in desired range
              # TODO: maybe it would be valuable to sample sub-utterances from long utterances
              if not (min_words <= num_words <= max_words):
                  num_skipped_words += num_words
                  num_skipped_utterances += 1
                  continue
              asr_texts.append(tmp_item["text"])
              if "tts_text_normalized" in tmp_item:
                  tts_texts.append(tmp_item["tts_text_normalized"])
              else:
                  tts_texts.append(tmp_item["tts_text"])
                  need_normalization = True
  
      if need_normalization:
          logging.warning(
              "TTS normalization is extremely slow! It is recommended to normalize TTS text"
          )
  
      if num_skipped_utterances:
          logging.warning(
              f"Skipped {num_skipped_utterances} utterances "
              f"with {num_skipped_words}"
          )
  
      num_utterances = len(asr_texts)
      # preprocessing is very costly, if we need only part - remove unnecessary utterances
      if num_parts > 1:
          # NB: floor division, full dataset can contain fewer utterances than original, like in tarred dataset
          num_utterances_part = num_utterances // num_parts
          start = num_utterances_part * current_part_index
          end = start + num_utterances_part
          logging.info(
              f"Taking part of the dataset: {current_part_index} index, total {num_parts} from {start} to {end}"
          )
          asr_texts = asr_texts[start:end]
          tts_texts = tts_texts[start:end]
          num_utterances = num_utterances_part
  
      self.data = [dict() for _ in range(num_utterances)]
  
      if len(asr_texts) == 0:
          # no data was loaded
          logging.warning("Text-to-text dataset is empty")
          return
  
      if tokenizer_workers == 1:
          logging.warning(
              "Preprocessing large text with tokenizer_workers=1 may be slow with TTS tokenizer. "
              "Prefer tokenizer_workers=(num_cpu_cores/num_gpus_per_node)"
          )
          for i, tokenized_text in enumerate(
              tqdm(
                  (
                      self._asr_text_to_tokens(text)
                      for text in asr_texts
                  ),
                  total=len(asr_texts),
              )
          ):
              self.data[i]["asr_text_tokens"] = tokenized_text
      else:
          # Multiprocessing hack: use global variables for every process (not really global in program context)
          def _init_asr_tokenize_process(tokenizer, bos_id, eos_id):
              global asr_tokenizer_global, asr_bos_id_global, asr_eos_id_global  # process-global
              # deepcopy to avoid serialization of parent models
              asr_tokenizer_global = copy.deepcopy(tokenizer)
              asr_bos_id_global = copy.deepcopy(bos_id)
              asr_eos_id_global = copy.deepcopy(eos_id)
  
          with concurrent.futures.ProcessPoolExecutor(
              initializer=_init_asr_tokenize_process,
              initargs=(
                  asr_tokenizer,
                  self.asr_bos_id,
                  self.asr_eos_id,
              ),
              max_workers=tokenizer_workers,
          ) as pool:
              # chunk size for pool map is empirically chosen as a trade-off between speed and responsiveness
              for i, tokenized_text in enumerate(
                  tqdm(
                      pool.map(
                          _asr_text_to_tokens, asr_texts, chunksize=1000
                      ),
                      total=len(asr_texts),
                  )
              ):
                  self.data[i]["asr_text_tokens"] = tokenized_text
      # force free memory
      del asr_texts
      gc.collect()
  
      if tokenizer_workers == 1:
          logging.warning(
              "Preprocessing large text with tokenizer_workers=1 may be slow with TTS tokenizer. "
              "Prefer tokenizer_workers=(num_cpu_cores/num_gpus_per_node)"
          )
          for i, tokenized_text in enumerate(
              tqdm(
                  (
                      self._tts_text_to_tokens(
                          text, normalize=need_normalization
                      )
                      for text in tts_texts
                  ),
                  total=len(tts_texts),
              )
          ):
              self.data[i]["tts_text_tokens"] = tokenized_text
      else:
          if need_normalization:
              # TODO: implement, if we really need normalization inplace
              raise NotImplementedError(
                  "Normalization with tokenizer_workers > 1 is not implemented. "
                  "It is not recommended to use normalization on the fly at all, since it's extremely slow"
              )
  
          def _init_tts_tokenize_process(tokenizer):
              global tts_tokenizer_global  # process-global
              tts_tokenizer_global = copy.deepcopy(tokenizer)
  
          with concurrent.futures.ProcessPoolExecutor(
              initializer=_init_tts_tokenize_process,
              initargs=(tts_parser,),
              max_workers=tokenizer_workers,
          ) as pool:
              # chunk size for pool map is empirically chosen as a trade-off between speed and responsiveness
              for i, tokenized_text in enumerate(
                  tqdm(
                      pool.map(
                          _tts_text_to_tokens, tts_texts, chunksize=1000
                      ),
                      total=len(tts_texts),
                  )
              ):
                  self.data[i]["tts_text_tokens"] = tokenized_text
      # force free memory
      del tts_texts
      gc.collect()
  
imports: from __future__ import annotations
  import concurrent.futures
  import copy
  import gc
  import json
  import math
  import random
  from pathlib import Path
  from typing import Any, Callable, Dict, Iterable, List, NamedTuple, Optional, Set, Union
  import numpy
  import torch
  import torch.utils.data
  from torch.nn.utils.rnn import pad_sequence
  from tqdm.auto import tqdm
  from nemo.collections.asr.data.audio_to_text import _speech_collate_fn
  from nemo.collections.common.tokenizers import TokenizerSpec
  from nemo.core.classes import Dataset, IterableDataset
  from nemo.utils import logging
label_desc: Use image node_id 1 to create a new TextToTextDatasetBase object  with example: obj = TextToTextDatasetBase(manifest_filepath, speakers_filepath, asr_tokenizer, asr_use_start_end_token, tts_parser, tts_text_pad_id, tts_text_normalizer, tts_text_normalizer_call_kwargs, min_words, max_words, tokenizer_workers, num_parts, current_part_index)