deepspace/fdir.py

"""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