docstr: Test AudioWithTargetWithEmbeddingDataset.
  
  In this use case, each line of the manifest file has the following format:
  ```
  {
      'input_filepath': 'path/to/input.wav',
      'target_filepath': 'path/to/path_to_target.wav',
      'embedding_filepath': 'path/to/path_to_embedding.npy',
      'duration': duration_of_input,
  }
  ```
docstr_tok: ["Test","AudioWithTargetWithEmbeddingDataset",".","In","this","use","case",",","each","line","of","the","manifest","file","has","the","following","format",":","``","`","{","'input_filepath","'",":","'path\/to\/input.wav","'",",","'target_filepath","'",":","'path\/to\/path_to_target.wav","'",",","'embedding_filepath","'",":","'path\/to\/path_to_embedding.npy","'",",","'duration","'",":","duration_of_input",",","}","``","`"]
code: def test_audio_to_target_with_embedding_dataset(self):
      """Test AudioWithTargetWithEmbeddingDataset.
  
      In this use case, each line of the manifest file has the following format:
      ```
      {
          'input_filepath': 'path/to/input.wav',
          'target_filepath': 'path/to/path_to_target.wav',
          'embedding_filepath': 'path/to/path_to_embedding.npy',
          'duration': duration_of_input,
      }
      ```
      """
      # Data setup
      random_seed = 42
      sample_rate = 16000
      num_examples = 25
      data_num_channels = {
          "input_signal": 4,
          "target_signal": 2,
          "embedding_vector": 1,
      }
      data_min_duration = 2.0
      data_max_duration = 8.0
      embedding_length = 64  # 64-dimensional embedding vector
      data_key = {
          "input_signal": "input_filepath",
          "target_signal": "target_filepath",
          "embedding_vector": "embedding_filepath",
      }
  
      # Tolerance
      atol = 1e-6
  
      # Generate random signals
      _rng = np.random.default_rng(seed=random_seed)
  
      # Input and target signals have the same duration
      data_duration = np.round(
          _rng.uniform(
              low=data_min_duration,
              high=data_max_duration,
              size=num_examples,
          ),
          3,
      )
      data_duration_samples = np.floor(
          data_duration * sample_rate
      ).astype(int)
  
      data = dict()
      for signal, num_channels in data_num_channels.items():
          data[signal] = []
          for n in range(num_examples):
              data_length = (
                  embedding_length
                  if signal == "embedding_vector"
                  else data_duration_samples[n]
              )
  
              if num_channels == 1:
                  random_signal = _rng.uniform(
                      low=-0.5, high=0.5, size=(data_length)
                  )
              else:
                  random_signal = _rng.uniform(
                      low=-0.5,
                      high=0.5,
                      size=(num_channels, data_length),
                  )
              data[signal].append(random_signal)
  
      with tempfile.TemporaryDirectory() as test_dir:
          # Build metadata for manifest
          metadata = []
  
          for n in range(num_examples):
              meta = dict()
  
              for signal in data:
                  if signal == "embedding_vector":
                      signal_filename = f"{signal}_{n:02d}.npy"
                      np.save(
                          os.path.join(test_dir, signal_filename),
                          data[signal][n],
                      )
  
                  else:
                      # filenames
                      signal_filename = f"{signal}_{n:02d}.wav"
  
                      # write audio files
                      sf.write(
                          os.path.join(test_dir, signal_filename),
                          data[signal][n].T,
                          sample_rate,
                          "float",
                      )
  
                  # update metadata
                  meta[data_key[signal]] = signal_filename
  
              meta["duration"] = data_duration[n]
              metadata.append(meta)
  
          # Save manifest
          manifest_filepath = os.path.join(test_dir, "manifest.json")
          write_manifest(manifest_filepath, metadata)
  
          # Test 1
          # - No constraints on channels or duration
          dataset = AudioToTargetWithEmbeddingDataset(
              manifest_filepath=manifest_filepath,
              input_key=data_key["input_signal"],
              target_key=data_key["target_signal"],
              embedding_key=data_key["embedding_vector"],
              sample_rate=sample_rate,
          )
  
          # Also test the corresponding factory
          config = {
              "manifest_filepath": manifest_filepath,
              "input_key": data_key["input_signal"],
              "target_key": data_key["target_signal"],
              "embedding_key": data_key["embedding_vector"],
              "sample_rate": sample_rate,
          }
          dataset_factory = audio_to_audio_dataset.get_audio_to_target_with_embedding_dataset(
              config
          )
  
          for n in range(num_examples):
              item = dataset.__getitem__(n)
              item_factory = dataset_factory.__getitem__(n)
  
              for signal in data:
                  item_signal = item[signal].cpu().detach().numpy()
                  golden_signal = data[signal][n]
                  assert (
                      item_signal.shape == golden_signal.shape
                  ), f"Signal {signal}: item shape {item_signal.shape} not matching reference shape {golden_signal.shape}"
                  assert np.allclose(
                      item_signal, golden_signal, atol=atol
                  ), f"Test 1: Failed for example {n}, signal {signal} (random seed {random_seed})"
  
                  item_factory_signal = (
                      item_factory[signal].cpu().detach().numpy()
                  )
                  assert np.allclose(
                      item_factory_signal, golden_signal, atol=atol
                  ), f"Test 1: Failed for factory example {n}, signal {signal} (random seed {random_seed})"
  
          # Test 2:
          # - Test collate_fn
          batch_size = 16
          batch = [dataset.__getitem__(n) for n in range(batch_size)]
          _ = dataset.collate_fn(batch)
  
imports: import copy
  import filecmp
  import json
  import os
  import shutil
  import tempfile
  from unittest import mock
  import numpy
  import pytest
  import soundfile
  import torch.cuda
  from omegaconf import DictConfig, OmegaConf
  from torch.utils.data import DataLoader
  from nemo.collections.asr.data import audio_to_audio_dataset, audio_to_text_dataset
  from nemo.collections.asr.data.audio_to_audio import ASRAudioProcessor, AudioToTargetDataset, AudioToTargetWithEmbeddingDataset, AudioToTargetWithReferenceDataset, _audio_collate_fn
  from nemo.collections.asr.data.audio_to_text import DataStoreObject, TarredAudioToBPEDataset, TarredAudioToCharDataset, cache_datastore_manifests
  from nemo.collections.asr.data.audio_to_text_dali import __DALI_MINIMUM_VERSION__, AudioToBPEDALIDataset, AudioToCharDALIDataset, is_dali_supported
  from nemo.collections.asr.data.audio_to_text_dataset import inject_dataloader_value_from_model_config
  from nemo.collections.asr.data.feature_to_text import FeatureToBPEDataset, FeatureToCharDataset
  from nemo.collections.asr.models.ctc_models import EncDecCTCModel
  from nemo.collections.asr.parts.utils.audio_utils import get_segment_start
  from nemo.collections.asr.parts.utils.manifest_utils import write_manifest
  from nemo.collections.common import tokenizers
  from nemo.utils import logging
label_desc: Use image node_id 7 for calling the TestAudioDatasets obj's underlying member method code with example usage: obj.test_audio_to_target_with_embedding_dataset() without return types