momentry_core/test_performance_quick.py

#!/opt/homebrew/bin/python3.11
"""Quick performance test for ASR processor."""

import subprocess
import tempfile
import os
import time
import sys
import json
import statistics


def test_small_video():
    """Test with small video to establish baseline."""
    test_video = "/Users/accusys/test_video/BigBuckBunny_320x180.mp4"
    if not os.path.exists(test_video):
        print(f"Small test video not found: {test_video}")
        return None

    print(f"Testing small video: {os.path.basename(test_video)}")

    runs = 3
    times = []

    for run in range(runs):
        with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
            output_path = f.name

        try:
            cmd = [
                "/opt/homebrew/bin/python3.11",
                "scripts/asr_processor.py",
                test_video,
                output_path,
                "--chunk-duration",
                "600",
            ]

            env = os.environ.copy()
            env["MOMENTRY_DISABLE_REDIS"] = "1"

            start = time.time()
            result = subprocess.run(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                env=env,
            )
            elapsed = time.time() - start

            if result.returncode == 0:
                times.append(elapsed)
                print(f"  Run {run + 1}: {elapsed:.1f}s")
            else:
                print(f"  Run {run + 1}: failed (code {result.returncode})")

        except Exception as e:
            print(f"  Run {run + 1}: error: {e}")
        finally:
            if os.path.exists(output_path):
                os.unlink(output_path)

    if times:
        avg = statistics.mean(times)
        stdev = statistics.stdev(times) if len(times) > 1 else 0
        print(f"\nResults: {len(times)} successful runs")
        print(f"  Average: {avg:.1f}s")
        print(f"  Std dev: {stdev:.1f}s")
        print(f"  Range: {min(times):.1f}s - {max(times):.1f}s")
        return avg
    else:
        print("\nNo successful runs")
        return None


def test_with_checkpoint():
    """Test checkpoint functionality."""
    test_video = "/Users/accusys/test_video/BigBuckBunny_320x180.mp4"
    if not os.path.exists(test_video):
        print(f"Test video not found: {test_video}")
        return

    print(f"\nTesting checkpoint functionality...")

    # Force chunked mode with small max_direct_duration
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
        output_path = f.name

    try:
        cmd = [
            "/opt/homebrew/bin/python3.11",
            "scripts/asr_processor.py",
            test_video,
            output_path,
            "--chunk-duration",
            "30",  # Small chunks for testing
            "--max-direct-duration",
            "1",  # Force chunked mode
        ]

        env = os.environ.copy()
        env["MOMENTRY_DISABLE_REDIS"] = "1"
        env["ASR_DEBUG"] = "1"

        print(f"Running with forced chunked mode...")
        start = time.time()
        result = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            env=env,
            timeout=60,
        )
        elapsed = time.time() - start

        print(f"  Time: {elapsed:.1f}s")
        print(f"  Return code: {result.returncode}")

        # Check for checkpoint file
        checkpoint_path = output_path + ".checkpoint"
        if os.path.exists(checkpoint_path):
            print(
                f"  Checkpoint file created: Yes ({os.path.getsize(checkpoint_path)} bytes)"
            )
            # Load and inspect checkpoint
            try:
                with open(checkpoint_path, "r") as f:
                    checkpoint = json.load(f)
                print(
                    f"  Checkpoint contains: {len(checkpoint.get('segments', []))} segments"
                )
                print(
                    f"  Processed chunks: {len(checkpoint.get('processed_chunks', []))}"
                )
            except Exception as e:
                print(f"  Failed to read checkpoint: {e}")
        else:
            print(f"  Checkpoint file created: No")

        # Check if checkpoint was cleaned up after completion
        # (should be cleaned up if processing completed)
        if result.returncode == 0:
            if not os.path.exists(checkpoint_path):
                print(f"  Checkpoint cleaned up after completion: Yes")
            else:
                print(f"  Checkpoint cleaned up after completion: No (still exists)")

    except subprocess.TimeoutExpired:
        print(f"  Timeout after 60 seconds")
    except Exception as e:
        print(f"  Error: {e}")
    finally:
        if os.path.exists(output_path):
            os.unlink(output_path)
        checkpoint_path = output_path + ".checkpoint"
        if os.path.exists(checkpoint_path):
            os.unlink(checkpoint_path)


if __name__ == "__main__":
    print("ASR Processor Quick Performance Test")
    print("=" * 60)

    # Baseline performance
    baseline = test_small_video()

    # Checkpoint test
    test_with_checkpoint()

    print("\n" + "=" * 60)
    if baseline:
        print(f"Baseline performance: {baseline:.1f}s")
        # Compare with previous baseline (~18.9s from earlier test)
        previous_baseline = 18.9
        change = ((baseline - previous_baseline) / previous_baseline) * 100
        print(f"Change from previous: {change:+.1f}%")

        if change <= 5:
            print("✅ Within 5% overhead limit")
        else:
            print(f"⚠️  Overhead exceeds 5% limit")

    print("\nCheckpoint functionality appears to be integrated.")