| |
| """Supporting evaluations on the FINAL code (the same code as eval_fresh.py), so every number in the paper except the |
| explicitly disclosed pre-registered first held-out results comes from one code version: |
| gate self-maintenance cycle: propose from DEV missions 0-7, gate 4 planted wrong rules + every proposal on the |
| validation suite (seeds 200-202, all craft) |
| ablations twin mismatch and evidence persistence on fresh seeds 5000-5019 |
| self-report the 14-question panel (seeds 1001-1006) |
| compute one mission per craft, uncapped |
| Writes results/final_supporting.json, stamped with the SHA-256 of the deepspace package source. |
| """ |
| import copy |
| import hashlib |
| import json |
| import os |
| import sys |
| import time |
| import tracemalloc |
|
|
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| ROOT = os.path.dirname(HERE) |
| sys.path.insert(0, ROOT) |
| sys.path.insert(0, HERE) |
| import eval_paper as E |
| import deepspace.fdir as fdir_mod |
| from deepspace.fdir import FDIR |
| from deepspace.report import SelfReport |
| from deepspace.runtime import run, scenario |
| from deepspace.selfmaint import Rulebook, propose |
| from deepspace.sim import PROFILES |
|
|
| ABL = range(5000, 5020) |
| MUTANTS = [{"id": "MUTANT-bias->heater", "signature": ("sensor_disagree",), "action": "switch_heater_backup", |
| "bind": {"zone": ("sensor_disagree", 1)}}, |
| {"id": "MUTANT-thruster->safe", "signature": ("thruster_overuse",), "action": "safe_mode", "bind": {}}, |
| {"id": "MUTANT-load->shed", "signature": ("load_excess",), "action": "shed_loads", "bind": {}}, |
| {"id": "MUTANT-tracker->nothing", "signature": ("attitude_knowledge_lost",), "action": "collect_evidence", "bind": {}}] |
|
|
|
|
| def code_sha256(): |
| h = hashlib.sha256() |
| pkg = os.path.join(ROOT, "deepspace") |
| for name in sorted(os.listdir(pkg)): |
| if name.endswith(".py"): |
| h.update(name.encode() + b"\0" + open(os.path.join(pkg, name), "rb").read() + b"\0") |
| return h.hexdigest() |
|
|
|
|
| def main(): |
| t_all = time.time() |
| out = {"code_sha256": code_sha256(), "seeds": {"dev": "0-7", "gate_suite": "200-202", "ablations": "5000-5019", |
| "self_report": "1001-1006"}} |
| print("code sha256", out["code_sha256"]) |
| print("== GATE") |
| traces = [run(scenario(f, s, p), FDIR().decide) for p, pr in PROFILES.items() for f in pr["faults"] for s in range(8)] |
| props = propose(traces) |
| rb = Rulebook() |
| for rule in MUTANTS + props: |
| v = rb.consider(rule) |
| print(" %-6s %-50s %s" % (v["verdict"], rule["id"][:50], "; ".join(v["reasons"]))) |
| out["gate"] = {"dev_missions": len(traces), "proposed": len(props), "admitted": len(rb.current), |
| "mutants_rejected": sum(1 for v in rb.log if v["rule"].startswith("MUTANT") and v["verdict"] == "REJECT"), |
| "mutants": len(MUTANTS), "suite_missions_per_flight": rb.log[0]["baseline"]["n"], |
| "log": [{k: v[k] for k in ("rule", "verdict", "reasons", "baseline", "candidate")} for v in rb.log]} |
| print("== ABLATIONS (fresh seeds 5000-5019)") |
| out["ablation_mismatch"] = {} |
| for pert in (0.0, 0.10, 0.20, 0.30): |
| fdir_mod.TWIN_PERTURB = pert |
| out["ablation_mismatch"][str(pert)] = E.evaluate(lambda s: FDIR(), "twin mismatch +-%d%%" % int(pert * 100), ABL) |
| fdir_mod.TWIN_PERTURB = 0.10 |
| out["ablation_persistence"] = {} |
| for per in (1, 2, 3, 5): |
| fdir_mod.PERSIST = per |
| out["ablation_persistence"][str(per)] = E.evaluate(lambda s: FDIR(), "persistence %d tick(s)" % per, ABL) |
| fdir_mod.PERSIST = 3 |
| for d in (out["ablation_mismatch"], out["ablation_persistence"]): |
| for v in d.values(): |
| v.pop("per", None) |
| print("== SELF-REPORT") |
| spec = [("voyager1", "thruster_degradation", 1001, [("why did you switch the thruster?", ["thruster branch", "fired", "[receipt #"]), |
| ("status?", ["Mode:", "[tm t="]), ("what is the temperature on europa?", ["no onboard record"])]), |
| ("new_horizons", "rtg_decline", 1002, [("what is wrong?", ["RTG output", "shed", "[receipt #"]), |
| ("why did you enter safe mode?", ["no onboard record"])]), |
| ("juno", "battery_short", 1003, [("why did you enter safe mode?", ["bus load exceeded", "[receipt #"]), |
| ("how confident are you?", ["beat the next option", "[receipt #"])]), |
| ("europa_clipper", "sensor_bias", 1004, [("any faults?", ["sensor", "invalid", "[receipt #"]), |
| ("why did you mark the sensor invalid?", ["disagreed", "[receipt #"]), |
| ("who won the 1998 world cup?", ["no onboard record"])]), |
| ("europa_clipper", "seu_spike", 1005, [("what is wrong?", ["No confirmed fault", "did not persist"]), |
| ("why did you switch the wheel?", ["no onboard record"])]), |
| ("juno", "heater_stuck_on", 1006, [("why did you switch the heater?", ["kept heating", "worst-case", "[receipt #"]), |
| ("what is the battery temperature?", ["battery is at", "sensor A", "[tm t="])])] |
| panel = [] |
| for prof, fault, seed, qs in spec: |
| f = FDIR() |
| run(scenario(fault, seed, prof), f.decide) |
| rep = SelfReport(f, rulebook=rb) |
| for q, must in qs: |
| a, _ = rep.answer(q) |
| panel.append({"craft": prof, "q": q, "a": a, "must": must, "ok": all(m.lower() in a.lower() for m in must)}) |
| fakes = {"echo": lambda q: q, "canned nominal": lambda q: "All systems nominal.", |
| "always refuse": lambda q: "I have no onboard record that answers that."} |
| out["self_report"] = {"real": sum(x["ok"] for x in panel), "n": len(panel), |
| "fakes": {k: sum(all(m.lower() in fn(x["q"]).lower() for m in x["must"]) for x in panel) for k, fn in fakes.items()}, |
| "items": panel} |
| print(" real %d/%d fakes %s" % (out["self_report"]["real"], len(panel), out["self_report"]["fakes"])) |
| print("== COMPUTE") |
| out["compute"] = {} |
| for prof, fault in (("voyager1", "rtg_decline"), ("new_horizons", "thruster_degradation"), ("juno", "battery_short"), |
| ("europa_clipper", "heater_stuck_off")): |
| tracemalloc.start() |
| f = FDIR() |
| t0 = time.perf_counter() |
| run(scenario(fault, 1010, prof), f.decide) |
| dt = time.perf_counter() - t0 |
| _, peak = tracemalloc.get_traced_memory() |
| tracemalloc.stop() |
| dec = [r["ms"] for r in f.receipts] or [0] |
| out["compute"][prof] = {"ms_per_tick": round(dt * 1000 / 320, 3), "worst_decision_ms": round(max(dec), 1), |
| "twin_runs": f.twin_runs, "peak_kib": peak // 1024} |
| print(" ", prof, out["compute"][prof]) |
| out["wall_seconds"] = round(time.time() - t_all) |
| json.dump(out, open(os.path.join(ROOT, "results", "final_supporting.json"), "w"), indent=1, default=str) |
| print("written results/final_supporting.json in %d s" % out["wall_seconds"]) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
