code: def dataio_prepare(hparams):
      """This function prepares the datasets to be used in the brain class.
      It also defines the data processing pipeline through user-defined functions.
      """
      codes_folder = pl.Path(hparams["codes_folder"])
  
      # Define audio pipeline. In this case, we simply read the audio contained
      # in the variable src_audio with the custom reader.
      @sb.utils.data_pipeline.takes("src_audio")
      @sb.utils.data_pipeline.provides("src_sig")
      def src_audio_pipeline(wav):
          """Load the source language audio signal.
          This is done on the CPU in the `collate_fn`
          """
          info = torchaudio.info(wav)
          sig = sb.dataio.dataio.read_audio(wav)
          sig = torchaudio.transforms.Resample(
              info.sample_rate,
              hparams["sample_rate"],
          )(sig)
          return sig
  
      @sb.utils.data_pipeline.takes("tgt_audio")
      @sb.utils.data_pipeline.provides("tgt_sig")
      def tgt_audio_pipeline(wav):
          """Load the target language audio signal.
          This is done on the CPU in the `collate_fn`.
          """
          info = torchaudio.info(wav)
          sig = sb.dataio.dataio.read_audio(wav)
          sig = torchaudio.transforms.Resample(
              info.sample_rate,
              hparams["sample_rate"],
          )(sig)
          return sig
  
      @sb.utils.data_pipeline.takes("id")
      @sb.utils.data_pipeline.provides("code_bos", "code_eos")
      def unit_pipeline(utt_id):
          """Load target codes"""
          code = np.load(codes_folder / f"{utt_id}_tgt.npy")
          code = torch.LongTensor(code)
          code = torch.unique_consecutive(code)
          code_bos = torch.cat(
              (torch.LongTensor([hparams["bos_index"]]), code)
          )
          yield code_bos
          code_eos = torch.cat(
              (code, torch.LongTensor([hparams["eos_index"]]))
          )
          yield code_eos
  
      datasets = {}
      for split in hparams["splits"]:
          datasets[
              split
          ] = sb.dataio.dataset.DynamicItemDataset.from_json(
              json_path=hparams[f"{split}_json"],
              dynamic_items=[
                  src_audio_pipeline,
                  tgt_audio_pipeline,
                  unit_pipeline,
              ],
              output_keys=[
                  "id",
                  "src_sig",
                  "tgt_sig",
                  "duration",
                  "code_bos",
                  "code_eos",
                  "tgt_text",
              ],
          )
  
      # 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":
          datasets["train"] = datasets["train"].filtered_sorted(
              sort_key="duration"
          )
          datasets["valid"] = datasets["valid"].filtered_sorted(
              sort_key="duration"
          )
  
          hparams["train_dataloader_opts"]["shuffle"] = False
          hparams["valid_dataloader_opts"]["shuffle"] = False
  
      elif hparams["sorting"] == "descending":
          datasets["train"] = datasets["train"].filtered_sorted(
              sort_key="duration", reverse=True
          )
          datasets["valid"] = datasets["valid"].filtered_sorted(
              sort_key="duration", reverse=True
          )
  
          hparams["train_dataloader_opts"]["shuffle"] = False
          hparams["valid_dataloader_opts"]["shuffle"] = False
  
      elif hparams["sorting"] == "random":
          hparams["train_dataloader_opts"]["shuffle"] = True
          hparams["valid_dataloader_opts"]["shuffle"] = False
  
      else:
          raise NotImplementedError(
              "sorting must be random, ascending or descending"
          )
  
      # Dynamic Batching is used, we instantiate the needed samplers.
      train_batch_sampler = None
      if hparams["dynamic_batching"]:
          from speechbrain.dataio.sampler import (
              DynamicBatchSampler,
          )  # noqa
  
          dynamic_hparams = hparams["dynamic_batch_sampler"]
          num_buckets = dynamic_hparams["num_buckets"]
  
          train_batch_sampler = DynamicBatchSampler(
              datasets["train"],
              dynamic_hparams["max_batch_len"],
              num_buckets=num_buckets,
              length_func=lambda x: x["duration"],
              shuffle=dynamic_hparams["shuffle_ex"],
              batch_ordering=dynamic_hparams["batch_ordering"],
              max_batch_ex=dynamic_hparams["max_batch_ex"],
          )
  
      return datasets, train_batch_sampler
  
imports: import sys
  import torch
  import logging
  import pathlib
  from hyperpyyaml import load_hyperpyyaml
  import speechbrain
  from speechbrain.inference.vocoders import UnitHIFIGAN
  from speechbrain.inference.ASR import EncoderDecoderASR
  import tqdm
  import torchaudio
  import numpy
  from torch.nn.parallel import DistributedDataParallel
label_desc: Use image node_id 1 for calling a global function with example usage: dataio_prepare(hparams) and returns: datasets,  train_batch_sampler, sig, sig