eval/eval_generalization.py

#!/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()