benchmark_data_reader.py

#!/usr/bin/env python3
"""
Benchmark Data Reader for LLM Inference Performance Dashboard

This module provides functionality to read benchmark result files and convert them
into a flattened Polars DataFrame for analysis and visualization.
"""

import json
import polars as pl
from pathlib import Path
from typing import List, Dict, Any, Optional
import logging

logger = logging.getLogger(__name__)


class BenchmarkDataReader:
    """Reader for benchmark result JSON files that flattens data into a Polars DataFrame."""

    def __init__(self, benchmark_dir: str = "benchmark_results"):
        """
        Initialize the benchmark data reader.

        Args:
            benchmark_dir: Directory containing benchmark result files
        """
        self.benchmark_dir = Path(benchmark_dir)

    def read_benchmark_files(self) -> pl.DataFrame:
        """
        Read all benchmark files and return a flattened Polars DataFrame.

        Returns:
            Polars DataFrame where each row represents a benchmark scenario with all metrics
        """
        all_records = []

        # Find all individual model benchmark files (exclude summary files)
        benchmark_files = list(self.benchmark_dir.rglob("*_benchmark_*.json"))
        benchmark_files = [f for f in benchmark_files if "summary" not in f.name]

        logger.info(f"Found {len(benchmark_files)} benchmark files")

        for file_path in benchmark_files:
            try:
                records = self._process_benchmark_file(file_path)
                all_records.extend(records)
                logger.debug(f"Processed {len(records)} scenarios from {file_path}")
            except Exception as e:
                logger.error(f"Error processing {file_path}: {e}")
                continue

        if not all_records:
            logger.warning("No benchmark data found")
            return pl.DataFrame()

        # Create DataFrame from all records
        df = pl.DataFrame(all_records)
        logger.info(f"Created DataFrame with {len(df)} rows and {len(df.columns)} columns")

        return df

    def _process_benchmark_file(self, file_path: Path) -> List[Dict[str, Any]]:
        """
        Process a single benchmark file and extract all scenarios.

        Args:
            file_path: Path to the benchmark JSON file

        Returns:
            List of flattened records, one per benchmark scenario
        """
        with open(file_path, 'r') as f:
            data = json.load(f)

        records = []
        model_name = data.get("model_name", "unknown")

        for scenario in data.get("benchmark_scenarios", []):
            record = self._flatten_scenario(scenario, model_name, file_path)
            records.append(record)

        return records

    def _flatten_scenario(self, scenario: Dict[str, Any], model_name: str, file_path: Path) -> Dict[str, Any]:
        """
        Flatten a single benchmark scenario into a flat record.

        Args:
            scenario: Scenario data from benchmark file
            model_name: Name of the model being benchmarked
            file_path: Path to the original file

        Returns:
            Flattened dictionary with all metrics and metadata
        """
        record = {
            # File metadata
            "file_path": str(file_path),
            "model_name": model_name,

            # Scenario metadata
            "scenario_name": scenario.get("scenario_name", "unknown"),
        }

        # Add metadata fields
        metadata = scenario.get("metadata", {})
        record.update({
            "timestamp": metadata.get("timestamp"),
            "commit_id": metadata.get("commit_id"),
        })

        # Add hardware info
        hw_info = metadata.get("hardware_info", {})
        record.update({
            "gpu_name": hw_info.get("gpu_name"),
            "gpu_memory_total_mb": hw_info.get("gpu_memory_total_mb"),
            "cpu_count": hw_info.get("cpu_count"),
            "memory_total_mb": hw_info.get("memory_total_mb"),
            "python_version": hw_info.get("python_version"),
            "torch_version": hw_info.get("torch_version"),
            "cuda_version": hw_info.get("cuda_version"),
        })

        # Add config info
        config = metadata.get("config", {})
        record.update({
            "config_name": config.get("name"),
            "model_id": config.get("model_id"),
            "variant": config.get("variant"),
            "warmup_iterations": config.get("warmup_iterations"),
            "measurement_iterations": config.get("measurement_iterations"),
            "num_tokens_to_generate": config.get("num_tokens_to_generate"),
            "device": config.get("device"),
            "torch_dtype": config.get("torch_dtype"),
            "compile_mode": config.get("compile_mode"),
            "use_cache": config.get("use_cache"),
            "batch_size": config.get("batch_size"),
            "sequence_length": config.get("sequence_length"),
            "attn_implementation": config.get("attn_implementation"),
            "sdpa_backend": config.get("sdpa_backend"),
        })

        # Add measurement statistics for each metric
        measurements = scenario.get("measurements", {})
        for metric_name, metric_data in measurements.items():
            if isinstance(metric_data, dict):
                # Add statistics for this metric
                for stat_name, stat_value in metric_data.items():
                    if stat_name != "measurements":  # Skip raw measurements array
                        record[f"{metric_name}_{stat_name}"] = stat_value

        # Add GPU metrics
        gpu_metrics = scenario.get("gpu_metrics", {})
        for gpu_metric, value in gpu_metrics.items():
            record[f"gpu_{gpu_metric}"] = value

        return record

    def get_summary_statistics(self, df: pl.DataFrame) -> Dict[str, Any]:
        """
        Generate summary statistics from the benchmark DataFrame.

        Args:
            df: Benchmark DataFrame

        Returns:
            Dictionary with summary statistics
        """
        if df.is_empty():
            return {}

        return {
            "total_scenarios": len(df),
            "unique_models": df["model_name"].n_unique(),
            "unique_scenarios": df["scenario_name"].n_unique(),
            "unique_hardware": df["gpu_name"].n_unique(),
            "date_range": {
                "earliest": df["timestamp"].min(),
                "latest": df["timestamp"].max(),
            },
            "performance_metrics": {
                "avg_latency_seconds": df.select(pl.col("latency_seconds_mean").mean()).item(),
                "avg_tokens_per_second": df.select(pl.col("tokens_per_second_mean").mean()).item(),
                "avg_time_to_first_token": df.select(pl.col("time_to_first_token_seconds_mean").mean()).item(),
            } if "latency_seconds_mean" in df.columns else None
        }


def main():
    """Example usage of the BenchmarkDataReader."""
    logging.basicConfig(level=logging.INFO)

    # Create reader and load data
    reader = BenchmarkDataReader()
    df = reader.read_benchmark_files()

    if df.is_empty():
        print("No benchmark data found!")
        return

    # Display basic info
    print(f"\nLoaded benchmark data: {len(df)} scenarios")
    print(f"Columns: {len(df.columns)}")
    print("\nColumn names:")
    for col in sorted(df.columns):
        print(f"  - {col}")

    # Show summary statistics
    summary = reader.get_summary_statistics(df)
    print(f"\nSummary Statistics:")
    for key, value in summary.items():
        print(f"  {key}: {value}")

    # Show sample data
    print(f"\nSample data (first 3 rows):")
    print(df.head(3))

    return df


if __name__ == "__main__":
    df = main()