#!/usr/bin/env python3 """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 # noqa: E402 from deepspace.fdir import FDIR, Facts, candidates, HYPOTHESES # noqa: E402 from deepspace.report import SelfReport # noqa: E402 from deepspace.runtime import PHYSICAL, run, scenario, score # noqa: E402 from deepspace.selfmaint import Rulebook, propose # noqa: E402 from deepspace.sim import CORRECT, HARMFUL, PROFILES, Spacecraft # noqa: E402 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)) # ---------------- baselines ---------------- 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 # ---------------- single-fault evaluation ---------------- 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 # ---------------- compound faults ---------------- 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) # harmful = an action wrong for one fault AND not a correct remedy for the other (v1 counted correct remedies) 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()