| 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)) |
|
|
