File size: 8,370 Bytes
bca45b0 | 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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | #!/usr/bin/env python3
"""Independent integrity and metric audit for a CV-Bench-500 evaluation."""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import re
from collections import Counter, defaultdict
from pathlib import Path
from PIL import Image
EXPECTED_SELECTION_HASH = "da5ca19b4a65b4df7f45bea069db43c76e48d9a33adc5a224d1abc64dfce1b92"
EXPECTED_SOURCE_REVISION = "bc284db50d036958861cb60cdd7b77612052ce0d"
TASK_COUNTS = {
("2D", "Count"): 137,
("2D", "Relation"): 113,
("3D", "Depth"): 125,
("3D", "Distance"): 125,
}
def _jsonl(path: Path) -> list[dict]:
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]
def _close(actual: float, expected: float) -> bool:
return math.isclose(float(actual), float(expected), rel_tol=0.0, abs_tol=1e-12)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--dataset-dir", type=Path, required=True)
parser.add_argument("--eval-dir", type=Path, required=True)
args = parser.parse_args()
dataset = args.dataset_dir.resolve()
evaluation = args.eval_dir.resolve()
manifest = _jsonl(dataset / "validation.jsonl")
summary = json.loads((dataset / "summary.json").read_text(encoding="utf-8"))
results = _jsonl(evaluation / "results.jsonl")
metrics_artifact = json.loads((evaluation / "metrics.json").read_text(encoding="utf-8"))
reported = metrics_artifact["metrics"]
if len(manifest) != 500 or len(results) != 500:
raise AssertionError(f"Expected 500 manifest/results rows, got {len(manifest)}/{len(results)}")
manifest_by_id = {row["sample_id"]: row for row in manifest}
result_by_id = {row["sample_id"]: row for row in results}
if len(manifest_by_id) != 500 or len(result_by_id) != 500:
raise AssertionError("Duplicate sample_id in manifest or results")
if set(manifest_by_id) != set(result_by_id):
raise AssertionError("Manifest/result sample_id coverage differs")
selection_items = [f"{row['config']}:{row['row_index']}" for row in manifest]
selection_hash = hashlib.sha256("\n".join(selection_items).encode()).hexdigest()
if selection_hash != EXPECTED_SELECTION_HASH or summary["selection_sha256"] != selection_hash:
raise AssertionError(f"Selection fingerprint mismatch: {selection_hash}")
if summary["source_revision"] != EXPECTED_SOURCE_REVISION:
raise AssertionError(f"Unexpected CV-Bench source revision: {summary['source_revision']}")
image_hashes: set[str] = set()
manifest_counts = Counter()
for row in manifest:
key = (str(row["config"]), str(row["task"]))
manifest_counts[key] += 1
image_path = (dataset / row["image"]).resolve()
if dataset not in image_path.parents or not image_path.is_file():
raise FileNotFoundError(f"Invalid image path: {image_path}")
raw = image_path.read_bytes()
image_hashes.add(hashlib.sha256(raw).hexdigest())
with Image.open(image_path) as image:
if image.size != (int(row["image_width"]), int(row["image_height"])):
raise AssertionError(f"Image dimension mismatch: {row['sample_id']}")
image.verify()
expected_match = re.fullmatch(r"\(([A-Z])\)", str(row["answer"]))
if not expected_match:
raise AssertionError(f"Malformed answer: {row['sample_id']}={row['answer']!r}")
expected_index = ord(expected_match.group(1)) - ord("A")
if expected_index >= len(row["choices"]):
raise AssertionError(f"Answer outside choices: {row['sample_id']}")
if manifest_counts != Counter(TASK_COUNTS) or len(image_hashes) != 500:
raise AssertionError(
f"Dataset coverage mismatch: tasks={dict(manifest_counts)}, unique_images={len(image_hashes)}"
)
shard_rows: list[dict] = []
for shard_path in sorted(evaluation.glob("rank_*.jsonl")):
shard_rows.extend(_jsonl(shard_path))
if len(shard_rows) != 500 or {row["sample_id"] for row in shard_rows} != set(result_by_id):
raise AssertionError("Per-rank shards do not cover the final result set exactly")
grouped: dict[tuple[str, str], list[dict]] = defaultdict(list)
correct_total = 0
parsed_total = 0
error_total = 0
for sample_id, result in result_by_id.items():
source = manifest_by_id[sample_id]
expected = re.fullmatch(r"\(([A-Z])\)", source["answer"]).group(1)
if result["config"] != source["config"] or result["task"] != source["task"]:
raise AssertionError(f"Result grouping mismatch: {sample_id}")
if result["expected"] != expected:
raise AssertionError(f"Result label mismatch: {sample_id}")
error_total += int(result["error"] is not None)
parsed_total += int(result["predicted"] is not None)
if result["predicted"] is not None and result["predicted"] not in [
chr(ord("A") + i) for i in range(len(source["choices"]))
]:
raise AssertionError(f"Predicted option outside choices: {sample_id}")
independently_correct = result["predicted"] == expected
if bool(result["correct"]) != independently_correct:
raise AssertionError(f"Incorrect correctness flag: {sample_id}")
correct_total += int(independently_correct)
grouped[(source["config"], source["task"])].append(result)
task_numerators = {
key: sum(bool(row["correct"]) for row in rows) for key, rows in grouped.items()
}
task_accuracy = {key: task_numerators[key] / TASK_COUNTS[key] for key in TASK_COUNTS}
independently_computed = {
"eval/cvbench_samples": 500.0,
"eval/cvbench_errors": float(error_total),
"eval/cvbench_parse_rate": parsed_total / 500,
"eval/cvbench_accuracy": correct_total / 500,
"eval/cvbench_2d_accuracy": (
task_numerators[("2D", "Count")] + task_numerators[("2D", "Relation")]
) / 250,
"eval/cvbench_3d_accuracy": (
task_numerators[("3D", "Depth")] + task_numerators[("3D", "Distance")]
) / 250,
"eval/cvbench_2d_count_accuracy": task_accuracy[("2D", "Count")],
"eval/cvbench_2d_relation_accuracy": task_accuracy[("2D", "Relation")],
"eval/cvbench_3d_depth_accuracy": task_accuracy[("3D", "Depth")],
"eval/cvbench_3d_distance_accuracy": task_accuracy[("3D", "Distance")],
}
independently_computed["eval/cvbench_2d_task_macro_accuracy"] = (
task_accuracy[("2D", "Count")] + task_accuracy[("2D", "Relation")]
) / 2
independently_computed["eval/cvbench_3d_task_macro_accuracy"] = (
task_accuracy[("3D", "Depth")] + task_accuracy[("3D", "Distance")]
) / 2
independently_computed["eval/cvbench_task_macro_accuracy"] = sum(task_accuracy.values()) / 4
for key, expected in independently_computed.items():
if key not in reported or not _close(reported[key], expected):
raise AssertionError(f"Metric mismatch {key}: reported={reported.get(key)}, expected={expected}")
swanlab = json.loads((evaluation / "swanlab.json").read_text(encoding="utf-8"))
if not swanlab.get("enabled") or not swanlab.get("run_id") or swanlab.get("error"):
raise AssertionError(f"SwanLab logging is incomplete: {swanlab}")
report = {
"assessment": "ready_to_share",
"dataset_rows": len(manifest),
"unique_sample_ids": len(manifest_by_id),
"unique_image_bytes": len(image_hashes),
"selection_sha256": selection_hash,
"source_revision": summary["source_revision"],
"task_denominators": {f"{key[0]}/{key[1]}": value for key, value in TASK_COUNTS.items()},
"task_numerators": {f"{key[0]}/{key[1]}": task_numerators[key] for key in TASK_COUNTS},
"correct_total": correct_total,
"parsed_total": parsed_total,
"error_total": error_total,
"metrics_verified": independently_computed,
"swanlab_run_id": swanlab["run_id"],
}
(evaluation / "validation_report.json").write_text(
json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
if __name__ == "__main__":
main()
|