code: def dataio_prep(hparams):
      """Creates data processing pipeline"""
      speech_dirs = [
          "read_speech",
          "german_speech",
          "french_speech",
          "italian_speech",
          "spanish_speech",
          "russian_speech",
      ]
      audio_length = hparams["audio_length"]
  
      train_shard_patterns = []
      for dir in speech_dirs:
          if not os.path.exists(
              os.path.join(hparams["train_data"], dir)
          ):
              dir = ""
          shard_pattern = os.path.join(
              hparams["train_data"], dir, "shard-*.tar"
          )
          shard_files = glob.glob(shard_pattern)
          train_shard_patterns.extend(shard_files)
  
      valid_shard_patterns = []
      for dir in speech_dirs:
          if not os.path.exists(
              os.path.join(hparams["valid_data"], dir)
          ):
              dir = ""
          shard_pattern = os.path.join(
              hparams["valid_data"], dir, "shard-*.tar"
          )
          shard_files = glob.glob(shard_pattern)
          valid_shard_patterns.extend(shard_files)
  
      def meta_loader(split_path):
          # Initialize the total number of samples
          total_samples = 0
  
          # Walk through the all subdirs
          # eg. german_speech, read_speech, ...
          for root, _, files in os.walk(split_path):
              for file in files:
                  if file == "meta.json":
                      meta_json_path = os.path.join(root, file)
                      with open(meta_json_path, "rb") as f:
                          meta = json.load(f)
                      total_samples += meta.get("num_data_samples", 0)
  
          return total_samples
  
      def train_audio_pipeline(sample_dict: Dict, random_chunk=True):
          key = sample_dict["__key__"]
          clean_wav = sample_dict["clean_file"]
          noise_wav = sample_dict["noise_file"]
          noisy_wav = sample_dict["noisy_file"]
          clean_sig = sample_dict["clean_audio.pth"].squeeze()
          noise_sig = sample_dict["noise_audio.pth"].squeeze()
          noisy_sig = sample_dict["noisy_audio.pth"].squeeze()
  
          return {
              "id": key,
              "clean_wav": clean_wav,
              "clean_sig": clean_sig,
              "noise_wav": noise_wav,
              "noise_sig": noise_sig,
              "noisy_wav": noisy_wav,
              "noisy_sig": noisy_sig,
          }
  
      def baseline_audio_pipeline(sample_dict: Dict, random_chunk=True):
          key = sample_dict["__key__"]
          noisy_sig = sample_dict["audio.pth"].squeeze()
  
          return {
              "id": key,
              "noisy_wav": key,
              "noisy_sig": noisy_sig,
          }
  
      def create_combined_dataset(shard_patterns, cache_dir):
          # mix multiple datasets, where each dataset consists of multiple shards
          # e.g. combine read_speech, german_speech etc. each with multiple shards.
          urls = [
              url
              for shard in shard_patterns
              for url in braceexpand.braceexpand(shard)
          ]
  
          combined_dataset = (
              wds.WebDataset(
                  urls,
                  shardshuffle=True,
                  cache_dir=cache_dir,
              )
              .repeat()
              .shuffle(1000)
              .decode("pil")
              .map(partial(train_audio_pipeline, random_chunk=True))
          )
  
          return combined_dataset
  
      train_data = create_combined_dataset(
          train_shard_patterns, hparams["shard_cache_dir"]
      )
      train_samples = meta_loader(hparams["train_data"])
      logger.info(f"Training data- Number of samples: {train_samples}")
      logger.info(
          f"Training data - Total duration: {train_samples * audio_length/ 3600:.2f} hours"
      )
  
      valid_data = create_combined_dataset(
          valid_shard_patterns, hparams["shard_cache_dir"]
      )
      valid_samples = meta_loader(hparams["valid_data"])
      logger.info(f"Valid data- Number of samples: {valid_samples}")
      logger.info(
          f"Valid data - Total duration: {valid_samples * audio_length  / 3600:.2f} hours"
      )
  
      baseline_data = (
          wds.WebDataset(
              hparams["baseline_shards"],
              cache_dir=hparams["shard_cache_dir"],
          )
          .repeat()
          .shuffle(1000)
          .decode("pil")
          .map(partial(baseline_audio_pipeline, random_chunk=True))
      )
  
      return (
          train_data,
          valid_data,
          train_samples,
          valid_samples,
          baseline_data,
      )
  
imports: import os
  import glob
  import sys
  import csv
  import json
  import logging
  import numpy
  from tqdm import tqdm
  from typing import Dict
  from functools import partial
  import torch
  import torchaudio
  import braceexpand
  import webdataset
  import torch.nn.functional
  import speechbrain
  from hyperpyyaml import load_hyperpyyaml
  from composite_eval import eval_composite
  import speechbrain.nnet.schedulers
  from speechbrain.utils.distributed import run_on_main
  from speechbrain.utils.metric_stats import MetricStats
  from speechbrain.processing.features import spectral_magnitude
  from speechbrain.dataio.batch import PaddedBatch
  from speechbrain.core import AMPConfig
  from pesq import pesq
  from pystoi import stoi
label_desc: Use image node_id 1 for calling a global function with example usage: dataio_prep(hparams) and returns: train_data,  valid_data,  train_samples,  valid_samples,  baseline_data, total_samples, dict, dict, combined_dataset