code: def get_audio_to_text_bpe_dataset_from_config(
      config,
      local_rank: int,
      global_rank: int,
      world_size: int,
      tokenizer,
      preprocessor_cfg: Optional[DictConfig] = None,
  ):
      """
      Construct Audio-To-Text BPE dataset from a config.
      Args:
          config: BPE dataset config
          local_rank: model local rank
          global_rank: model global rand
          world_size: world size
          tokenizer: BPE tokenizer
          preprocessor_cfg: preprocessor config, for DALI BPE dataset
  
      Returns:
          constructed dataset or None if dataset config is invalid or nothing to load
      """
      if "augmentor" in config:
          augmentor = process_augmentations(
              config["augmentor"],
              global_rank=global_rank,
              world_size=world_size,
          )
      else:
          augmentor = None
  
      if "hf_data_cfg" in config:
          return get_hf_audio_to_text_bpe_dataset(
              config=config,
              global_rank=global_rank,
              world_size=world_size,
              tokenizer=tokenizer,
              augmentor=augmentor,
          )
  
      is_concat = config.get("is_concat", False)
      if is_concat:
          if (
              "concat_sampling_technique" in config
              and config["concat_sampling_technique"] is None
          ):
              logging.warning(
                  f"Concat dataset requires `concat_sampling_technique` but it was not provided. Config: {config}"
              )
              return None
  
          if config["concat_sampling_technique"] == "random":
              if not "concat_sampling_probabilities" in config:
                  logging.warning(
                      f"Concat dataset requires `concat_sampling_probabilities` list. Config: {config}"
                  )
                  return None
              else:
                  if not isclose(
                      sum(config["concat_sampling_probabilities"]),
                      1,
                      abs_tol=1e-6,
                  ):
                      logging.warning(
                          f"`concat_sampling_probabilities` need to sum to 1. Config: {config}"
                      )
                      return None
  
      shuffle = config["shuffle"]
      device = "gpu" if torch.cuda.is_available() else "cpu"
      if config.get("use_dali", False):
          device_id = local_rank if device == "gpu" else None
          dataset = get_dali_bpe_dataset(
              config=config,
              tokenizer=tokenizer,
              shuffle=shuffle,
              device_id=device_id,
              global_rank=global_rank,
              world_size=world_size,
              preprocessor_cfg=preprocessor_cfg,
          )
          return dataset
  
      # Instantiate a code-switched dataset if config is present
      if config.get("is_code_switched", False):
          if (
              "manifest_filepath" in config
              and config["manifest_filepath"] is None
          ):
              logging.warning(
                  f"Could not load dataset as `manifest_filepath` was None. Provided config : {config}"
              )
              return None
          if not (
              "code_switched" in config
              and config["code_switched"] is not None
          ):
              logging.warning(
                  f"Code switched dataset requires `*_ds.code_switched.*` dict but it was not provided. Config: {config}"
              )
              return None
          if (
              ("probs" in config["code_switched"])
              and (config["code_switched"]["probs"] is not None)
              and (
                  not isclose(
                      sum(config["code_switched"]["probs"]),
                      1,
                      abs_tol=1e-6,
                  )
              )
          ):
              logging.warning(
                  f"`.code_switched.probs` need to sum to 1. Config: {config['code_switched']}"
              )
              return None
  
          shuffle_n = (
              config.get("shuffle_n", 4 * config["batch_size"])
              if shuffle
              else 0
          )
          dataset = get_code_switched_dataset(
              config=config,
              shuffle_n=shuffle_n,
              global_rank=global_rank,
              world_size=world_size,
              tokenizer=tokenizer,
              augmentor=augmentor,
          )
      # Instantiate tarred dataset loader or normal dataset loader
      elif config.get("is_tarred", False):
          if (
              "tarred_audio_filepaths" in config
              and config["tarred_audio_filepaths"] is None
          ) or (
              "manifest_filepath" in config
              and config["manifest_filepath"] is None
          ):
              logging.warning(
                  "Could not load dataset as `manifest_filepath` was None or "
                  f"`tarred_audio_filepaths` is None. Provided config : {config}"
              )
              return None
  
          shuffle_n = (
              config.get("shuffle_n", 4 * config["batch_size"])
              if shuffle
              else 0
          )
          if is_concat:
              dataset = get_concat_tarred_dataset(
                  config=config,
                  tokenizer=tokenizer,
                  shuffle_n=shuffle_n,
                  global_rank=global_rank,
                  world_size=world_size,
                  augmentor=augmentor,
              )
          else:
              dataset = get_tarred_dataset(
                  config=config,
                  tokenizer=tokenizer,
                  shuffle_n=shuffle_n,
                  global_rank=global_rank,
                  world_size=world_size,
                  augmentor=augmentor,
              )
      else:
          if (
              "manifest_filepath" in config
              and config["manifest_filepath"] is None
          ):
              logging.warning(
                  f"Could not load dataset as `manifest_filepath` was None. Provided config : {config}"
              )
              return None
          if is_concat:
              dataset = get_concat_bpe_dataset(
                  config=config,
                  global_rank=global_rank,
                  world_size=world_size,
                  tokenizer=tokenizer,
                  augmentor=augmentor,
              )
          else:
              dataset = get_bpe_dataset(
                  config=config,
                  tokenizer=tokenizer,
                  augmentor=augmentor,
              )
      return dataset
  
imports: import copy
  import json
  import random
  from math import isclose
  from typing import Any, List, Optional, Union
  import torch
  from omegaconf import DictConfig, OmegaConf, open_dict
  from omegaconf.listconfig import ListConfig
  from pytorch_lightning.callbacks import BasePredictionWriter
  from torch.utils.data import ChainDataset
  from nemo.collections.asr.data import audio_to_text, audio_to_text_dali
  from nemo.collections.asr.data.huggingface.hf_audio_to_text_dataset import get_hf_audio_to_text_bpe_dataset, get_hf_audio_to_text_char_dataset
  from nemo.collections.asr.parts.preprocessing.perturb import process_augmentations
  from nemo.collections.common.data.dataset import CodeSwitchedDataset, ConcatDataset
  from nemo.utils import logging
label_desc: Use image node_id 12 for calling a global function with example usage: get_audio_to_text_bpe_dataset_from_config(config, local_rank, global_rank, world_size, tokenizer, preprocessor_cfg) and returns: dataset, get_hf_audio_to_text_bpe_dataset, dataset, None, None, None, None, None, None, None, None