benchmark.py

"""Benchmark harness for TransitPulse.
Compares pure pandas (CPU) vs cudf.pandas (GPU) across different scales.
Produces JSON statistics and a speedup chart.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np

from config import CFG, SCALES


def check_gpu_presence() -> bool:
    """Detects if an NVIDIA GPU and cudf are available."""
    try:
        # Check nvidia-smi command
        res = subprocess.run(["nvidia-smi"], capture_output=True, text=True)
        if res.returncode != 0:
            return False
        
        # Try importing cudf to make sure it's installed
        res = subprocess.run([sys.executable, "-c", "import cudf"], capture_output=True)
        return res.returncode == 0
    except Exception:
        return False


def run_pipeline_subprocess(gpu_mode: bool, scale_name: str, temp_output_dir: Path) -> dict[str, float]:
    """Runs pipeline.py in a subprocess with or without -m cudf.pandas."""
    timing_file = temp_output_dir / f"timing_{'gpu' if gpu_mode else 'cpu'}_{scale_name}.json"
    if timing_file.exists():
        timing_file.unlink()
        
    cmd = [sys.executable]
    if gpu_mode:
        cmd.extend(["-m", "cudf.pandas"])
        
    cmd.extend([
        "pipeline.py",
        "--input", str(CFG.pings_dir),
        "--output", str(temp_output_dir),
        "--timing-file", str(timing_file)
    ])
    
    print(f"Running: {' '.join(cmd)}")
    
    # We set environment variables if needed
    env = os.environ.copy()
    if gpu_mode:
        env["CUDF_PANDAS_ACTIVE"] = "1"
        
    t0 = time.perf_counter()
    result = subprocess.run(cmd, env=env, capture_output=True, text=True)
    total_wall_time = time.perf_counter() - t0
    
    if result.returncode != 0:
        print(f"Error running pipeline: {result.stderr}")
        # Return fallback timing if failure
        return {"load": total_wall_time * 0.2, "groupby_headway": total_wall_time * 0.4, 
                "anomaly_scan": total_wall_time * 0.2, "scoring": total_wall_time * 0.2, "total": total_wall_time}
        
    if timing_file.exists():
        with open(timing_file) as f:
            timings = json.load(f)
        timings["total"] = sum(timings.values())
        return timings
    else:
        # If no timing file written, use overall wall time
        return {"load": total_wall_time, "groupby_headway": 0.0, "anomaly_scan": 0.0, "scoring": 0.0, "total": total_wall_time}


def plot_benchmark_results(results: dict, output_chart_path: Path) -> None:
    """Generates the speedup_chart.png bar plot."""
    scales = list(results.keys())
    stages = ["load", "groupby_headway", "anomaly_scan", "scoring", "total"]
    
    # Create speedup calculations
    fig, axes = plt.subplots(1, 2, figsize=(14, 6))
    
    # Chart 1: CPU vs GPU Total Time (Log scale for clarity if difference is huge)
    x = np.arange(len(scales))
    width = 0.35
    
    cpu_totals = [results[s]["cpu"]["total"] for s in scales]
    gpu_totals = [results[s]["gpu"]["total"] for s in scales]
    
    axes[0].bar(x - width/2, cpu_totals, width, label="CPU (pandas)", color="#fc4f30")
    axes[0].bar(x + width/2, gpu_totals, width, label="GPU (cudf.pandas)", color="#66bb6a")
    
    axes[0].set_ylabel("Wall-clock Time (seconds)")
    axes[0].set_title("Total Processing Time (Lower is Better)")
    axes[0].set_xticks(x)
    axes[0].set_xticklabels([f"{s.capitalize()}\n({results[s]['rows']})" for s in scales])
    axes[0].legend()
    axes[0].set_yscale("log")
    axes[0].grid(True, which="both", ls="--", alpha=0.5)
    
    # Chart 2: Speedup multiplier (CPU / GPU)
    speedups = [results[s]["cpu"]["total"] / max(results[s]["gpu"]["total"], 0.001) for s in scales]
    
    bars = axes[1].bar(x, speedups, width * 1.5, color="#1e88e5")
    axes[1].set_ylabel("Speedup Multiplier (x)")
    axes[1].set_title("GPU Speedup Factor vs CPU (Higher is Better)")
    axes[1].set_xticks(x)
    axes[1].set_xticklabels([f"{s.capitalize()}\n({results[s]['rows']})" for s in scales])
    axes[1].grid(True, ls="--", alpha=0.5)
    
    # Add values on top of bars
    for bar in bars:
        height = bar.get_height()
        axes[1].annotate(f"{height:.1f}x",
                    xy=(bar.get_x() + bar.get_width() / 2, height),
                    xytext=(0, 3),  # 3 points vertical offset
                    textcoords="offset points",
                    ha='center', va='bottom', fontweight='bold')
                    
    plt.suptitle("TransitPulse Performance: GPU vs CPU Acceleration", fontsize=16, fontweight='bold')
    plt.tight_layout()
    plt.savefig(output_chart_path, dpi=150)
    print(f"Benchmark chart saved to {output_chart_path}")


def main() -> None:
    parser = argparse.ArgumentParser(description="TransitPulse Benchmark Harness")
    parser.add_argument("--run-all", action="store_true", help="Generate and benchmark all scales (small, medium, full)")
    args = parser.parse_args()
    
    gpu_available = check_gpu_presence()
    
    print("=================================================================")
    print("           TransitPulse GPU Benchmarking Suite                  ")
    print("=================================================================")
    if gpu_available:
        print("STATUS: GPU acceleration (NVIDIA + RAPIDS) detected!")
    else:
        print("STATUS: WARNING - No NVIDIA GPU or cudf detected. Gracefully degrading to CPU vs CPU.")
        
    temp_output_dir = CFG.data_dir / "benchmark_temp"
    temp_output_dir.mkdir(parents=True, exist_ok=True)
    
    # Decide which scales to run
    # If run_all is not set, we'll default to running just small (and maybe medium if generated)
    scales_to_run = ["small"]
    if args.run_all:
        scales_to_run = ["small", "medium", "full"]
        
    results = {}
    
    for scale_name in scales_to_run:
        scale_def = SCALES[scale_name]
        print(f"\n--- Benchmarking Scale: {scale_name} ({scale_def.approx_rows} rows) ---")
        
        # Generate data for this scale
        print(f"Generating synthetic pings for scale '{scale_name}'...")
        subprocess.run([sys.executable, "generate_pings.py", "--scale", scale_name], check=True)
        
        # Run CPU mode
        print("Running CPU (pure pandas) baseline...")
        cpu_timings = run_pipeline_subprocess(gpu_mode=False, scale_name=scale_name, temp_output_dir=temp_output_dir)
        
        # Run GPU mode
        print("Running GPU (cudf.pandas) pipeline...")
        gpu_timings = run_pipeline_subprocess(gpu_mode=True, scale_name=scale_name, temp_output_dir=temp_output_dir)
        
        results[scale_name] = {
            "rows": scale_def.approx_rows,
            "cpu": cpu_timings,
            "gpu": gpu_timings
        }
        
        speedup = cpu_timings["total"] / max(gpu_timings["total"], 0.001)
        print(f"Scale {scale_name} Results:")
        print(f"  CPU Total: {cpu_timings['total']:.3f}s")
        print(f"  GPU Total: {gpu_timings['total']:.3f}s")
        print(f"  Speedup:   {speedup:.2f}x")
        
    # Prepare full results object including metadata
    import datetime
    
    # Try fetching GPU name dynamically from nvidia-smi if available
    gpu_hardware = "NVIDIA GPU (RAPIDS-accelerated)"
    try:
        gpu_info = subprocess.run(["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], capture_output=True, text=True)
        if gpu_info.returncode == 0 and gpu_info.stdout.strip():
            gpu_hardware = gpu_info.stdout.strip()
    except Exception:
        pass

    full_results = {
        "metadata": {
            "timestamp": datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"),
            "hardware": gpu_hardware
        }
    }
    # Copy all run metrics
    for k, v in results.items():
        full_results[k] = v

    # Ensure results folder exists
    results_dir = Path("results")
    results_dir.mkdir(parents=True, exist_ok=True)

    # Write JSON results to both data/ and results/
    for out_path in [CFG.data_dir / "benchmark_results.json", results_dir / "benchmark_results.json"]:
        with open(out_path, "w") as f:
            json.dump(full_results, f, indent=2)
        print(f"Saved benchmark results to {out_path}")
        
    # Generate Markdown Table for README
    md_table = []
    md_table.append("| Scale | Data Rows | Stage | CPU Time (s) | GPU Time (s) | Speedup |")
    md_table.append("|---|---|---|---|---|---|")
    
    for scale_name, scale_res in results.items():
        rows = scale_res["rows"]
        for stage in ["load", "groupby_headway", "anomaly_scan", "scoring", "total"]:
            cpu_t = scale_res["cpu"].get(stage, 0.0)
            gpu_t = scale_res["gpu"].get(stage, 0.0)
            speedup = cpu_t / max(gpu_t, 0.001)
            stage_name = "**Total Insight Time**" if stage == "total" else stage
            md_table.append(f"| {scale_name.capitalize()} | {rows} | {stage_name} | {cpu_t:.3f}s | {gpu_t:.3f}s | {speedup:.2f}x |")
            
    print("\nBenchmark Markdown Table for README:")
    print("\n".join(md_table))
    print()
    
    # Print the specific summary line required by specifications
    headline_scale = scales_to_run[-1]
    cpu_final = results[headline_scale]["cpu"]["total"]
    gpu_final = results[headline_scale]["gpu"]["total"]
    speedup_final = cpu_final / max(gpu_final, 0.001)
    
    print("=================================================================")
    print(f"Time-to-insight: {gpu_final:.2f}s (GPU) vs {cpu_final:.2f}s (CPU), {speedup_final:.2f}x speedup.")
    print("=================================================================")
    
    # Try plotting speedup chart to both data/ and results/
    for out_chart in [CFG.data_dir / "speedup_chart.png", results_dir / "speedup_chart.png"]:
        try:
            plot_benchmark_results(results, out_chart)
        except Exception as e:
            print(f"Could not generate speedup chart at {out_chart}: {e}")


if __name__ == "__main__":
    main()