Create and Load Custom Audio Manifests

1
import json
2
3
# Example: Create a manifest from a list of audio files
4
audio_data = [
5
    {"audio_filepath": "/data/audio/sample1.wav", "text": "hello world"},
6
    {"audio_filepath": "/data/audio/sample2.wav", "text": "good morning"},
7
    {"audio_filepath": "/data/audio/sample3.wav", "text": "how are you"}
8
]
9
10
# Write JSONL manifest
11
with open("my_audio_manifest.jsonl", "w") as f:
12
    for entry in audio_data:
13
        f.write(json.dumps(entry) + "\n")

1
from nemo_curator.pipeline import Pipeline
2
from nemo_curator.stages.text.io.reader import JsonlReader
3
from nemo_curator.stages.audio.inference.asr_nemo import InferenceAsrNemoStage
4
from nemo_curator.stages.audio.metrics.get_wer import GetPairwiseWerStage
5
6
# Create pipeline
7
pipeline = Pipeline(name="custom_audio_processing")
8
9
# Load custom manifest (produces DocumentBatch)
10
pipeline.add_stage(
11
    JsonlReader(
12
        file_paths="my_audio_manifest.jsonl",
13
        fields=["audio_filepath", "text"]
14
    )
15
)
16
17
# ASR inference (consumes DocumentBatch, produces AudioTask)
18
pipeline.add_stage(
19
    InferenceAsrNemoStage(
20
        model_name="nvidia/stt_en_fastconformer_hybrid_large_pc",
21
        filepath_key="audio_filepath",
22
        pred_text_key="pred_text"
23
    )
24
)
25
26
# Calculate WER between ground truth and prediction
27
pipeline.add_stage(
28
    GetPairwiseWerStage(
29
        text_key="text", 
30
        pred_text_key="pred_text", 
31
        wer_key="wer"
32
    )
33
)

1
# Validation occurs when stages process AudioTask objects
2
# Files are checked for existence during ASR inference
3
# Invalid files generate warnings but don't stop processing
4
5
# Pipeline stages handle validation automatically
6
if audio_task.validate():
7
    print("All audio files exist")
8
else:
9
    print("Some audio files are missing")

1
from nemo_curator.pipeline import Pipeline
2
from nemo_curator.stages.text.io.reader import JsonlReader
3
from nemo_curator.stages.audio.inference.asr_nemo import InferenceAsrNemoStage
4
from nemo_curator.stages.audio.metrics.get_wer import GetPairwiseWerStage
5
from nemo_curator.stages.audio.common import GetAudioDurationStage, PreserveByValueStage
6
from nemo_curator.stages.audio.io.convert import AudioToDocumentStage
7
from nemo_curator.stages.text.io.writer import JsonlWriter
8
9
def create_custom_audio_pipeline(manifest_path: str, output_path: str) -> Pipeline:
10
    """Create pipeline for custom audio manifest processing."""
11
    
12
    pipeline = Pipeline(name="custom_audio_processing")
13
    
14
    # Load custom manifest
15
    pipeline.add_stage(JsonlReader(
16
        file_paths=manifest_path,
17
        fields=["audio_filepath", "text"]
18
    ))
19
    
20
    # ASR processing
21
    pipeline.add_stage(InferenceAsrNemoStage(
22
        model_name="nvidia/stt_en_fastconformer_hybrid_large_pc"
23
    ))
24
    
25
    # Quality assessment
26
    pipeline.add_stage(GetPairwiseWerStage())
27
    pipeline.add_stage(GetAudioDurationStage(
28
        audio_filepath_key="audio_filepath",
29
        duration_key="duration"
30
    ))
31
    
32
    # Filter by quality (keep WER <= 40%)
33
    pipeline.add_stage(PreserveByValueStage(
34
        input_value_key="wer",
35
        target_value=40.0,
36
        operator="le"
37
    ))
38
    
39
    # Export results
40
    pipeline.add_stage(AudioToDocumentStage())
41
    pipeline.add_stage(JsonlWriter(path=output_path))
42
    
43
    return pipeline
44
45
# Usage
46
pipeline = create_custom_audio_pipeline(
47
    manifest_path="my_audio_manifest.jsonl",
48
    output_path="processed_results"
49
)