code: def dataio_prepare(hparams, tokenizer):
      """This function prepares the datasets to be used in the brain class.
      It also defines the data processing pipeline through user-defined functions.
      """
  
      # Define audio pipeline. In this case, we simply read the path contained
      # in the variable wav with the audio reader.
      @sb.utils.data_pipeline.takes("path")
      @sb.utils.data_pipeline.provides("sig")
      def audio_pipeline(wav):
          """Load the audio signal. This is done on the CPU in the `collate_fn`."""
          sig = sb.dataio.dataio.read_audio(wav)
          return sig
  
      @sb.utils.data_pipeline.takes("path")
      @sb.utils.data_pipeline.provides("sig")
      def sp_audio_pipeline(wav):
          """Load the audio signal. This is done on the CPU in the `collate_fn`."""
          sig = sb.dataio.dataio.read_audio(wav)
          sig = sig.unsqueeze(0)
          sig = hparams["speed_perturb"](sig)
          sig = sig.squeeze(0)
          return sig
  
      # Define text processing pipeline. We start from the raw text and then
      # encode it using the tokenizer. The tokens with BOS are used for feeding
      # decoder during training, the tokens with EOS for computing the cost function.
      @sb.utils.data_pipeline.takes("trans")
      @sb.utils.data_pipeline.provides(
          "trans",
          "tokens_list",
          "tokens_bos",
          "tokens_eos",
      )
      def reference_text_pipeline(translation):
          """Processes the transcriptions to generate proper labels"""
          yield translation
          labels = tokenizer(
              text_target=translation.replace("\n", ""),
              return_tensors="pt",
          )
          tokens_list = labels["input_ids"].tolist()[-1]
          yield tokens_list
          tokens_bos = torch.LongTensor(tokens_list[0:-1])
          yield tokens_bos
          tokens_eos = torch.LongTensor(tokens_list[1:])
          yield tokens_eos
  
      datasets = {}
      data_folder = hparams["data_folder"]
      for dataset in ["train", "valid"]:
          json_path = hparams[f"annotation_{dataset}"]
  
          is_use_sp = dataset == "train" and "speed_perturb" in hparams
          audio_pipeline_func = (
              sp_audio_pipeline if is_use_sp else audio_pipeline
          )
  
          datasets[
              dataset
          ] = sb.dataio.dataset.DynamicItemDataset.from_json(
              json_path=json_path,
              replacements={"data_root": data_folder},
              dynamic_items=[
                  audio_pipeline_func,
                  reference_text_pipeline,
              ],
              output_keys=[
                  "id",
                  "sig",
                  "duration",
                  "trans",
                  "tokens_list",
                  "tokens_bos",
                  "tokens_eos",
              ],
          )
  
      for dataset in ["test"]:
          json_path = hparams[f"annotation_{dataset}"]
          datasets[
              dataset
          ] = sb.dataio.dataset.DynamicItemDataset.from_json(
              json_path=json_path,
              replacements={"data_root": data_folder},
              dynamic_items=[audio_pipeline, reference_text_pipeline],
              output_keys=[
                  "id",
                  "sig",
                  "duration",
                  "trans",
                  "tokens_list",
                  "tokens_bos",
                  "tokens_eos",
              ],
          )
  
      # Sorting training data with ascending order makes the code  much
      # faster  because we minimize zero-padding. In most of the cases, this
      # does not harm the performance.
      if hparams["sorting"] == "ascending":
          # use smaller dataset to debug the model
          if hparams["debug"]:
              datasets["train"] = datasets["train"].filtered_sorted(
                  key_min_value={"duration": 1},
                  key_max_value={"duration": 3},
                  sort_key="duration",
                  reverse=True,
              )
              datasets["valid"] = datasets["valid"].filtered_sorted(
                  key_min_value={"duration": 1},
                  key_max_value={"duration": 3},
                  sort_key="duration",
                  reverse=True,
              )
          else:
              datasets["train"] = datasets["train"].filtered_sorted(
                  sort_key="duration"
              )
              datasets["valid"] = datasets["valid"].filtered_sorted(
                  sort_key="duration"
              )
  
          hparams["dataloader_options"]["shuffle"] = False
          hparams["dataloader_options"]["shuffle"] = False
      elif hparams["sorting"] == "descending":
          # use smaller dataset to debug the model
          if hparams["debug"]:
              datasets["train"] = datasets["train"].filtered_sorted(
                  key_min_value={"duration": 1},
                  key_max_value={"duration": 3},
                  sort_key="duration",
                  reverse=True,
              )
              datasets["valid"] = datasets["valid"].filtered_sorted(
                  key_min_value={"duration": 1},
                  key_max_value={"duration": 3},
                  sort_key="duration",
                  reverse=True,
              )
          else:
              datasets["train"] = datasets["train"].filtered_sorted(
                  sort_key="duration", reverse=True
              )
              datasets["valid"] = datasets["valid"].filtered_sorted(
                  sort_key="duration", reverse=True
              )
  
          hparams["dataloader_options"]["shuffle"] = False
          hparams["dataloader_options"]["shuffle"] = False
      elif hparams["sorting"] == "random":
          # use smaller dataset to debug the model
          if hparams["debug"]:
              datasets["train"] = datasets["train"].filtered_sorted(
                  key_min_value={"duration": 1},
                  key_max_value={"duration": 3},
                  sort_key="duration",
              )
              datasets["valid"] = datasets["valid"].filtered_sorted(
                  key_min_value={"duration": 1},
                  key_max_value={"duration": 3},
              )
  
          hparams["dataloader_options"]["shuffle"] = True
      else:
          raise NotImplementedError(
              "sorting must be random, ascending or descending"
          )
  
      return datasets
  
file_path: speechbrain/recipes/IWSLT22_lowresource/AST/transformer/train_with_w2v_mbart.py
imports: import sys
  import torch
  import logging
  import speechbrain
  from speechbrain.utils.distributed import run_on_main
  from hyperpyyaml import load_hyperpyyaml
  from sacremoses import MosesDetokenizer
  from torch.nn.parallel import DistributedDataParallel
label_desc: Use image node_id 1 for calling a global function with example usage: dataio_prepare(hparams, tokenizer) and returns: datasets, sig, sig