Quality Assessment for Audio Data

1
from nemo_curator.stages.audio.metrics.get_wer import GetPairwiseWerStage
2
3
# Calculate WER for each audio sample
4
wer_stage = GetPairwiseWerStage(
5
    text_key="text",           # Ground truth transcription
6
    pred_text_key="pred_text", # ASR prediction
7
    wer_key="wer"             # Output WER field
8
)

1
from nemo_curator.stages.audio.metrics.get_wer import get_cer
2
3
# Calculate CER between two strings (for testing/validation)
4
cer_value = get_cer("hello world", "helo world")  # Returns 9.09

1
from nemo_curator.stages.audio.metrics.get_wer import get_wordrate, get_charrate
2
3
# Calculate words per second
4
word_rate = get_wordrate("hello world example", 2.5)  # 1.2 words/second
5
6
# Calculate characters per second
7
char_rate = get_charrate("hello world", 2.0)  # 5.5 chars/second

1
from nemo_curator.stages.audio.common import PreserveByValueStage
2
3
# Keep samples with WER <= 30% (high quality)
4
high_quality_filter = PreserveByValueStage(
5
    input_value_key="wer",
6
    target_value=30.0,
7
    operator="le"  # less than or equal
8
)
9
10
# Remove samples with WER >= 80% (very poor quality)
11
poor_quality_filter = PreserveByValueStage(
12
    input_value_key="wer",
13
    target_value=80.0,
14
    operator="lt"  # less than
15
)
16
# Preserves only entries with WER < 80%

1
from nemo_curator.stages.audio.common import PreserveByValueStage
2
3
# Keep samples between 1-30 seconds
4
duration_min_filter = PreserveByValueStage(
5
    input_value_key="duration",
6
    target_value=1.0,
7
    operator="ge"  # greater than or equal
8
)
9
10
duration_max_filter = PreserveByValueStage(
11
    input_value_key="duration",
12
    target_value=30.0,
13
    operator="le"  # less than or equal
14
)

1
from nemo_curator.pipeline import Pipeline
2
from nemo_curator.stages.audio.metrics.get_wer import GetPairwiseWerStage
3
from nemo_curator.stages.audio.common import GetAudioDurationStage, PreserveByValueStage
4
5
# Create multi-stage quality pipeline
6
quality_pipeline = Pipeline(name="audio_quality_assessment")
7
8
# Calculate all metrics
9
quality_pipeline.add_stage(GetPairwiseWerStage())
10
quality_pipeline.add_stage(GetAudioDurationStage(
11
    audio_filepath_key="audio_filepath",
12
    duration_key="duration"
13
))
14
15
# Apply filters in sequence
16
filters = [
17
    PreserveByValueStage("wer", 50.0, "le"),        # WER <= 50%
18
    PreserveByValueStage("duration", 1.0, "ge"),     # Duration >= 1s
19
    PreserveByValueStage("duration", 20.0, "le"),    # Duration <= 20s
20
]
21
22
for filter_stage in filters:
23
    quality_pipeline.add_stage(filter_stage)

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, PreserveByValueStage
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 complete quality assessment pipeline
12
pipeline = Pipeline(name="audio_quality_assessment")
13
14
# 1. Load data
15
pipeline.add_stage(CreateInitialManifestFleursStage(
16
    lang="hy_am",
17
    split="dev",
18
    raw_data_dir="./audio_data"
19
).with_(batch_size=4))
20
21
# 2. ASR inference
22
pipeline.add_stage(InferenceAsrNemoStage(
23
    model_name="nvidia/stt_hy_fastconformer_hybrid_large_pc"
24
).with_(resources=Resources(gpus=1.0)))
25
26
# 3. Calculate quality metrics
27
pipeline.add_stage(GetPairwiseWerStage())
28
pipeline.add_stage(GetAudioDurationStage(
29
    audio_filepath_key="audio_filepath",
30
    duration_key="duration"
31
))
32
33
# 4. Apply quality filters
34
pipeline.add_stage(PreserveByValueStage(
35
    input_value_key="wer",
36
    target_value=75.0,
37
    operator="le"  # Keep WER <= 75%
38
))
39
40
pipeline.add_stage(PreserveByValueStage(
41
    input_value_key="duration",
42
    target_value=1.0,
43
    operator="ge"  # Keep duration >= 1s
44
))
45
46
pipeline.add_stage(PreserveByValueStage(
47
    input_value_key="duration",
48
    target_value=30.0,
49
    operator="le"  # Keep duration <= 30s
50
))
51
52
# 5. Export high-quality results
53
pipeline.add_stage(AudioToDocumentStage())
54
pipeline.add_stage(JsonlWriter(path="./high_quality_audio"))
55
56
# Execute pipeline
57
executor = XennaExecutor()
58
pipeline.run(executor)