analysis.py

#!/usr/bin/env python3
"""
Analyze sortbloom JSONL output and render readable terminal summaries and a
single relative-time line plot.

Input format (one JSON object per line):
  - First line: {"kind": "run_meta", ...}
  - Subsequent lines: {"args": {...}, "bf_results": [{...}]}

Groups and comparisons:
  - Grouping is by filter_size only (aggregates all runs at that size).
  - Within each group, compare all (hasher, sort) variants.

Outputs:
  - Terminal: sorted table per group + ASCII bars (best-normalized).
  - Plot: a single line chart across filter sizes with Y = Time Ratio
    (NoSort / variant) computed using milliseconds, X = Bloom Filter Size
    (log2 bits). The NoSort baseline is shown as a grey dashed line at 1.0
    and included in the legend.
  - Plot: a second line chart with Y = Cache Miss Ratio (NoSort / variant),
    using normalized misses per kilo-probe, with the same X axis.
  - CLI: pass --no-plots to skip generating the plot and only print terminal
    summaries.

Dependencies:
  - Python standard library (always used)
  - matplotlib (for plots). If unavailable, terminal output still works.
"""

from __future__ import annotations

import argparse
import json
import math
import os
import sys
from collections import defaultdict
import re
import subprocess
from dataclasses import dataclass
from typing import Any, Callable, Dict, Iterable, List, Tuple, Optional

import matplotlib.pyplot as plt
import statistics

_EXPECTED_RUN_COUNT: Optional[int] = None

# -------- Data structures --------


@dataclass(frozen=True)
class ArgsKey:
    filter_size: int
    num_entries: int
    probe_count: int


@dataclass
class Record:
    key: ArgsKey
    hasher: str
    sort: str
    pos_queries: Optional[float]
    l2: int
    l3: int
    bfkind: str
    milliseconds: float
    cache_count: int
    l2_load_miss_count: Optional[int]
    cycle_count: int
    found_count: int

    @property
    def ns_per_probe(self) -> float:
        return (self.milliseconds * 1e6) / max(1, self.key.probe_count)

    @property
    def l3_misses_per_killoprobe(self) -> float:
        return (
            self.cache_count / (self.key.probe_count / 1000.0)
            if self.key.probe_count
            else float("nan")
        )

    @property
    def l2_misses_per_killoprobe(self) -> float:
        if self.l2_load_miss_count is None:
            return float("nan")
        return (
            self.l2_load_miss_count / (self.key.probe_count / 1000.0)
            if self.key.probe_count
            else float("nan")
        )

    @property
    def cycle_per_probe(self) -> float:
        return (self.cycle_count) / max(1, self.key.probe_count)


# -------- Parsing --------


def load_records(path: str) -> List[Record]:
    recs: List[Record] = []
    with open(path, "r") as fp:
        for line in fp:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                print(
                    f"Warning: skipping invalid JSON line in {path}: {line!r}"
                )
                continue
            if obj.get("kind") in ("run_meta", "experiment_end"):
                continue

            args = obj.get("args", {})
            results = obj.get("bf_results", [])
            if not results:
                print(
                    "Warning: encountered record without 'bf_results'; skipping."
                )
                continue

            r0 = results[0]
            key = ArgsKey(
                filter_size=int(args["filter_size"]),
                num_entries=int(args["num_entries"]),
                probe_count=int(args["probe_count"]),
            )
            recs.append(
                Record(
                    key=key,
                    hasher=str(args.get("hasher", "")),
                    sort=str(args.get("sort", "")),
                    pos_queries=(
                        float(args["pos_queries"])
                        if "pos_queries" in args and args["pos_queries"] is not None
                        else None
                    ),
                    l2=args.get("l2"),
                    l3=args.get("l3"),
                    bfkind=args.get("bfkind"),
                    milliseconds=float(r0.get("milliseconds", 0)),
                    cache_count=int(r0.get("cache_count", 0)),
                    found_count=int(r0.get("found_count", 0)),
                    l2_load_miss_count=(lambda v: int(v) if v is not None else None)(
                        r0.get("l2_load_miss_count")
                    ),
                    cycle_count=int(r0.get("cycle_count", 0)),
                )
            )
    return recs


def load_run_meta(path: str) -> Optional[Dict[str, Any]]:
    try:
        with open(path, "r") as fp:
            for line in fp:
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                except json.JSONDecodeError:
                    print(
                        f"Warning: skipping invalid JSON line while reading run_meta from {path}: {line!r}"
                    )
                    continue
                if obj.get("kind") == "run_meta":
                    return obj
    except Exception as e:
        print(f"Warning: failed to load run_meta from {path}: {e}")
        return None
    return None


# -------- Grouping & Metrics --------


def group_records(recs: Iterable[Record]) -> Dict[int, List[Record]]:
    groups: Dict[int, List[Record]] = defaultdict(list)
    for r in recs:
        key = r.key.filter_size
        groups[key].append(r)
    return groups


# Takes the mean of the same variant sizes as well here.
def aggregate_records_by_variant(rows: Iterable[Record]) -> List[Record]:
    buckets: Dict[Tuple[str, str], Dict[str, float]] = {}
    exemplars: Dict[Tuple[str, str], Record] = {}
    counts: Dict[Tuple[str, str], int] = {}
    for r in rows:
        key = (r.hasher, r.sort, r.l2, r.l3, r.bfkind)
        counts[key] = counts.get(key, 0) + 1
        exemplars.setdefault(key, r)
        acc = buckets.setdefault(
            key,
            {
                "milliseconds": 0.0,
                "cache_count": 0.0,
                "found_count": 0.0,
                "cycle_count": 0.0,
                "l2_load_miss_count": 0.0,
                "l2_samples": 0,
            },
        )
        acc["milliseconds"] += r.milliseconds
        acc["cache_count"] += r.cache_count
        acc["cycle_count"] += r.cycle_count
        acc["found_count"] += r.found_count
        if r.l2_load_miss_count is not None:
            acc["l2_load_miss_count"] += r.l2_load_miss_count
            acc["l2_samples"] += 1

    aggregated: List[Record] = []
    for key, exemplar in exemplars.items():
        total = counts[key]
        if _EXPECTED_RUN_COUNT is not None and total != _EXPECTED_RUN_COUNT:
            # raise RuntimeError(
            #     "Run count mismatch for variant {}: expected {}, saw {}".format(
            #         key, _EXPECTED_RUN_COUNT, total
            #     )
            # )
            print(
                "Run count mismatch for variant {}: expected {}, saw {}".format(
                    key, _EXPECTED_RUN_COUNT, total
                )
            )
        if total <= 0:
            raise RuntimeError(
                f"Internal error: aggregated count for variant {key} is non-positive: {total}"
            )

        sums = buckets[key]
        aggregated.append(
            Record(
                key=exemplar.key,
                hasher=exemplar.hasher,
                sort=exemplar.sort,
                pos_queries=exemplar.pos_queries,
                l2=exemplar.l2,
                l3=exemplar.l3,
                bfkind=exemplar.bfkind,
                milliseconds=sums["milliseconds"] / total,
                cache_count=sums["cache_count"] / total,
                found_count=sums["found_count"] / total,
                l2_load_miss_count=(
                    sums["l2_load_miss_count"] / sums["l2_samples"]
                    if sums["l2_samples"] > 0
                    else None
                ),
                cycle_count=sums["cycle_count"] / total,
            )
        )
    return aggregated


def human_bits(nbits: int) -> str:
    return f"2^{nbits:,}"


def human_mega_bytes(bits: int) -> str:
    mega_bytes = 1024 * 1024 * 8
    mb = bits // mega_bytes
    return f"{mb}"


def human_filter_size(bits: int) -> str:
    if bits <= 0:
        return str(bits)
    nbits = int(round(math.log2(bits)))
    return f"{human_bits(nbits)} ({human_mega_bytes(bits)}MB) ({bits:,})"


def humanize_kb(kb: Optional[int]) -> str:
    if kb is None:
        return "unknown"
    try:
        val = float(kb)
    except Exception:
        return str(kb)
    if val >= 1024.0:
        mb = val / 1024.0
        if abs(mb - round(mb)) < 0.05:
            return f"{int(round(mb))} MB"
        return f"{mb:.1f} MB"
    return f"{int(round(val))} KB"


def humanize_bytes(n: Optional[int]) -> str:
    if n is None:
        return "unknown"
    try:
        v = float(n)
    except Exception:
        return str(n)
    units = ["B", "KB", "MB", "GB", "TB"]
    i = 0
    while v >= 1024.0 and i < len(units) - 1:
        v /= 1024.0
        i += 1
    if abs(v - round(v)) < 0.05:
        return f"{int(round(v))} {units[i]}"
    return f"{v:.1f} {units[i]}"


def humanize_seconds(s: Optional[float]) -> str:
    if s is None:
        return "unknown"
    try:
        t = int(round(float(s)))
    except Exception:
        return str(s)
    days, rem = divmod(t, 86400)
    hrs, rem = divmod(rem, 3600)
    mins, secs = divmod(rem, 60)
    parts: List[str] = []
    if days:
        parts.append(f"{days}d")
    if hrs:
        parts.append(f"{hrs}h")
    if mins:
        parts.append(f"{mins}m")
    if not parts:
        parts.append(f"{secs}s")
    return " ".join(parts)


def variant_label(r: Record) -> str:
    # Prefer explicit hasher/sort enums; otherwise return a generic label
    assert r.hasher and r.sort
    return f"{r.hasher.lower()}/{get_sort_norm(r)}"


def camel_to_kebab(s: str) -> str:
    # Convert PascalCase/mixed identifiers to kebab-case, also handling underscores and digits.
    # Examples: "NoSort" -> "no-sort", "PartialLut2_0_0_20" -> "partial-lut2-0-0-20".
    assert s
    out = s.replace("_", "-")
    out = out.replace(",0", "")
    out = out.replace(",", "-")
    out = re.sub(r"(?<=[a-z])([A-Z])", r"-\1", out)
    out = re.sub(r"-+", "-", out)
    return out.lower()


# -------- Terminal Rendering --------

# ANSI color helpers (8-color foreground; text only)
_ANSI_PALETTE = [37, 36, 35, 32, 33]  # white, cyan, magenta, geen, yellow
_HASHER_COLOR_CACHE: Dict[str, int] = {}
_COLOR = 0


def _colorize(text: str, color_code: int) -> str:
    return f"\x1b[{color_code}m{text}\x1b[0m"


def _color_for_hasher(hasher: str) -> int:
    global _COLOR

    if hasher in _HASHER_COLOR_CACHE:
        return _HASHER_COLOR_CACHE[hasher]

    code = _ANSI_PALETTE[_COLOR % len(_ANSI_PALETTE)]
    _HASHER_COLOR_CACHE[hasher] = code
    _COLOR += 1
    return code


def _select_baseline(
    rows: List[Record], baseline_pref: Optional[str] = None
) -> Tuple[Record, float]:
    """Select baseline using milliseconds (ms).

    Behavior is controlled by `baseline_pref`:
      - "slowest": choose the slowest overall by milliseconds.
      - any other string: treated as a sort label (kebab/pascal/underscore accepted);
        choose the slowest row whose sort matches that label; if none exist, fall back
        to the default behavior described above.
    Returns (record, baseline_ms).
    """
    assert rows

    pref = (baseline_pref or "").strip().lower()
    if pref == "slowest":
        base = max(rows, key=lambda r: r.milliseconds)
        return base, base.milliseconds
    else:
        # Treat as specific sort label
        target_sort = camel_to_kebab(pref)
        cand = [r for r in rows if get_sort_norm(r) == target_sort]
        assert cand

        if cand:
            # Choose the slowest among the baseline
            base = max(cand, key=lambda r: r.milliseconds)
            return base, base.milliseconds


def get_sort_norm(r: Record) -> str:
    name = r.sort
    if r.l2:
        name += f"_{r.l2}"
    if r.l3:
        name += f"_{r.l3}"
    if r.bfkind and r.bfkind != "Bf":
        name += f"_{r.bfkind}"
    return camel_to_kebab(name)


def print_group_summary(
    key: Any,
    rows: List[Record],
    allowed_sorts: List[str],
    baseline_pref: Optional[str],
    width: int = 80,
) -> None:
    # If no pos_queries info is present, fall back to legacy behavior:
    # one summary per filter size.
    has_any_pos = any(r.pos_queries is not None for r in rows)
    if not has_any_pos:
        _print_group_summary_single(
            key=key,
            rows=rows,
            allowed_sorts=allowed_sorts,
            baseline_pref=baseline_pref,
            width=width,
            pos_queries=None,
        )
        return

    # Otherwise, render a separate summary for each distinct pos_queries value.
    by_pos: Dict[Optional[float], List[Record]] = defaultdict(list)
    for r in rows:
        by_pos[r.pos_queries].append(r)

    # Sort by pos_queries value; keep None (if any) first as "unknown".
    def _pos_sort_key(v: Optional[float]) -> Tuple[int, float]:
        if v is None:
            return (0, float("-inf"))
        return (1, float(v))

    for pos_val in sorted(by_pos.keys(), key=_pos_sort_key):
        _print_group_summary_single(
            key=key,
            rows=by_pos[pos_val],
            allowed_sorts=allowed_sorts,
            baseline_pref=baseline_pref,
            width=width,
            pos_queries=pos_val,
        )


def _print_group_summary_single(
    key: Any,
    rows: List[Record],
    allowed_sorts: List[str],
    baseline_pref: Optional[str],
    width: int,
    pos_queries: Optional[float],
) -> None:
    if not rows:
        print("no data")
        return

    aggregated = aggregate_records_by_variant(rows)
    if not aggregated:
        print("no data after aggregation")
        return

    # Choose baseline from all variants by milliseconds; compute ns/probe for display
    base_rec, _base_ms = _select_baseline(aggregated, baseline_pref)
    base_ns = base_rec.ns_per_probe

    display_rows: List[Record] = []
    for r in aggregated:
        s_norm = get_sort_norm(r)
        if s_norm in allowed_sorts:
            display_rows.append(r)
    aggregated = display_rows

    if not aggregated:
        # Nothing to display for this group after filtering
        if pos_queries is None:
            title = f"Filter size {human_filter_size(int(key))}"
        else:
            title = (
                f"Filter size {human_filter_size(int(key))} "
                f"(pos_queries={pos_queries})"
            )
        print("\n" + title)
        print("-" * min(len(title), width))
        print("No variants enabled by --sort-options.")
        return

    # Sort by ns/probe ascending (post-filter)
    rows_sorted = sorted(aggregated, key=lambda r: r.ns_per_probe)

    # Header
    if pos_queries is None:
        title = f"Filter size {human_filter_size(int(key))}"
    else:
        title = (
            f"Filter size {human_filter_size(int(key))} "
            f"(pos_queries={pos_queries})"
        )
    print("\n" + title)
    print("-" * min(len(title), width))
    if base_rec is not None:
        print(f"Baseline: {variant_label(base_rec)} @ {base_ns:.2f} ns/probe")

    # Header aligned with the data rows:
    # - variant label column: 45 chars (same as label_field)
    # - 2-char baseline marker column, plus a space, before numeric columns
    print(
        f"{'variant':<45}   {'ms':>8} {'ns/probe':>10} {'L3mpk':>8} {'L3_rel':>8} "
        f"{'L2mpk':>8} {'L2_rel':>8} {'fpr%':>8} {'speedup':>7}"
    )

    # Render rows and ASCII bars
    bar_width = max(10, min(40, width - 76))
    for r in rows_sorted:
        idx = base_ns / r.ns_per_probe if r.ns_per_probe > 0 else 0.0
        base_l3 = (
            base_rec.l3_misses_per_killoprobe if base_rec is not None else float("nan")
        )
        base_l2 = (
            base_rec.l2_misses_per_killoprobe if base_rec is not None else float("nan")
        )
        # Guard divide-by-zero and non-finite cases; display placeholder when invalid
        if (
            r.l3_misses_per_killoprobe
            and r.l3_misses_per_killoprobe > 0
            and math.isfinite(base_l3)
        ):
            _miss_rel_val = base_l3 / r.l3_misses_per_killoprobe
            miss_rel_display = (
                f"{_miss_rel_val:8.2f}" if math.isfinite(_miss_rel_val) else f"{'—':>8}"
            )
        else:
            miss_rel_display = f"{'—':>8}"
        if (
            r.l2_misses_per_killoprobe
            and r.l2_misses_per_killoprobe > 0
            and math.isfinite(base_l2)
        ):
            _miss_rel_l2_val = base_l2 / r.l2_misses_per_killoprobe
            miss_rel_l2_display = (
                f"{_miss_rel_l2_val:8.2f}"
                if math.isfinite(_miss_rel_l2_val)
                else f"{'—':>8}"
            )
        else:
            miss_rel_l2_display = f"{'—':>8}"
        # raw_len = int(round(idx * bar_width))
        # bar_len = min(bar_width, max(0, raw_len))
        # clipped = raw_len > bar_width
        # bar = "█" * bar_len + ("»" if clipped else
        #                        "") + " " * (bar_width - bar_len -
        #                                     (1 if clipped else 0))
        label_plain = variant_label(r)
        # Color only the variant column (text) by hasher color
        label_field = f"{label_plain:<45}"
        label_colored = _colorize(label_field, _color_for_hasher(r.hasher))
        base_mark = " *" if base_rec is not None and r is base_rec else "  "
        found_ratio = (
            r.found_count / r.key.probe_count if r.key.probe_count else float("nan")
        )
        found_ratio_percent = (
            found_ratio * 100 if math.isfinite(found_ratio) else float("nan")
        )
        print(
            f"{label_colored}{base_mark} {r.milliseconds:>8.1f} {r.ns_per_probe:>10.2f} {r.l3_misses_per_killoprobe:>8.2f} {miss_rel_display} {r.l2_misses_per_killoprobe:>8.2f} {miss_rel_l2_display} {found_ratio_percent:>8.3f} {idx:>7.2f}"  # |{bar}|"
        )


# -------- Plotting (relative-time line) --------


def get_aggregated_rows(
    all_groups: Dict[int, List[Record]],
    record_fn: Optional[Callable[[Record]], float],
) -> Dict[Tuple[str, str, Optional[float]], Dict[int, List[float]]]:
    rows_by_variant: Dict[Tuple[str, str, Optional[float]], Dict[int, List[float]]] = {}

    for sz_bits, row in all_groups.items():
        for r in row:
            s_norm = get_sort_norm(r)
            key = (r.hasher, s_norm, r.pos_queries)
            if key not in rows_by_variant:
                rows_by_variant[key] = {}
            if sz_bits not in rows_by_variant[key]:
                rows_by_variant[key][sz_bits] = []

            if record_fn:
                rows_by_variant[key][sz_bits].append(record_fn(r))
            else:
                rows_by_variant[key][sz_bits].append(r)

    return rows_by_variant


def get_plot_data(
    all_groups: Dict[int, List[Record]],
    allowed_sorts: List[str],
    record_fn: Callable[[Record], float],
    baseline_pref: Optional[str],
) -> Tuple[
    Dict[str, Tuple[List[int], List[float]]],
    Dict[str, List[float]],
    Dict[str, Optional[float]],
]:
    # Use filter_size-only grouping.
    sizes_bits: List[int] = sorted(all_groups.keys())
    assert sizes_bits

    samples_by_variant = get_aggregated_rows(all_groups, record_fn)

    # Baseline per (filter size, pos_queries) (may be any hasher)
    baselines_variants: Dict[
        Tuple[int, Optional[float]], Tuple[str, str, Optional[float]]
    ] = {}
    for sz_bits, row in all_groups.items():
        by_pos: Dict[Optional[float], List[Record]] = defaultdict(list)
        for r in row:
            by_pos[r.pos_queries].append(r)
        for pos_val, records in by_pos.items():
            base_rec, _ = _select_baseline(records, baseline_pref)
            baselines_variants[(sz_bits, pos_val)] = (
                base_rec.hasher,
                get_sort_norm(base_rec),
                base_rec.pos_queries,
            )

    # Build series per (variant, pos_queries): X = log2(filter bits), Y = baseline / variant
    # Also compute optional Y error bars as stdev of per-run ratio when run_count > 1
    series: Dict[str, Tuple[List[int], List[float]]] = {}
    series_err: Dict[str, List[float]] = {}
    series_pos: Dict[str, Optional[float]] = {}
    allowed_set = set(allowed_sorts)

    for (hasher, sort_lbl, pos_val), per_size in samples_by_variant.items():
        if sort_lbl not in allowed_set:
            continue

        base_label = f"{hasher}--{sort_lbl}"
        if pos_val is None:
            label = base_label
        else:
            label = f"{base_label}--pos-{pos_val:g}"

        xs_mb: List[int] = []
        ys_rel: List[float] = []
        yerrs: List[float] = []
        for sz_bits in sorted(sizes_bits):
            baseline_key = (sz_bits, pos_val)
            if baseline_key not in baselines_variants:
                continue
            variant_key = baselines_variants[baseline_key]
            baseline_vals = samples_by_variant.get(variant_key, {}).get(sz_bits, [])
            current_vals = per_size.get(sz_bits, [])
            if not baseline_vals or not current_vals:
                continue

            ratio_samples: List[float] = []
            for bs, cs in zip(baseline_vals, current_vals):
                if (
                    cs is None
                    or cs == 0
                    or not math.isfinite(cs)
                    or bs is None
                    or not math.isfinite(bs)
                ):
                    continue
                ratio_samples.append(bs / cs)

            if not ratio_samples:
                continue

            # Prefer integer tick positions for consistent spacing
            nbits: int = int(round(math.log2(sz_bits)))

            xs_mb.append(nbits)
            ys_rel.append(statistics.mean(ratio_samples))
            if len(ratio_samples) > 1:
                yerrs.append(statistics.stdev(ratio_samples))
            else:
                yerrs.append(0.0)

        if xs_mb and ys_rel:
            series[label] = (xs_mb, ys_rel)
            series_err[label] = yerrs
            series_pos[label] = pos_val

    return series, series_err, series_pos


def get_raw_metric_by_variant(
    all_groups: Dict[int, List[Record]],
    allowed_sorts: List[str],
    metric_fn: Callable[[Record], float],
) -> Dict[Tuple[str, Optional[float]], Dict[int, Tuple[float, float]]]:
    """Return raw metric values per (hasher, sort, pos_queries) and filter size.

    Keys are (base_label, pos_queries) where base_label is "<hasher>--<sort>".
    Inner dict maps X position (log2(filter bits)) to a (mean, stdev) pair
    for the metric across all runs for that variant, filter size, and
    pos_queries value. The standard deviation is 0.0 when only a single
    sample is available.
    """
    rows_by_variant = get_aggregated_rows(all_groups, None)
    allowed_set = set(allowed_sorts)
    raw: Dict[Tuple[str, Optional[float]], Dict[int, Tuple[float, float]]] = {}

    for (hasher, sort_lbl, pos_val), per_size in rows_by_variant.items():
        if sort_lbl not in allowed_set:
            continue

        base_label = f"{hasher}--{sort_lbl}"
        key = (base_label, pos_val)

        per_x: Dict[int, List[float]] = {}
        for sz_bits, records in per_size.items():
            if sz_bits <= 0:
                continue
            x = int(round(math.log2(sz_bits)))
            for r in records:
                v = metric_fn(r)
                if v is None or not math.isfinite(v):
                    continue
                per_x.setdefault(x, []).append(v)

        if per_x:
            per_x_stats: Dict[int, Tuple[float, float]] = {}
            for x, vals in per_x.items():
                if not vals:
                    continue
                mean_v = statistics.mean(vals)
                if len(vals) > 1:
                    std_v = statistics.stdev(vals)
                else:
                    std_v = 0.0
                per_x_stats[x] = (mean_v, std_v)
            if per_x_stats:
                raw[key] = per_x_stats

    return raw


def plot_relative_lines(
    out_path: str,
    all_groups: Dict[int, List[Record]],
    allowed_sorts: List[str],
    xlabel: str,
    ylabel: str,
    title: str,
    record_fn: Callable[[Record], float],
    baseline_pref: Optional[str],
) -> None:
    assert plt and all_groups

    series, series_err, _series_pos = get_plot_data(
        all_groups, allowed_sorts, record_fn, baseline_pref
    )
    assert series

    # Determine all X tick positions across series
    x_data = sorted({x for (_, (xs, _)) in series.items() for x in xs})
    # Plot
    fig_w = max(6.0, min(14.0, 4.0 + 0.6 * len(x_data)))
    fig, ax = plt.subplots(figsize=(fig_w, 5))
    print((fig_w, 5))

    for lbl in sorted(series.keys()):
        xs, ys = series[lbl]
        # Ensure points are sorted by X (and keep yerr aligned if present)
        if lbl in series_err:
            yerr = series_err[lbl]
            ax.errorbar(
                xs,
                ys,
                yerr=yerr,
                fmt="o-",
                linewidth=1.8,
                markersize=5,
                capsize=4,
                label=lbl,
            )
        else:
            ax.plot(xs, ys, marker="o", linewidth=1.8, label=lbl)

    # Baseline horizontal line at y=1.0
    # ax.axhline(1.0,
    #            color="gray",
    #            linestyle="--",
    #            linewidth=1.5,
    #            label="NoSort baseline")

    ax.set_xlabel(xlabel)
    ax.set_ylabel(ylabel)
    ax.set_title(title)

    # X ticks on MB values present
    ax.set_xticks(x_data)
    ax.set_xticklabels(x_data)
    ax.grid(True, which="both", axis="y", alpha=0.3)

    # Y limits
    ymax = 1.0
    ymin = 1.0
    for _, (_, ys) in series.items():
        if ys:
            ymax = max(ymax, max(ys))
            ymin = min(ymin, min(ys))

    ax.set_ylim(ymin * 0.9, ymax * 1.10)

    ax.legend(loc="best", fontsize=9)
    fig.tight_layout()
    os.makedirs(os.path.dirname(out_path), exist_ok=True)
    fig.savefig(out_path, dpi=500)
    plt.close(fig)


def plot_milliseconds(
    out_path: str,
    all_groups: Dict[int, List[Record]],
    allowed_sorts: List[str],
    baseline_pref: Optional[str],
) -> None:
    def _get_ms(record: Record) -> float:
        return record.milliseconds

    xlabel = "Bloom Filter Size (log2 bits)"
    ylabel = "Time Ratio (baseline / variant)"
    title = "Time Ratio vs baseline"
    plot_relative_lines(
        out_path,
        all_groups,
        allowed_sorts,
        xlabel,
        ylabel,
        title,
        _get_ms,
        baseline_pref,
    )


def plot_cache_miss_ratio_lines(
    out_path: str,
    all_groups: Dict[int, List[Record]],
    allowed_sorts: List[str],
    baseline_pref: Optional[str],
) -> None:
    assert plt and all_groups

    def _get_misses_per_kp(record: Record) -> float:
        return record.l3_misses_per_killoprobe

    xlabel = "Bloom Filter Size (log2 bits)"
    ylabel = "Cache Miss Ratio (baseline / variant)"
    title = "Cache Miss Ratio vs baseline"
    plot_relative_lines(
        out_path,
        all_groups,
        allowed_sorts,
        xlabel,
        ylabel,
        title,
        _get_misses_per_kp,
        baseline_pref,
    )


def generate_gnuplot_relative_time_plot(
    input_path: str,
    all_groups: Dict[int, List[Record]],
    allowed_sorts: List[str],
    baseline_pref: Optional[str],
) -> None:
    def _get_ms(record: Record) -> float:
        return record.milliseconds

    series, series_err, series_pos = get_plot_data(
        all_groups, allowed_sorts, _get_ms, baseline_pref
    )
    if not series:
        print(
            f"Warning: no series data for gnuplot relative-time plot; skipping ({input_path})."
        )
        return

    # Raw ns/probe values (mean per variant, filter size, and pos_queries)
    raw_ns_per_probe = get_raw_metric_by_variant(
        all_groups,
        allowed_sorts,
        lambda r: r.ns_per_probe,
    )

    base = os.path.basename(input_path)
    stem = base.rsplit(".", 1)[0] if "." in base else base
    out_dir = os.path.join("results", "plots", stem)
    os.makedirs(out_dir, exist_ok=True)

    x_values_set = set()
    for _, (xs, ys) in series.items():
        for x in xs:
            x_values_set.add(x)
    if not x_values_set:
        print(
            f"Warning: no X values for gnuplot relative-time plot; skipping ({input_path})."
        )
        return

    x_values = sorted(x_values_set)

    per_file_dat_lines: Dict[str, List[str]] = {}
    per_file_raw_lines: Dict[str, List[str]] = {}
    for sort_label, (xs, ys) in series.items():
        yerrs = series_err.get(sort_label, [])
        if len(yerrs) != len(xs):
            yerrs = [0.0] * len(xs)
        pos_val = series_pos.get(sort_label)
        pos_str = "nan" if pos_val is None else f"{pos_val}"
        # Base label is always "<hasher>--<sort>"; keep filenames using this.
        base_label = (
            sort_label.split("--pos-", 1)[0] if pos_val is not None else sort_label
        )
        per_x_raw = raw_ns_per_probe.get((base_label, pos_val), {})

        for x, y, e in zip(xs, ys, yerrs):
            per_file_dat_lines.setdefault(base_label, []).append(
                f"{x} {y} {e} {pos_str}\n"
            )
            mean_raw, std_raw = per_x_raw.get(
                x, (float("nan"), float("nan"))
            )
            per_file_raw_lines.setdefault(base_label, []).append(
                f"{x} {mean_raw:.2f} {std_raw:.2f} {pos_str}\n"
            )

    data_files: List[Tuple[str, str]] = []
    for base_label, lines in per_file_dat_lines.items():
        data_path = os.path.join(out_dir, f"{base_label}.dat")
        raw_path = os.path.join(out_dir, f"{base_label}.raw")
        data_files.append((base_label, data_path))
        with open(data_path, "w") as f_dat:
            f_dat.writelines(lines)
        with open(raw_path, "w") as f_raw:
            f_raw.writelines(per_file_raw_lines.get(base_label, []))

    xtics_parts = [f'"{x}" {x}' for x in x_values]
    xtics_str = ", ".join(xtics_parts)

    plot_lines: List[str] = []
    for idx, (sort_label, data_path) in enumerate(data_files):
        comma = "," if idx < len(data_files) - 1 else ""
        plot_lines.append(
            f'    "{data_path}" using 1:2:3 with yerrorlines title "{sort_label}"{comma}'
        )
    plot_block = " \\\n".join(plot_lines)

    script_path = os.path.join(out_dir, f"plot_{stem}.gp")
    script_content = (
        'set title "Time Ratio vs baseline"\n'
        'set ylabel "Time Ratio (baseline / variant)"\n'
        'set xlabel "Bloom Filter Size (log2 bits)"\n'
        "set grid ytics\n"
        "set key left top\n"
        "set tics font ',10'\n"
        "set terminal pdfcairo size 6in,4in enhanced color font 'Helvetica,11' linewidth 2\n"
        f'set output "results/plots/{stem}/relative-lines-{stem}.pdf"\n'
        f"set xtics ({xtics_str})\n"
        "plot \\\n"
        f"{plot_block}\n"
    )

    with open(script_path, "w") as f:
        f.write(script_content)

    subprocess.run(["gnuplot", script_path], check=True)


def generate_gnuplot_cache_ratio_plot(
    input_path: str,
    all_groups: Dict[int, List[Record]],
    allowed_sorts: List[str],
    baseline_pref: Optional[str],
) -> None:
    def _get_misses_per_kp(record: Record) -> float:
        return record.l3_misses_per_killoprobe

    series, series_err, series_pos = get_plot_data(
        all_groups, allowed_sorts, _get_misses_per_kp, baseline_pref
    )
    if not series:
        print(
            f"Warning: no series data for gnuplot cache-ratio plot; skipping ({input_path})."
        )
        return

    # Restrict to sizes with log2(bits) >= 28 (e.g., >= 32 MiB).
    filtered_series: Dict[str, Tuple[List[int], List[float]]] = {}
    filtered_err: Dict[str, List[float]] = {}
    for label, (xs, ys) in series.items():
        yerrs = series_err.get(label, [])
        keep_xs: List[int] = []
        keep_ys: List[float] = []
        keep_errs: List[float] = []
        for idx, (x, y) in enumerate(zip(xs, ys)):
            if x >= 27:
                keep_xs.append(x)
                keep_ys.append(y)
                if idx < len(yerrs):
                    keep_errs.append(yerrs[idx])
                else:
                    keep_errs.append(0.0)
        if keep_xs:
            filtered_series[label] = (keep_xs, keep_ys)
            filtered_err[label] = keep_errs

    if not filtered_series:
        print(
            f"Warning: no cache-ratio data after filtering for gnuplot; skipping ({input_path})."
        )
        return

    # Raw L3 misses per kilo-probe (mean and stdev per variant, filter size, and pos_queries)
    raw_l3_mpk = get_raw_metric_by_variant(
        all_groups,
        allowed_sorts,
        lambda r: r.l3_misses_per_killoprobe,
    )

    base = os.path.basename(input_path)
    stem = base.rsplit(".", 1)[0] if "." in base else base
    out_dir = os.path.join("results", "plots", f"{stem}_cache")
    os.makedirs(out_dir, exist_ok=True)

    x_values_set = set()
    for _, (xs, ys) in filtered_series.items():
        for x in xs:
            x_values_set.add(x)
    if not x_values_set:
        print(
            f"Warning: no X values for gnuplot cache-ratio plot; skipping ({input_path})."
        )
        return

    x_values = sorted(x_values_set)

    per_file_dat_lines: Dict[str, List[str]] = {}
    per_file_raw_lines: Dict[str, List[str]] = {}
    for sort_label, (xs, ys) in filtered_series.items():
        yerrs = filtered_err.get(sort_label, [])
        if len(yerrs) != len(xs):
            yerrs = [0.0] * len(xs)
        pos_val = series_pos.get(sort_label)
        pos_str = "nan" if pos_val is None else f"{pos_val}"
        base_label = (
            sort_label.split("--pos-", 1)[0] if pos_val is not None else sort_label
        )
        per_x_raw = raw_l3_mpk.get((base_label, pos_val), {})

        for x, y, e in zip(xs, ys, yerrs):
            per_file_dat_lines.setdefault(base_label, []).append(
                f"{x} {y} {e} {pos_str}\n"
            )
            mean_raw, std_raw = per_x_raw.get(
                x, (float("nan"), float("nan"))
            )
            per_file_raw_lines.setdefault(base_label, []).append(
                f"{x} {mean_raw:.2f} {std_raw:.2f} {pos_str}\n"
            )

    data_files: List[Tuple[str, str]] = []
    for base_label, lines in per_file_dat_lines.items():
        data_path = os.path.join(out_dir, f"{base_label}.dat")
        raw_path = os.path.join(out_dir, f"{base_label}.raw")
        data_files.append((base_label, data_path))
        with open(data_path, "w") as f_dat:
            f_dat.writelines(lines)
        with open(raw_path, "w") as f_raw:
            f_raw.writelines(per_file_raw_lines.get(base_label, []))

    xtics_parts = [f'"{x}" {x}' for x in x_values]
    xtics_str = ", ".join(xtics_parts)

    plot_lines: List[str] = []
    for idx, (sort_label, data_path) in enumerate(data_files):
        comma = "," if idx < len(data_files) - 1 else ""
        plot_lines.append(
            f'    "{data_path}" using 1:2:3 with yerrorlines title "{sort_label}"{comma}'
        )
    plot_block = " \\\n".join(plot_lines)

    script_path = os.path.join(out_dir, f"plot_{stem}_cache.gp")
    script_content = (
        'set title "Cache Miss Ratio vs baseline"\n'
        'set ylabel "Cache Miss Ratio (baseline / variant)"\n'
        'set xlabel "Bloom Filter Size (log2 bits)"\n'
        "set grid ytics\n"
        "set key left top\n"
        "set tics font ',10'\n"
        "set terminal pdfcairo size 6in,4in enhanced color font 'Helvetica,11' linewidth 2\n"
        f'set output "results/plots/{stem}_cache/cache-miss-ratio-{stem}.pdf"\n'
        f"set xtics ({xtics_str})\n"
        "plot \\\n"
        f"{plot_block}\n"
    )

    with open(script_path, "w") as f:
        f.write(script_content)

    subprocess.run(["gnuplot", script_path], check=True)


# -------- Main --------


def print_meta(meta: Dict[str, Any]) -> None:
    """Pretty-print experiment/run metadata from the first JSONL line.

    This keeps `main()` slim and centralizes formatting. It intentionally
    mirrors the previous output for stability.
    """

    def _print_header(meta: Dict[str, Any]) -> None:
        exp = meta.get("experiment")
        ts = meta.get("ts")
        host = meta.get("host", {}) or {}
        cpu_model = host.get("cpu_model")
        hostname = host.get("hostname")
        os_name = host.get("os")
        kernel = host.get("kernel")
        arch = host.get("arch")

        print("Experiment summary:")
        print(f"  experiment: {exp}")
        print(f"  ts: {ts}")
        print(f"  cpu_model: {cpu_model}")

        if hostname or os_name or kernel or arch:
            print(f"  host: {hostname} | {os_name} {kernel} | {arch}")

    def _print_lscpu(meta: Dict[str, Any]) -> None:
        lscpu = meta.get("lscpu") if isinstance(meta, dict) else None
        if isinstance(lscpu, dict) and lscpu.get("available") is not False:
            vendor_id = lscpu.get("vendor_id")
            mhz_min = lscpu.get("mhz_min")
            mhz_max = lscpu.get("mhz_max")
            feats: list[str] = []
            if lscpu.get("avx2") is True:
                feats.append("avx2")
            if lscpu.get("avx512") is True:
                avx512_list = lscpu.get("avx512_features") or []
                try:
                    n512 = len(avx512_list)
                except Exception:
                    n512 = 0
                feats.append("avx512" + (f"({n512})" if n512 else ""))
            if feats:
                print(f"  cpu features: {', '.join(feats)}")
            parts_meta: list[str] = []
            if vendor_id:
                parts_meta.append(str(vendor_id))
            if isinstance(mhz_min, (int, float)) or isinstance(mhz_max, (int, float)):

                def _fmt(x: float) -> str:
                    try:
                        xv = float(x)
                        return (
                            f"{int(round(xv))}"
                            if abs(xv - round(xv)) < 0.05
                            else f"{xv:.1f}"
                        )
                    except Exception:
                        return str(x)

                if mhz_min and mhz_max:
                    parts_meta.append(f"{_fmt(mhz_min)}-{_fmt(mhz_max)} MHz")
                elif mhz_min:
                    parts_meta.append(f"min {_fmt(mhz_min)} MHz")
                elif mhz_max:
                    parts_meta.append(f"max {_fmt(mhz_max)} MHz")
            if parts_meta:
                print("  lscpu: " + " | ".join(parts_meta))
            topo = lscpu.get("topology") if isinstance(lscpu, dict) else None
            if isinstance(topo, dict):
                topo_parts: list[str] = []
                if topo.get("cpus") is not None:
                    topo_parts.append(f"{int(topo['cpus'])} cpus")
                if topo.get("threads_per_core") is not None:
                    topo_parts.append(
                        f"{int(topo['threads_per_core'])} threads_per_core"
                    )
                if topo.get("cores_per_socket") is not None:
                    topo_parts.append(
                        f"{int(topo['cores_per_socket'])} cores_per_socket"
                    )
                if topo.get("sockets") is not None:
                    topo_parts.append(f"{int(topo['sockets'])} sockets")
                if topo.get("numa_nodes") is not None:
                    topo_parts.append(f"{int(topo['numa_nodes'])} numa")
                if topo_parts:
                    print("  topology: " + ", ".join(topo_parts))

    def _print_cpu_policy(meta: Dict[str, Any]) -> None:
        # Parse governor and frequency range from cpupower frequency-info output
        cpu_freq = meta.get("cpu_freq") if isinstance(meta, dict) else None
        if not cpu_freq:
            return  # older versions: no cpu_freq -> print nothing
        text = str(cpu_freq)
        # current policy line contains range; governor usually on following line
        # Example:
        # current policy: frequency should be within 1.20 GHz and 4.30 GHz.
        #                 The governor "powersave" may decide which speed to use
        m_range = re.search(
            r"current policy:.*?within\s+([0-9]+(?:\.[0-9]+)?\s*[kKmMgG][hH][zZ])\s+and\s+([0-9]+(?:\.[0-9]+)?\s*[kKmMgG][hH][zZ])",
            text,
            re.S,
        )
        m_gov = re.search(r'The governor\s+"([^"]+)"', text)

        if not (m_range and m_gov):
            assert False, "Failed to parse cpu_freq current policy"

        lo, hi = m_range.group(1), m_range.group(2)
        gov = m_gov.group(1)

        print(f"  cpu policy: {gov}, {lo} – {hi}")

    def _print_rustc(meta: Dict[str, Any]) -> None:
        rustc = meta.get("rustc_version")
        if rustc:
            print(f"  rustc: {rustc}")

    def _print_run_count(meta: Dict[str, Any]) -> None:
        run_count = meta.get("run_count")
        if run_count is not None:
            print(f"  run_count: {run_count}")
            global _EXPECTED_RUN_COUNT
            _EXPECTED_RUN_COUNT = int(run_count)

    def _print_caches_and_temp(meta: Dict[str, Any]) -> None:
        cache = meta.get("cpu_cache", {}) or {}
        data_kb = (cache.get("data_kb") or {}) if isinstance(cache, dict) else {}
        l1 = data_kb.get("L1")
        l2 = data_kb.get("L2")
        l3 = data_kb.get("L3")
        if any(v is not None for v in (l1, l2, l3)):
            parts: List[str] = []
            if l1 is not None:
                parts.append(f"L1: {humanize_kb(int(l1))}")
            if l2 is not None:
                parts.append(f"L2: {humanize_kb(int(l2))}")
            if l3 is not None:
                parts.append(f"L3: {humanize_kb(int(l3))}")
            print("  data caches: " + ", ".join(parts))
        else:
            print("  data caches: unknown")
        cpu_temp_c = meta.get("cpu_temp_c")
        if cpu_temp_c is not None:
            try:
                print(f"  temperature: {float(cpu_temp_c):.1f} C")
            except Exception:
                print(f"  temperature: {cpu_temp_c} C")
        else:
            print("  temperature: unknown")

    def _print_utilization(meta: Dict[str, Any]) -> None:
        sysutil = meta.get("system_utilization", {}) or {}
        cpu = sysutil.get("cpu", {}) if isinstance(sysutil, dict) else {}
        mem = sysutil.get("memory", {}) if isinstance(sysutil, dict) else {}
        disk_root = sysutil.get("disk_root", {}) if isinstance(sysutil, dict) else {}
        uptime_s = sysutil.get("uptime_seconds") if isinstance(sysutil, dict) else None
        if sysutil:
            util_parts: List[str] = []
            util_pct = cpu.get("util_percent")
            if isinstance(util_pct, (int, float)):
                util_parts.append(f"cpu {util_pct:.0f}%")
            load = cpu.get("loadavg") or {}
            if isinstance(load, dict) and load.get("1m") is not None:
                try:
                    util_parts.append(
                        f"load {float(load['1m']):.2f}/{float(load.get('5m', 0)):.2f}/{float(load.get('15m', 0)):.2f}"
                    )
                except Exception:
                    pass
            if mem:
                used_pct = mem.get("used_percent")
                used_kb = mem.get("used_kb")
                total_kb = mem.get("total_kb")
                if used_pct is not None and total_kb is not None:
                    util_parts.append(
                        f"mem {used_pct:.0f}% ({humanize_kb(used_kb)}/{humanize_kb(total_kb)})"
                    )
            if disk_root:
                d_used_pct = disk_root.get("used_percent")
                if d_used_pct is not None:
                    util_parts.append(
                        f"disk / {d_used_pct:.0f}% ({humanize_bytes(disk_root.get('used_bytes'))}/{humanize_bytes(disk_root.get('total_bytes'))})"
                    )
            if uptime_s is not None:
                util_parts.append(f"uptime {humanize_seconds(uptime_s)}")
            if util_parts:
                print("  util: " + ", ".join(util_parts))

    def _print_users(meta: Dict[str, Any]) -> None:
        w_output = meta.get("w_output")
        if isinstance(w_output, str) and w_output.strip():
            first_line = w_output.splitlines()[0] if "\n" in w_output else w_output
            m = re.search(r"(\d+)\s+users?,", first_line)
            users_text = None
            if m:
                try:
                    nusers = int(m.group(1))
                    users_text = f"{nusers} user" + ("s" if nusers != 1 else "")
                except Exception:
                    pass
            if users_text:
                print(f"  logged in: {users_text}")
            else:
                print("  logged in: unknown")

    # Orchestrate
    _print_header(meta)
    _print_caches_and_temp(meta)
    _print_cpu_policy(meta)
    _print_run_count(meta)
    print()
    _print_lscpu(meta)
    _print_rustc(meta)
    _print_utilization(meta)
    _print_users(meta)
    print("")


def main() -> None:
    ap = argparse.ArgumentParser(
        description="Analyze sortbloom results JSONL and summarize tables + relative-time plot"
    )
    ap.add_argument(
        "--input",
        default="results/latest.jsonl",
        help="Path to JSONL file (default: results/latest.jsonl)",
    )
    ap.add_argument(
        "--no-plot",
        default=False,
        action="store_true",
        help="Do not generate plots; print terminal summaries only",
    )
    ap.add_argument(
        "--sort-options",
        default=None,
        help=(
            "Optional path to a file listing enabled sort variants (kebab-case). "
            "Lines starting with # are ignored. If omitted, all variants are shown."
        ),
    )
    ap.add_argument(
        "--hasher",
        default=None,
        help="Comma-separated hasher names to include (case-insensitive). If omitted, include all",
    )
    ap.add_argument(
        "--baseline",
        default="no-sort",
        help=(
            "Baseline selector: 'nosort' (default), 'slowest', or a specific sort label "
            "(kebab/camel/underscore accepted), e.g. 'partial-lut2-0-0-20'"
        ),
    )
    args = ap.parse_args()

    if not os.path.exists(args.input):
        print(f"ERROR: Input file not found: {args.input}", file=sys.stderr)
        sys.exit(2)

    # Experiment summary (from run_meta line)
    meta = load_run_meta(args.input)
    if meta:
        print_meta(meta)

    recs = load_records(args.input)
    # Optional filter by hasher(s)
    if args.hasher:
        # Accept comma-separated list; trim spaces; case-insensitive match
        raw = [p.strip() for p in args.hasher.split(",") if p.strip()]
        allowed_hashers = {p.lower() for p in raw}
        recs = [r for r in recs if r.hasher and r.hasher.lower() in allowed_hashers]
    if not recs:
        print("No records found.")
        return

    # Load allowed sort options (kebab-case), if provided.
    # If no file is provided, allow all variants present in the data.
    if args.sort_options is None:
        # No sort-options file: show every variant.
        allowed_sorts: List[str] = sorted({get_sort_norm(r) for r in recs})
    else:
        # User provided a sort-options file: assert it exists, then load it.
        if not os.path.exists(args.sort_options):
            raise AssertionError(f"Sort options file not found: {args.sort_options}")

        allowed_sorts = []
        try:
            with open(args.sort_options, "r") as f:
                for line in f:
                    s = line.strip()
                    if not s or s.startswith("#"):
                        continue
                    allowed_sorts.append(camel_to_kebab(s))
        except Exception as e:
            print(f"Warning: failed to read sort options from {args.sort_options}: {e}")
    groups = group_records(recs)
    # Stable group order by filter size ascending
    group_keys = sorted(groups.keys())

    for k in group_keys:
        print_group_summary(k, groups[k], allowed_sorts, args.baseline)
    # Optionally produce the single relative-time line plot (ms vs NoSort baseline).
    if not args.no_plot:
        base = os.path.basename(args.input)
        stem = base.rsplit(".", 1)[0] if "." in base else base
        plot_out1 = os.path.join("results", "plots", f"relative-lines-{stem}.pdf")
        plot_milliseconds(plot_out1, groups, allowed_sorts, args.baseline)
        print(f"Plot saved: {os.getcwd()}/{plot_out1}")

        plot_out2 = os.path.join("results", "plots", f"cache-miss-ratio-{stem}.pdf")
        plot_cache_miss_ratio_lines(plot_out2, groups, allowed_sorts, args.baseline)
        print(f"Plot saved: {os.getcwd()}/{plot_out2}")

    try:
        generate_gnuplot_relative_time_plot(
            args.input, groups, allowed_sorts, args.baseline
        )
    except Exception as e:
        print(f"Warning: failed to run gnuplot plot: {e}", file=sys.stderr)
    try:
        generate_gnuplot_cache_ratio_plot(
            args.input, groups, allowed_sorts, args.baseline
        )
    except Exception as e:
        print(
            f"Warning: failed to run gnuplot cache plot: {e}",
            file=sys.stderr,
        )


if __name__ == "__main__":
    main()