Spaces:
Paused
Paused
File size: 6,744 Bytes
6b3874e | 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 | """Deterministic scripted synthesis (mock backend path).
Rule-driven hypothesis generation assembled from the ACTUAL evidence in the
artifact — the same pattern as the platform's offline MITL copilot
(sensorflow/evaluation/copilot.py). Never invents telemetry: hypotheses cite
supporting evidence keys and explicitly list the evidence that is missing.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from sensorflow.retro.scorecard import RootCauseHypothesis, Severity
def scripted_hypotheses(failure_type: str, log: Dict[str, Any],
missing_fields: List[str]) -> List[RootCauseHypothesis]:
gt = log.get("ground_truth") or {}
pred = log.get("prediction") or {}
scenario = log.get("scenario") or {}
sensors = log.get("sensor_metadata") or {}
tracks = log.get("tracks") or []
hyps: List[RootCauseHypothesis] = []
gt_class = (gt.get("class") or "").lower()
weather = (scenario.get("weather") or "").lower()
flicker = any(t.get("class_flicker") for t in tracks)
if failure_type == "FALSE_POSITIVE" and gt_class in (
"plastic_bag", "debris", "cardboard", "balloon"):
hyps.append(RootCauseHypothesis(
hypothesis="Training-distribution gap for wind-blown deformable "
"low-mass objects: the detector assigned a hazardous "
"class with high confidence to a non-hazard object, "
"consistent with the historical phantom-braking cluster "
"(RETRO-2025-014 [SYNTHETIC]).",
confidence=0.65,
supporting_evidence_keys=["gt_class", "pred_class", "pred_confidence"]
+ (["track_class_flicker"] if flicker else []),
missing_evidence=["camera embeddings for the misclassified frames",
"training-set frequency of deformable clutter"]))
if "velocity_note" in pred or any(
t.get("velocity_variance") == "high" for t in tracks):
hyps.append(RootCauseHypothesis(
hypothesis="Tracker velocity latch: bag flutter was integrated "
"into a spurious closing velocity, converting a "
"static-clutter FP into a phantom emergency brake.",
confidence=0.55,
supporting_evidence_keys=["pred_relative_velocity",
"gt_relative_velocity",
"track_velocity_variance"],
missing_evidence=["raw radar doppler track for the object"]))
elif failure_type == "FALSE_NEGATIVE" and gt_class in (
"pedestrian", "cyclist", "motorcycle", "wheelchair"):
if weather in ("rain", "snow", "fog"):
cam = (sensors.get("camera") or {}).get("status", "")
lid = (sensors.get("lidar") or {}).get("status", "")
hyps.append(RootCauseHypothesis(
hypothesis=f"Fusion confidence starvation under {weather}: "
f"camera '{cam}' and LiDAR '{lid}' each degraded, "
"and the fusion confidence stayed below the tracking "
"threshold instead of applying weather-conditioned "
"weighting (pattern matches RETRO-2025-021 "
"[SYNTHETIC]).",
confidence=0.6,
supporting_evidence_keys=["sensor_camera_status",
"sensor_lidar_status",
"pred_confidence",
"pred_detection_delay"],
missing_evidence=["per-frame fusion confidence trace",
"confidence threshold config for this release"]))
hyps.append(RootCauseHypothesis(
hypothesis="Training-data coverage gap for this VRU appearance/"
"condition slice; the distribution-shift slice data "
"shows the regression concentrated rather than uniform.",
confidence=0.5,
supporting_evidence_keys=["distribution_shift", "metric_delta"],
missing_evidence=["slice-level training data counts"]))
elif failure_type == "FALSE_POSITIVE":
hyps.append(RootCauseHypothesis(
hypothesis="Static-scene classification ambiguity: a fixed roadside "
"asset was transiently classified as a dynamic hazard; "
"map-prior fusion did not veto the detection.",
confidence=0.5,
supporting_evidence_keys=["gt_class", "pred_class",
"pred_confidence"],
missing_evidence=["map-prior veto decision log"]))
elif failure_type == "FALSE_NEGATIVE":
hyps.append(RootCauseHypothesis(
hypothesis="Detection miss on a low-salience static object; range "
"sparsity is the most consistent factor given nominal "
"sensor status.",
confidence=0.45,
supporting_evidence_keys=["gt_distance", "sensor_lidar_status"],
missing_evidence=["point count inside the ground-truth box"]))
if not hyps:
hyps.append(RootCauseHypothesis(
hypothesis="Insufficient patterns matched for a specific causal "
"hypothesis; requires human investigation.",
confidence=0.3, supporting_evidence_keys=[],
missing_evidence=["additional telemetry"]))
if missing_fields:
for h in hyps:
for f in missing_fields:
marker = f"artifact field {f}"
if marker not in h.missing_evidence:
h.missing_evidence.append(marker)
return hyps
def scripted_severity_proposal(log: Dict[str, Any],
failure_type: str) -> Optional[Severity]:
"""The mock 'LLM proposal' — a deliberately simpler heuristic than the
policy engine, so the policy-validate/override flow is real."""
collision = (log.get("collision") or {})
planner = (log.get("planner_response") or {})
traffic = ((log.get("traffic_context") or {}).get("following_vehicle") or {})
if collision.get("occurred"):
return Severity.FATAL
if collision.get("near_miss"):
return Severity.CRITICAL
decel = planner.get("decel_mps2") or 0.0
gap = traffic.get("time_gap_s")
if decel >= 3.4 and gap is not None and gap < 1.5:
return Severity.CRITICAL
if decel >= 3.0:
return Severity.DISRUPTIVE
return Severity.BENIGN
|