"""oracle7-deepspace fault card: trafficcop for Fault Detection, Isolation and Recovery (FDIR). telemetry window -> FACTS (limits, redundancy disagreement, model disagreement, persistence) -> HYPOTHESES (which fault could produce these facts) -> ADMISSIBLE RESPONSES (each operator's preconditions checked against the echoed state) -> fast path: an ADMITTED RULE for this fact signature (learned, gated, rollback-able), or slow path: LOOKAHEAD on the onboard DIGITAL TWIN - every candidate response is simulated under every hypothesis still consistent with the facts; the response with the best WORST case is chosen -> a RECEIPT naming the facts, hypotheses, scored alternatives and the path taken A single-tick anomaly never licenses a physical action: the honest answer is collect_evidence. The twin is built from telemetry only (what the craft can know), with physics perturbed +-10%, never from the truth. """ from __future__ import annotations import time from .sim import BANDS, PROFILES, SURVIVAL, THERMOSTAT, ZONES, Spacecraft ACTION_COST = {"switch_thruster_branch": 2.0, "no_action": 0.0, "collect_evidence": 0.0, "mark_sensor_invalid": 1.0, "switch_heater_backup": 2.0, "switch_to_spare_wheel": 2.0, "switch_attitude_sensor": 2.0, "shed_loads": 8.0, "safe_mode": 40.0} LATENT_WEIGHT = {"battery_short": 50.0} TWIN_PERTURB = 0.10 # twin model mismatch (thermal and source constants); swept in the ablation WAIT_EXPIRY = 30 # minutes after which a decision to wait is re-examined PERSIST = 3 # ticks an anomaly must persist before it is a CONFIRMED fact HORIZON = 90 # twin lookahead ticks (90 min) def _persist(): return PERSIST class Facts: """derives facts from a sliding telemetry window; remembers what the model predicted, to judge sensors""" def __init__(self): self.hist = [] self.counts = {} def _bump(self, key, on): self.counts[key] = self.counts.get(key, 0) + 1 if on else 0 return self.counts[key] def update(self, tm): self.hist.append(tm) del self.hist[:-30] invalid = set(tm["sensor_invalid"]) facts = [] for z in ZONES: a, b = tm["%s_A" % z], tm["%s_B" % z] va, vb = ("%s.A" % z) not in invalid, ("%s.B" % z) not in invalid if va and vb: n = self._bump(("disagree", z), abs(a - b) > 6.0) if n >= _persist(): suspect = self._suspect(z) facts.append(("sensor_disagree", z, suspect)) elif n == 1: spiked = "A" if abs(a - self._prev(z)) > abs(b - self._prev(z)) else "B" facts.append(("transient_spike", z, spiked)) est = self.zone_temp(tm, z) lo, hi = BANDS[z] cmd = tm["%s_heater_cmd" % z] rising = self._trend(z) > 0.02 falling = self._trend(z) < -0.02 # uncommanded heating: above the thermostat's off-point, commanded OFF, still rising if self._bump(("hot_uncmd", z), est > THERMOSTAT[z][1] + 3 and not cmd and rising) >= _persist(): facts.append(("heater_on_uncommanded", z)) if self._bump(("cold_uncmd", z), est < THERMOSTAT[z][0] - 3 and cmd and falling) >= _persist(): facts.append(("heater_off_uncommanded", z)) if self._bump(("band", z), not lo <= est <= hi) >= _persist(): facts.append(("zone_out_of_band", z, "high" if est > hi else "low")) for w in tm["wheels_active"]: if self._bump(("wheel", w), tm["wheel%d_current" % w] > 1.0) >= 2: facts.append(("wheel_overcurrent", w)) if self._bump("tracker", (not tm["star_tracker_valid"]) and tm["attitude_sensor"] == "star_tracker") >= 2: facts.append(("attitude_knowledge_lost",)) pr = PROFILES[tm.get("profile", "europa_clipper")] rtg = tm.get("power_source") == "rtg" ratio = 0.93 if rtg else 0.85 if self._bump("source", tm["i_source"] < ratio * tm["i_source_nominal"] and not tm["safe_mode"]) >= _persist(): facts.append(("rtg_low",) if rtg else ("solar_low",)) if tm.get("actuator") == "thrusters" and self._bump("thruster", tm["thruster_duty"] > 0.15) >= _persist(): facts.append(("thruster_overuse", tm["thruster_branch"])) expected_load = pr["base_load"] + pr["heater_amp"] * sum(1 for z in ZONES if tm["%s_heater_cmd" % z]) \ + (pr["science_amp"] if tm["science_on"] else 0) + sum(tm["wheel%d_current" % w] for w in tm["wheels_active"]) \ + (0.6 * tm["thruster_duty"] if tm.get("actuator") == "thrusters" else 0.0) if self._bump("load", tm["i_load"] - expected_load > 2.5 and not tm["safe_mode"]) >= _persist(): facts.append(("load_excess", round(tm["i_load"] - expected_load, 2))) return facts def _prev(self, z): if len(self.hist) < 2: return self.zone_temp(self.hist[-1], z) return self.zone_temp(self.hist[-2], z) def zone_temp(self, tm, z): inv = set(tm["sensor_invalid"]) vals = [tm["%s_%s" % (z, s)] for s in "AB" if ("%s.%s" % (z, s)) not in inv] if len(vals) == 2 and abs(vals[0] - vals[1]) > 6.0 and len(self.hist) > 3: # the two disagree: trust the one that tracks the zone's own recent history ref = sum(self.zone_temp_raw(h, z) for h in self.hist[-6:-3]) / 3.0 vals = [min(vals, key=lambda v: abs(v - ref))] return sum(vals) / len(vals) def zone_temp_raw(self, tm, z): inv = set(tm["sensor_invalid"]) vals = [tm["%s_%s" % (z, s)] for s in "AB" if ("%s.%s" % (z, s)) not in inv] return sorted(vals)[0] if vals else 0.0 def _trend(self, z): h = self.hist[-6:] if len(h) < 4: return 0.0 return (self.zone_temp(h[-1], z) - self.zone_temp(h[0], z)) / (len(h) - 1) def _suspect(self, z): """the sensor that jumped: compare each sensor now with the consensus before the disagreement began""" before = [h for h in self.hist[:-_persist()] if abs(h["%s_A" % z] - h["%s_B" % z]) <= 6.0] ref = (before[-1]["%s_A" % z] + before[-1]["%s_B" % z]) / 2.0 if before else None tm = self.hist[-1] if ref is None: return "A" return "A" if abs(tm["%s_A" % z] - ref) > abs(tm["%s_B" % z] - ref) else "B" HYPOTHESES = { "sensor_disagree": lambda f: [("sensor_bias", {"zone": f[1], "sensor": f[2]})], "heater_on_uncommanded": lambda f: [("heater_stuck_on", {"zone": f[1]})], "heater_off_uncommanded": lambda f: [("heater_stuck_off", {"zone": f[1]})], "wheel_overcurrent": lambda f: [("wheel_friction", {"wheel": f[1]})], "attitude_knowledge_lost": lambda f: [("star_tracker_dropout", {})], "solar_low": lambda f: [("solar_degradation", {}), ("battery_short", {})], "load_excess": lambda f: [("battery_short", {}), ("solar_degradation", {})], "rtg_low": lambda f: [("rtg_decline", {})], "thruster_overuse": lambda f: [("thruster_degradation", {"branch": f[1]})], } def candidates(hyp, tm): """admissible responses for a hypothesis: each operator's PRECONDITIONS are checked against echoed state""" kind, arg = hyp out = [("collect_evidence", {})] if kind == "sensor_bias": other = "B" if arg["sensor"] == "A" else "A" if ("%s.%s" % (arg["zone"], other)) not in tm["sensor_invalid"]: out.append(("mark_sensor_invalid", dict(arg))) if tm["%s_heater_circuit" % arg["zone"]] == "primary": out.append(("switch_heater_backup", {"zone": arg["zone"]})) if kind in ("heater_stuck_on", "heater_stuck_off") and tm["%s_heater_circuit" % arg["zone"]] == "primary": out.append(("switch_heater_backup", dict(arg))) if kind == "thruster_degradation" and tm["thruster_branch"] == arg.get("branch", "A"): out.append(("switch_thruster_branch", {})) if kind == "wheel_friction" and 3 not in tm["wheels_active"]: out.append(("switch_to_spare_wheel", dict(arg))) if kind == "star_tracker_dropout" and tm["attitude_sensor"] == "star_tracker": out.append(("switch_attitude_sensor", {})) if kind in ("solar_degradation", "battery_short", "wheel_friction", "rtg_decline") and tm["science_on"]: out.append(("shed_loads", {})) if not tm["safe_mode"]: out.append(("safe_mode", {})) return out def twin_from_telemetry(tm, facts, hyp, seed=99): """the onboard model: state ESTIMATED from telemetry, physics perturbed +-10%, the hypothesised fault injected""" sc = Spacecraft(seed=seed, profile=tm.get("profile", "europa_clipper")).clone(rng_seed=seed, perturb=TWIN_PERTURB) sc.t = tm["t"] sc.soc = tm["soc"] for z in ZONES: sc.temp[z] = facts.zone_temp(tm, z) sc.heater_circuit[z] = tm["%s_heater_circuit" % z] sc.heater_cmd[z] = tm["%s_heater_cmd" % z] sc.sensor_invalid = {tuple(s.split(".")) for s in tm["sensor_invalid"]} sc.wheels_active = list(tm["wheels_active"]) sc.attitude_sensor = tm["attitude_sensor"] sc.star_tracker_ok = tm["star_tracker_valid"] sc.science_on, sc.safe = tm["science_on"], tm["safe_mode"] sc.pointing_err = tm["pointing_err"] sc.thruster_branch = tm.get("thruster_branch", "A") if sc.actuator == "thrusters": # calibrate the active branch's authority from observed firing duty (duty = 0.05 / efficiency) sc.thruster_eff[sc.thruster_branch] = min(1.0, 0.05 / max(0.05, tm["thruster_duty"])) kind, arg = hyp if kind == "sensor_bias": # the good sensor carries the zone; the suspect one reads what it reads now z, s = arg["zone"], arg["sensor"] good = tm["%s_%s" % (z, "B" if s == "A" else "A")] sc.temp[z] = good sc.sensor_bias[(z, s)] = tm["%s_%s" % (z, s)] - good elif kind != "seu": sc.inject(kind, **{k: v for k, v in arg.items() if k in ("zone", "wheel", "sensor")}, magnitude=0.5) # CALIBRATE the twin to what is observed before predicting (measured 2026-09-24: an uncalibrated +-10% solar # parameter erased a real 0.5-2 A deficit). Mismatch then remains only in what telemetry cannot see (thermal). import math cos_now = math.cos(math.radians(min(tm["pointing_err"], 80.0))) if sc.power == "solar" else 1.0 sc.source_health = max(0.2, min(1.2, tm["i_source"] / max(0.1, sc.p["source_amp"] * cos_now))) explained = sc.p["heater_amp"] * sum(1 for z in ZONES if tm["%s_heater_cmd" % z]) \ + (sc.p["science_amp"] if tm["science_on"] else 0.0) + sum(tm["wheel%d_current" % w] for w in tm["wheels_active"]) \ + (0.6 * tm["thruster_duty"] if sc.actuator == "thrusters" else 0.0) if kind == "battery_short": sc.short_amp = max(0.0, tm["i_load"] - (sc.p["base_load"] + explained)) else: sc.p["base_load"] = max(3.0, tm["i_load"] - explained) sc.latent = hyp return sc def fault_unremedied(sc, hyp): """is the hypothesised fault still active at the end of the horizon? (latent hazard, priced even if calm now)""" kind, arg = hyp if kind == "sensor_bias": return (arg["zone"], arg["sensor"]) not in sc.sensor_invalid if kind in ("heater_stuck_on", "heater_stuck_off"): return sc.heater_circuit[arg["zone"]] == "primary" if kind == "wheel_friction": return arg["wheel"] in sc.wheels_active if kind == "star_tracker_dropout": return sc.attitude_sensor == "star_tracker" if kind == "thruster_degradation": return sc.thruster_branch == arg.get("branch", "A") if kind == "battery_short": # an unexplained multi-amp draw is a hazard beyond its energy cost until the faulted string is isolated # (safe mode is the isolation path in this model); only when the twin actually carries a short return sc.short_amp > 1.0 and not sc.safe return False def score_future(sc, action, arg): """simulate HORIZON ticks after applying the response; lower is better""" if action not in ("collect_evidence", "no_action"): sc.command(action, **arg) soc0 = sc.soc s = ACTION_COST[action] net = 0.0 for _ in range(HORIZON): sc.step() net += sc.i_source - sc.i_load for z in ZONES: lo, hi = BANDS[z] if not lo <= sc.temp[z] <= hi: s += 1.0 slo, shi = SURVIVAL[z] if not slo <= sc.temp[z] <= shi: s += 50.0 if sc.pointing_err > 1.5: s += 0.5 if sc.soc < 0.3: s += 5.0 s += max(0.0, soc0 - sc.soc) * 300.0 # a draining battery is a slow emergency # a sustained deficit (AVERAGE over the horizon; heater cycling makes one tick misleading) is a certain loss # later, however slow: price it deficit = -net / HORIZON if deficit > 0.1: s += 25.0 + 20.0 * deficit # a confirmed fault left un-remedied is a latent hazard even when the next 90 minutes look calm if getattr(sc, "latent", None) and fault_unremedied(sc, sc.latent): # hazard class: an unisolated electrical short is SURVIVAL-class (priced like a survival-temperature breach); # other un-remedied faults are degraded-redundancy class s += LATENT_WEIGHT.get(sc.latent[0], 10.0) return s class FDIR: def __init__(self, rulebook=None, lookahead=True): self.facts = Facts() self.rulebook = rulebook if rulebook is not None else {} self.lookahead = lookahead self.handled = set() self.waiting = {} # fact -> tick at which a decision to WAIT expires and the fact is re-decided self.receipts = [] self.twin_runs = 0 def decide(self, tm): t0 = time.perf_counter() facts = self.facts.update(tm) confirmed = [f for f in facts if f[0] not in ("transient_spike", "zone_out_of_band")] # a decision to wait EXPIRES (context changes: a repaired heater restores a load that a deficit was hiding) fresh = [f for f in confirmed if f not in self.handled and self.waiting.get(f, -1) <= tm["t"]] if not fresh: if any(f[0] == "transient_spike" for f in facts): return self._receipt(tm, facts, [], "collect_evidence", {}, "evidence", None, t0, note="single-tick anomaly: no physical action until it persists") return None sig = tuple(sorted(f[0] for f in fresh)) rule = self.rulebook.get(sig) if rule is not None: action, arg = rule["action"], _bind(rule["bind"], fresh) for f in fresh: self.handled.add(f) return self._receipt(tm, facts, [], action, arg, "rule", rule["id"], t0) hyps = [h for f in fresh for h in HYPOTHESES.get(f[0], lambda f: [])(f)] if not hyps: return None cand = {} for h in hyps: for a, arg in candidates(h, tm): cand.setdefault((a, tuple(sorted(arg.items()))), (a, arg)) scored = [] for key, (a, arg) in cand.items(): worst = max(score_future(twin_from_telemetry(tm, self.facts, h), a, arg) for h in hyps) self.twin_runs += len(hyps) scored.append((round(worst, 2), a, arg)) scored.sort(key=lambda x: (x[0], ACTION_COST[x[1]])) best = scored[0] for f in fresh: # a fact is handled only by an action that is one of ITS admissible responses (or by a decision to wait); # otherwise a second simultaneous fault would be silently dropped own = {(a, tuple(sorted(g.items()))) for h in HYPOTHESES.get(f[0], lambda f: [])(f) for a, g in candidates(h, tm)} if best[1] in ("collect_evidence", "no_action"): self.waiting[f] = tm["t"] + WAIT_EXPIRY elif (best[1], tuple(sorted(best[2].items()))) in own: self.handled.add(f) return self._receipt(tm, facts, hyps, best[1], best[2], "lookahead", None, t0, alternatives=[{"action": a, "arg": g, "worst_case": s} for s, a, g in scored[:5]]) def _receipt(self, tm, facts, hyps, action, arg, path, rule_id, t0, note=None, alternatives=None): r = {"id": len(self.receipts), "t": tm["t"], "facts": [list(f) for f in facts], "hypotheses": [[h[0], h[1]] for h in hyps], "action": action, "arg": arg, "path": path, "rule": rule_id, "alternatives": alternatives or [], "note": note, "ms": round((time.perf_counter() - t0) * 1000, 3)} self.receipts.append(r) return r def _bind(template, facts): """fill a rule's action arguments from the facts that matched it (zone / wheel / sensor)""" arg = {} for key, (fact_type, idx) in template.items(): for f in facts: if f[0] == fact_type and len(f) > idx: arg[key] = f[idx] return arg