Spaces:
Paused
Paused
| """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)) | |