"""Reproduce the comparison tables from the public inputs, using Python's standard library.

Keep input folders as listed in v1.json. Use --product ID for a single table.
No network call, package installation or hardware workload runs here.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
import math
from collections import defaultdict
from datetime import datetime
from itertools import pairwise
from pathlib import Path
from statistics import median

ROOT = Path(__file__).resolve().parents[1]
MANIFEST = "benchmark-continuations/v1.json"
PRODUCTS = (
    "llm-server-compare",
    "mcp-server-bench",
    "rtx-5080-llm-throughput",
    "rtx-5080-llm-power-efficiency",
    "rog-astral-rtx-5080-oc",
    "ha-voice-benchmark",
    "local-coding-evaluation",
    "fps-calculator-catalogue",
)


def require(condition, message):
    if not condition:
        raise ValueError(message)


def sha(raw):
    return hashlib.sha256(raw.replace(b"\r\n", b"\n")).hexdigest()


class Inputs:
    def __init__(self, root):
        self.root = root
        self.files = []

    def raw(self, name):
        raw = (self.root / name).read_bytes()
        self.files.append({"path": name, "sha256_canonical_lf": sha(raw)})
        return raw

    def csv(self, name, comments=False):
        parsed = csv.DictReader(
            io.StringIO(self.raw(name).decode("utf-8-sig"), newline=""), strict=True
        )
        require(
            parsed.fieldnames and len(set(parsed.fieldnames)) == len(parsed.fieldnames),
            f"invalid CSV columns: {name}",
        )
        rows = list(parsed)
        if comments:
            rows = [r for r in rows if not r[parsed.fieldnames[0]].startswith("#")]
        require(
            rows and all(None not in row and None not in row.values() for row in rows),
            f"invalid CSV: {name}",
        )
        return rows

    def json(self, name):
        return json.loads(self.raw(name).decode("utf-8-sig"))


def single(rows, **match):
    hits = [r for r in rows if all(r[k] == v for k, v in match.items())]
    require(len(hits) == 1, f"expected one matching record: {match}, got {len(hits)}")
    return hits[0]


def table(title, context, headers, rows, limit, retest):
    require(
        rows and all(len(row) == len(headers) for row in rows), "empty or uneven table"
    )
    return {
        "title": title,
        "context": context,
        "headers": headers,
        "rows": rows,
        "limit": limit,
        "retest": retest,
    }


def server(inputs):
    first = inputs.csv("llm-server-compare-2026-08.csv", comments=True)
    second = inputs.csv("llm-server-compare-2026-08-26-lmstudio.csv")
    result = []
    for label, model, alias in [
        ("Llama 3.2 3B", "llama3.2:3b", "llama-3.2-3b-q4km"),
        ("Qwen 2.5 14B", "qwen2.5:14b", "qwen2.5-14b-q4km"),
    ]:
        pairs = [
            (
                "Aug 20",
                "llama.cpp b10507",
                single(first, model=model, server="ollama 0.32.1"),
                single(first, model=model, server="llama.cpp b10507"),
            ),
            (
                "Aug 26",
                "LM Studio 0.4.21",
                single(
                    second,
                    model=model,
                    app="ollama",
                    condition="normalized",
                    context="4096",
                ),
                single(
                    second,
                    model=alias,
                    app="lms",
                    condition="normalized",
                    context="4096",
                ),
            ),
        ]
        for day, engine, left, right in pairs:
            a, b = float(left["decode_tps_median"]), float(right["decode_tps_median"])
            result.append(
                [
                    day,
                    label,
                    f"{a:.2f}",
                    engine,
                    f"{b:.2f}",
                    f"{(b / a - 1) * 100:+.1f}%",
                ]
            )
    return table(
        "Which server decoded the same model faster?",
        "Each row compares the same dense model within one August 2026 capture at 4,096 context. Ollama is the denominator. Rates are median output tokens per second.",
        ["Capture", "Model", "Ollama", "Other server", "Other rate", "Difference"],
        result,
        "The August 20 Ollama version was 0.32.1; August 26 used 0.32.15. Do not combine different capture dates into an upgrade result. The gpt-oss comparison is excluded because the August 20 upstream GGUF differed.",
        "Match the complete GGUF hash, runtime and backend versions, offload, context, prompt set, seed and cache policy. Retain a wall-clock check beside engine counters.",
    )


def mcp_samples(inputs, receipt, script):
    cohort = inputs.json("benchmark-continuations/" + receipt)
    require(
        cohort["probe_sha256_canonical_lf"]
        == sha(inputs.raw("benchmark-continuations/" + script)),
        "probe differs from recorded instrument",
    )
    samples = cohort["samples"]
    require(
        len(samples) == 5
        and all(
            r["conversion_checked"] is True and "convert_time" in r["tools"]
            for r in samples
        ),
        "incomplete MCP workload",
    )
    return cohort


def mcp(inputs):
    initial = mcp_samples(
        inputs, "mcp-time-2026-11-04-initial.json", "installed_time_probe_initial.py"
    )
    cohort = mcp_samples(inputs, "mcp-time-2026-11-04.json", "installed_time_probe.py")
    require(
        all(
            initial[k] == cohort[k]
            for k in ("method", "target", "workload", "requested_protocol")
        ),
        "installed time-server cohorts differ in workload or target",
    )
    rows = []
    for key, label in [
        ("ready_ms", "Fresh process to ready (ms)"),
        ("list_ms", "List tools (ms)"),
        ("call_ms", "Convert UTC 12:34 to UTC (ms)"),
        ("rss_after_list_mib", "Process-tree RSS after listing (MiB)"),
        ("rss_after_call_mib", "Process-tree RSS after conversion (MiB)"),
    ]:
        for run in (initial, cohort):
            values = [r[key] for r in run["samples"]]
            require(
                all(
                    isinstance(v, (float, int))
                    and not isinstance(v, bool)
                    and math.isfinite(v)
                    and v > 0
                    for v in values
                )
                and median(values) == run["medians"][key],
                "invalid MCP sample/median",
            )
        repeat_values = [r[key] for r in cohort["samples"]]
        rows.append(
            [
                label,
                *(
                    f"{value:.3f}"
                    for value in (
                        initial["medians"][key],
                        min(repeat_values),
                        cohort["medians"][key],
                        max(repeat_values),
                    )
                ),
            ]
        )
    target = cohort["target"]
    return table(
        "What does one installed time-server call cost?",
        f"November 4, 2026: mcp-server-time {target['package_version']}, MCP SDK {target['sdk_version']}, Python {target['python']} on Windows. Each run used one discarded warmup and five fresh processes. Each process listed tools and verified a conversion from UTC 12:34 to UTC 12:34. The first run started at 16:45 UTC. We repeated it at 16:57 after formatting the downloadable probe changed its file hash. Both medians are shown separately; minimum and maximum are from the repeat.",
        [
            "Operation or sample",
            "First median",
            "Repeat min",
            "Repeat median",
            "Repeat max",
        ],
        rows,
        "RSS is the process tree's resident working set at two specific points, not peak or incremental allocation. This direct interpreter run does not use the August uvx wrapper and cannot establish a speedup over that cohort.",
        "Use the linked probe with a separate installed interpreter. Keep interpreter, package and SDK versions, transport, startup path, call arguments and repetition count with the result.",
    )


def throughput(inputs):
    all_rows = inputs.csv("rtx-5080-llm-throughput.csv")
    groups = defaultdict(dict)
    key_fields = (
        "capture_date_utc",
        "method_version",
        "gpu",
        "driver",
        "ollama_version",
        "model",
        "model_digest12",
        "quantization",
        "prompt_tokens",
        "reps",
    )
    for row in all_rows:
        group = groups[tuple(row[k] for k in key_fields)]
        require(row["num_ctx"] not in group, "duplicate throughput condition")
        group[row["num_ctx"]] = row
    selected = {}
    for key, by_ctx in groups.items():
        if not {"4096", "16384"} <= set(by_ctx):
            continue
        identity = key[:8] + (key[9],)
        if identity not in selected or int(key[8]) < int(selected[identity][0][8]):
            selected[identity] = (key, by_ctx)
    rows = []
    for key, by_ctx in sorted(selected.values()):
        a, b = [float(by_ctx[k]["median_decode_tok_s"]) for k in ("4096", "16384")]
        rows.append(
            [
                key[0],
                key[5],
                key[8],
                f"{a:.2f}",
                f"{b:.2f}",
                f"{(b / a - 1) * 100:+.1f}%",
            ]
        )
    return table(
        "What changed when context rose from 4K to 16K?",
        "The rows pair the shortest shared prompt for each model and capture. Model digest, quantization, driver, runtime, GPU, method, prompt token count and repetitions must match before a comparison appears.",
        [
            "Capture (UTC)",
            "Model",
            "Prompt tokens",
            "4K decode (tok/s)",
            "16K decode (tok/s)",
            "Difference",
        ],
        rows,
        "These are configured context lengths; the prompts did not fill both windows. Small differences in medians are not a confidence interval or proof that a larger window makes a model faster.",
        "Keep full model identity, quantization, model and prompt hashes, context, runtime, GPU and driver. Record prefill separately from decode; retain all repetitions.",
    )


def whole_trace(rows):
    samples = [
        (datetime.fromisoformat(row["timestamp"]), float(row["power.draw"]))
        for row in rows
    ]
    require(
        len(samples) >= 2
        and all(
            t.tzinfo is not None and math.isfinite(p) and p >= 0 for t, p in samples
        ),
        "invalid power samples",
    )
    energy_j = 0.0
    for (a, wa), (b, wb) in pairwise(samples):
        seconds = (b - a).total_seconds()
        require(0 < seconds <= 5, "out-of-order or excessive telemetry gap")
        energy_j += (wa + wb) * 0.5 * seconds
    duration = (samples[-1][0] - samples[0][0]).total_seconds()
    return duration, energy_j / 3600, energy_j / duration


def power(inputs):
    rows = []
    for model, slug in [
        ("Llama 3.2 3B", "llama3-2-3b"),
        ("Qwen 2.5 7B", "qwen2-5-7b"),
        ("Qwen 2.5 14B", "qwen2-5-14b"),
        ("gpt-oss 20B", "gpt-oss-20b"),
    ]:
        samples = inputs.csv(f"rtx-5080-llm-power-efficiency-raw/stock448-{slug}-1.csv")
        duration, wh, mean = whole_trace(samples)
        rows.append(
            [model, len(samples), f"{duration:.1f}", f"{mean:.1f}", f"{wh:.3f}"]
        )
    return table(
        "How much board energy is in each recorded trace?",
        "These August traces include every interval between the first and last GPU power sample, including quieter intervals. For each adjacent pair, average the watts and multiply by elapsed seconds; sum the joules and divide by 3,600.",
        ["Model profile", "Samples", "Duration (s)", "Mean board W", "Board Wh"],
        rows,
        "Each row covers a different duration. This whole-trace integral is not energy per output token, a wall-meter reading or a replacement for the published loaded-watts and decode-rate comparison.",
        "Record token boundaries and timestamps against the same telemetry clock for energy per workload. Use a physical AC meter to measure the whole system; board telemetry cannot establish that value.",
    )


def astral(inputs):
    data = inputs.csv("gpu/rog-astral-rtx-5080-oc-2026-06-09/dataset.csv")
    charts = sorted(
        {
            row["chart_id"]
            for row in data
            if row["chart_id"].startswith("02") and row["metric"] == "score"
        }
    )
    output = []
    for chart in charts:
        records = [
            single(data, chart_id=chart, label=tune, metric="score", unit="score")
            for tune in ("Stock", "Factory OC", "Manual OC")
        ]
        a, b, c = [int(row["value"]) for row in records]
        output.append(
            [
                records[0]["chart_title"].split(" — ")[0],
                a,
                b,
                c,
                f"{(c / a - 1) * 100:+.1f}%",
            ]
        )
    return table(
        "How much did the manual tune change each 3DMark score?",
        "Each row stays within one named synthetic test on the owned Astral card. The percentage compares the manual result with stock in that same test.",
        ["Test", "Stock score", "Factory OC", "Manual OC", "Manual vs stock"],
        output,
        "Scores from different tests use different scales. A score gain is not a game FPS gain; generated display frames, native frames and input latency must remain separate.",
        "Add new stock and tuned runs as a separate dated cohort with test build, driver, power limit, fan policy, ambient conditions and repeated runs. Preserve original chart and telemetry files.",
    )


def voice(inputs):
    data = inputs.csv("ha-voice-brain-bench-2026-08.csv")
    require(
        len(
            {
                (
                    r["capture_date_utc"],
                    r["corpus_version"],
                    r["method_version"],
                    r["device"],
                    r["warm_reps"],
                    r["accuracy_n"],
                )
                for r in data
            }
        )
        == 1,
        "voice cohort conditions differ",
    )
    rows = [
        [r["model"], r["warm_total_ms_median"], r["accuracy_pct"], r["warm_reps"]]
        for r in sorted(data, key=lambda r: float(r["warm_total_ms_median"]))
    ]
    return table(
        "Which model traded response time for command accuracy?",
        "Same August 26 GPU voice-brain workload, corpus 1.0.1. Accuracy uses 24 scored commands; latency is the median of five warm requests. Read both columns before choosing a model.",
        ["Model", "Warm median (ms)", "Command accuracy (%)", "Warm requests"],
        rows,
        "This step measures the voice brain. Microphone capture, speech recognition, speech synthesis and network delay are outside these latencies.",
        "Keep the corpus, scorer, chat template, GGUF, offload and server build; publish speech-to-text and text-to-speech measurements as their own stages.",
    )


def coding(inputs):
    data = inputs.csv("rtx-5080-coding-eval-v1.csv")
    models = defaultdict(lambda: defaultdict(dict))
    for r in data:
        key = (r["model"], r["model_digest12"], r["num_ctx"], r["eval_version"])
        task = models[key][r["task"]]
        require(
            r["rep"] not in task and r["passed"] in ("True", "False"),
            "invalid repetition",
        )
        task[r["rep"]] = r["passed"] == "True"
    rows = []
    for (model, _, ctx, _), tasks in sorted(models.items()):
        require(
            all(set(reps) == {"0", "1"} for reps in tasks.values()),
            "incomplete paired repetitions",
        )
        counts = [sum(reps.values()) for reps in tasks.values()]
        rows.append(
            [model, ctx, len(counts), counts.count(2), counts.count(1), counts.count(0)]
        )
    return table(
        "Did each coding task pass on both attempts?",
        "This groups the two recorded attempts by task, model digest, evaluation version and configured context. A split result means that exactly one of the two attempts passed.",
        ["Model", "Context", "Tasks", "Both passed", "Split", "Neither passed"],
        rows,
        "Two attempts are too few to estimate a stable pass probability. These eight tasks do not measure overall coding ability, and context differs between models.",
        "Keep the same tasks, scorer, executable tests, prompts, model digest and independent repetitions. Preserve split results when publishing a new cohort.",
    )


def fps(inputs):
    data = inputs.json("fps-calculator/catalog.v1.json")
    by_provider = defaultdict(list)
    for row in data["rows"]:
        require(
            row["test_date"]["state"] in ("specified", "unspecified"),
            "unrecognized test date state",
        )
        by_provider[(row["dataset_id"], row["provider_name"])].append(row)
    result = []
    for (_, provider), rows in sorted(by_provider.items()):
        exact = sum(r["test_date"]["state"] == "specified" for r in rows)
        result.append(
            [
                provider,
                len(rows),
                len({r["game_id"] for r in rows}),
                exact,
                len(rows) - exact,
            ]
        )
    return table(
        "Which results actually state a test date?",
        "The lookup preserves each source separately. A publication date can tell you when a chart appeared; it does not establish the test date when the provider did not report one.",
        [
            "Provider",
            "Measured rows",
            "Games",
            "Test date stated",
            "Test date unspecified",
        ],
        result,
        "These rows are exact lookup tuples, not a prediction model. Do not blend providers, game scenes or rendering settings to fill an unsupported combination.",
        "Keep provider, chart hash, hardware, scene, preset, resolution, metric and date basis. Validate the source transcription separately from any formula experiment.",
    )


BUILDERS = dict(
    zip(
        PRODUCTS,
        (server, mcp, throughput, power, astral, voice, coding, fps),
        strict=True,
    )
)


def build(root, products=PRODUCTS):
    result = {}
    for product in products:
        inputs = Inputs(root)
        result[product] = {**BUILDERS[product](inputs), "inputs": inputs.files}
    return {
        "schema_version": 1,
        "hash_basis": "SHA-256 after CRLF to LF normalization",
        "products": result,
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--data-root", type=Path, default=ROOT)
    action = parser.add_mutually_exclusive_group()
    action.add_argument(
        "--write", action="store_true", help="regenerate the complete local manifest"
    )
    action.add_argument(
        "--check",
        action="store_true",
        help="compare the committed manifest with all source files (default)",
    )
    action.add_argument(
        "--product",
        choices=PRODUCTS,
        help="print one comparison and its exact input hashes",
    )
    args = parser.parse_args()
    result = build(args.data_root, (args.product,) if args.product else PRODUCTS)
    output = json.dumps(result, indent=2, ensure_ascii=False) + "\n"
    if args.product:
        print(output, end="")
    elif args.write:
        (args.data_root / MANIFEST).write_text(output, encoding="utf-8", newline="\n")
    else:
        require(
            (args.data_root / MANIFEST).read_text(encoding="utf-8") == output,
            "manifest differs from raw inputs or derived comparisons",
        )
        print(f"{len(PRODUCTS)} comparison contracts match their source files.")


if __name__ == "__main__":
    main()