sensorflow/rotr/attribution.py

"""Per-violation causal-layer attribution.

THE invariant of this module: a downstream failure NEVER auto-implies an
upstream cause. Every causal layer gets an INDEPENDENT evidence test that
returns SUPPORTED, RULED_OUT, or UNKNOWN; a layer is implicated only by its
own positive evidence, never by another layer's silence. Missing evidence
is UNKNOWN — an honest triage outcome, not a defect.

The planning test is the keystone of perception/planning separation: the
plan is re-judged by the SAME rule engine, but against the stack's OWN
world view (mapped context, detections it actually made, intents it
actually predicted, in its believed localization frame). A compliant plan
rules planning out; a non-compliant plan given the stack's own information
implicates planning regardless of what perception did.
"""

from __future__ import annotations

from typing import Dict, List, Optional

from sensorflow.rotr import SOFTWARE_VERSION
from sensorflow.rotr.models import (
    CAUSAL_LAYERS, Actor, FailureAttribution, LayerEvidence, Provenance,
    ROTRScenario, ROTRViolation,
)
from sensorflow.rotr.rules import plan_violations

ATTRIBUTION_VERSION = "rotr-attr-1.0.0"

# ILLUSTRATIVE evidence thresholds (synthetic substrate).
DETECTION_RATE_RULED_OUT = 0.8
DETECTION_RATE_SUPPORTED = 0.5
POSITION_ERROR_SUPPORTED_M = 1.0
LOCALIZATION_ERROR_SUPPORTED_M = 1.0
CONTROL_POS_DEV_SUPPORTED_M = 1.5
CONTROL_SPEED_DEV_SUPPORTED_MPS = 2.0


def _material_actors(scenario: ROTRScenario,
                     violation: ROTRViolation) -> List[Actor]:
    ids = set(violation.actor_ids)
    return [a for a in scenario.actors if a.actor_id in ids]


# ------------------------------------------------------------ layer tests


def _test_perception(scenario: ROTRScenario,
                     violation: ROTRViolation) -> LayerEvidence:
    actors = _material_actors(scenario, violation)
    if not actors:
        return LayerEvidence(
            layer="perception", status="RULED_OUT", confidence=0.7,
            evidence="rule trigger is ego/map-geometric; no actor whose "
                     "perception could have caused it")
    # Judge detection quality in the window that mattered: from the actor
    # becoming relevant (within 60 m) up to the violation instant.
    worst: Optional[LayerEvidence] = None
    for actor in actors:
        n_frames = n_detected = n_absent = 0
        pos_err_sum = 0.0
        for i, s in enumerate(actor.states):
            e = scenario.ego[min(i, len(scenario.ego) - 1)]
            if s.t > violation.t_end + 0.5:
                break
            if abs(s.x - e.x) > 60 or abs(s.y - e.y) > 30:
                continue
            n_frames += 1
            if s.observation is None:
                n_absent += 1
            elif s.observation.detected:
                n_detected += 1
                if s.observation.x is not None:
                    pos_err_sum += abs(s.observation.x - s.x) + \
                        abs((s.observation.y or s.y) - s.y)
        if n_frames == 0:
            continue
        if n_absent / n_frames > 0.5:
            cand = LayerEvidence(
                layer="perception", status="UNKNOWN", confidence=0.0,
                evidence=f"{actor.actor_id}: perception telemetry missing for "
                         f"{n_absent}/{n_frames} relevant frames")
        else:
            assessable = n_frames - n_absent
            rate = n_detected / assessable if assessable else 0.0
            mean_err = pos_err_sum / n_detected if n_detected else 0.0
            if rate < DETECTION_RATE_SUPPORTED:
                cand = LayerEvidence(
                    layer="perception", status="SUPPORTED", confidence=0.9,
                    evidence=f"{actor.actor_id}: missed detection — GT visible "
                             f"{assessable} frames, detected {n_detected} "
                             f"(rate {rate:.2f})")
            elif mean_err > POSITION_ERROR_SUPPORTED_M:
                cand = LayerEvidence(
                    layer="perception", status="SUPPORTED", confidence=0.75,
                    evidence=f"{actor.actor_id}: detected but mislocalized "
                             f"(mean |err| {mean_err:.2f} m)")
            elif rate >= DETECTION_RATE_RULED_OUT:
                cand = LayerEvidence(
                    layer="perception", status="RULED_OUT", confidence=0.9,
                    evidence=f"{actor.actor_id}: detected {n_detected}/"
                             f"{assessable} relevant frames (rate {rate:.2f}, "
                             f"mean |err| {mean_err:.2f} m) — GT-vs-detected "
                             f"diff is clean")
            else:
                cand = LayerEvidence(
                    layer="perception", status="SUPPORTED", confidence=0.6,
                    evidence=f"{actor.actor_id}: degraded detection "
                             f"(rate {rate:.2f})")
        worst = _worse(worst, cand)
    return worst or LayerEvidence(layer="perception", status="UNKNOWN",
                                  evidence="no relevant frames to assess")


def _worse(a: Optional[LayerEvidence], b: LayerEvidence) -> LayerEvidence:
    """Prefer SUPPORTED > UNKNOWN > RULED_OUT when merging actor evidence."""
    if a is None:
        return b
    order = {"SUPPORTED": 2, "UNKNOWN": 1, "RULED_OUT": 0}
    if order[b.status] > order[a.status]:
        return b
    if order[b.status] == order[a.status] and b.confidence > a.confidence:
        return b
    return a


def _test_prediction(scenario: ROTRScenario,
                     violation: ROTRViolation) -> LayerEvidence:
    actors = _material_actors(scenario, violation)
    if not actors:
        return LayerEvidence(layer="prediction", status="RULED_OUT",
                             confidence=0.7,
                             evidence="no actor whose intent needed predicting")
    worst: Optional[LayerEvidence] = None
    for actor in actors:
        if actor.predicted_intent is None:
            cand = LayerEvidence(layer="prediction", status="UNKNOWN",
                                 evidence=f"{actor.actor_id}: no recorded "
                                          "intent prediction")
        elif actor.predicted_intent != actor.intent:
            cand = LayerEvidence(
                layer="prediction", status="SUPPORTED", confidence=0.85,
                evidence=f"{actor.actor_id}: GT intent {actor.intent} but "
                         f"stack predicted {actor.predicted_intent}")
        else:
            cand = LayerEvidence(
                layer="prediction", status="RULED_OUT", confidence=0.85,
                evidence=f"{actor.actor_id}: predicted intent matches GT "
                         f"({actor.intent})")
        worst = _worse(worst, cand)
    return worst


def _test_planning(scenario: ROTRScenario,
                   violation: ROTRViolation) -> LayerEvidence:
    if not scenario.planned:
        return LayerEvidence(layer="planning", status="UNKNOWN",
                             evidence="no planned trajectory recorded")
    hits = plan_violations(scenario)
    same_rule = [h for h in hits if h["rule_id"] == violation.rule_id]
    if same_rule:
        return LayerEvidence(
            layer="planning", status="SUPPORTED", confidence=0.85,
            evidence=f"plan violates {violation.rule_id} given the stack's "
                     f"OWN world view (detections + predicted intents + "
                     f"believed frame): {same_rule[0]['evidence']}")
    if hits:
        return LayerEvidence(
            layer="planning", status="SUPPORTED", confidence=0.6,
            evidence="plan violates a different rule given the stack's own "
                     f"view: {hits[0]['rule_id']}")
    return LayerEvidence(
        layer="planning", status="RULED_OUT", confidence=0.85,
        evidence="plan is rule-compliant given the stack's own world view — "
                 "the plan was reasonable for what the stack knew")


def _test_localization(scenario: ROTRScenario,
                       violation: ROTRViolation) -> LayerEvidence:
    errs = []
    for e in scenario.ego:
        if e.believed_x is None or e.believed_y is None:
            return LayerEvidence(layer="localization", status="UNKNOWN",
                                 evidence="no believed-pose telemetry")
        errs.append(abs(e.believed_x - e.x) + abs(e.believed_y - e.y))
    max_err = max(errs) if errs else 0.0
    lane_mismatch = any(e.believed_lane_id != e.lane_id for e in scenario.ego
                        if e.lane_id and e.believed_lane_id)
    if max_err > LOCALIZATION_ERROR_SUPPORTED_M or lane_mismatch:
        return LayerEvidence(
            layer="localization", status="SUPPORTED", confidence=0.9,
            evidence=f"believed-vs-true pose error up to {max_err:.2f} m"
                     + ("; believed lane differs from true lane"
                        if lane_mismatch else ""))
    return LayerEvidence(layer="localization", status="RULED_OUT",
                         confidence=0.9,
                         evidence=f"pose error <= {max_err:.2f} m; lane "
                                  "association consistent")


def _test_map(scenario: ROTRScenario,
              violation: ROTRViolation) -> LayerEvidence:
    m, a = scenario.map_context, scenario.actual_context
    diffs = []
    if m.control != a.control:
        diffs.append(f"control mapped={m.control!r} actual={a.control!r}")
    if m.signal_state_for_ego != a.signal_state_for_ego:
        diffs.append(f"signal mapped={m.signal_state_for_ego!r} "
                     f"actual={a.signal_state_for_ego!r}")
    a_lanes = {l.lane_id: l for l in a.lanes}
    for lane in m.lanes:
        al = a_lanes.get(lane.lane_id)
        if al is None:
            diffs.append(f"lane {lane.lane_id} not in actual geometry")
            continue
        if lane.restricted_to != al.restricted_to:
            diffs.append(f"lane {lane.lane_id} restriction mapped="
                         f"{lane.restricted_to!r} actual={al.restricted_to!r}")
        if set(lane.permitted_maneuvers) != set(al.permitted_maneuvers):
            diffs.append(f"lane {lane.lane_id} permitted maneuvers differ")
    if diffs:
        return LayerEvidence(layer="map", status="SUPPORTED", confidence=0.9,
                             evidence="map-vs-actual mismatch: " + "; ".join(diffs))
    return LayerEvidence(layer="map", status="RULED_OUT", confidence=0.9,
                         evidence="mapped context matches as-built geometry")


def _test_control(scenario: ROTRScenario,
                  violation: ROTRViolation) -> LayerEvidence:
    if not scenario.planned:
        return LayerEvidence(layer="control", status="UNKNOWN",
                             evidence="no planned trajectory to compare "
                                      "execution against")
    n = min(len(scenario.planned), len(scenario.ego))
    max_pos = max_spd = 0.0
    for k in range(n):
        p, e = scenario.planned[k], scenario.ego[k]
        max_pos = max(max_pos, abs(p.x - e.x) + abs(p.y - e.y))
        max_spd = max(max_spd, abs(p.v - e.speed))
    if max_pos > CONTROL_POS_DEV_SUPPORTED_M or \
            max_spd > CONTROL_SPEED_DEV_SUPPORTED_MPS:
        return LayerEvidence(
            layer="control", status="SUPPORTED", confidence=0.85,
            evidence=f"executed trajectory deviates from plan: max position "
                     f"dev {max_pos:.2f} m, max speed dev {max_spd:.2f} m/s")
    return LayerEvidence(
        layer="control", status="RULED_OUT", confidence=0.85,
        evidence=f"execution tracks the plan (max pos dev {max_pos:.2f} m, "
                 f"max speed dev {max_spd:.2f} m/s)")


def _test_data_label(scenario: ROTRScenario,
                     violation: ROTRViolation) -> LayerEvidence:
    problems = []
    ts = [e.t for e in scenario.ego]
    if any(t2 <= t1 for t1, t2 in zip(ts, ts[1:])):
        problems.append("ego timestamps not strictly increasing")
    for a in scenario.actors:
        if len(a.states) != len(scenario.ego):
            problems.append(f"{a.actor_id}: state array length mismatch")
    if problems:
        return LayerEvidence(layer="data_label", status="SUPPORTED",
                             confidence=0.95,
                             evidence="record integrity failed: "
                                      + "; ".join(problems))
    return LayerEvidence(layer="data_label", status="RULED_OUT",
                         confidence=0.9,
                         evidence="record integrity checks pass "
                                  "(monotone timestamps, consistent arrays)")


def _test_policy_rule(others: Dict[str, LayerEvidence]) -> LayerEvidence:
    statuses = {le.status for le in others.values()}
    if statuses == {"RULED_OUT"}:
        return LayerEvidence(
            layer="policy_rule", status="SUPPORTED", confidence=0.4,
            evidence="every stack layer independently RULED_OUT — the rule "
                     "itself needs HITL review (possible rule misfire)")
    if "SUPPORTED" in statuses:
        return LayerEvidence(layer="policy_rule", status="RULED_OUT",
                             confidence=0.7,
                             evidence="a stack layer has positive evidence; "
                                      "no indication the rule misfired")
    return LayerEvidence(layer="policy_rule", status="UNKNOWN",
                         evidence="stack evidence incomplete; rule quality "
                                  "cannot be judged")


# ------------------------------------------------------------ entry point


def attribute(scenario: ROTRScenario,
              violation: ROTRViolation) -> FailureAttribution:
    layers: Dict[str, LayerEvidence] = {
        "perception": _test_perception(scenario, violation),
        "prediction": _test_prediction(scenario, violation),
        "planning": _test_planning(scenario, violation),
        "localization": _test_localization(scenario, violation),
        "map": _test_map(scenario, violation),
        "control": _test_control(scenario, violation),
        "data_label": _test_data_label(scenario, violation),
    }
    layers["policy_rule"] = _test_policy_rule(layers)

    supported = [(name, le) for name, le in layers.items()
                 if le.status == "SUPPORTED"]
    primary = None
    if supported:
        supported.sort(key=lambda kv: (-kv[1].confidence,
                                       CAUSAL_LAYERS.index(kv[0])))
        primary = supported[0][0]
    note = ("no layer has positive evidence — triaged to HITL, NOT "
            "defaulted to perception" if primary is None else
            f"primary={primary} by highest-confidence positive evidence; "
            f"{len(supported)} layer(s) SUPPORTED")

    return FailureAttribution(
        violation_id=violation.violation_id,
        scenario_id=scenario.scenario_id,
        layers=layers, primary_layer=primary, note=note,
        provenance=Provenance(
            scenario_id=scenario.scenario_id,
            dataset_version=scenario.provenance.dataset_version,
            model_version=scenario.provenance.model_version,
            software_version=f"{SOFTWARE_VERSION}/{ATTRIBUTION_VERSION}",
            source=scenario.provenance.source))