scripts/run_demo.py

#!/usr/bin/env python3
"""
scripts/run_demo.py — End-to-end execution demo of the Voice AI Speech Data Pipeline.

Creates a temporary dummy wave file, executes all stages of the pipeline
using zero-dependency fallback configurations, and shows stats/outputs.
"""

import os
import tempfile
import wave
import struct
import math
from pathlib import Path

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


def create_dummy_wav(path: Path, duration_s: float = 5.0, sample_rate: int = 16000) -> None:
    """Create a dummy PCM 16-bit 16kHz mono WAV file containing a sine wave."""
    num_samples = int(duration_s * sample_rate)
    with wave.open(str(path), "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2) # 16-bit
        w.setframerate(sample_rate)
        
        # Write wave values (mixing two sine waves for more energy diversity)
        for i in range(num_samples):
            val1 = 32767.0 * 0.4 * math.sin(2.0 * math.pi * 440.0 * i / sample_rate)
            val2 = 32767.0 * 0.2 * math.sin(2.0 * math.pi * 880.0 * i / sample_rate)
            val = int(val1 + val2)
            data = struct.pack("<h", val)
            w.writeframes(data)


def main() -> None:
    print("=" * 80)
    print("      Voice AI SPEECH DATA PROCESSING PIPELINE — END-TO-END DEMO RUN      ")
    print("=" * 80)

    # 1. Initialize logging
    setup_logging(log_level="INFO", structured=False)

    # 2. Setup a clean sandbox directory in workspace/Voice AI-int/data/demo_sandbox
    import shutil
    sandbox_dir = Path("/workspace/Voice AI-int/data/demo_sandbox").resolve()
    if sandbox_dir.exists():
        shutil.rmtree(sandbox_dir)
    sandbox_dir.mkdir(parents=True, exist_ok=True)
    
    dummy_wav = sandbox_dir / "demo_tone.wav"
    create_dummy_wav(dummy_wav, duration_s=6.0)
    print(f"Created dummy audio tone: {dummy_wav.name} ({dummy_wav.stat().st_size / 1024:.1f} KB)")

    # 3. Configure env variables to use sandbox paths and fallbacks
    os.environ["VOICE_AUDIO_OUTPUT_DIR"] = str(sandbox_dir / "processed")
    os.environ["VOICE_VAD_OUTPUT_DIR"] = str(sandbox_dir / "segments")
    os.environ["VOICE_DIARIZATION_OUTPUT_DIR"] = str(sandbox_dir / "annotations")
    os.environ["VOICE_EXPORT_OUTPUT_DIR"] = str(sandbox_dir / "exports")
    os.environ["VOICE_REPORTING_OUTPUT_DIR"] = str(sandbox_dir / "reports")
    os.environ["VOICE_PIPELINE_CHECKPOINT_DIR"] = str(sandbox_dir / ".checkpoints")
    
    os.environ["VOICE_DIARIZATION_BACKEND"] = "simple"
    os.environ["VOICE_EMOTION_BACKEND"] = "rule_based"
    os.environ["VOICE_ASR_ENABLED"] = "false"
    os.environ["VOICE_VAD_BACKEND"] = "energy"
    os.environ["VOICE_AUDIO_NOISE_REDUCE"] = "false"

    # Relax validation metrics so pure tone passes quality checks
    os.environ["VOICE_VALIDATION_MIN_SNR_DB"] = "-10.0"
    os.environ["VOICE_VALIDATION_MAX_SILENCE_RATIO"] = "1.0"
    os.environ["VOICE_VALIDATION_MIN_SPEECH_RATIO"] = "0.0"
    os.environ["VOICE_VALIDATION_MIN_EMOTION_CONFIDENCE"] = "0.0"
    os.environ["VOICE_VALIDATION_MIN_DIARIZATION_CONFIDENCE"] = "0.0"

    # 4. Instantiate and run pipeline
    print("\nStarting pipeline orchestrator...")
    pipeline = SpeechDataPipeline()
    export_dir = pipeline.run_on_file(dummy_wav)

    print("\n" + "=" * 80)
    print("                             PIPELINE RUN OUTPUTS                           ")
    print("=" * 80)
    print(f"Export Directory: {export_dir}")
    print("\nExported Files:")
    for file in sorted(export_dir.glob("*")):
        if file.is_file():
            print(f"  - {file.name} ({file.stat().st_size} bytes)")
        elif file.is_dir():
            print(f"  - {file.name}/ ({len(list(file.glob('*')))} files)")

    print("\nSegments Manifest (segments.jsonl) Content:")
    with open(export_dir / "segments.jsonl") as f:
        print(f.read())
        
    print("=" * 80)


if __name__ == "__main__":
    main()