Load Local Audio Files

import os
import json
def create_audio_manifest(audio_dir: str, manifest_path: str):
    """Create a basic manifest for local audio files."""
    manifest_entries = []
    # Find all audio files in directory
    for filename in os.listdir(audio_dir):
        if filename.endswith(('.wav', '.flac', '.mp3', '.ogg')):
            audio_path = os.path.abspath(os.path.join(audio_dir, filename))
            # Basic entry without a reference transcript
            entry = {
                "audio_filepath": audio_path,
                "text": ""  # ASR writes pred_text; it does not populate text
            }
            manifest_entries.append(entry)
    # Write manifest file
    with open(manifest_path, 'w') as f:
        for entry in manifest_entries:
            f.write(json.dumps(entry) + '\n')
    print(f"Created manifest with {len(manifest_entries)} entries: {manifest_path}")
# Usage
create_audio_manifest(
    audio_dir="/path/to/your/audio/files",
    manifest_path="/absolute/path/to/local_audio_manifest.jsonl"
)

from pathlib import Path
from nemo_curator.pipeline import Pipeline
from nemo_curator.stages.audio.common import GetAudioDurationStage, ManifestReader, PreserveByValueStage
from nemo_curator.stages.audio.inference.asr.stage import ASRStage
from nemo_curator.stages.audio.io.convert import AudioToDocumentStage
from nemo_curator.stages.text.io.writer import JsonlWriter
def process_local_audio_manifest(manifest_path: str, output_dir: str):
    """Process local audio files using a manifest."""
    manifest_path = str(Path(manifest_path).expanduser().resolve())
    output_dir = str(Path(output_dir).expanduser().resolve())
    pipeline = Pipeline(name="local_audio_processing")
    # Load manifest
    pipeline.add_stage(ManifestReader(manifest_path=manifest_path))
    # ASR processing
    pipeline.add_stage(ASRStage(
        adapter_target="nemo_curator.models.asr.nemo_asr.NeMoASRAdapter",
        model_id="nvidia/stt_en_fastconformer_hybrid_large_pc",
        max_audio_sec_per_actor=240.0,
        max_inference_duration_s=120.0,
        local_bucketing=True,
        audio_filepath_key="audio_filepath",
    ))
    # Calculate duration and filter
    pipeline.add_stage(GetAudioDurationStage(
        audio_filepath_key="audio_filepath",
        duration_key="duration"
    ))
    # Keep files between 1-30 seconds
    pipeline.add_stage(PreserveByValueStage(
        input_value_key="duration",
        target_value=1.0,
        operator="ge"
    ))
    pipeline.add_stage(PreserveByValueStage(
        input_value_key="duration",
        target_value=30.0,
        operator="le"
    ))
    # Export results
    pipeline.add_stage(AudioToDocumentStage())
    pipeline.add_stage(JsonlWriter(path=output_dir))
    # Execute pipeline
    pipeline.run()
# Usage: First create manifest, then process
create_audio_manifest("/path/to/audio/files", "/absolute/path/to/local_manifest.jsonl")
process_local_audio_manifest("/absolute/path/to/local_manifest.jsonl", "/output/processed_audio")

import os
import json
def create_manifest_with_transcripts(audio_dir: str, transcript_dir: str, manifest_path: str):
    """Create manifest pairing audio files with existing transcriptions."""
    manifest_entries = []
    for filename in os.listdir(audio_dir):
        if filename.endswith(('.wav', '.flac', '.mp3', '.ogg')):
            # Find matching transcript file
            base_name = os.path.splitext(filename)[0]
            transcript_file = os.path.join(transcript_dir, f"{base_name}.txt")
            audio_path = os.path.abspath(os.path.join(audio_dir, filename))
            # Read transcript if it exists
            text = ""
            if os.path.exists(transcript_file):
                with open(transcript_file, 'r', encoding='utf-8') as f:
                    text = f.read().strip()
            entry = {
                "audio_filepath": audio_path,
                "text": text
            }
            manifest_entries.append(entry)
    # Write manifest
    with open(manifest_path, 'w') as f:
        for entry in manifest_entries:
            f.write(json.dumps(entry) + '\n')
    print(f"Created manifest with {len(manifest_entries)} entries: {manifest_path}")
# Usage
create_manifest_with_transcripts(
    audio_dir="/path/to/audio/files",
    transcript_dir="/path/to/transcripts",
    manifest_path="/absolute/path/to/paired_manifest.jsonl"
)

import os
def validate_manifest(manifest_path: str):
    """Check that all audio files in manifest exist."""
    missing_files = []
    valid_count = 0
    with open(manifest_path, 'r') as f:
        for line_num, line in enumerate(f, 1):
            entry = json.loads(line.strip())
            audio_path = entry.get("audio_filepath", "")
            if not os.path.exists(audio_path):
                missing_files.append(f"Line {line_num}: {audio_path}")
            else:
                valid_count += 1
    if missing_files:
        print(f"Warning: {len(missing_files)} missing files:")
        for missing in missing_files[:5]:  # Show first 5
            print(f"  {missing}")
    print(f"Validation complete: {valid_count} valid files")
    return len(missing_files) == 0
# Validate before processing
if validate_manifest("/absolute/path/to/local_manifest.jsonl"):
    process_local_audio_manifest("/absolute/path/to/local_manifest.jsonl", "/output")