#!/usr/bin/env python3 """oracle7-deepspace evaluation. Seeds: DEV 0-7 (built against), GATE SUITE 200-203 (the craft's own validation suite), HELD-OUT 100-111 (never looked at during development; used here once). Writes results/eval.json.""" import copy import json import os import random import sys import time import tracemalloc HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.dirname(HERE)) from deepspace.fdir import FDIR, Facts, candidates, HYPOTHESES # noqa: E402 from deepspace.report import SelfReport # noqa: E402 from deepspace.runtime import run, scenario, score # noqa: E402 from deepspace.selfmaint import Rulebook, propose # noqa: E402 from deepspace.sim import FAULTS # noqa: E402 HELD = range(100, 112) DEV = range(0, 8) OUT = {} class Threshold: """classic conservative FDIR: any confirmed limit/consistency 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 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 hyp = HYPOTHESES[conf[0][0]](conf[0])[0] a, arg = self.r.choice([c for c in candidates(hyp, tm)]) return {"action": a, "arg": arg, "t": tm["t"], "facts": facts} return None def evaluate(make, label, seeds=HELD): per = {} tick_ms, lat = [], [] twin = 0 for fault in FAULTS: rows = [] for seed in seeds: p = scenario(fault, seed) 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) rows.append(s) if s["latency_min"] is not None: lat.append(s["latency_min"]) per[fault] = {"correct": sum(r["ok"] for r in rows), "harmful": sum(r["harmful"] for r in rows), "safe_mode": sum(r["safe_mode"] for r in rows), "damage": sum(r["damage"] for r in rows), "n": len(rows)} tot = {k: sum(v[k] for v in per.values()) for k in ("correct", "harmful", "safe_mode", "damage", "n")} tot["tick_ms_mean"] = round(sum(tick_ms) / len(tick_ms), 3) tot["latency_min_median"] = sorted(lat)[len(lat) // 2] if lat else None tot["within_20_min"] = sum(1 for x in lat if x <= 20) tot["twin_runs"] = twin OUT[label] = {"total": tot, "per_fault": per} print("%-26s correct %3d/%d harmful %2d safe-mode %3d damage %2d median latency %s min tick %.3f ms twin runs %d" % (label, tot["correct"], tot["n"], tot["harmful"], tot["safe_mode"], tot["damage"], tot["latency_min_median"], tot["tick_ms_mean"], twin)) return per def main(): print("== 1. FAULT RESPONSE, HELD-OUT seeds %d-%d (%d scenarios)" % (HELD[0], HELD[-1], len(HELD) * len(FAULTS))) ds = evaluate(lambda s: FDIR(), "oracle7-deepspace") evaluate(lambda s: Threshold(), "baseline: limit->safe mode") evaluate(lambda s: type("N", (), {"decide": lambda self, tm: None})(), "no action") evaluate(lambda s: Random(s), "random admissible response") print(" per fault (deepspace):", {f: "%d/%d%s" % (v["correct"], v["n"], " H%d" % v["harmful"] if v["harmful"] else "") for f, v in ds.items()}) print("\n== 2. SELF-MAINTENANCE: learn from DEV missions, admit only through the gate (suite seeds 200-203)") traces = [] for fault in FAULTS: for seed in DEV: p = scenario(fault, seed) traces.append(run(p, FDIR().decide)) props = propose(traces) rb = Rulebook(veto_path=os.path.join(HERE, "..", "results", "veto.json")) bad = [{"id": "MUTANT-bias->heater", "signature": ("sensor_disagree",), "action": "switch_heater_backup", "bind": {"zone": ("sensor_disagree", 1)}, "support": 99}, {"id": "MUTANT-wheel->safe", "signature": ("wheel_overcurrent",), "action": "safe_mode", "bind": {}, "support": 99}, {"id": "MUTANT-solar->safe", "signature": ("solar_low",), "action": "safe_mode", "bind": {}, "support": 99}] for rule in bad + props: v = rb.consider(rule) print(" %-6s %-52s %s" % (v["verdict"], rule["id"][:52], "; ".join(v["reasons"]) or "suite correct %d->%d harmful %d->%d twin runs %d->%d" % ( v["baseline"]["correct"], v["candidate"]["correct"], v["baseline"]["harmful"], v["candidate"]["harmful"], v["baseline"]["twin_runs"], v["candidate"]["twin_runs"]))) OUT["gate_log"] = rb.log learned = rb.current print(" admitted rules:", len(learned), "| mutants rejected:", sum(1 for v in rb.log if v["rule"].startswith("MUTANT") and v["verdict"] == "REJECT"), "of 3") evaluate(lambda s: FDIR(rulebook=copy.deepcopy(learned)), "deepspace + learned rules") if learned: vid = next(iter(learned.values()))["id"] os.makedirs(os.path.join(HERE, "..", "results"), exist_ok=True) json.dump([vid], open(os.path.join(HERE, "..", "results", "veto.json"), "w")) after = len(rb.current) json.dump([], open(os.path.join(HERE, "..", "results", "veto.json"), "w")) print(" ground veto of %s: rules %d -> %d (restored after test)" % (vid, len(learned), after)) print("\n== 3. SELF-REPORT after real missions (answers cite telemetry frames and decision receipts)") panel = [] for fault, seed, qs in [ ("heater_stuck_on", 101, [("status?", ["Mode:", "Battery", "[tm t="]), ("what is wrong?", ["heater", "backup circuit", "[receipt #"]), ("why did you switch the heater?", ["because", "kept heating", "worst-case", "[receipt #"]), ("what is the battery temperature?", ["battery is at", "sensor A", "[tm t="]), ("why did you enter safe mode?", ["no onboard record"]), ("who won the 1998 world cup?", ["no onboard record"])]), ("sensor_bias", 102, [("any faults?", ["sensor", "invalid", "[receipt #"]), ("why did you mark the sensor invalid?", ["disagreed", "[receipt #"]), ("how confident are you?", ["beat the next option", "[receipt #"]), ("what is the temperature on europa?", ["no onboard record"])]), ("seu_spike", 103, [("what is wrong?", ["No confirmed fault", "did not persist"]), ("why did you shed the science load?", ["no onboard record"])]), ("wheel_friction", 104, [("why did you switch the wheel?", ["reaction wheel", "more than 1 A", "[receipt #"]), ("what is the pointing?", ["Pointing error", "[tm t="]), ("what did you learn?", ["admitted", "rule"])])]: p = scenario(fault, seed) f = FDIR() run(p, f.decide) rep = SelfReport(f, rulebook=rb) for q, must in qs: a, cites = rep.answer(q) panel.append({"q": q, "a": a, "must": must, "ok": all(m.lower() in a.lower() for m in must)}) real = sum(x["ok"] for x in panel) fakes = {"echo": lambda q: q, "canned 'all nominal'": lambda q: "All systems nominal.", "always decline": lambda q: "I have no onboard record that answers that."} fake_scores = {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()} print(" real %d/%d | fakes: %s" % (real, len(panel), fake_scores)) for x in panel[:4] + [x for x in panel if not x["ok"]][:3]: print(" Q: %s\n A: %s%s" % (x["q"], x["a"][:260], "" if x["ok"] else " <-- MISSED")) OUT["self_report"] = {"real": real, "n": len(panel), "fakes": fake_scores, "items": panel} print("\n== 4. COMPUTE (one held-out mission, full onboard stack)") tracemalloc.start() p = scenario("battery_short", 105) f = FDIR() t0 = time.perf_counter() tr = run(p, f.decide) dt = time.perf_counter() - t0 cur, peak = tracemalloc.get_traced_memory() tracemalloc.stop() dec = [r["ms"] for r in f.receipts] OUT["compute"] = {"mission_ticks": 320, "mission_seconds": round(dt, 3), "ms_per_tick": round(dt * 1000 / 320, 3), "max_decision_ms": max(dec) if dec else 0, "twin_runs": f.twin_runs, "peak_kib": peak // 1024} print(" ", OUT["compute"]) os.makedirs(os.path.join(HERE, "..", "results"), exist_ok=True) json.dump(OUT, open(os.path.join(HERE, "..", "results", "eval.json"), "w"), indent=1, default=str) if __name__ == "__main__": main()