#!/usr/bin/env python3 """Audit and mechanically select the frozen ResNet P1 learning rates.""" import argparse import glob import hashlib import json import math import os import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) from experiments.resnet_crossover_grid import p1_jobs, registry_sha256 GRIDDED = ("fa", "dfa", "pepita", "ep", "dualprop") def sha256(path): digest = hashlib.sha256() with open(path, "rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def read_json(path): with open(path, encoding="utf-8") as handle: return json.load(handle) def history_metrics(record, method): if method == "ff": layers = record["layers"] completed = sum(len(layer["epochs"]) for layer in layers) expected = record["architecture"]["depth"] assert len(layers) == expected assert all(len(layer["epochs"]) == 10 for layer in layers) losses = [ float(epoch["loss"]) for layer in layers for epoch in layer["epochs"]] validation = float(record["final"]["accuracy"]) final_loss = losses[-1] best_epoch = None elif record["args"].get("method") in ( "pepita", "ep", "dualprop"): epochs = record["epochs"] assert len(epochs) == 10 validations = [ float(epoch["validation"]["accuracy"]) for epoch in epochs] losses = [float(epoch["train_loss"]) for epoch in epochs] validation = float(record["final"]["accuracy"]) completed = len(epochs) best_epoch = max(range(len(validations)), key=validations.__getitem__) + 1 final_loss = float(record["final"]["loss"]) else: epochs = record["epochs"] assert len(epochs) == 10 validations = [float(epoch["eval_accuracy"]) for epoch in epochs] losses = [float(epoch["train_loss"]) for epoch in epochs] validation = float(record["final"]["accuracy"]) completed = len(epochs) best_epoch = max(range(len(validations)), key=validations.__getitem__) + 1 final_loss = float(record["final"]["loss"]) if method == "ff": best_validation = validation else: best_validation = max(validations + [validation]) return { "best_validation_accuracy": best_validation, "final_validation_accuracy": validation, "final_validation_loss": final_loss, "best_epoch": best_epoch, "epochs_completed": completed, "all_training_losses_finite": all( math.isfinite(value) for value in losses), } def audit_job(job, expected_source): manifest_path = job["output"] + ".manifest.json" assert os.path.isfile(manifest_path), ( f"missing manifest for {job['experiment_name']}") manifest = read_json(manifest_path) for key in ( "stage", "method", "architecture", "rate", "experiment_name", "output", "timeout_seconds", "command"): assert manifest[key] == job[key], ( f"{job['experiment_name']}: drift in {key}") assert manifest["source"] == expected_source hardware = manifest["hardware_lock"] assert hardware["physical_gpu_index"] in (5, 7) assert hardware["physical_gpu_uuid"] common = { "method": job["method"], "rate": job["rate"], "experiment_name": job["experiment_name"], "manifest": os.path.relpath(manifest_path, ROOT), "status": manifest["status"], "output_sha256": manifest["output_sha256"], "driver_wall_seconds": manifest["driver_wall_seconds"], "physical_gpu_index": hardware["physical_gpu_index"], "physical_gpu_uuid": hardware["physical_gpu_uuid"], } if manifest["status"] != "completed": assert manifest["status"] in { "timeout", "nonzero_exit", "missing_output"} return { **common, "finite": False, "best_validation_accuracy": None, "final_validation_accuracy": None, "final_validation_loss": None, "best_epoch": None, "epochs_completed": None, "all_training_losses_finite": False, } assert manifest["output_exists"] is True assert manifest["output_sha256"] == sha256(job["output"]) record = read_json(job["output"]) assert record["provenance"]["git_commit"] == expected_source["git_commit"] assert record["provenance"]["git_tracked_dirty"] is False assert record["args"]["depth"] == 20 assert record["args"]["seed"] == 0 assert record["args"]["eval_split"] == "validation" assert record["evaluation_protocol"]["test_evaluations"] == 0 assert record["evaluation_protocol"]["test_used_for_selection"] is False assert record["split"]["validation_examples"] == 5000 assert record["split"]["split_seed"] == 2027 if job["method"] in ("pepita", "ff", "ep", "dualprop"): assert record["args"]["method"] == job["method"] else: expected_mode = { "bp": "bp", "fa": "hfa", "dfa": "dfa", "clean_kp": "kp", "sdil": "kp_traffic", }[job["method"]] assert record["args"]["mode"] == expected_mode metrics = history_metrics(record, job["method"]) return { **common, "finite": ( metrics["all_training_losses_finite"] and math.isfinite(metrics["final_validation_loss"])), **metrics, } def choose(candidates): return max(candidates, key=lambda row: ( int(row["finite"]), ( row["best_validation_accuracy"] if row["best_validation_accuracy"] is not None else -math.inf), int( row["final_validation_loss"] is not None and math.isfinite(row["final_validation_loss"])), -row["rate"], )) def main(): parser = argparse.ArgumentParser() parser.add_argument( "--out", default="results/resnet_crossover/p1_selector.json") parser.add_argument("--allow-partial", action="store_true") args = parser.parse_args() jobs = p1_jobs() manifests = [ job["output"] + ".manifest.json" for job in jobs if os.path.isfile(job["output"] + ".manifest.json")] missing = [ job["experiment_name"] for job in jobs if not os.path.isfile(job["output"] + ".manifest.json")] if missing and not args.allow_partial: raise AssertionError(f"missing P1 jobs: {missing}") sources = [read_json(path)["source"] for path in manifests] if sources: expected_source = sources[0] assert all(source == expected_source for source in sources) else: expected_source = None records = [ audit_job(job, expected_source) for job in jobs if os.path.isfile(job["output"] + ".manifest.json")] selected = {} launch = None if not missing: launch_path = os.path.join( ROOT, "results", "resnet_crossover", "p1_launch.json") assert os.path.isfile(launch_path), "missing ResNet P1 launch lock" launch = read_json(launch_path) assert launch["stage"] == "p1" assert launch["source"] == expected_source assert launch["registry_sha256"] == registry_sha256(jobs) assert launch["num_jobs"] == len(jobs) assert launch["allowed_physical_gpus"] == [5, 7] for method in dict((job["method"], None) for job in jobs): candidates = [ record for record in records if record["method"] == method] winner = ( choose(candidates) if method in GRIDDED else candidates[0]) selected[method] = { "rate": winner["rate"], "best_validation_accuracy": winner["best_validation_accuracy"], "experiment_name": winner["experiment_name"], } report = { "gate": "pass" if not missing else "partial", "stage": "resnet_crossover_p1", "selection_rule": "maximum best validation accuracy, then finite final loss, " "then lower learning rate", "source": expected_source, "num_expected_records": len(jobs), "num_audited_records": len(records), "missing_experiments": missing, "test_policy": "none", "launch_lock": launch, "records": records, "selected": selected, } os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True) with open(args.out, "w", encoding="utf-8") as handle: json.dump(report, handle, indent=2, sort_keys=True) handle.write("\n") print(json.dumps({ "gate": report["gate"], "num_audited_records": len(records), "missing_experiments": missing, "selected": selected, }, indent=2, sort_keys=True)) if __name__ == "__main__": main()