analyzer.py

#!/usr/bin/env python3
"""Analyzer for C3 demo.

This script reads one or two summary JSON files generated by the demo stub and
produces a human‑readable text summary as well as a comparison plot. It can
operate in two modes:

* When both --baseline and --obfuscated are provided, it compares the two
  runs and writes both a summary.txt and a plot.png.
* When only one summary is provided, it still writes a summary and produces
  a single‑series plot.

The plot uses matplotlib with the Agg backend so it can run headlessly.
"""

import argparse
import json
import os
import sys
from typing import Any, Dict, Optional


def load_summary(path: str) -> Dict[str, Any]:
    """Load a summary.json file and return its contents."""
    with open(path, 'r', encoding='utf-8') as f:
        return json.load(f)


def compute_summary_lines(baseline: Dict[str, Any], obfuscated: Optional[Dict[str, Any]], preset: str) -> str:
    """Produce a concise multi‑line summary comparing baseline and obfuscated runs.

    Returns a string with 6–10 lines as required by the specification.
    """
    lines = []
    lines.append(f"preset={preset}")
    duration = baseline.get('duration_s')
    req_baseline = baseline.get('req_total')
    req_obfuscated = obfuscated.get('req_total') if obfuscated else None
    lines.append(f"duration_s={duration}  req_total_baseline={req_baseline}" +
                 (f"  req_total_obfuscated={req_obfuscated}" if req_obfuscated is not None else ""))
    lines.append(f"baseline_p50_ms={baseline.get('p50_latency_ms')}  baseline_p95_ms={baseline.get('p95_latency_ms')}")
    if obfuscated:
        lines.append(
            f"obfuscated_p50_ms={obfuscated.get('p50_latency_ms')}  obfuscated_p95_ms={obfuscated.get('p95_latency_ms')}")
    lines.append(f"baseline_bytes_in={baseline.get('bytes_in')}  baseline_bytes_out={baseline.get('bytes_out')}")
    if obfuscated:
        lines.append(
            f"obfuscated_bytes_in={obfuscated.get('bytes_in')}  obfuscated_bytes_out={obfuscated.get('bytes_out')}")
    lines.append("status=OK")
    return "\n".join(lines)


def generate_plot(baseline: Dict[str, Any], obfuscated: Optional[Dict[str, Any]], out_path: str) -> None:
    """Generate a PNG plot comparing baseline and obfuscated distributions.

    The figure contains two stacked subplots: the top shows packet size
    distributions and the bottom shows inter‑packet delay (IPD) distributions.
    The plot is saved to out_path/plot.png.
    """
    import matplotlib
    # Force the Agg backend for headless environments
    matplotlib.use('Agg')
    import matplotlib.pyplot as plt

    sizes_base = baseline.get('sizes', [])
    ipd_base = baseline.get('ipd_ms', [])
    sizes_obf = obfuscated.get('sizes', []) if obfuscated else []
    ipd_obf = obfuscated.get('ipd_ms', []) if obfuscated else []

    # Create a figure with two stacked subplots
    fig, axes = plt.subplots(2, 1, figsize=(6, 8))
    # Packet sizes
    ax1 = axes[0]
    if sizes_base:
        ax1.hist(sizes_base, bins=30, alpha=0.6, label='baseline', color='tab:blue')
    if sizes_obf:
        ax1.hist(sizes_obf, bins=30, alpha=0.6, label='obfuscated', color='tab:orange')
    ax1.set_title('Packet size distribution')
    ax1.set_xlabel('bytes')
    ax1.set_ylabel('count')
    if sizes_obf:
        ax1.legend()

    # IPD distributions
    ax2 = axes[1]
    if ipd_base:
        ax2.hist(ipd_base, bins=30, alpha=0.6, label='baseline', color='tab:blue')
    if ipd_obf:
        ax2.hist(ipd_obf, bins=30, alpha=0.6, label='obfuscated', color='tab:orange')
    ax2.set_title('Inter‑packet delay distribution')
    ax2.set_xlabel('milliseconds')
    ax2.set_ylabel('count')
    if ipd_obf:
        ax2.legend()

    plt.tight_layout()
    os.makedirs(out_path, exist_ok=True)
    fig.savefig(os.path.join(out_path, 'plot.png'), dpi=150)
    plt.close(fig)


def main() -> None:
    parser = argparse.ArgumentParser(description='C3 analyzer: compare baseline and obfuscated runs')
    parser.add_argument('--baseline', type=str, help='Path to baseline summary.json', required=False)
    parser.add_argument('--obfuscated', type=str, help='Path to obfuscated summary.json', required=False)
    parser.add_argument('--out', type=str, help='Output directory', required=False)
    args = parser.parse_args()

    # Determine input summaries
    baseline_path = args.baseline
    obfuscated_path = args.obfuscated
    # If not provided, try to find them automatically in the latest artifacts folder
    if not baseline_path or not args.out:
        # Attempt to infer from current working directory
        # Find the latest artifacts directory
        runs = []
        if os.path.isdir('artifacts'):
            runs = sorted([
                os.path.join('artifacts', d) for d in os.listdir('artifacts')
                if os.path.isdir(os.path.join('artifacts', d))
            ], reverse=True)
        if runs:
            latest = runs[0]
            if not baseline_path:
                candidate = os.path.join(latest, 'baseline_summary.json')
                if os.path.isfile(candidate):
                    baseline_path = candidate
            if not args.out:
                args.out = latest
    if not baseline_path:
        print('No baseline summary provided and none found automatically.', file=sys.stderr)
        sys.exit(1)
    baseline = load_summary(baseline_path)
    obfuscated = None
    if obfuscated_path and os.path.isfile(obfuscated_path):
        obfuscated = load_summary(obfuscated_path)

    # Determine preset (stored in both summaries)
    preset = baseline.get('preset', 'unknown')
    if obfuscated and obfuscated.get('preset'):
        preset = obfuscated['preset']
    out_dir = args.out or os.path.dirname(baseline_path)
    os.makedirs(out_dir, exist_ok=True)
    # Write summary.txt
    summary_lines = compute_summary_lines(baseline, obfuscated, preset)
    with open(os.path.join(out_dir, 'summary.txt'), 'w', encoding='utf-8') as f:
        f.write(summary_lines + '\n')
    # Generate plot
    generate_plot(baseline, obfuscated, out_dir)
    # Also print to stdout for convenience
    print(summary_lines)


if __name__ == '__main__':
    main()