Spaces:
Sleeping
Sleeping
File size: 5,192 Bytes
5146e76 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """CSV export for leaderboard entries.
Two flavors:
* leaderboard_csv(entries) - one row per (method) with aggregate score
+ flattened per-partner columns. Suitable for paper tables.
* detailed_csv(entry) - one row per (method, partner) pair, with
full statistics (mean, std, CI, n_episodes, normalized_mean).
Suitable for plots and statistical work.
"""
from __future__ import annotations
import csv
import io
from typing import Any
def leaderboard_csv(entries: list[dict[str, Any]]) -> str:
"""One row per method, partner scores as columns."""
if not entries:
return ""
partner_keys: list[str] = []
seen = set()
for e in entries:
for k in e.get("per_partner", {}):
if k not in seen:
partner_keys.append(k)
seen.add(k)
partner_keys.sort()
fieldnames = [
"rank", "agent_name", "env", "version", "aggregate_score",
"aggregate_ci_low", "aggregate_ci_high",
"num_episodes", "eval_seed", "wall_clock_seconds", "notes",
] + [f"{k}__norm" for k in partner_keys] + [f"{k}__mean" for k in partner_keys]
sorted_entries = sorted(
entries, key=lambda e: e.get("aggregate_score", 0), reverse=True,
)
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
for rank, e in enumerate(sorted_entries, start=1):
agg = e.get("aggregate", {})
row = {
"rank": rank,
"agent_name": e.get("agent_name", ""),
"env": e.get("env", ""),
"version": e.get("version", ""),
"aggregate_score": e.get("aggregate_score", agg.get("mean", "")),
"aggregate_ci_low": agg.get("ci_low", ""),
"aggregate_ci_high": agg.get("ci_high", ""),
"num_episodes": e.get("num_episodes", ""),
"eval_seed": e.get("eval_seed", ""),
"wall_clock_seconds": e.get("wall_clock_seconds", ""),
"notes": (e.get("notes") or "").replace("\n", " ").replace("\r", " "),
}
per_partner = e.get("per_partner", {})
for pk in partner_keys:
row[f"{pk}__norm"] = per_partner.get(pk, {}).get("normalized_mean", "")
row[f"{pk}__mean"] = per_partner.get(pk, {}).get("mean", "")
writer.writerow(row)
return buf.getvalue()
def detailed_csv(entry: dict[str, Any]) -> str:
"""One row per (method, partner) pair, with full statistics."""
fieldnames = [
"agent_name", "env", "version", "partner_key", "partner_display_name",
"mean", "std", "ci_low", "ci_high",
"normalized_mean", "normalized_ci_low", "normalized_ci_high",
"n_episodes", "mean_steps",
]
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
for pk, stats in (entry.get("per_partner") or {}).items():
row = {
"agent_name": entry.get("agent_name", ""),
"env": entry.get("env", ""),
"version": entry.get("version", ""),
"partner_key": pk,
"partner_display_name": stats.get("display_name", pk),
"mean": stats.get("mean", ""),
"std": stats.get("std", ""),
"ci_low": stats.get("ci_low", ""),
"ci_high": stats.get("ci_high", ""),
"normalized_mean": stats.get("normalized_mean", ""),
"normalized_ci_low": stats.get("normalized_ci_low", ""),
"normalized_ci_high": stats.get("normalized_ci_high", ""),
"n_episodes": stats.get("n_episodes", ""),
"mean_steps": stats.get("mean_steps", ""),
}
writer.writerow(row)
return buf.getvalue()
def comparison_markdown(entries_by_env: dict[str, list[dict[str, Any]]]) -> str:
"""Markdown table: rows = methods, columns = envs, cells = aggregate score."""
if not entries_by_env:
return "_no entries yet_\n"
method_to_env_score: dict[str, dict[str, tuple[float, float, float]]] = {}
for env, entries in entries_by_env.items():
for e in entries:
name = e.get("agent_name", "?")
agg = e.get("aggregate", {})
score = float(e.get("aggregate_score", agg.get("mean", 0)))
ci_low = float(agg.get("ci_low", score))
ci_high = float(agg.get("ci_high", score))
method_to_env_score.setdefault(name, {})[env] = (score, ci_low, ci_high)
envs = sorted(entries_by_env.keys())
methods = sorted(method_to_env_score.keys(),
key=lambda m: -sum(s[0] for s in method_to_env_score[m].values()))
lines = []
lines.append("| method | " + " | ".join(envs) + " |")
lines.append("|" + "---|" * (len(envs) + 1))
for m in methods:
row = [m]
for env in envs:
cell = method_to_env_score[m].get(env)
if cell is None:
row.append("—")
else:
score, lo, hi = cell
row.append(f"{score:.3f} ({lo:.3f}–{hi:.3f})")
lines.append("| " + " | ".join(row) + " |")
return "\n".join(lines) + "\n"
|