scripts/test_pipeline_single.py

#!/usr/bin/env python3
import os
import sys
import shutil
from pathlib import Path

# Ensure we run from repo root
REPO_ROOT = Path(__file__).resolve().parent.parent
os.chdir(REPO_ROOT)
sys.path.insert(0, str(REPO_ROOT))

# Import pipeline
import voice_pipeline
from voice_pipeline.pipeline import SpeechDataPipeline
from voice_pipeline.utils.logger import setup_logging

def main():
    setup_logging(log_level="INFO")
    
    # Run test on 6Ak15wj9ePw.wav
    audio_file = Path("/root/audiodown/6Ak15wj9ePw.wav")
    if not audio_file.exists():
        print(f"Error: {audio_file} does not exist!")
        sys.exit(1)
        
    print(f"Starting test pipeline on: {audio_file}")
    
    # We will use the pipeline_config.yaml config directly
    config_path = REPO_ROOT / "voice_pipeline" / "configs" / "pipeline_config.yaml"
    
    # Clear validation checkpoints just for this file so it runs completely
    from voice_pipeline.utils.checkpoint import CheckpointManager
    for stage in ["download", "ingest", "audio_normalize", "vad", "diarization", "emotion", "asr", "validation"]:
        ckpt = CheckpointManager(REPO_ROOT / "data" / ".checkpoints", stage)
        # Clear if exists
        ckpt_path = Path(REPO_ROOT / "data" / ".checkpoints" / f"{stage}_done.jsonl")
        if ckpt_path.exists():
            # Filter out this file's entries (hash of file_path)
            from voice_pipeline.utils.file_utils import get_file_hash
            file_hash = get_file_hash(audio_file, algo="md5")
            file_id = file_hash[:16]
            lines = []
            with open(ckpt_path, "r") as f:
                for line in f:
                    if file_id not in line:
                        lines.append(line)
            with open(ckpt_path, "w") as f:
                f.writelines(lines)
            print(f"Cleared checkpoint for stage {stage} for file {file_id}")

    pipeline = SpeechDataPipeline(config_path=config_path)
    
    print("Running pipeline...")
    export_dir = pipeline.run_on_file(
        file_path=audio_file,
        language="hi",
        content_type="debate_podcast"
    )
    
    print("\nPipeline run finished successfully!")
    print(f"Export directory: {export_dir}")
    
    # Verify outputs
    segments_file = export_dir / "segments.jsonl"
    if segments_file.exists():
        print("\nFirst few lines of segments.jsonl:")
        with open(segments_file, "r") as f:
            for _ in range(5):
                line = f.readline()
                if not line:
                    break
                print(line.strip())
    else:
        print("\nWarning: segments.jsonl was not generated.")

if __name__ == "__main__":
    main()