nemo_curator.stages.audio.common

Module Contents

Classes

Name	Description
`GetAudioDurationStage`	Compute audio duration from the file at audio_filepath_key and
`ManifestReader`	Composite stage for reading JSONL manifests.
`ManifestReaderStage`	Read JSONL manifest files from a FileGroupTask and emit one AudioTask per line.
`ManifestWriterStage`	Append a single AudioTask to a JSONL manifest file.
`PreserveByValueStage`	Filter entries by comparing input_value_key against target_value.

Functions

Name	Description
`ensure_mono`	Convert multi-channel waveform to mono. Assumes 2D (channels, samples) input.
`ensure_waveform_2d`	Ensure waveform is a torch.Tensor in 2D (channels, samples) format.
`get_audio_duration`	Get the duration of the audio file in seconds.
`load_audio_file`	Load audio file and return waveform tensor (channels, samples) and sample rate.
`resolve_model_path`	Resolve a relative model path using the reference file’s directory and module subdirectory.
`resolve_waveform_from_item`	Resolve (waveform, sample_rate) from an item dict, loading from file if needed.

API

class nemo_curator.stages.audio.common.GetAudioDurationStage(
    name: str = 'GetAudioDurationStage',
    audio_filepath_key: str = 'audio_filepath',
    duration_key: str = 'duration'
)

Dataclass

Bases: ProcessingStage[AudioTask, AudioTask]

Compute audio duration from the file at audio_filepath_key and store the result under duration_key.

Parameters:

audio_filepath_key

strDefaults to 'audio_filepath'

Key to get path to wav file.

duration_key

strDefaults to 'duration'

Key to put audio duration.

audio_filepath_key

str = 'audio_filepath'

duration_key

str = 'duration'

name

str = 'GetAudioDurationStage'

nemo_curator.stages.audio.common.GetAudioDurationStage.inputs() -> tuple[list[str], list[str]]

nemo_curator.stages.audio.common.GetAudioDurationStage.outputs() -> tuple[list[str], list[str]]

nemo_curator.stages.audio.common.GetAudioDurationStage.process(
    task: nemo_curator.tasks.AudioTask
) -> nemo_curator.tasks.AudioTask

nemo_curator.stages.audio.common.GetAudioDurationStage.setup(
    _worker_metadata: nemo_curator.backends.base.WorkerMetadata | None = None
) -> None

class nemo_curator.stages.audio.common.ManifestReader(
    manifest_path: str | list[str],
    name: str = 'manifest_reader',
    files_per_partition: int | None = 1,
    blocksize: int | str | None = None,
    file_extensions: list[str] = (lambda: ['.jsonl', '.json'...,
    storage_options: dict[str, typing.Any] | None = None
)

Dataclass

Bases: CompositeStage[_EmptyTask, AudioTask]

Composite stage for reading JSONL manifests.

Decomposes into:

FilePartitioningStage — discovers and partitions manifest files
ManifestReaderStage — reads each partition line-by-line (no Pandas)

Parameters:

manifest_path

str | list[str]

Path or list of paths to JSONL manifests (local or cloud).

files_per_partition

int | NoneDefaults to 1

Number of manifest files per partition. Defaults to 1.

blocksize

int | str | NoneDefaults to None

Target size per partition (e.g., “100MB”). Ignored if files_per_partition is set.

file_extensions

list[str]Defaults to (lambda: ['.jsonl', '.json'])()

File extensions to filter. Defaults to [“.jsonl”, “.json”].

storage_options

dict[str, Any] | NoneDefaults to None

Storage options for cloud paths (S3, GCS credentials, endpoints).

blocksize

int | str | None = None

file_extensions

list[str]

files_per_partition

int | None = 1

manifest_path

str | list[str]

name

str = 'manifest_reader'

storage_options

dict[str, Any] | None = None

nemo_curator.stages.audio.common.ManifestReader.__post_init__() -> None

nemo_curator.stages.audio.common.ManifestReader.decompose() -> list[nemo_curator.stages.base.ProcessingStage]

nemo_curator.stages.audio.common.ManifestReader.get_description() -> str

class nemo_curator.stages.audio.common.ManifestReaderStage(
    name: str = 'manifest_reader_stage'
)

Dataclass

Bases: ProcessingStage[FileGroupTask, AudioTask]

Read JSONL manifest files from a FileGroupTask and emit one AudioTask per line.

Uses line-by-line streaming via fsspec (no Pandas) to keep memory at ~1x file size. Supports local and cloud paths (S3, GCS).

name

str = 'manifest_reader_stage'

nemo_curator.stages.audio.common.ManifestReaderStage.process(
    task: nemo_curator.tasks.FileGroupTask
) -> list[nemo_curator.tasks.AudioTask]

nemo_curator.stages.audio.common.ManifestReaderStage.ray_stage_spec() -> dict[str, typing.Any]

nemo_curator.stages.audio.common.ManifestReaderStage.xenna_stage_spec() -> dict[str, typing.Any]

class nemo_curator.stages.audio.common.ManifestWriterStage(
    output_path: str,
    name: str = 'manifest_writer'
)

Dataclass

Bases: ProcessingStage[AudioTask, AudioTask]

Append a single AudioTask to a JSONL manifest file.

The output file is truncated once in setup() (called on the driver) so repeated pipeline runs produce a clean output. setup_on_node() only creates the parent directory — it never truncates, so multi-node deployments do not erase each other’s data.

.. note:: Because all nodes append to the same path, callers in multi-node setups should either use a shared filesystem or provide a node-unique output_path.

Supports local and cloud paths via fsspec.

Parameters:

output_path

str

Destination JSONL path (local or cloud).

name

str = 'manifest_writer'

output_path

str

nemo_curator.stages.audio.common.ManifestWriterStage.__post_init__() -> None

nemo_curator.stages.audio.common.ManifestWriterStage.num_workers() -> int | None

nemo_curator.stages.audio.common.ManifestWriterStage.process(
    task: nemo_curator.tasks.AudioTask
) -> nemo_curator.tasks.AudioTask

nemo_curator.stages.audio.common.ManifestWriterStage.setup(
    _worker_metadata: nemo_curator.backends.base.WorkerMetadata | None = None
) -> None

Truncate the output file once on the driver before processing starts.

nemo_curator.stages.audio.common.ManifestWriterStage.setup_on_node(
    _node_info: nemo_curator.backends.base.NodeInfo | None = None,
    _worker_metadata: nemo_curator.backends.base.WorkerMetadata | None = None
) -> None

Ensure parent directory exists on each node (no truncation).

nemo_curator.stages.audio.common.ManifestWriterStage.xenna_stage_spec() -> dict[str, typing.Any]

class nemo_curator.stages.audio.common.PreserveByValueStage(
    input_value_key: str,
    target_value: int | str,
    operator: str = 'eq'
)

Bases: ProcessingStage[AudioTask, AudioTask]

Filter entries by comparing input_value_key against target_value.

Returns None from process() to drop entries that fail the comparison, matching the text-modality filter convention.

Parameters:

input_value_key

str

The field in the dataset entries to evaluate.

target_value

int | str

The value to compare with.

operator

strDefaults to 'eq'

Comparison operator (lt, le, eq, ne, ge, gt).

name

str = 'PreserveByValueStage'

operator

= ops[operator]

nemo_curator.stages.audio.common.PreserveByValueStage.inputs() -> tuple[list[str], list[str]]

nemo_curator.stages.audio.common.PreserveByValueStage.outputs() -> tuple[list[str], list[str]]

nemo_curator.stages.audio.common.PreserveByValueStage.process(
    task: nemo_curator.tasks.AudioTask
) -> nemo_curator.tasks.AudioTask | None

nemo_curator.stages.audio.common.PreserveByValueStage.process_batch(
    tasks: list[nemo_curator.tasks.AudioTask]
) -> list[nemo_curator.tasks.AudioTask]

nemo_curator.stages.audio.common.ensure_mono(
    waveform: torch.Tensor
) -> torch.Tensor

Convert multi-channel waveform to mono. Assumes 2D (channels, samples) input.

nemo_curator.stages.audio.common.ensure_waveform_2d(
    waveform: typing.Any
) -> torch.Tensor

Ensure waveform is a torch.Tensor in 2D (channels, samples) format.

nemo_curator.stages.audio.common.get_audio_duration(
    audio_filepath: str
) -> float

Get the duration of the audio file in seconds.

nemo_curator.stages.audio.common.load_audio_file(
    audio_path: str,
    mono: bool = True
) -> tuple[torch.Tensor, int]

Load audio file and return waveform tensor (channels, samples) and sample rate.

nemo_curator.stages.audio.common.resolve_model_path(
    model_path: str,
    reference_file: str,
    module_subdir: str
) -> str

Resolve a relative model path using the reference file’s directory and module subdirectory.

nemo_curator.stages.audio.common.resolve_waveform_from_item(
    item: dict[str, typing.Any],
    task_id: str,
    mono: bool = True
) -> tuple[torch.Tensor, int] | None

Resolve (waveform, sample_rate) from an item dict, loading from file if needed.

Checks item[‘waveform’] + item[‘sample_rate’], falls back to loading from item[‘audio_filepath’], resolves missing sample_rate from file header. Updates item in-place when loading from file. Returns None if resolution fails.