class_docstr: Automatically trains tokenizers and preprocesses machine translation data based on the MTEncDecModelConfig.
  For training NMT models with datasets larger than 5M sentence pairs, 
  it can be inefficient to train them without first creating a tarred dataset. 
  If the user wants to change the tokenizer, vocab size, or batch size, for example, 
  they must reprocess the data with the correct configuration. 
  With MTDataPreproc users can sweep through data configurations and the tarred dataset will 
  be automatically created according to the model configuration.
  To train tokenizer model and create tarred dataset specify in configuration:
      model.preproc_out_dir=/path/to/preproc_out
      model.encoder_tokenizer.vocab_size=32000
      model.decoder_tokenizer.vocab_size=32000 
      model.train_ds.use_tarred_dataset=True 
      model.train_ds.src_file_name=/path/to/src.txt
      model.train_ds.tgt_file_name=/path/to/tgt.txt
      model.train_ds.tokens_in_batch=16000 
  Once a dataset has been constructed based on this configuration, MTDataPreproc will not process it again.
  If a previously trained tokenizer model or tarred dataset is found, MTDataPreproc will not preprocess the data.
  
  Note: the only tokenizer currently supported is YouTokenToMe.
code: def __init__(
      self, cfg: MTEncDecModelConfig, trainer: Trainer = None
  ) -> None:
      self._cfg = cfg
      self.global_rank = 0
      self.world_size = 1
      if trainer is not None:
          self.global_rank = (
              trainer.node_rank * trainer.num_devices
          ) + trainer.local_rank
          self.world_size = trainer.num_nodes * trainer.num_devices
  
      if hasattr(cfg, "train_ds"):
          supported_tokenizers = [
              "yttm",
              "huggingface",
              "sentencepiece",
              "megatron",
              "byte-level",
          ]
          supported_multilingual_tokenizers = [
              "sentencepiece",
              "byte-level",
          ]
          supported_train_tokenizers = ["yttm", "sentencepiece"]
  
          if (
              cfg.encoder_tokenizer.get("library")
              not in supported_tokenizers
              or cfg.decoder_tokenizer.get("library")
              not in supported_tokenizers
          ):
              raise NotImplementedError(
                  f"Currently we only support {supported_tokenizers}."
              )
  
          if cfg.get("multilingual") and (
              cfg.encoder_tokenizer.get("library")
              not in supported_multilingual_tokenizers
              or cfg.decoder_tokenizer.get("library")
              not in supported_multilingual_tokenizers
          ):
              raise NotImplementedError(
                  f"Currently we only support {supported_multilingual_tokenizers} for multilingual models."
              )
  
          if cfg.get("shared_tokenizer") and cfg.encoder_tokenizer.get(
              "library"
          ) != cfg.decoder_tokenizer.get("library"):
              raise ValueError(
                  "Shared tokenizers cannot be from different libraries."
              )
  
          # Prepare tokenizers
          if (
              cfg.encoder_tokenizer.get("library")
              in supported_train_tokenizers
              or cfg.decoder_tokenizer.get("library")
              in supported_train_tokenizers
          ):
              # Train tokenizer models if using yttm or sentencepiece and they don't exist
              if (
                  cfg.encoder_tokenizer.get("library")
                  in supported_train_tokenizers
                  and cfg.encoder_tokenizer.get("tokenizer_model")
                  is None
              ) or (
                  cfg.decoder_tokenizer.get("library")
                  in supported_train_tokenizers
                  and cfg.decoder_tokenizer.get("tokenizer_model")
                  is None
              ):
                  if cfg.get("preproc_out_dir") is None:
                      raise ValueError(
                          "Tokenizer model training required but cfg.preproc_out_dir is None."
                      )
                  if (
                      cfg.train_ds.get("src_file_name") is None
                      or cfg.train_ds.get("tgt_file_name") is None
                  ):
                      raise ValueError(
                          "src_file_name and tgt_file_name needed to train tokenizers but could not be found."
                      )
  
                  src_fname = cfg.train_ds.get("src_file_name")
                  tgt_fname = cfg.train_ds.get("tgt_file_name")
                  src_language = cfg.get("src_language")
                  tgt_language = cfg.get("tgt_language")
                  spt_symbols = None
                  tempdir = tempfile.TemporaryDirectory()
  
                  if cfg.get("multilingual"):
                      spt_symbols = []
                      if isinstance(src_fname, ListConfig):
                          fnames = (" ").join(src_fname)
                          src_fname = os.path.join(
                              tempdir.name, "src.txt"
                          )
                          os.system("cat %s > %s" % (fnames, src_fname))
  
                      if isinstance(tgt_fname, ListConfig):
                          fnames = (" ").join(tgt_fname)
                          tgt_fname = os.path.join(
                              tempdir.name, "tgt.txt"
                          )
                          os.system("cat %s > %s" % (fnames, tgt_fname))
  
                      if isinstance(src_language, ListConfig):
                          for lng in src_language:
                              spt_symbols.append("<" + lng + ">")
  
                      if isinstance(tgt_language, ListConfig):
                          for lng in tgt_language:
                              spt_symbols.append("<" + lng + ">")
  
                  # train tokenizer model on training data
                  (
                      self.encoder_tokenizer_model,
                      self.decoder_tokenizer_model,
                  ) = MTDataPreproc.train_tokenizers(
                      out_dir=cfg.get("preproc_out_dir"),
                      src_fname=src_fname,
                      tgt_fname=tgt_fname,
                      shared_tokenizer=cfg.get("shared_tokenizer"),
                      encoder_tokenizer_vocab_size=cfg.encoder_tokenizer.get(
                          "vocab_size"
                      ),
                      decoder_tokenizer_vocab_size=cfg.decoder_tokenizer.get(
                          "vocab_size"
                      ),
                      encoder_tokenizer_name=cfg.encoder_tokenizer.get(
                          "library"
                      ),
                      decoder_tokenizer_name=cfg.decoder_tokenizer.get(
                          "library"
                      ),
                      encoder_tokenizer_coverage=cfg.encoder_tokenizer.get(
                          "coverage", 0.999
                      ),
                      decoder_tokenizer_coverage=cfg.decoder_tokenizer.get(
                          "coverage", 0.999
                      ),
                      global_rank=self.global_rank,
                      encoder_training_sample_size=cfg.encoder_tokenizer.get(
                          "training_sample_size", -1
                      ),
                      decoder_training_sample_size=cfg.decoder_tokenizer.get(
                          "training_sample_size", -1
                      ),
                      encoder_special_tokens=OmegaConf.to_container(
                          cfg.encoder_tokenizer.special_tokens
                      )
                      if cfg.encoder_tokenizer.special_tokens
                      else None,
                      decoder_special_tokens=OmegaConf.to_container(
                          cfg.decoder_tokenizer.special_tokens
                      )
                      if cfg.decoder_tokenizer.special_tokens
                      else None,
                      spt_symbols=spt_symbols,
                  )
                  # update config
                  self._cfg.encoder_tokenizer.tokenizer_model = (
                      self.encoder_tokenizer_model
                  )
                  self._cfg.decoder_tokenizer.tokenizer_model = (
                      self.decoder_tokenizer_model
                  )
  
                  tempdir.cleanup()
              else:
                  self.encoder_tokenizer_model = (
                      cfg.encoder_tokenizer.get("tokenizer_model")
                  )
                  self.decoder_tokenizer_model = (
                      cfg.decoder_tokenizer.get("tokenizer_model")
                  )
  
          (
              self.encoder_tokenizer,
              self.decoder_tokenizer,
          ) = self.get_enc_dec_tokenizers(
              encoder_tokenizer_name=cfg.encoder_tokenizer.get(
                  "library"
              ),
              encoder_model_name=cfg.encoder.get("model_name"),
              encoder_tokenizer_model=getattr(
                  self, "encoder_tokenizer_model", None
              ),
              encoder_bpe_dropout=cfg.encoder_tokenizer.get(
                  "bpe_dropout", 0.0
              ),
              encoder_r2l=cfg.encoder_tokenizer.get("r2l", False),
              decoder_tokenizer_name=cfg.decoder_tokenizer.get(
                  "library"
              ),
              decoder_model_name=cfg.decoder.get("model_name"),
              decoder_tokenizer_model=getattr(
                  self, "decoder_tokenizer_model", None
              ),
              decoder_bpe_dropout=cfg.decoder_tokenizer.get(
                  "bpe_dropout", 0.0
              ),
              decoder_r2l=cfg.decoder_tokenizer.get("r2l", False),
              encoder_tokenizer_legacy=cfg.encoder_tokenizer.get(
                  "sentencepiece_legacy", False
              ),
              decoder_tokenizer_legacy=cfg.decoder_tokenizer.get(
                  "sentencepiece_legacy", False
              ),
          )
  
          # If using tarred dataset for training, automatically create it if needed
          if cfg.train_ds.get("use_tarred_dataset"):
              if (
                  cfg.train_ds.get("tar_files") is None
                  and cfg.train_ds.get("metadata_file") is None
              ):
                  if cfg.get("preproc_out_dir") is None:
                      raise ValueError(
                          "Data preprocessing required but cfg.preproc_out_dir is None."
                      )
                  if (
                      cfg.train_ds.get("src_file_name") is None
                      or cfg.train_ds.get("tgt_file_name") is None
                  ):
                      raise ValueError(
                          "src_file_name and tgt_file_name needed to create tarred dataset but could not be found."
                      )
                  # Preprocess data and cache for use during training
                  if self.global_rank == 0:
                      logging.info(
                          f"Creating tarred dataset for src: {cfg.train_ds.get('src_file_name')} and tgt: {cfg.train_ds.get('tgt_file_name')}"
                      )
  
                  if isinstance(cfg.train_ds.get("src_file_name"), str):
                      src_file_list = [
                          cfg.train_ds.get("src_file_name")
                      ]
                      tgt_file_list = [
                          cfg.train_ds.get("tgt_file_name")
                      ]
                      outdir_list = [cfg.get("preproc_out_dir")]
                  else:
                      src_file_list = cfg.train_ds.get("src_file_name")
                      tgt_file_list = cfg.train_ds.get("tgt_file_name")
                      if isinstance(
                          cfg.get("src_language"), ListConfig
                      ):
                          langs = cfg.get("src_language")
                      elif isinstance(
                          cfg.get("tgt_language"), ListConfig
                      ):
                          langs = cfg.get("tgt_language")
                      outdir_list = []
                      for lang in langs:
                          outdir_list.append(
                              os.path.join(
                                  cfg.get("preproc_out_dir"), lang
                              )
                          )
  
                  if len(src_file_list) != len(tgt_file_list) or len(
                      src_file_list
                  ) != len(outdir_list):
                      raise ValueError(
                          "Number of source files, target files, and multilingual language pairs must be the same."
                      )
  
                  # TODO: have to get tokenizers instide .preprocess_parallel because they can't be pickled
                  metadata_file_list = []
                  for idx, src_file in enumerate(src_file_list):
                      (
                          self.train_tar_files,
                          self.train_metadata_file,
                      ) = MTDataPreproc.preprocess_parallel_dataset(
                          clean=cfg.train_ds.clean,
                          src_fname=src_file,
                          tgt_fname=tgt_file_list[idx],
                          out_dir=outdir_list[idx],
                          encoder_tokenizer_name=cfg.encoder_tokenizer.get(
                              "library"
                          ),
                          encoder_model_name=cfg.encoder.get(
                              "model_name"
                          ),
                          encoder_tokenizer_model=getattr(
                              self, "encoder_tokenizer_model", None
                          ),
                          encoder_bpe_dropout=cfg.encoder_tokenizer.get(
                              "bpe_dropout", 0.0
                          ),
                          encoder_tokenizer_r2l=cfg.encoder_tokenizer.get(
                              "r2l", False
                          ),
                          encoder_tokenizer_legacy=cfg.encoder_tokenizer.get(
                              "sentencepiece_legacy", False
                          ),
                          decoder_tokenizer_name=cfg.decoder_tokenizer.get(
                              "library"
                          ),
                          decoder_model_name=cfg.decoder.get(
                              "model_name"
                          ),
                          decoder_tokenizer_model=getattr(
                              self, "decoder_tokenizer_model", None
                          ),
                          decoder_bpe_dropout=cfg.decoder_tokenizer.get(
                              "bpe_dropout", 0.0
                          ),
                          decoder_tokenizer_r2l=cfg.decoder_tokenizer.get(
                              "r2l", False
                          ),
                          decoder_tokenizer_legacy=cfg.decoder_tokenizer.get(
                              "sentencepiece_legacy", False
                          ),
                          max_seq_length=cfg.train_ds.get(
                              "max_seq_length", 512
                          ),
                          tokens_in_batch=cfg.train_ds.get(
                              "tokens_in_batch", 8192
                          ),
                          lines_per_dataset_fragment=cfg.train_ds.get(
                              "lines_per_dataset_fragment", 1000000
                          ),
                          num_batches_per_tarfile=cfg.train_ds.get(
                              "num_batches_per_tarfile", 1000
                          ),
                          min_seq_length=1,
                          global_rank=self.global_rank,
                          world_size=self.world_size,
                          n_jobs=cfg.train_ds.get("n_preproc_jobs", -2),
                          tar_file_prefix=cfg.train_ds.get(
                              "tar_file_prefix", "parallel"
                          ),
                      )
                      metadata_file_list.append(
                          self.train_metadata_file
                      )
                  # update config
                  # self._cfg.train_ds.tar_files = self.tar_files_to_string(self.train_tar_files)
                  # self._cfg.train_ds.tar_files = self.train_tar_files
                  if isinstance(cfg.train_ds.get("metadata_file"), str):
                      self._cfg.train_ds.metadata_file = (
                          metadata_file_list[0]
                      )
                  else:
                      self._cfg.train_ds.metadata_file = (
                          metadata_file_list
                      )
  
                  logging.info(
                      f"Using tarred dataset created in folder(s) {outdir_list} and metadata created at {self._cfg.train_ds.metadata_file}"
                  )
  
              elif (
                  cfg.train_ds.get("tar_files") is not None
                  and cfg.train_ds.get("metadata_file") is None
              ):
                  raise ValueError(
                      "A metadata file is required for tarred dataset but cfg.metadata_file is None."
                  )
              elif (
                  cfg.train_ds.get("tar_files") is None
                  and cfg.train_ds.get("metadata_file") is not None
              ):
                  if isinstance(cfg.train_ds.get("metadata_file"), str):
                      metadata_file_list = [
                          cfg.train_ds.get("metadata_file")
                      ]
                  else:
                      metadata_file_list = cfg.train_ds.get(
                          "metadata_file"
                      )
  
                  for metadata_file in metadata_file_list:
                      with open(metadata_file) as metadata_reader:
                          metadata = json.load(metadata_reader)
                      if metadata["tar_files"]:
                          logging.info(
                              f"Using tarred dataset: {metadata['tar_files']}"
                          )
                      else:
                          raise ValueError(
                              f"tar_files not provided and metadata does not have tar files"
                          )
              else:
                  self.train_tar_files = cfg.train_ds.get("tar_files")
                  self.train_metadata_file = cfg.train_ds.get(
                      "metadata_file"
                  )
                  logging.info(
                      f"Using tarred dataset from config at {self.train_tar_files} and metadata from {self.train_metadata_file}"
                  )
  
imports: import glob
  import json
  import os
  import pickle
  import tarfile
  import tempfile
  import youtokentome
  from joblib import Parallel, delayed
  from omegaconf import ListConfig, OmegaConf
  from pytorch_lightning import Trainer
  from nemo.collections.common.tokenizers.sentencepiece_tokenizer import SentencePieceTokenizer, create_spt_model
  from nemo.collections.nlp.data.language_modeling.sentence_dataset import SentenceDataset
  from nemo.collections.nlp.data.machine_translation.machine_translation_dataset import TranslationDataset
  from nemo.collections.nlp.models.machine_translation.mt_enc_dec_config import MTEncDecModelConfig
  from nemo.collections.nlp.modules.common.tokenizer_utils import get_nmt_tokenizer, get_tokenizer
  from nemo.utils import logging
label_desc: Use image node_id 1 to create a new MTDataPreproc object  with example: obj = MTDataPreproc(cfg, trainer)