|
|
| """Paper evaluation for oracle7-deepspace. Seeds (fixed before any held-out run):
|
| DEV 0-7 (built against) | GATE SUITE 200-202 (the craft's own validation suite) | HELD-OUT 1000-1049 (main)
|
| COMPOUND 3000-3019 | ablations use the first 20 held-out seeds (1000-1019).
|
| Writes results/paper_eval.json. Every rate is reported with a Wilson 95% interval."""
|
| import copy
|
| import json
|
| import math
|
| import os
|
| import random
|
| import sys
|
| import time
|
| import tracemalloc
|
|
|
| HERE = os.path.dirname(os.path.abspath(__file__))
|
| ROOT = os.path.dirname(HERE)
|
| sys.path.insert(0, ROOT)
|
| import deepspace.fdir as fdir_mod
|
| from deepspace.fdir import FDIR, Facts, candidates, HYPOTHESES
|
| from deepspace.report import SelfReport
|
| from deepspace.runtime import PHYSICAL, run, scenario, score
|
| from deepspace.selfmaint import Rulebook, propose
|
| from deepspace.sim import CORRECT, HARMFUL, PROFILES, Spacecraft
|
|
|
| HELD = range(1000, 1050)
|
| COMPOUND_SEEDS = range(int(os.environ.get("COMPOUND_START", "3000")), int(os.environ.get("COMPOUND_START", "3000")) + 20)
|
| ABL = range(1000, 1020)
|
| OUT = {"seeds": {"dev": "0-7", "gate_suite": "200-202", "held_out": "1000-1049", "compound": "3000-3019"}}
|
|
|
|
|
| def wilson(k, n, z=1.96):
|
| if n == 0:
|
| return (0.0, 0.0, 0.0)
|
| p = k / n
|
| d = 1 + z * z / n
|
| c = (p + z * z / (2 * n)) / d
|
| h = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / d
|
| return (round(p, 4), round(max(0, c - h), 4), round(min(1, c + h), 4))
|
|
|
|
|
|
|
| class Threshold:
|
| """classic conservative fault protection: any confirmed violation -> safe mode"""
|
| def __init__(self):
|
| self.facts, self.done = Facts(), False
|
|
|
| def decide(self, tm):
|
| facts = self.facts.update(tm)
|
| if not self.done and [f for f in facts if f[0] != "transient_spike"]:
|
| self.done = True
|
| return {"action": "safe_mode", "arg": {}, "t": tm["t"], "facts": facts}
|
| return None
|
|
|
|
|
| EXPERT = {"sensor_disagree": ("mark_sensor_invalid", {"zone": 1, "sensor": 2}),
|
| "heater_on_uncommanded": ("switch_heater_backup", {"zone": 1}),
|
| "heater_off_uncommanded": ("switch_heater_backup", {"zone": 1}),
|
| "wheel_overcurrent": ("switch_to_spare_wheel", {"wheel": 1}),
|
| "attitude_knowledge_lost": ("switch_attitude_sensor", {}),
|
| "thruster_overuse": ("switch_thruster_branch", {}),
|
| "solar_low": ("shed_loads", {}), "rtg_low": ("shed_loads", {}), "load_excess": ("safe_mode", {})}
|
|
|
|
|
| class ExpertTable:
|
| """a hand-authored fault table (fact -> designed response): what well-designed flight fault protection does for
|
| ANTICIPATED faults. Same facts module as oracle7-deepspace, so the comparison isolates the decision layer."""
|
| def __init__(self):
|
| self.facts, self.handled = Facts(), set()
|
|
|
| def decide(self, tm):
|
| facts = self.facts.update(tm)
|
| for f in facts:
|
| if f[0] in EXPERT and f not in self.handled:
|
| self.handled.add(f)
|
| a, bind = EXPERT[f[0]]
|
| return {"action": a, "arg": {k: f[i] for k, i in bind.items()}, "t": tm["t"], "facts": facts}
|
| return None
|
|
|
|
|
| class Random:
|
| def __init__(self, seed):
|
| self.facts, self.done, self.r = Facts(), False, random.Random(seed)
|
|
|
| def decide(self, tm):
|
| facts = self.facts.update(tm)
|
| conf = [f for f in facts if f[0] in HYPOTHESES]
|
| if not self.done and conf:
|
| self.done = True
|
| a, arg = self.r.choice(candidates(HYPOTHESES[conf[0][0]](conf[0])[0], tm))
|
| return {"action": a, "arg": arg, "t": tm["t"], "facts": facts}
|
| return None
|
|
|
|
|
| class NoAction:
|
| def decide(self, tm):
|
| return None
|
|
|
|
|
|
|
| def evaluate(make, label, seeds=HELD, quiet=False):
|
| per, lat = {}, []
|
| k = harm = safe = dmg = n = trap_act = trap_n = 0
|
| twin = 0
|
| tick_ms = []
|
| for prof, pr in PROFILES.items():
|
| for fault in pr["faults"]:
|
| c = h = 0
|
| for seed in seeds:
|
| p = scenario(fault, seed, prof)
|
| agent = make(seed)
|
| t0 = time.perf_counter()
|
| tr = run(p, agent.decide)
|
| tick_ms.append((time.perf_counter() - t0) * 1000 / 320)
|
| twin += getattr(agent, "twin_runs", 0)
|
| s = score(p, tr)
|
| k += s["ok"]; c += s["ok"]; harm += s["harmful"]; h += s["harmful"]
|
| safe += s["safe_mode"]; dmg += s["damage"]; n += 1
|
| if fault in ("nominal", "seu_spike", "sensor_bias"):
|
| trap_n += 1
|
| trap_act += bool(tr["actions"]) and not s["ok"]
|
| if s["latency_min"] is not None:
|
| lat.append(s["latency_min"])
|
| per["%s/%s" % (prof, fault)] = {"correct": c, "harmful": h, "n": len(seeds)}
|
| res = {"correct": wilson(k, n), "correct_n": k, "harmful": wilson(harm, n), "harmful_n": harm,
|
| "safe_mode_n": safe, "damage_n": dmg, "n": n, "trap_wrong_action": wilson(trap_act, trap_n),
|
| "latency_min_median": sorted(lat)[len(lat) // 2] if lat else None,
|
| "latency_min_p90": sorted(lat)[int(len(lat) * 0.9)] if lat else None, "twin_runs": twin,
|
| "tick_ms_mean": round(sum(tick_ms) / len(tick_ms), 4), "per": per}
|
| if not quiet:
|
| print(" %-30s correct %4d/%d = %.3f [%.3f,%.3f] harmful %3d safe-mode %4d damage %3d trap-wrong %.3f lat med %s p90 %s"
|
| % (label, k, n, *res["correct"], harm, safe, dmg, res["trap_wrong_action"][0],
|
| res["latency_min_median"], res["latency_min_p90"]))
|
| return res
|
|
|
|
|
|
|
| PAIRS = {"voyager1": [("heater_stuck_off", "rtg_decline"), ("sensor_bias", "thruster_degradation"),
|
| ("heater_stuck_on", "star_tracker_dropout")],
|
| "new_horizons": [("heater_stuck_off", "thruster_degradation"), ("sensor_bias", "rtg_decline"),
|
| ("heater_stuck_on", "star_tracker_dropout")],
|
| "juno": [("heater_stuck_off", "solar_degradation"), ("sensor_bias", "thruster_degradation"),
|
| ("battery_short", "heater_stuck_on")],
|
| "europa_clipper": [("heater_stuck_off", "wheel_friction"), ("sensor_bias", "star_tracker_dropout"),
|
| ("battery_short", "heater_stuck_on")]}
|
|
|
|
|
| def run_compound(prof, f1, f2, seed, onboard):
|
| p1, p2 = scenario(f1, seed, prof), scenario(f2, seed + 7777, prof)
|
| if f1 in ("sensor_bias",) and f2 in ("heater_stuck_on", "heater_stuck_off") and p1["zone"] == p2["zone"]:
|
| p2["zone"] = "tank" if p1["zone"] != "tank" else "battery"
|
| on2 = p1["onset"] + 30
|
| sc = Spacecraft(seed=seed * 7 + 1, profile=prof)
|
| acts = []
|
| for t in range(360):
|
| if t == p1["onset"]:
|
| sc.inject(f1, zone=p1["zone"], wheel=p1["wheel"], sensor=p1["sensor"], magnitude=p1["magnitude"])
|
| if t == on2:
|
| sc.inject(f2, zone=p2["zone"], wheel=p2["wheel"], sensor=p2["sensor"], magnitude=p2["magnitude"])
|
| sc.step()
|
| r = onboard(sc.telemetry())
|
| if r and r["action"] in PHYSICAL:
|
| sc.command(r["action"], **r["arg"])
|
| acts.append(r["action"])
|
| need = [CORRECT[f1], CORRECT[f2]]
|
| both = all(any(a in c for a in acts) for c in need)
|
|
|
| bad = any((a in HARMFUL.get(f1, set()) and a not in CORRECT[f2]) or (a in HARMFUL.get(f2, set()) and a not in CORRECT[f1]) for a in acts)
|
| return both, bad, bool(sc.damage), "safe_mode" in acts
|
|
|
|
|
| def compound(make, label):
|
| k = bad = dmg = safe = n = 0
|
| for prof, pairs in PAIRS.items():
|
| for f1, f2 in pairs:
|
| for seed in COMPOUND_SEEDS:
|
| both, b, d, s = run_compound(prof, f1, f2, seed, make(seed).decide)
|
| k += both; bad += b; dmg += d; safe += s; n += 1
|
| res = {"both_remedied": wilson(k, n), "both_n": k, "harmful_n": bad, "damage_n": dmg, "safe_mode_n": safe, "n": n}
|
| print(" %-30s both faults remedied %3d/%d = %.3f [%.3f,%.3f] harmful %d damage %d safe-mode %d"
|
| % (label, k, n, *res["both_remedied"], bad, dmg, safe))
|
| return res
|
|
|
|
|
| def main():
|
| t_all = time.time()
|
| print("== 1. SINGLE FAULTS, HELD-OUT (4 craft x their fault classes x 50 seeds)")
|
| OUT["main"] = {
|
| "oracle7-deepspace": evaluate(lambda s: FDIR(), "oracle7-deepspace"),
|
| "expert fault table": evaluate(lambda s: ExpertTable(), "expert fault table (hand-authored)"),
|
| "limit -> safe mode": evaluate(lambda s: Threshold(), "limit -> safe mode"),
|
| "random admissible": evaluate(lambda s: Random(s), "random admissible response"),
|
| "no action": evaluate(lambda s: NoAction(), "no action")}
|
|
|
| print("\n== 2. SELF-MAINTENANCE: learn from DEV missions, admit only via the gate (suite 200-202, all craft)")
|
| traces = [run(scenario(f, s, prof), FDIR().decide) for prof, pr in PROFILES.items() for f in pr["faults"] for s in range(8)]
|
| props = propose(traces)
|
| rb = Rulebook()
|
| 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": {}}]
|
| for rule in mutants + props:
|
| v = rb.consider(rule)
|
| print(" %-6s %-55s %s" % (v["verdict"], rule["id"][:55], "; ".join(v["reasons"]) or "suite correct %d->%d harmful %d->%d twin %d->%d" % (
|
| v["baseline"]["correct"], v["candidate"]["correct"], v["baseline"]["harmful"], v["candidate"]["harmful"],
|
| v["baseline"]["twin_runs"], v["candidate"]["twin_runs"])))
|
| learned = rb.current
|
| OUT["gate"] = {"proposed": len(props), "admitted": len(learned),
|
| "mutants_rejected": sum(1 for v in rb.log if v["rule"].startswith("MUTANT") and v["verdict"] == "REJECT"),
|
| "mutants": len(mutants), "log": [{k: v[k] for k in ("rule", "verdict", "reasons", "baseline", "candidate")} for v in rb.log]}
|
| OUT["main"]["oracle7-deepspace + learned rules"] = evaluate(lambda s: FDIR(rulebook=copy.deepcopy(learned)),
|
| "oracle7-deepspace + learned rules")
|
|
|
| print("\n== 3. COMPOUND FAULTS (two faults, 30 min apart; 12 pairs x 20 seeds)")
|
| OUT["compound"] = {"oracle7-deepspace": compound(lambda s: FDIR(), "oracle7-deepspace"),
|
| "expert fault table": compound(lambda s: ExpertTable(), "expert fault table"),
|
| "limit -> safe mode": compound(lambda s: Threshold(), "limit -> safe mode")}
|
|
|
| print("\n== 4. ABLATIONS (held-out seeds 1000-1019)")
|
| OUT["ablation_mismatch"] = {}
|
| for pert in (0.0, 0.10, 0.20, 0.30):
|
| fdir_mod.TWIN_PERTURB = pert
|
| OUT["ablation_mismatch"][str(pert)] = 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)] = evaluate(lambda s: FDIR(), "persistence %d tick(s)" % per, ABL)
|
| fdir_mod.PERSIST = 3
|
|
|
| print("\n== 5. SELF-REPORT across craft")
|
| panel = []
|
| 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="])])]
|
| 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"]))
|
| for x in panel:
|
| if not x["ok"]:
|
| print(" MISSED:", x["craft"], x["q"], "->", x["a"][:200])
|
|
|
| print("\n== 6. COMPUTE per craft (uncapped, one held-out mission each)")
|
| 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)
|
| os.makedirs(os.path.join(ROOT, "results"), exist_ok=True)
|
| json.dump(OUT, open(os.path.join(ROOT, "results", "paper_eval.json"), "w"), indent=1, default=str)
|
| print("\nwritten results/paper_eval.json in %d s" % OUT["wall_seconds"])
|
|
|
|
|
| if __name__ == "__main__":
|
| main()
|
|
|
