code: def get_lhotse_dataloader_from_config(
      config: DictConfig,
      global_rank: int,
      world_size: int,
      dataset: torch.utils.data.Dataset,
  ) -> torch.utils.data.DataLoader:
      """
      Set up a Lhotse training dataloder.
  
      Expects a typical NeMo dataset configuration format, with additional fields: "use_lhotse=True" and "lhotse: <dict>".
      Some fields in the original NeMo configuration may be ignored.
  
      The ``dataset`` parameter should be an instance of a Lhotse-compatible PyTorch Dataset class.
      It only needs to define the following method ``__getitem__(self, cuts: CutSet) -> Dict[str, torch.Tensor]``.
      This dataset is not expected to hold a reference to any actual data; it may be interpreted as a function
      mapping a Lhotse CutSet into a mini-batch of tensors.
  
      For example, see: :class:`nemo.collections.asr.data.audio_to_text_lhotse.LhotseSpeechToTextBpeDataset`,
      which is constructed from just a tokenizer and essentially loads and collates audio and tokenizes the transcript.
      """
      logging.info("We will be using a Lhotse DataLoader.")
  
      config = make_structured_with_schema_warnings(config)
  
      # 1. Load a manifest as a Lhotse CutSet.
      cuts, is_tarred = read_cutset_from_config(config)
  
      # Resample as a safeguard; it's a no-op when SR is already OK
      cuts = cuts.resample(config.sample_rate)
  
      # Duration filtering, same as native NeMo dataloaders.
      min_dur, max_dur = config.min_duration, config.max_duration
      cuts = cuts.filter(lambda c: min_dur <= c.duration <= max_dur)
  
      # Safeguard against utterances with identical IDs across different datasets
      # that would make Lhotse complain otherwise.
      cuts = cuts.modify_ids(create_id_randomizer(config.seed))
  
      # 2. Optional augmentations.
      # 2.a. Noise mixing.
      if config.noise_path is not None:
          noise = CutSet.from_file(config.noise_path)
          cuts = cuts.mix(
              cuts=noise,
              snr=config.noise_snr,
              mix_prob=config.noise_mix_prob,
              seed="trng",
              random_mix_offset=True,
          )
  
      # 2.b. On-the-fly speed perturbation.
      #    mux here ensures it's uniformly distributed throughout sampling,
      #    and applying it here (before sampler/dataset) ensures optimal
      #    bucket allocation.
      if config.perturb_speed:
          cuts = CutSet.mux(
              cuts,
              cuts.perturb_speed(0.9),
              cuts.perturb_speed(1.1),
          )
  
      # 3. The sampler.
      if config.use_bucketing:
          # Bucketing. Some differences from NeMo's native bucketing:
          #    - we can tweak the number of buckets and bucket duration bins using the configuration
          #    - batch size is dynamic and configurable via a single param: max_duration (config: batch_duration)
          #    - quadratic_duration introduces a penalty to balance batch sizes for quadratic time complexity models
          logging.info(
              f"Creating a Lhotse DynamicBucketingSampler "
              f"(max_batch_duration={config.batch_duration} max_batch_size={config.batch_size})"
          )
          sampler = DynamicBucketingSampler(
              cuts,
              max_duration=config.batch_duration,
              max_cuts=config.batch_size,
              shuffle=config.shuffle,
              drop_last=config.drop_last,
              shuffle_buffer_size=config.shuffle_buffer_size,
              quadratic_duration=config.quadratic_duration,
              seed=config.seed,
              num_buckets=config.num_buckets,
              duration_bins=config.bucket_duration_bins,
              num_cuts_for_bins_estimate=config.num_cuts_for_bins_estimate,
              buffer_size=config.bucket_buffer_size,
              rank=0 if is_tarred else global_rank,
              world_size=1 if is_tarred else world_size,
          )
      else:
          # Non-bucketing sampler, similar to original NeMo dataloading without bucketing,
          # but we also use batch_duration instead of batch_size here.
          # Recommended for dev/test.
          logging.info(
              f"Creating a Lhotse DynamicCutSampler (bucketing is disabled, "
              f"(max_batch_duration={config.batch_duration} max_batch_size={config.batch_size})"
          )
          sampler = DynamicCutSampler(
              cuts,
              max_duration=config.batch_duration,
              max_cuts=config.batch_size,
              shuffle=config.shuffle,
              drop_last=config.drop_last,
              shuffle_buffer_size=config.shuffle_buffer_size,
              quadratic_duration=config.quadratic_duration,
              seed=config.seed,
              rank=0 if is_tarred else global_rank,
              world_size=1 if is_tarred else world_size,
          )
  
      if config.concatenate_samples:
          # Cut concatenation will produce longer samples out of shorter samples
          # by gluing them together from the shortest to longest not to exceed a duration
          # of longest_cut * duration_factor (greedy knapsack algorithm for minimizing padding).
          # Useful e.g. for simulated code-switching in multilingual setups.
          # We follow concatenation by ``merge_supervisions`` which creates a single supervision
          # object with texts joined by a whitespace so that "regular" dataset classes don't
          # have to add a special support for multi-supervision cuts.
          sampler = sampler.map(
              CutConcatenate(
                  gap=config.concatenate_gap_seconds,
                  duration_factor=config.concatenate_duration_factor,
              )
          )
          if config.db_norm is not None:
              sampler = sampler.map(
                  lambda cuts: cuts.normalize_loudness(
                      config.db_norm, mix_first=False
                  )
              )
          if config.concatenate_merge_supervisions:
              sampler = sampler.map(
                  lambda cuts: cuts.merge_supervisions()
              )
  
      # 4. Creating dataloader.
      if is_tarred:
          # Wrapper here is necessary when using NeMo tarred data or Lhotse Shar data,
          # because then I/O happens upon sampler iteration. Normally, the sampler resides
          # in the training loop process, but when we use iterable dataset, we can move it to
          # the dataloading worker process.
          # We use lhotse's own worker_init_fn which leverages information such as rank, world_size,
          # worker_id, etc. to set a different random seed for each (node, worker) combination.
          # This together with infinite datasets removes the need to split data across nodes/workers.
          dloader_kwargs = dict(
              dataset=IterableDatasetWrapper(
                  dataset=dataset, sampler=sampler
              ),
              worker_init_fn=make_worker_init_fn(
                  rank=global_rank, world_size=world_size
              ),
              persistent_workers=config.num_workers
              > 0,  # helps Lhotse Shar maintain shuffling state
          )
      else:
          # For non-tarred data, the sampler resides in the training loop process and
          # reads only light-weight JSON objects; it samples mini-batches and passes
          # the meta-data to Dataset, which performs the actual I/O inside its __getitem__ method.
          dloader_kwargs = dict(dataset=dataset, sampler=sampler)
      dloader = torch.utils.data.DataLoader(
          **dloader_kwargs,
          batch_size=None,
          num_workers=config.num_workers,
          pin_memory=config.pin_memory,
      )
  
      return dloader
  
file_path: NeMo/nemo/collections/common/data/lhotse/dataloader.py
imports: import logging
  import random
  import warnings
  from dataclasses import dataclass
  from typing import Any, Callable, Optional
  import torch
  from lhotse import CutSet
  from lhotse.dataset import CutConcatenate, DynamicBucketingSampler, DynamicCutSampler, IterableDatasetWrapper, make_worker_init_fn
  from omegaconf import DictConfig, OmegaConf
  from nemo.collections.common.data.lhotse.cutset import read_cutset_from_config
label_desc: Use image node_id 1 for calling a global function with example usage: get_lhotse_dataloader_from_config(config, global_rank, world_size, dataset) and returns: dloader