Spaces:
Paused
Paused
Download sensorflow/retro/agent/scripted.py from whassan/SensorflowStudio: direct link, hf CLI and curl.
- Browser
- Download file 6.74 kB
-
https://huggingface.co/spaces/whassan/SensorflowStudio/resolve/main/sensorflow/retro/agent/scripted.py
- Command line
-
hf download hf://spaces/whassan/SensorflowStudio/sensorflow/retro/agent/scripted.py
-
curl -L -o scripted.py https://huggingface.co/spaces/whassan/SensorflowStudio/resolve/main/sensorflow/retro/agent/scripted.py
6.74 kB
| """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 | |