AudioTask

1
from nemo_curator.tasks import AudioTask

1
from dataclasses import dataclass
2
3
@dataclass
4
class AudioTask(Task[dict]):
5
    """Task containing a single audio manifest entry for processing.
6
7
    Attributes:
8
        task_id: Framework-managed deterministic lineage identifier.
9
        dataset_name: Name of the source dataset.
10
        data: Audio manifest entry (single dict, stored as _AttrDict).
11
    """
12
13
    dataset_name: str
14
    data: dict  # _AttrDict subclass — supports attribute-style access
15
    # task_id is inherited from Task with init=False.

1
@property
2
def num_items(self) -> int:
3
    """Returns 1."""

1
from nemo_curator.tasks import AudioTask
2
3
# Single manifest entry
4
task = AudioTask(
5
    dataset_name="speech_dataset",
6
    data={
7
        "audio_filepath": "/data/audio/sample.wav",
8
        "duration": 5.2,
9
        "text": "Hello world",
10
    },
11
)
12
13
# Access fields via attribute or dict style
14
task.data["audio_filepath"]   # "/data/audio/sample.wav"
15
task.data.audio_filepath      # "/data/audio/sample.wav"

1
from dataclasses import dataclass
2
from nemo_curator.stages.base import ProcessingStage
3
from nemo_curator.tasks import AudioTask
4
5
@dataclass
6
class DurationFilterStage(ProcessingStage[AudioTask, AudioTask]):
7
    """Filter audio by duration."""
8
9
    name: str = "DurationFilter"
10
    min_duration: float = 1.0
11
    max_duration: float = 30.0
12
13
    def inputs(self) -> tuple[list[str], list[str]]:
14
        return ["data"], []
15
16
    def outputs(self) -> tuple[list[str], list[str]]:
17
        return ["data"], []
18
19
    def process(self, task: AudioTask) -> AudioTask | None:
20
        duration = task.data.get("duration", 0)
21
        if self.min_duration <= duration <= self.max_duration:
22
            return task
23
        return None

1
from dataclasses import dataclass, field
2
from nemo_curator.stages.resources import Resources
3
4
@dataclass
5
class MyGpuStage(ProcessingStage[AudioTask, AudioTask]):
6
    """GPU stage using process_batch."""
7
8
    name: str = "MyGpuStage"
9
    resources: Resources = field(
10
        default_factory=lambda: Resources(gpu_memory_gb=16.0)
11
    )
12
13
    def process(self, task: AudioTask) -> AudioTask:
14
        raise NotImplementedError("Use process_batch for GPU stages")
15
16
    def process_batch(self, tasks: list[AudioTask]) -> list[AudioTask]:
17
        # Batched GPU inference
18
        paths = [t.data["audio_filepath"] for t in tasks]
19
        results = self.model.infer(paths)
20
        for task, result in zip(tasks, results):
21
            task.data["pred_text"] = result
22
        return tasks

1
def process(self, task: AudioTask) -> AudioTask:
2
    audio_path = task.data["audio_filepath"]
3
    task.data["pred_text"] = self.asr_model.transcribe(audio_path)
4
    return task

1
def process(self, task: AudioTask) -> AudioTask:
2
    if "text" in task.data and "pred_text" in task.data:
3
        task.data["wer"] = compute_wer(task.data["text"], task.data["pred_text"])
4
    return task