diff options
Diffstat (limited to 'experiments/import_native_baseline.py')
| -rw-r--r-- | experiments/import_native_baseline.py | 349 |
1 files changed, 349 insertions, 0 deletions
diff --git a/experiments/import_native_baseline.py b/experiments/import_native_baseline.py new file mode 100644 index 0000000..ae61727 --- /dev/null +++ b/experiments/import_native_baseline.py @@ -0,0 +1,349 @@ +#!/usr/bin/env python3 +"""Import completed author-code runs into deterministic, audited JSON records. + +Run this script with the baseline's own Python environment. BurstCCN needs the +``wandb`` package to decode its offline event file; Dual Prop needs ``numpy`` +to read the author's ``hist.npy``. The importer never evaluates or retrains a +model and refuses incomplete histories, dirty source trees, revision drift, +non-finite metrics, and protocol/config mismatches. +""" +import argparse +import hashlib +import json +import math +import os +import subprocess +import sys + + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +DEFAULT_PROTOCOLS = os.path.join( + ROOT, "experiments", "native_baseline_protocols.json") + + +def file_record(path): + digest = hashlib.sha256() + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return { + "path": os.path.abspath(path), + "bytes": os.path.getsize(path), + "sha256": digest.hexdigest(), + } + + +def run_text(command, cwd=None): + return subprocess.run( + command, cwd=cwd, check=True, capture_output=True, text=True).stdout.strip() + + +def git_record(path): + return { + "path": os.path.abspath(path), + "revision": run_text(["git", "rev-parse", "HEAD"], cwd=path), + "tracked_dirty": bool(run_text( + ["git", "status", "--porcelain", "--untracked-files=no"], cwd=path)), + } + + +def finite_number(value, label): + value = float(value) + if not math.isfinite(value): + raise ValueError(f"{label} is non-finite: {value!r}") + return value + + +def finite_list(values, label): + return [finite_number(value, f"{label}[{index}]") + for index, value in enumerate(values)] + + +def summarize_burst_history(rows, expected_epochs): + by_epoch = {} + for row in rows: + if "epoch" not in row: + continue + epoch = int(row["epoch"]) + if "epoch/top1_error/test" in row: + by_epoch[epoch] = row + expected = list(range(expected_epochs)) + observed = sorted(by_epoch) + if observed != expected: + raise ValueError( + f"BurstCCN history incomplete: expected epochs 0..{expected_epochs - 1}, " + f"found {len(observed)} rows ending at {observed[-1] if observed else None}") + + test_errors = finite_list( + [by_epoch[e]["epoch/top1_error/test"] for e in expected], "test top1 error") + val_errors = finite_list( + [by_epoch[e]["epoch/top1_error/val"] for e in expected], "val top1 error") + final = expected[-1] + best_test = min(range(expected_epochs), key=lambda e: test_errors[e]) + best_val = min(range(expected_epochs), key=lambda e: val_errors[e]) + runtime = finite_number(by_epoch[final].get("_runtime"), "W&B runtime") + return { + "epochs_observed": expected_epochs, + "optimization_epochs": expected_epochs - 1, + "initial_evaluation_epoch": 0, + "primary_final_test_accuracy_percent": 100.0 - test_errors[final], + "primary_final_epoch": final, + "final_validation_accuracy_percent": 100.0 - val_errors[final], + "validation_selected_epoch": best_val, + "validation_selected_validation_accuracy_percent": 100.0 - val_errors[best_val], + "validation_selected_test_accuracy_percent": 100.0 - test_errors[best_val], + "test_selected_epoch": best_test, + "test_selected_test_accuracy_percent": 100.0 - test_errors[best_test], + "test_evaluations": expected_epochs, + "wall_s": runtime, + "wall_definition": "W&B runtime from run initialization through the final epoch log", + } + + +def read_burst_history(path): + try: + from wandb.proto import wandb_internal_pb2 + from wandb.sdk.internal.datastore import DataStore + except ImportError as exc: + raise RuntimeError( + "BurstCCN import must run in an environment containing wandb") from exc + + store = DataStore() + store.open_for_scan(path) + rows = [] + while True: + data = store.scan_data() + if data is None: + break + record = wandb_internal_pb2.Record() + record.ParseFromString(data) + if record.WhichOneof("record_type") != "history": + continue + row = {} + for item in record.history.item: + key = item.key or "/".join(item.nested_key) + if key: + row[key] = json.loads(item.value_json) + rows.append(row) + return rows + + +def summarize_dualprop_hist(hist, expected_epochs): + required_vectors = ( + "train_loss", "train_accuracy", "train_time", "val_loss", + "val_accuracy", "val_top5accuracy", "val_time") + vectors = {} + for key in required_vectors: + if key not in hist: + raise ValueError(f"Dual Prop history lacks {key!r}") + values = list(hist[key]) + if len(values) != expected_epochs: + raise ValueError( + f"Dual Prop {key} has {len(values)} epochs, expected {expected_epochs}") + vectors[key] = finite_list(values, key) + for key in ("test_loss", "test_accuracy", "test_top5accuracy", "test_time"): + if key not in hist: + raise ValueError(f"Dual Prop history lacks {key!r}") + test_accuracy = finite_number(hist["test_accuracy"], "test_accuracy") + test_top5 = finite_number(hist["test_top5accuracy"], "test_top5accuracy") + test_loss = finite_number(hist["test_loss"], "test_loss") + test_time = finite_number(hist["test_time"], "test_time") + best_val = max(range(expected_epochs), key=lambda e: vectors["val_accuracy"][e]) + component_wall = (sum(vectors["train_time"]) + sum(vectors["val_time"]) + + test_time) + return { + "epochs_observed": expected_epochs, + "final_epoch": expected_epochs, + "final_train_accuracy_percent": vectors["train_accuracy"][-1], + "final_validation_accuracy_percent": vectors["val_accuracy"][-1], + "validation_selected_epoch": best_val + 1, + "validation_selected_validation_accuracy_percent": vectors["val_accuracy"][best_val], + "validation_selected_test_accuracy_percent": test_accuracy, + "primary_test_accuracy_percent": test_accuracy, + "primary_test_top5_accuracy_percent": test_top5, + "primary_test_loss": test_loss, + "test_evaluations": 1, + "test_selected_test_accuracy_percent": None, + "wall_s": component_wall, + "training_wall_s": sum(vectors["train_time"]), + "validation_wall_s": sum(vectors["val_time"]), + "test_wall_s": test_time, + "wall_definition": ( + "sum of author-measured train, validation, and final test runtimes; " + "excludes setup, checkpoint I/O, and diagnostic plotting"), + } + + +def read_dualprop_hist(path): + try: + import numpy as np + except ImportError as exc: + raise RuntimeError( + "Dual Prop import must run in an environment containing numpy") from exc + value = np.load(path, allow_pickle=True) + if value.shape != (): + raise ValueError(f"expected scalar object hist.npy, found shape {value.shape}") + hist = value.item() + if not isinstance(hist, dict): + raise ValueError(f"expected dict in hist.npy, found {type(hist).__name__}") + return hist + + +def nested_value(mapping, dotted_key): + value = mapping + for key in dotted_key.split("."): + if not isinstance(value, dict) or key not in value: + raise ValueError(f"resolved config lacks {dotted_key!r}") + value = value[key] + return value + + +def read_and_validate_config(path, required): + try: + from omegaconf import OmegaConf + except ImportError as exc: + raise RuntimeError( + "resolved Hydra config import must run in the BurstCCN environment") from exc + config = OmegaConf.to_container(OmegaConf.load(path), resolve=True) + mismatches = [] + for key, expected in required.items(): + actual = nested_value(config, key) + if actual != expected: + mismatches.append(f"{key}: expected {expected!r}, found {actual!r}") + if mismatches: + raise ValueError("resolved config mismatch:\n" + "\n".join(mismatches)) + return config + + +def environment_record(include_freeze=True): + record = { + "python_executable": os.path.abspath(sys.executable), + "python_version": sys.version, + } + if include_freeze: + freeze = run_text([sys.executable, "-m", "pip", "freeze"]).splitlines() + freeze = sorted(line for line in freeze if line.strip()) + encoded = ("\n".join(freeze) + "\n").encode() + record["pip_freeze"] = freeze + record["pip_freeze_sha256"] = hashlib.sha256(encoded).hexdigest() + return record + + +def gpu_record(index): + query = run_text([ + "nvidia-smi", f"--id={index}", + "--query-gpu=index,name,uuid,driver_version,memory.total", + "--format=csv,noheader,nounits", + ]) + fields = [value.strip() for value in query.split(",")] + if len(fields) != 5: + raise ValueError(f"unexpected nvidia-smi record: {query!r}") + return dict(zip(("physical_index", "name", "uuid", "driver_version", + "memory_total_mib"), fields)) + + +def load_protocol(path, protocol_id): + with open(path) as handle: + ledger = json.load(handle) + if ledger.get("schema_version") != 1: + raise ValueError("unsupported native protocol schema") + try: + return ledger["protocols"][protocol_id] + except KeyError as exc: + raise ValueError(f"unknown protocol {protocol_id!r}") from exc + + +def importer_provenance(): + return git_record(ROOT) + + +def import_result(args): + protocol = load_protocol(args.protocols, args.protocol) + source = git_record(args.source_dir) + if source["revision"] != protocol["source_revision"]: + raise ValueError( + f"source revision mismatch: expected {protocol['source_revision']}, " + f"found {source['revision']}") + if source["tracked_dirty"]: + raise ValueError("author source tree has tracked modifications") + local_source = importer_provenance() + if local_source["tracked_dirty"]: + raise ValueError("SDIL importer source tree has tracked modifications") + + if protocol["method"] == "BurstCCN": + if not args.resolved_config: + raise ValueError("BurstCCN requires --resolved-config") + resolved = read_and_validate_config( + args.resolved_config, protocol["required_config"]) + history = read_burst_history(args.artifact) + metrics = summarize_burst_history(history, protocol["expected_epochs"]) + elif protocol["method"] == "Dual Propagation": + if args.resolved_config: + raise ValueError("Dual Prop protocol is frozen directly in the ledger") + resolved = protocol["resolved_config"] + history = read_dualprop_hist(args.artifact) + metrics = summarize_dualprop_hist(history, protocol["expected_epochs"]) + else: + raise ValueError(f"unsupported native method {protocol['method']!r}") + + artifacts = [file_record(args.artifact)] + if args.resolved_config: + artifacts.append(file_record(args.resolved_config)) + dataset_artifacts = [file_record(path) for path in args.dataset_artifact] + result = { + "schema_version": 1, + "protocol_id": args.protocol, + "method": protocol["method"], + "dataset": protocol["dataset"], + "seed": protocol["seed"], + "protocol": protocol, + "resolved_config": resolved, + "metrics": metrics, + "audit": { + "complete": True, + "finite": True, + "expected_epochs": protocol["expected_epochs"], + "source_revision_match": True, + "source_tracked_clean": True, + "selection_semantics_frozen": True, + }, + "source_provenance": source, + "importer_provenance": local_source, + "runtime_environment": environment_record(not args.skip_package_freeze), + "gpu": gpu_record(args.gpu_index), + "dataset_artifacts": dataset_artifacts, + "run_artifacts": artifacts, + } + if os.path.exists(args.output) and not args.overwrite: + raise FileExistsError(f"refusing to overwrite {args.output}") + os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) + with open(args.output, "w") as handle: + json.dump(result, handle, indent=2, sort_keys=True) + handle.write("\n") + print(json.dumps({ + "output": os.path.abspath(args.output), + "method": result["method"], + "metrics": result["metrics"], + }, indent=2, sort_keys=True)) + + +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument("--protocol", required=True) + parser.add_argument("--protocols", default=DEFAULT_PROTOCOLS) + parser.add_argument("--source-dir", required=True) + parser.add_argument("--artifact", required=True, + help="BurstCCN run-*.wandb or Dual Prop hist.npy") + parser.add_argument("--resolved-config", + help="BurstCCN's .hydra/config.yaml") + parser.add_argument("--dataset-artifact", action="append", default=[]) + parser.add_argument("--gpu-index", required=True, type=int) + parser.add_argument("--output", required=True) + parser.add_argument("--skip-package-freeze", action="store_true", + help="development only; final audited imports retain the freeze") + parser.add_argument("--overwrite", action="store_true") + return parser.parse_args() + + +if __name__ == "__main__": + import_result(parse_args()) |