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
|
#!/usr/bin/env python3
"""Apply the frozen A1 exact-BP viability gate."""
import argparse
import json
import math
import os
def read(path, variant):
with open(path) as handle:
record = json.load(handle)
args = record["args"]
expected = {
"mode": "bp", "depth": 20, "width": 16, "seed": 0,
"epochs": 200, "val_examples": 5000, "eval_split": "validation",
"normalization": "batchnorm",
}
for key, value in expected.items():
if args.get(key) != value:
raise ValueError(f"{path}: {key}={args.get(key)!r}, expected {value!r}")
if record["provenance"]["git_tracked_dirty"]:
raise ValueError(f"tracked-dirty result: {path}")
if record["split"]["validation_index_sha256"] != (
"8328b206a97c420e49e54e3eca4abe3274c4756b084355784ea3fb8059e4515b"):
raise ValueError(f"split drift: {path}")
accuracy = float(record["final"]["accuracy"])
loss = float(record["final"]["loss"])
return {
"variant": variant, "path": path,
"source_commit": record["provenance"]["git_commit"],
"accuracy": accuracy, "loss": loss,
"finite": record["final"]["finite"] and math.isfinite(accuracy + loss),
"test_evaluations": record["evaluation_protocol"]["test_evaluations"],
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--primary", default="results/oral_a_dev/bp_reference_primary.json")
parser.add_argument("--recovery", default="results/oral_a_dev/bp_reference_recovery.json")
parser.add_argument("--out", default="results/oral_a_bp_selection.json")
args = parser.parse_args()
primary = read(args.primary, "primary")
rows = [primary]
if primary["finite"] and primary["accuracy"] >= 0.90:
status = "passed_primary"
selected = primary
elif not os.path.exists(args.recovery):
status = "recovery_required"
selected = None
else:
recovery = read(args.recovery, "recovery")
rows.append(recovery)
selected = max(rows, key=lambda row: row["accuracy"] if row["finite"] else -1.0)
status = ("passed_recovery" if selected["finite"] and selected["accuracy"] >= 0.90
else "failed_no_viable_reference")
if any(row["test_evaluations"] for row in rows):
raise ValueError("A1 must not evaluate test")
output = {
"protocol": "oral_a_A1_v1", "status": status,
"rows": rows, "selected": selected,
"confirmation_test_seeds_touched": False,
}
os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
with open(args.out, "w") as handle:
json.dump(output, handle, indent=2, sort_keys=True)
handle.write("\n")
print(json.dumps(output, indent=2))
if __name__ == "__main__":
main()
|
