"""Independent boundary/source audit. Runs metric functions, never neural code.""" import ast from decimal import Decimal import hashlib import itertools import json from pathlib import Path from types import SimpleNamespace from typing import Optional import minimal_eval as local ROOT = Path(__file__).resolve().parent OUT = ROOT / "results" SNAP = ROOT / "source_snapshots" def load_functions(filename, names, namespace): """Select only reviewed function ASTs; no upstream imports or main execution.""" tree = ast.parse((SNAP / filename).read_text()) selected = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in names] assert len(selected) == len(names) exec(compile(ast.Module(body=selected, type_ignores=[]), filename, "exec"), namespace) return namespace def slow_distance(a, b): """Full-matrix reference, independent of the local rolling-row implementation.""" grid = [[0] * (len(b)+1) for _ in range(len(a)+1)] for i in range(len(a)+1): grid[i][0] = i for j in range(len(b)+1): grid[0][j] = j for i in range(1, len(a)+1): for j in range(1, len(b)+1): grid[i][j] = min(grid[i-1][j]+1, grid[i][j-1]+1, grid[i-1][j-1]+int(a[i-1] != b[j-1])) return grid[-1][-1] def main(): for entry in json.loads((SNAP / "manifest.json").read_text()): assert hashlib.sha256((SNAP / entry["file"]).read_bytes()).hexdigest() == entry["sha256"] chart = load_functions("evaluate_vqa.py", {"relaxed_correctness", "evaluate_relaxed_accuracy"}, {"Optional": Optional}) doc = load_functions("infographicsvqa_eval.py", {"levenshtein_distance", "evaluate_method", "validate_data"}, {"json": json, "question_ids_to_exclude": []}) edges = [] # These outputs are checked against hand-derived asymmetric denominators. for gold, pred, expected_source, expected_local in [ ("100", "105.2", 1, 0), ("100", "95", 0, 1), ("0", "0.0", 0, 1), ("50%", "0.5", 1, 1), ("50%", "50", 0, 0), ("-100", "-105.2", 1, 0), ("100", "The answer is 100", 0, 0), ]: actual_source = chart["evaluate_relaxed_accuracy"]([{"answer": pred, "annotation": [gold]}]) actual_local = local.numeric([gold], pred) assert actual_source == expected_source and actual_local == expected_local edges.append({"gold": gold, "prediction": pred, "source_chart": actual_source, "local_gold_denominator": actual_local}) golds = ["ab", "abcd", "abcd", "AB123456"] preds = ["ac", "abxx", "abxx" + " " * 16, "AB123450"] gt = {"dataset_name": "synthetic_metric_edges", "data": [ {"questionId": i, "question": "Metric-only boundary fixture", "answers": [g]} for i, g in enumerate(golds)]} submissions = [{"questionId": i, "answer": p} for i, p in enumerate(preds)] local.dump(OUT / "anls_edge_gold.json", gt) local.dump(OUT / "anls_edge_predictions.json", submissions) doc["validate_data"](OUT / "anls_edge_gold.json", OUT / "anls_edge_predictions.json") source = doc["evaluate_method"](OUT / "anls_edge_gold.json", OUT / "anls_edge_predictions.json", SimpleNamespace(answer_types=False, anls_threshold=.5)) anls_edges = [] for i, (g, p, expected) in enumerate(zip(golds, preds, [.5, .5, .9, .875])): source_score = source["per_sample_result"][str(i)]["score"] assert abs(source_score - expected) < 1e-12 anls_edges.append({"gold": g, "prediction": p, "source_anls": source_score, "local_normalized_strict": local.anls([g], p)}) assert [e["local_normalized_strict"] for e in anls_edges] == [0, 0, 0, .875] strings = ["".join(p) for n in range(5) for p in itertools.product("ab", repeat=n)] for a, b in itertools.product(strings, repeat=2): expected = slow_distance(a, b) assert local.distance(a, b) == expected == doc["levenshtein_distance"](a, b) test = local.read_jsonl(OUT / "test.jsonl") dev = local.read_jsonl(OUT / "dev.jsonl") assert not {r["doc_id"] for r in test} & {r["doc_id"] for r in dev} assert len(test) == 60 and len(dev) == 20 # Rebuild all gold answers from source specs, without calling the generator. for r in test + dev: s = r["spec"] a, b = s["series"]["cyan"], s["series"]["amber"] oracle = {"ocr": s["identifier"], "lookup": str(a), "difference": str(a-b), "percentage": format(100*a/(a+b), ".2f") + "%"} assert r["answers"] == [oracle[r["task"]]] report = json.loads((OUT / "summary.json").read_text()) rows_checked = 0 for mode, summary in report["modes"].items(): rows = json.loads((OUT / f"{mode}_rows.json").read_text()) numeric_rows = [r for r in rows if r["task"] != "ocr"] for row in numeric_rows: def parse(s): return Decimal(s[:-1])/100 if s.endswith("%") else Decimal(s) try: g, p = parse(row["answers"][0]), parse(row["prediction"]) expected = float(abs(g-p) <= abs(g)/20) except Exception: expected = 0.0 assert row["relaxed"] == expected rows_checked += 1 mapping = {"exact_match_all": (rows, "exact_match"), "anls_ocr": ([r for r in rows if r["task"] == "ocr"], "anls"), "relaxed_numeric": (numeric_rows, "relaxed"), "numeric_exact": (numeric_rows, "numeric_exact")} for metric, (subset, key) in mapping.items(): assert summary[metric]["n"] == len(subset) assert abs(summary[metric]["mean"] - sum(r[key] for r in subset)/len(subset)) < 1e-12 valid = local.predictions(test, "gold_fixture") bad_cases = [valid[:-1], valid + [valid[0]], [{**valid[0], "question_id": "unknown"}] + valid[1:]] for bad in bad_cases: try: local.evaluate(test, bad) except ValueError: pass else: raise AssertionError("Invalid submission was silently accepted") assert local.evaluate(test, list(reversed(valid)))[0] == local.evaluate(test, valid)[0] assert local.anls(["", "United States", "US"], "us") == 1 assert local.anls(["text"], "") == 0 assert local.number("NaN") is None and local.number("inf") is None result = {"status": "passed", "source_metric_functions_executed": True, "full_model_executed": False, "levenshtein_pairs": len(strings)**2, "gold_answers_rebuilt": len(test)+len(dev), "numeric_rows_independently_checked": rows_checked, "invalid_submission_checks": 3, "reordered_submission_invariance": True, "chart_edges": edges, "anls_edges": anls_edges, "warning": "Snapshot-specific metric behavior; not benchmark impact or model quality."} local.dump(OUT / "audit.json", result) print(json.dumps(result, indent=2)) if __name__ == "__main__": main()