File size: 3,322 Bytes
20c251e | 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 | from __future__ import annotations
import csv
import subprocess
import sys
from pathlib import Path
from dovla_cil.experiments.reports import generate_eval_report
from dovla_cil.utils.io import write_json
def test_report_eval_works_with_fake_metrics(tmp_path: Path) -> None:
metrics_dir = tmp_path / "runs"
_write_fake_metrics(metrics_dir)
out_dir = tmp_path / "report"
result = subprocess.run(
[
sys.executable,
"scripts/report_eval.py",
"--inputs",
str(metrics_dir / "*" / "metrics.json"),
"--out",
str(out_dir),
"--name",
"fake_scaling",
],
check=True,
text=True,
capture_output=True,
)
assert "num runs: 3" in result.stdout
assert (out_dir / "aggregate_metrics.csv").exists()
assert (out_dir / "report.md").exists()
assert (out_dir / "success_rate.png").exists()
assert (out_dir / "ranking_accuracy.png").exists()
assert (out_dir / "score_vs_k.png").exists()
def test_eval_report_markdown_and_csv_content(tmp_path: Path) -> None:
metrics_dir = tmp_path / "runs"
_write_fake_metrics(metrics_dir)
out_dir = tmp_path / "report"
summary = generate_eval_report(
[metrics_dir / "*" / "metrics.json"],
out_dir,
experiment_name="fake_scaling",
)
rows = _read_csv(out_dir / "aggregate_metrics.csv")
markdown = (out_dir / "report.md").read_text(encoding="utf-8")
assert summary["num_runs"] == 3
assert rows[0]["k"] == "1"
assert rows[1]["k"] == "2"
assert rows[2]["k"] == "4"
assert rows[1]["ranking_acc"] == "0.7"
assert "# fake_scaling" in markdown
assert "Best K by ranking_acc: `2`" in markdown
assert "Best K by success: `4`" in markdown
assert "beta_log_k" in markdown
assert "Warning: ranking_acc does not improve monotonically with K." in markdown
def _write_fake_metrics(root: Path) -> None:
payloads = [
{
"run_name": "k1",
"k": 1,
"task_success_rate": 0.30,
"pairwise_ranking_accuracy": 0.50,
"top1_action_selection": 0.40,
"instruction_switch_accuracy": 0.20,
"effect_prediction_mae": 0.90,
"regret_calibration_error": 0.30,
},
{
"run_name": "k2",
"k": 2,
"task_success_rate": 0.50,
"pairwise_ranking_accuracy": 0.70,
"top1_action_selection": 0.60,
"instruction_switch_accuracy": 0.40,
"effect_prediction_mae": 0.70,
"regret_calibration_error": 0.20,
},
{
"run_name": "k4",
"k": 4,
"success_rate": 0.60,
"ranking_acc": 0.65,
"top1_action_selection": 0.70,
"instruction_switch_acc": 0.50,
"effect_mae": 0.60,
"regret_ece": 0.15,
"regression": {"ranking_acc": {"beta_log_k": 0.1}},
},
]
for payload in payloads:
out = root / f"k_{int(payload['k']):04d}" / "metrics.json"
write_json(payload, out)
def _read_csv(path: Path) -> list[dict[str, str]]:
with path.open("r", encoding="utf-8", newline="") as handle:
return list(csv.DictReader(handle))
|