Text Integration for Audio Data

1
from nemo_curator.stages.audio.io.convert import AudioToDocumentStage
2
from nemo_curator.tasks import AudioTask
3
4
# Convert audio data to DocumentBatch format
5
converter = AudioToDocumentStage()
6
7
# Input: AudioTask with audio processing results
8
audio_task = AudioTask(data={
9
    "audio_filepath": "/data/audio/sample.wav",
10
    "text": "ground truth text",
11
    "pred_text": "asr predicted text",
12
    "wer": 12.5,
13
    "duration": 3.2
14
})
15
16
# Output: DocumentBatch with pandas DataFrame
17
document_batches = converter.process(audio_task)
18
document_batch = document_batches[0]
19
20
# Access the converted data
21
print(f"Converted {len(document_batch.data)} audio records to DocumentBatch")

1
# All audio processing results are maintained:
2
# - audio_filepath: Original audio file reference
3
# - text: Ground truth transcription (if available)
4
# - pred_text: ASR prediction
5
# - wer: Word Error Rate (if calculated)
6
# - duration: Audio duration (if calculated)
7
# - Any other metadata fields you've added

1
from nemo_curator.pipeline import Pipeline
2
from nemo_curator.backends.xenna import XennaExecutor
3
from nemo_curator.stages.audio.datasets.fleurs.create_initial_manifest import CreateInitialManifestFleursStage
4
from nemo_curator.stages.audio.inference.asr_nemo import InferenceAsrNemoStage
5
from nemo_curator.stages.audio.metrics.get_wer import GetPairwiseWerStage
6
from nemo_curator.stages.audio.common import GetAudioDurationStage
7
from nemo_curator.stages.audio.io.convert import AudioToDocumentStage
8
from nemo_curator.stages.text.io.writer import JsonlWriter
9
from nemo_curator.stages.resources import Resources
10
11
# Create pipeline that processes audio and exports results
12
pipeline = Pipeline(name="audio_processing_with_export")
13
14
# 1. Load audio data
15
pipeline.add_stage(CreateInitialManifestFleursStage(
16
    lang="en_us",
17
    split="test",
18
    raw_data_dir="./audio_data"
19
).with_(batch_size=8))
20
21
# 2. Run ASR inference
22
pipeline.add_stage(InferenceAsrNemoStage(
23
    model_name="nvidia/stt_en_fastconformer_hybrid_large_pc",
24
    pred_text_key="pred_text"
25
).with_(resources=Resources(gpus=1.0)))
26
27
# 3. Calculate quality metrics
28
pipeline.add_stage(GetPairwiseWerStage(
29
    text_key="text",
30
    pred_text_key="pred_text",
31
    wer_key="wer"
32
))
33
pipeline.add_stage(GetAudioDurationStage(
34
    audio_filepath_key="audio_filepath",
35
    duration_key="duration"
36
))
37
38
# 4. Convert to DocumentBatch for export
39
pipeline.add_stage(AudioToDocumentStage())
40
41
# 5. Export to JSONL format
42
pipeline.add_stage(JsonlWriter(path="/output/processed_audio_results"))
43
44
# Execute pipeline
45
executor = XennaExecutor()
46
pipeline.run(executor)

1
from nemo_curator.stages.function_decorators import processing_stage
2
from nemo_curator.tasks import DocumentBatch
3
import pandas as pd
4
5
@processing_stage(name="custom_transcription_filter")
6
def filter_transcriptions(document_batch: DocumentBatch) -> DocumentBatch:
7
    """Custom filtering of ASR transcriptions."""
8
9
    # Access the pandas DataFrame
10
    df = document_batch.data
11
12
    # Example: Filter by transcription length
13
    df = df[df['pred_text'].str.len() > 10]  # Keep transcriptions >10 chars
14
15
    # Example: Filter by WER if available
16
    if 'wer' in df.columns:
17
        df = df[df['wer'] < 50.0]  # Keep WER < 50%
18
19
    return DocumentBatch(
20
        data=df,
21
        task_id=document_batch.task_id,
22
        dataset_name=document_batch.dataset_name
23
    )

1
# Example output structure
2
document_batch.data  # pandas DataFrame with columns:
3
# - audio_filepath: "/path/to/audio.wav"
4
# - text: "ground truth transcription"
5
# - pred_text: "asr prediction"
6
# - wer: 15.2
7
# - duration: 3.4
8
# - [any other fields from your audio processing]