File size: 9,146 Bytes
3538060 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | #!/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()
|