Digital Drosophila Archive
76 lines
5.4 KiB
Python
76 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Seal Phase 4M after run-log closure and qualify whole-brain causal language."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import yaml
|
|
|
|
from phase4h_audit import sha256, write_json
|
|
from phase4k import seal
|
|
from phase4m import frozen_checks, frozen_hashes
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(); parser.add_argument("--config", type=Path, required=True); args = parser.parse_args(); mcfg = yaml.safe_load(args.config.read_text())
|
|
bpath = Path(mcfg["frozen_phase4l_config"]); cfg = yaml.safe_load(bpath.read_text()); cfg["paths"]["phase4l_root"] = cfg["paths"]["experiment_root"]; cfg["paths"]["phase4l_model_root"] = str(bpath.parent)
|
|
exp, model = Path(mcfg["experiment_root"]), Path(mcfg["model_root"]); summary_path = exp / "audit/final-summary.json"; summary = json.loads(summary_path.read_text())
|
|
if not summary["integrity"]["all_passed"]: raise RuntimeError("Phase 4M completion integrity was not PASS")
|
|
checks = frozen_checks(cfg)
|
|
if frozen_hashes(cfg) != summary["integrity"]["after"]: raise RuntimeError("frozen input changed after Phase 4M run")
|
|
whole = summary["whole_brain"]; whole["operational_gate_label_only"] = True
|
|
whole["feedback_causally_isolated_by_this_experiment"] = False
|
|
whole["interpretation"] = ("The prescribed WHOLE-BRAIN-FEEDBACK-MASKS-MEMORY label denotes isolated transfer PASS with whole-brain A/B specificity FAIL. "
|
|
"It does not prove feedback is the unique cause; shared plastic edges and baseline network input may also contribute.")
|
|
frame = pd.read_parquet(exp / "whole-brain-replay/results.parquet")
|
|
whole["endpoint_100_KC_recruitment_change"] = {mech: {odor: float(part.KC_spike_L1.mean()) for odor, part in group.groupby("odor")}
|
|
for mech, group in frame.groupby("mechanism")}
|
|
|
|
edge = pd.read_parquet(Path(cfg["paths"]["phase4l_root"]) / "eligible-edges/edge-identity.parquet")
|
|
categories = edge.odor_eligibility_category.astype(str).to_numpy(); static = {}; seed_rows = []
|
|
for seed in map(int, mcfg["protocol"]["train_seeds"]):
|
|
data = np.load(exp / "efficacy-overlay" / f"E-seed-{seed}-pairings-100.npz")
|
|
for mech, key in (("multiplicative", "E_multiplicative"), ("bounded_multiplicative", "E_bounded_multiplicative")):
|
|
E = data[key]
|
|
for cat in ("A-specific", "B-specific", "shared", "inactive"):
|
|
sel = categories == cat; seed_rows.append({"seed": seed, "mechanism": mech, "edge_category": cat,
|
|
"changed_edges": int(np.count_nonzero(E[sel] < 1 - 1e-7)), "E_reduction_mean": float((1 - E[sel]).mean()) if sel.any() else 0.0,
|
|
"E_reduction_sum": float((1 - E[sel]).sum())})
|
|
data.close()
|
|
audit = pd.DataFrame(seed_rows); audit.to_parquet(exp / "whole-brain-replay/endpoint-E-identity.parquet", index=False)
|
|
for mech, group in audit.groupby("mechanism"):
|
|
static[mech] = {cat: {"changed_edges_mean": float(g.changed_edges.mean()), "E_reduction_mean": float(g.E_reduction_mean.mean()),
|
|
"E_reduction_sum_mean": float(g.E_reduction_sum.mean())} for cat, g in group.groupby("edge_category")}
|
|
whole["endpoint_100_E_identity"] = static
|
|
whole["direct_feedback_vs_shared_edge_causal_disambiguation"] = "NOT-ESTABLISHED; no feedback edge ablation was authorized"
|
|
write_json(exp / "whole-brain-replay/summary.json", whole); summary["whole_brain"] = whole
|
|
|
|
iso = summary["isolated"]; iso["functional_transfer_gate_scope"] = "PASS at one or more predeclared pairing checkpoints, not necessarily retained at 100 pairings"
|
|
iso["endpoint_100_majority_pass"] = {mech: bool(gate["endpoint_100_specific_functional_seeds"] >= int(mcfg["gates"]["seed_majority"]))
|
|
for mech, gate in iso["mechanism_gates"].items()}
|
|
iso["nonmonotonic_functional_output_despite_monotonic_E_depression"] = True
|
|
write_json(exp / "isolated-replay/summary.json", iso); summary["isolated"] = iso
|
|
summary["integrity"]["postrun_frozen_recheck_passed"] = checks["all_passed"]
|
|
write_json(summary_path, summary)
|
|
version = json.loads((model / "VERSION.json").read_text()); version["summary_sha256"] = sha256(summary_path); write_json(model / "VERSION.json", version)
|
|
write_json(exp / "audit/post-run-qualification.json", {"only_phase4m_reports_edited": True, "brain_rerun": False,
|
|
"run_log_sealed_after_process_exit": True, "whole_brain_feedback_label_is_operational_not_causal_proof": True,
|
|
"frozen_recheck_passed": checks["all_passed"]})
|
|
shutil.copy2(Path(__file__).resolve(), exp / "audit/phase4m_finalize.py"); shutil.copy2(Path(__file__).resolve(), model / "phase4m_finalize.py")
|
|
seal(model, model / "MANIFEST.json", model / "checksums.sha256")
|
|
seal(exp, exp / "manifests/manifest.json", exp / "manifests/checksums.sha256")
|
|
for root, checksum in ((model, "checksums.sha256"), (exp, "manifests/checksums.sha256")):
|
|
check = subprocess.run(["sha256sum", "-c", checksum], cwd=root, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
|
if check.returncode: raise RuntimeError(f"self-checksum failure {root}: {check.stderr[-1000:]}")
|
|
print(json.dumps({"frozen": checks["all_passed"], "model_SHA256": "PASS", "experiment_SHA256": "PASS",
|
|
"endpoint_100_isolated_majority": iso["endpoint_100_majority_pass"]}))
|
|
|
|
|
|
if __name__ == "__main__": main()
|