#!/usr/bin/env python3 """Four questions a fault-protection reviewer asks, answered with the flight code FROZEN (SHA-256 recorded). Written, then run once on seeds 8000-8049 (never used before). A NEW SPACECRAFT four profiles added after the freeze (eval/new_craft.py): Voyager 2, Parker Solar Probe (hot), OSIRIS-APEX, MRO. Their fault classes x 50 seeds, with the baselines, and with the rules the gate admitted from the ORIGINAL four craft only (do learned rules transfer?). B NOVEL FAULTS five faults the code has no hypothesis for (eval/novel.py), on all eight spacecraft. C DETERMINISM record every telemetry frame, replay it through fresh flight code, compare every receipt; and the same missions in separate processes with different hash seeds must produce identical receipts. D SOAK each of the eight craft flies 60 simulated days continuously, a fault every 36-60 h (until its redundancy is spent) and a radiation upset every ~8 h: decisions, per-tick time and memory over time. Writes results/generalization_eval.json. """ import copy import hashlib import json import os import random import subprocess 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 # noqa: E402 from eval_final import code_sha256 # noqa: E402 from deepspace.fdir import FDIR # 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, ZONES # noqa: E402 SEEDS = range(8000, 8050) def learned_rules_from_original_craft(): """the gate cycle exactly as in eval_final.py, BEFORE the new profiles are registered""" traces = [run(scenario(f, s, p), FDIR().decide) for p, pr in PROFILES.items() for f in pr["faults"] for s in range(8)] rb = Rulebook() for rule in propose(traces): rb.consider(rule) return rb.current def outcome_ok(s, tr): return not s["damage"] and not s["harmful"] # ---------------- A: new spacecraft ---------------- def section_a(rules): import new_craft out = {} systems = {"oracle7-deepspace": lambda s: FDIR(), "oracle7-deepspace + rules learned on the original four": lambda s: FDIR(rulebook=copy.deepcopy(rules)), "expert fault table": lambda s: E.ExpertTable(), "limit -> safe mode (simplified)": lambda s: E.Threshold()} for prof in new_craft.NEW: out[prof] = {} for name, make in systems.items(): k = harm = dmg = safe = n = 0 per = {} for fault in PROFILES[prof]["faults"]: c = 0 for seed in SEEDS: p = scenario(fault, seed, prof) s = score(p, run(p, make(seed).decide)) k += s["ok"]; c += s["ok"]; harm += s["harmful"]; dmg += s["damage"]; safe += s["safe_mode"]; n += 1 per[fault] = c out[prof][name] = {"correct": E.wilson(k, n), "correct_n": k, "n": n, "harmful_n": harm, "damage_n": dmg, "safe_mode_n": safe, "per": per} print(" A %-19s %-58s correct %3d/%d = %.3f harmful %3d damage %3d" % (prof, name, k, n, k / n, harm, dmg), flush=True) return out # ---------------- B: novel faults ---------------- def fly_novel(sc_params, onboard, ticks=320): from novel import NovelCraft sc = NovelCraft(seed=sc_params["seed"] * 7 + 1, profile=sc_params["profile"]) acts = [] for t in range(ticks): if t == sc_params["onset"]: sc.inject_novel(sc_params["kind"], zone=sc_params["zone"], sensor=sc_params["sensor"]) sc.step() r = onboard(sc.telemetry()) if r and r["action"] in PHYSICAL: sc.command(r["action"], **r["arg"]) acts.append((sc.t, r["action"], r["arg"])) return acts, sc.damage def section_b(): from novel import NOVEL, novel_scenario systems = {"oracle7-deepspace": lambda: FDIR(), "expert fault table": lambda: E.ExpertTable(), "limit -> safe mode (simplified)": lambda: E.Threshold()} out = {} for kind, spec in NOVEL.items(): out[kind] = {} for name, make in systems.items(): strict = harm = dmg = n = 0 for prof in PROFILES: for seed in SEEDS: p = novel_scenario(kind, seed, prof) acts, damage = fly_novel(p, make().decide) after = [(t, a, g) for t, a, g in acts if t > p["onset"]] bad = [a for _, a, _ in after if a in spec["harmful"]] if spec["remedy"]: ok = any(a in spec["remedy"] and (g.get("zone") in (None, p["zone"])) and (kind != "sensor_stuck" or g.get("sensor") == p["sensor"]) for _, a, g in after) else: ok = not bad strict += ok and not bad; harm += bool(bad); dmg += bool(damage); n += 1 # outcome (no harmful act and no damage in the same run) is computed per run in section_b_outcome out[kind][name] = {"strict": E.wilson(strict, n), "strict_n": strict, "harmful_n": harm, "damage_n": dmg, "n": n, "outcome_ok": None} print(" B %-21s %-34s strict %3d/%d = %.3f harmful %3d damage %3d" % (kind, name, strict, n, strict / n, harm, dmg), flush=True) return out def section_b_outcome(): """outcome per run: no harmful action and no damage""" from novel import NOVEL, novel_scenario systems = {"oracle7-deepspace": lambda: FDIR(), "expert fault table": lambda: E.ExpertTable(), "limit -> safe mode (simplified)": lambda: E.Threshold()} out = {} for kind, spec in NOVEL.items(): out[kind] = {} for name, make in systems.items(): good = n = 0 for prof in PROFILES: for seed in SEEDS: p = novel_scenario(kind, seed, prof) acts, damage = fly_novel(p, make().decide) bad = [a for t, a, _ in acts if t > p["onset"] and a in spec["harmful"]] good += not bad and not damage; n += 1 out[kind][name] = E.wilson(good, n) return out # ---------------- C: determinism ---------------- def receipts_digest(seeds, profiles): h = hashlib.sha256() n = 0 for prof in profiles: for fault in PROFILES[prof]["faults"]: for seed in seeds: f = FDIR() run(scenario(fault, seed, prof), f.decide) for r in f.receipts: h.update(json.dumps({k: v for k, v in r.items() if k != "ms"}, sort_keys=True, default=str).encode()) n += 1 return h.hexdigest(), n def section_c(): from deepspace.sim import Spacecraft seeds = range(8000, 8010) same = total = 0 for prof in list(PROFILES): for fault in PROFILES[prof]["faults"]: for seed in seeds: p = scenario(fault, seed, prof) frames = [] f1 = FDIR() def rec(tm, f1=f1): frames.append(json.loads(json.dumps(tm))) return f1.decide(tm) run(p, rec) f2 = FDIR() for tm in frames: f2.decide(tm) a = [{k: v for k, v in r.items() if k != "ms"} for r in f1.receipts] b = [{k: v for k, v in r.items() if k != "ms"} for r in f2.receipts] same += json.dumps(a, sort_keys=True, default=str) == json.dumps(b, sort_keys=True, default=str) total += 1 digests = [] for hs in ("1", "4242", "random"): env = dict(os.environ, PYTHONHASHSEED=hs) code = ("import sys; sys.path[:0]=[%r,%r]; import new_craft, eval_generalization as G; " "d,n=G.receipts_digest(range(8000,8004), list(G.PROFILES)); print(d, n)") % (ROOT, HERE) o = subprocess.run([sys.executable, "-B", "-c", code], env=env, capture_output=True, text=True, cwd=HERE).stdout.split() digests.append({"PYTHONHASHSEED": hs, "digest": o[0], "receipts": int(o[1])}) print(" C replay identical %d/%d; process digests %s" % (same, total, [d["digest"][:12] for d in digests]), flush=True) return {"replay_identical": same, "replay_n": total, "process_digests": digests, "processes_agree": len({d["digest"] for d in digests}) == 1} # ---------------- D: soak ---------------- # Harness v2 (the first soak run mis-scored; the flight code is unchanged): each physical action is credited to the open # fault it remedies; Earth isolates a battery short and exits safe mode 72 h after a safe-mode entry, restoring the # science state the craft had chosen before safe mode (the first run left # the craft in safe mode for weeks, where several monitors are suspended by design); memory is the flight code's own # allocations (tracemalloc filtered to deepspace/fdir.py and report.py), with the timing buffer allocated before tracing. def section_d(days=60): from array import array from deepspace.sim import Spacecraft out = {} for prof in PROFILES: rng = random.Random(zlib_crc("soak:" + prof)) sc = Spacecraft(seed=zlib_crc(prof) % 10000, profile=prof) f = FDIR() ticks = days * 1440 tick_ms = array("d", bytes(8 * ticks)) next_fault = rng.randint(36 * 60, 60 * 60) next_seu = rng.randint(6 * 60, 10 * 60) used, used_zones = set(), set() faults, actions, mem, recoveries = [], [], [], 0 safe_since, science_seen, science_before = None, True, True tracemalloc.start() for t in range(ticks): if t == next_seu: sc.inject("seu_spike", zone=rng.choice(ZONES), sensor=rng.choice("AB")) next_seu += rng.randint(6 * 60, 10 * 60) if t == next_fault: cand = [x for x in PROFILES[prof]["faults"] if x not in ("nominal", "seu_spike") and (x not in used or x in ("heater_stuck_on", "heater_stuck_off", "sensor_bias"))] zones = [z for z in ZONES if z not in used_zones] cand = [x for x in cand if x not in ("heater_stuck_on", "heater_stuck_off", "sensor_bias") or [z for z in zones if x == "sensor_bias" or z != "avionics"]] if cand: kind = rng.choice(cand) zone = rng.choice([z for z in zones if kind == "sensor_bias" or z != "avionics"]) if kind in ( "heater_stuck_on", "heater_stuck_off", "sensor_bias") else "battery" if kind in ("heater_stuck_on", "heater_stuck_off", "sensor_bias"): used_zones.add(zone) else: used.add(kind) p = {"fault": kind, "zone": zone, "wheel": rng.randint(0, 2), "sensor": rng.choice("AB"), "magnitude": round(rng.random(), 3)} sc.inject(kind, zone=p["zone"], wheel=p["wheel"], sensor=p["sensor"], magnitude=p["magnitude"]) faults.append({"t": t, **p}) next_fault += rng.randint(36 * 60, 60 * 60) # Earth's recovery: 72 h after entering safe mode, isolate a short and return to science if sc.safe and safe_since is None: safe_since, science_before = t, science_seen if safe_since is not None and t - safe_since >= 72 * 60: # restore the science state the craft had chosen before safe mode (a load it shed stays shed) sc.short_amp, sc.safe, sc.science_on, safe_since = 0.0, False, science_before, None recoveries += 1 if not sc.safe: science_seen = sc.science_on t0 = time.perf_counter() sc.step() r = f.decide(sc.telemetry()) tick_ms[t] = (time.perf_counter() - t0) * 1000 if r and r["action"] in PHYSICAL: sc.command(r["action"], **r["arg"]) actions.append((t, r["action"], r["arg"])) if t % 1440 == 1439: snap = tracemalloc.take_snapshot().filter_traces( [tracemalloc.Filter(True, "*fdir.py"), tracemalloc.Filter(True, "*report.py")]) mem.append(sum(st.size for st in snap.statistics("filename")) // 1024) tracemalloc.stop() scored = [] for i, fl in enumerate(faults): end = faults[i + 1]["t"] if i + 1 < len(faults) else ticks fix = [(t, a, g) for t, a, g in actions if fl["t"] < t and a in CORRECT[fl["fault"]] and (fl["fault"] not in ("heater_stuck_on", "heater_stuck_off", "sensor_bias") or g.get("zone") == fl["zone"]) and (fl["fault"] != "wheel_friction" or g.get("wheel") == fl["wheel"])] bad = [a for t, a, _ in actions if fl["t"] < t <= end and a in HARMFUL.get(fl["fault"], set())] scored.append({"t": fl["t"], "fault": fl["fault"], "zone": fl["zone"], "remedied": bool(fix), "minutes": (fix[0][0] - fl["t"]) if fix else None, "harmful": bad}) days_ms = [sorted(tick_ms[d * 1440:(d + 1) * 1440]) for d in range(days)] out[prof] = {"days": days, "faults": len(faults), "remedied": sum(x["remedied"] for x in scored), "harmful": sum(bool(x["harmful"]) for x in scored), "earth_recoveries": recoveries, "damage_kinds": sorted({d[1] for d in sc.damage}), "receipts": len(f.receipts), "twin_runs": f.twin_runs, "tick_ms_p50_day1": round(days_ms[0][720], 4), "tick_ms_p50_last": round(days_ms[-1][720], 4), "tick_ms_p99_day1": round(days_ms[0][1425], 3), "tick_ms_p99_last": round(days_ms[-1][1425], 3), "tick_ms_max": round(max(tick_ms), 1), "flight_mem_kib_day1": mem[0], "flight_mem_kib_last": mem[-1], "fault_log": scored} o = out[prof] print(" D %-19s faults %d remedied %d harmful %d earth-recoveries %d damage %s | tick p50 %.3f->%.3f ms p99 %.2f->%.2f max %.0f ms | flight-code memory %d->%d KiB | receipts %d" % (prof, o["faults"], o["remedied"], o["harmful"], recoveries, o["damage_kinds"], o["tick_ms_p50_day1"], o["tick_ms_p50_last"], o["tick_ms_p99_day1"], o["tick_ms_p99_last"], o["tick_ms_max"], o["flight_mem_kib_day1"], o["flight_mem_kib_last"], o["receipts"]), flush=True) return out def zlib_crc(s): import zlib return zlib.crc32(s.encode()) def main(): t0 = time.time() out = {"code_sha256": code_sha256(), "seeds": "8000-8049"} print("code", out["code_sha256"], flush=True) rules = learned_rules_from_original_craft() out["rules_learned_on_original_four"] = sorted(v["id"] for v in rules.values()) print("== A. NEW SPACECRAFT (profiles added after the freeze)", flush=True) out["new_craft"] = section_a(rules) print("== B. NOVEL FAULTS (no hypothesis in the flight code), all eight spacecraft", flush=True) out["novel"] = section_b() oc = section_b_outcome() for kind in out["novel"]: for name in out["novel"][kind]: out["novel"][kind][name]["outcome_ok"] = oc[kind][name] for kind, v in out["novel"].items(): print(" B outcome %-21s %s" % (kind, {n: round(x["outcome_ok"][0], 3) for n, x in v.items()}), flush=True) print("== C. DETERMINISM", flush=True) out["determinism"] = section_c() print("== D. SOAK (60 simulated days per craft)", flush=True) out["soak"] = section_d() out["wall_seconds"] = round(time.time() - t0) json.dump(out, open(os.path.join(ROOT, "results", "generalization_eval.json"), "w"), indent=1, default=str) print("written results/generalization_eval.json in %d s" % out["wall_seconds"]) if __name__ == "__main__": if sys.argv[1:] == ["--soak-only"]: import new_craft # noqa: F401 path = os.path.join(ROOT, "results", "generalization_eval.json") doc = json.load(open(path)) doc["soak_first_run_misscored"] = doc.pop("soak", None) print("== D. SOAK v2 (60 simulated days per craft)", flush=True) doc["soak"] = section_d() json.dump(doc, open(path, "w"), indent=1, default=str) print("merged into results/generalization_eval.json") else: main()