Duration Calculation

1
from nemo_curator.stages.audio.common import GetAudioDurationStage
2
from nemo_curator.tasks import AudioBatch
3
4
# Initialize duration calculator
5
duration_stage = GetAudioDurationStage(
6
    audio_filepath_key="audio_filepath",
7
    duration_key="duration"
8
)
9
10
# Process audio data
11
audio_data = {"audio_filepath": "/path/to/audio.wav", "text": "transcription"}
12
audio_batch = AudioBatch(data=[audio_data])
13
result_batch = duration_stage.process(audio_batch)
14
15
# Access duration information
16
duration = result_batch[0].data[0]["duration"]
17
print(f"Audio duration: {duration:.3f} seconds")

1
from nemo_curator.stages.audio.common import GetAudioDurationStage
2
3
# Configure duration calculation
4
duration_stage = GetAudioDurationStage(
5
    audio_filepath_key="audio_filepath",  # Field containing audio file paths
6
    duration_key="duration"               # Output field for duration values
7
)

1
# Use custom field names for your data format
2
duration_stage = GetAudioDurationStage(
3
    audio_filepath_key="wav_file_path",   # Custom input field
4
    duration_key="audio_length_seconds"   # Custom output field
5
)

1
from nemo_curator.stages.audio.common import GetAudioDurationStage
2
from nemo_curator.tasks import AudioBatch
3
4
# Sample audio data
5
audio_samples = [
6
    {"audio_filepath": "/path/to/sample1.wav", "text": "Hello world"},
7
    {"audio_filepath": "/path/to/sample2.wav", "text": "How are you"},
8
    {"audio_filepath": "/path/to/sample3.wav", "text": "Good morning"}
9
]
10
11
# Create duration calculation stage
12
duration_stage = GetAudioDurationStage(
13
    audio_filepath_key="audio_filepath",
14
    duration_key="duration"
15
)
16
17
# Process each sample
18
for sample in audio_samples:
19
    audio_batch = AudioBatch(data=[sample])
20
    result_batch = duration_stage.process(audio_batch)
21
22
    processed_sample = result_batch[0].data[0]
23
    print(f"File: {processed_sample['audio_filepath']}")
24
    print(f"Duration: {processed_sample['duration']:.3f} seconds")

1
from nemo_curator.pipeline import Pipeline
2
from nemo_curator.stages.audio.common import GetAudioDurationStage, PreserveByValueStage
3
4
# Create audio processing pipeline
5
pipeline = Pipeline(name="audio_duration_pipeline")
6
7
# Add duration calculation stage
8
pipeline.add_stage(GetAudioDurationStage(
9
    audio_filepath_key="audio_filepath",
10
    duration_key="duration"
11
))
12
13
# Add duration-based filtering (1-30 seconds)
14
pipeline.add_stage(PreserveByValueStage(
15
    input_value_key="duration",
16
    target_value=1.0,
17
    operator="ge"  # greater than or equal
18
))
19
20
pipeline.add_stage(PreserveByValueStage(
21
    input_value_key="duration",
22
    target_value=30.0,
23
    operator="le"  # less than or equal
24
))

1
from nemo_curator.stages.audio.common import GetAudioDurationStage
2
from nemo_curator.tasks import AudioBatch
3
4
# Process multiple samples in a batch
5
audio_data_list = [
6
    {"audio_filepath": "/path/to/file1.wav", "text": "Sample 1"},
7
    {"audio_filepath": "/path/to/file2.wav", "text": "Sample 2"},
8
    {"audio_filepath": "/path/to/file3.wav", "text": "Sample 3"}
9
]
10
11
# Create batch
12
audio_batch = AudioBatch(data=audio_data_list)
13
14
# Process entire batch
15
duration_stage = GetAudioDurationStage(
16
    audio_filepath_key="audio_filepath",
17
    duration_key="duration"
18
)
19
20
# Process returns list of AudioBatch objects
21
result_batches = duration_stage.process(audio_batch)
22
23
# Extract processed data
24
for batch in result_batches:
25
    for sample in batch.data:
26
        print(f"File: {sample['audio_filepath']}")
27
        print(f"Duration: {sample['duration']:.3f} seconds")

1
# Non-existent files result in duration = -1.0
2
sample = {"audio_filepath": "/nonexistent/file.wav", "text": "test"}
3
audio_batch = AudioBatch(data=[sample])
4
result = duration_stage.process(audio_batch)
5
# result[0].data[0]["duration"] == -1.0

1
# Corrupted files are logged and marked with duration = -1.0
2
# Check logs for specific error messages
3
import logging
4
logging.basicConfig(level=logging.WARNING)
5
6
# Process will continue with other files
7
result = duration_stage.process(audio_batch)

1
from nemo_curator.stages.audio.common import PreserveByValueStage
2
3
# Filter out files with calculation errors
4
error_filter = PreserveByValueStage(
5
    input_value_key="duration",
6
    target_value=0.0,
7
    operator="gt"  # greater than (excludes -1.0 error values)
8
)

1
from nemo_curator.pipeline import Pipeline
2
from nemo_curator.stages.audio.common import GetAudioDurationStage, PreserveByValueStage
3
4
# Create comprehensive quality pipeline
5
pipeline = Pipeline(name="audio_quality_assessment")
6
7
# Step 1: Calculate durations
8
pipeline.add_stage(GetAudioDurationStage(
9
    audio_filepath_key="audio_filepath",
10
    duration_key="duration"
11
))
12
13
# Step 2: Filter by duration range (optimal for ASR training)
14
pipeline.add_stage(PreserveByValueStage(
15
    input_value_key="duration",
16
    target_value=1.0,      # Minimum 1 second
17
    operator="ge"
18
))
19
20
pipeline.add_stage(PreserveByValueStage(
21
    input_value_key="duration",
22
    target_value=15.0,     # Maximum 15 seconds
23
    operator="le"
24
))
25
26
# Step 3: Remove error files
27
pipeline.add_stage(PreserveByValueStage(
28
    input_value_key="duration",
29
    target_value=0.0,      # Exclude -1.0 error values
30
    operator="gt"
31
))

1
# Check supported formats
2
import soundfile as sf
3
print("Supported formats:", sf.available_formats())
4
5
# Common supported formats: WAV, FLAC, OGG, AIFF
6
# MP3 support depends on your system's libsndfile build