#!/usr/bin/env python3 """Summarize full PlantGeneBench scored rows.""" from __future__ import annotations import argparse import csv import json import random from collections import Counter, defaultdict from pathlib import Path from typing import Any def load_jsonl(path: Path) -> list[dict[str, Any]]: rows = [] with path.open("r", encoding="utf-8") as handle: for line in handle: line = line.strip() if line: rows.append(json.loads(line)) return rows def mean(values: list[float]) -> float: return sum(values) / len(values) if values else 0.0 def bootstrap_ci(values: list[float], *, iterations: int = 2000, seed: int = 20260608) -> tuple[float, float]: if not values: return 0.0, 0.0 if len(values) == 1: return values[0], values[0] rng = random.Random(seed) n = len(values) means = [] for _ in range(iterations): sample = [values[rng.randrange(n)] for _ in range(n)] means.append(mean(sample)) means.sort() lo = means[int(0.025 * (iterations - 1))] hi = means[int(0.975 * (iterations - 1))] return lo, hi def write_tsv(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fields = ["task_id", "provider", "layer", "score", "backend_status", "review_required", "errors", "predicted_stages", "expected_stages"] with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields, delimiter="\t") writer.writeheader() for row in rows: writer.writerow({ "task_id": row["task_id"], "provider": row.get("provider", ""), "layer": row.get("layer", ""), "score": row.get("score", 0), "backend_status": row.get("backend_status", ""), "review_required": row.get("review_required", False), "errors": ";".join(row.get("errors") or []), "predicted_stages": ",".join(row.get("predicted_stages") or []), "expected_stages": ",".join(row.get("expected_stages") or []), }) def write_aggregate_tsv(path: Path, rows: list[dict[str, Any]]) -> None: groups: dict[tuple[str, str], list[float]] = defaultdict(list) for row in rows: groups[(row.get("provider", ""), row.get("layer", ""))].append(float(row.get("score") or 0.0)) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.writer(handle, delimiter="\t") writer.writerow(["provider", "layer", "n", "mean_score"]) for (provider, layer), values in sorted(groups.items()): writer.writerow([provider, layer, len(values), f"{mean(values):.6f}"]) def write_error_tsv(path: Path, rows: list[dict[str, Any]]) -> None: counter = Counter() for row in rows: counter.update(row.get("errors") or []) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.writer(handle, delimiter="\t") writer.writerow(["error", "count"]) for error, count in counter.most_common(): writer.writerow([error, count]) def write_report(path: Path, rows: list[dict[str, Any]]) -> None: provider_scores: dict[str, list[float]] = defaultdict(list) layer_scores: dict[tuple[str, str], list[float]] = defaultdict(list) backend_counter = Counter() errors = Counter() llm_json_counts: dict[str, Counter] = defaultdict(Counter) llm_tokens: dict[str, list[int]] = defaultdict(list) for row in rows: provider = row.get("provider", "") layer = row.get("layer", "") score = float(row.get("score") or 0.0) provider_scores[provider].append(score) layer_scores[(provider, layer)].append(score) backend_counter.update([row.get("backend_status", "") or "unknown"]) errors.update(row.get("errors") or []) pred = row.get("prediction") or {} if provider.startswith("deepseek"): valid = pred.get("llm_json_valid") llm_json_counts[provider].update([str(valid)]) usage = pred.get("llm_usage") or {} try: total = int(usage.get("total_tokens") or 0) except Exception: total = 0 if total: llm_tokens[provider].append(total) lines = ["# PlantGeneBench Full Summary", "", f"- Rows: {len(rows)}", f"- Review required: {sum(1 for r in rows if r.get('review_required'))}", ""] lines += ["## Provider Scores", "", "| Provider | N | Mean score | 95% bootstrap CI |", "|---|---:|---:|---:|"] for provider, values in sorted(provider_scores.items()): lo, hi = bootstrap_ci(values) lines.append(f"| {provider} | {len(values)} | {mean(values):.3f} | {lo:.3f}-{hi:.3f} |") lines += ["", "## Layer Scores", "", "| Provider | Layer | N | Mean score |", "|---|---|---:|---:|"] for (provider, layer), values in sorted(layer_scores.items()): lines.append(f"| {provider} | {layer} | {len(values)} | {mean(values):.3f} |") if llm_json_counts: lines += ["", "## DeepSeek Diagnostics", "", "| Provider | JSON valid/None/False counts | Mean total tokens |", "|---|---|---:|"] for provider in sorted(llm_json_counts): counts = ", ".join(f"{key}={value}" for key, value in sorted(llm_json_counts[provider].items())) values = llm_tokens.get(provider, []) lines.append(f"| {provider} | {counts} | {mean(values):.1f} |") lines += ["", "## Backend Status", "", "| Status | Count |", "|---|---:|"] for status, count in backend_counter.most_common(): lines.append(f"| {status} | {count} |") lines += ["", "## Error Types", "", "| Error | Count |", "|---|---:|"] for error, count in errors.most_common(): lines.append(f"| {error} | {count} |") path.write_text("\n".join(lines) + "\n", encoding="utf-8") def run() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--scored", type=Path, nargs="+", required=True) parser.add_argument("--out-prefix", type=Path, required=True) args = parser.parse_args() rows: list[dict[str, Any]] = [] for path in args.scored: rows.extend(load_jsonl(path)) prefix = args.out_prefix prefix.parent.mkdir(parents=True, exist_ok=True) write_tsv(Path(str(prefix) + ".full_summary.tsv"), rows) write_aggregate_tsv(Path(str(prefix) + ".layer_scores.tsv"), rows) write_error_tsv(Path(str(prefix) + ".error_types.tsv"), rows) write_report(Path(str(prefix) + ".final_report.md"), rows) print(f"Wrote {prefix}.full_summary.tsv") print(f"Wrote {prefix}.layer_scores.tsv") print(f"Wrote {prefix}.error_types.tsv") print(f"Wrote {prefix}.final_report.md") return 0 if __name__ == "__main__": raise SystemExit(run())