Skip to content

feat(audio-advanced-pipelines): make AudioDataFilterStage agent-ready #2342

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shubhamNvidia wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(audio-advanced-pipelines): make AudioDataFilterStage agent-ready #2342

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion nemo_curator/stages/audio/advanced_pipelines/audio_data_filter/audio_data_filter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,17 +46,19 @@

from loguru import logger

from nemo_curator.stages.audio._agent._agent_ready import AgentReady, Gates, StageContract
from nemo_curator.stages.audio.filtering import BandFilterStage, SIGMOSFilterStage, UTMOSFilterStage
from nemo_curator.stages.audio.postprocessing import TimestampMapperStage
from nemo_curator.stages.audio.preprocessing import MonoConversionStage, SegmentConcatenationStage
from nemo_curator.stages.audio.segmentation import SpeakerSeparationStage, VADSegmentationStage
from nemo_curator.stages.base import CompositeStage, ProcessingStage
from nemo_curator.stages.resources import Resources
from nemo_curator.tasks import AudioTask

from .config import _deep_merge, get_enabled_stages, load_config


class AudioDataFilterStage(CompositeStage):
class AudioDataFilterStage(AgentReady, CompositeStage[AudioTask, AudioTask]):
"""Complete audio data filtering and curation pipeline (CompositeStage).

Decomposes into independent stages that the executor can schedule with
Expand Down Expand Up @@ -90,6 +92,16 @@ def __init__(
if config:
self._cfg = _deep_merge(self._cfg, config)

def describe(self) -> StageContract:
return StageContract(
wrappable=False,
# True for every topology because it is true of each delegate: mono conversion, VAD,
# the three quality filters, concatenation, speaker separation and the timestamp
# mapper all work from the row they are handed. The factories below leave
# ``write_to_disk`` unset throughout, so none of them even opens a shared directory.
gates=Gates(per_row_independent=True),
)

def decompose(self) -> list[ProcessingStage]:
"""Build a self-consistent pipeline topology based on enabled features."""
cfg = self._cfg
Expand Down
Loading