code: def _testimpl_rnnt_data_pipeline(
      device,
      pad_amount=0,
      preemph_coeff=0.97,
      window_size=0.02,
      window_stride=0.01,
      window="hann",
      nfeatures=64,
      n_fft=512,
      frame_splicing_stack=1,
      frame_splicing_subsample=1,
      lowfreq=0.0,
      highfreq=None,
      normalize_type="per_feature",
      batch_size=32,
  ):
      sample_rate = npy_files_sr
      speed_perturb = False
      silence_trim = False
  
      ref_pipeline = FilterbankFeatures(
          sample_rate=sample_rate,
          window_size=window_size,
          window_stride=window_stride,
          window=window,
          normalize=normalize_type,
          n_fft=n_fft,
          pad_amount=pad_amount,
          preemph=preemph_coeff,
          nfilt=nfeatures,
          lowfreq=lowfreq,
          highfreq=highfreq,
          log=True,
          frame_splicing_stack=frame_splicing_stack,
          frame_splicing_subsample=frame_splicing_subsample,
      )
      reference_data = []
      for i in range(nrecordings):
          reference_data.append(
              ref_pipeline.forward(
                  torch.tensor([recordings[i]]),
                  torch.tensor([recordings[i].shape[0]]),
              )
          )
  
      pipe = rnnt_train_pipe(
          audio_files,
          sample_rate,
          pad_amount,
          preemph_coeff,
          window_size,
          window_stride,
          window,
          nfeatures,
          n_fft,
          frame_splicing_stack,
          frame_splicing_subsample,
          lowfreq,
          highfreq,
          normalize_type,
          speed_perturb,
          silence_trim,
          device,
          seed=42,
          batch_size=batch_size,
      )
      pipe.build()
      nbatches = (nrecordings + batch_size - 1) // batch_size
      i = 0
      for b in range(nbatches):
          dali_out = list(pipe.run())
          for s in range(batch_size):
              if i >= nrecordings:
                  break
  
              (
                  norm_log_features,
                  log_features_spliced,
                  log_features,
                  mel_spec,
                  spec,
                  preemph_audio,
                  padded_audio,
                  audio,
              ) = [to_array(out[s]) for out in dali_out]
  
              ref = np.array(reference_data[i].squeeze(0))
              assert (
                  ref.shape == norm_log_features.shape
              ), f"{ref.shape}, {norm_log_features.shape}"
              nfeatures, seq_len = ref.shape
  
              audio_ref = recordings[i]
              np.testing.assert_allclose(audio, audio_ref, atol=1e-4)
  
              padded_audio_ref = torch_reflect_pad(audio, pad_amount)
              np.testing.assert_equal(padded_audio, padded_audio_ref)
  
              preemph_audio_ref = torch_preemphasis(
                  padded_audio_ref, preemph=preemph_coeff
              )
              np.testing.assert_allclose(
                  preemph_audio, preemph_audio_ref, atol=1e-4
              )
  
              spec_ref = torch_spectrogram(
                  preemph_audio_ref,
                  npy_files_sr,
                  window_size=window_size,
                  window_stride=window_stride,
                  center=True,
                  pad_mode="reflect",
                  window=window,
                  n_fft=n_fft,
              )
              np.testing.assert_allclose(spec, spec_ref, atol=1e-4)
  
              mel_spec_ref = torch_mel_fbank(spec_ref, npy_files_sr)
              np.testing.assert_allclose(
                  mel_spec, mel_spec_ref, atol=1e-4
              )
  
              log_features_ref = torch_log(mel_spec_ref)
              np.testing.assert_allclose(
                  log_features, log_features_ref, atol=1e-3
              )
              log_features_ref2 = torch_log(mel_spec)
              np.testing.assert_allclose(
                  log_features, log_features_ref2, atol=1e-4
              )
  
              log_features_spliced_ref = torch_frame_splicing(
                  log_features_ref,
                  stacking=frame_splicing_stack,
                  subsampling=frame_splicing_subsample,
              )
              np.testing.assert_allclose(
                  log_features_spliced,
                  log_features_spliced_ref,
                  atol=1e-3,
              )
  
              log_features_spliced_ref2 = torch_frame_splicing(
                  log_features,
                  stacking=frame_splicing_stack,
                  subsampling=frame_splicing_subsample,
              )
              np.testing.assert_allclose(
                  log_features_spliced,
                  log_features_spliced_ref2,
                  atol=1e-4,
              )
  
              norm_log_features_ref = torch_normalize(
                  log_features_spliced_ref, normalize_type
              )
              np.testing.assert_allclose(
                  norm_log_features, norm_log_features_ref, atol=1e-3
              )
  
              norm_log_features_ref2 = torch_normalize(
                  log_features_spliced, normalize_type
              )
              np.testing.assert_allclose(
                  norm_log_features, norm_log_features_ref2, atol=1e-4
              )
  
              # Full pipeline
              np.testing.assert_allclose(
                  norm_log_features, ref, atol=1e-3
              )
  
              i += 1
  
imports: import nose_utils
  import nvidia.dali.fn
  from nvidia.dali import pipeline_def
  import nvidia.dali.types
  from test_utils import get_files, to_array
  import numpy
  import librosa
  import torch
  import math
  import random
  import os
  from nose.tools import nottest
label_desc: Use image node_id 15 for calling a global function with example usage: _testimpl_rnnt_data_pipeline(device, pad_amount, preemph_coeff, window_size, window_stride, window, nfeatures, n_fft, frame_splicing_stack, frame_splicing_subsample, lowfreq, highfreq, normalize_type, batch_size) without return types