Spaces:
Paused
Paused
| """SensorFusionVerificationAgent — cross-modality coherence check. | |
| Re-simulates the bevfusion camera and LiDAR sensor models on the failure | |
| frames (deterministic seeded reuse of sensorflow.bevfusion.sensors) and | |
| cross-checks camera / LiDAR / tracking / ground-truth coherence for each | |
| failure instance. | |
| Verdicts: | |
| multi_modal_supported both modalities independently corroborate the GT | |
| object (the misclassification is well-evidenced) | |
| single_modality_only only one modality carries evidence — weaker claim | |
| modality_conflict camera and LiDAR genuinely disagree -> MANDATORY | |
| human review (enforced via escalation triggers) | |
| verification_failed the check itself could not run -> INDETERMINATE-ward | |
| """ | |
| from __future__ import annotations | |
| import math | |
| from typing import Any, Dict, List, Tuple | |
| from sensorflow.agentic import data as data_mod | |
| from sensorflow.agentic.agents.base import BaseAgent, compact, escalate, no_escalation | |
| from sensorflow.agentic.models import AgentEscalation, FailureEvent | |
| from sensorflow.bevfusion import scenes as scenes_mod | |
| from sensorflow.bevfusion import sensors as sensors_mod | |
| POSITION_AGREEMENT_M = 2.0 | |
| CONE_DIMS = (0.4, 0.4, 0.75) # reference template for the cone hypothesis | |
| VRU = {"pedestrian", "cyclist", "motorcycle"} | |
| def _dims_log_error(dims: List[float], template: Tuple[float, float, float]) -> float: | |
| return sum(math.log(max(d, 0.05) / t) ** 2 for d, t in zip(dims, template)) | |
| class SensorFusionVerificationAgent(BaseAgent): | |
| name = "sensor_fusion_verification" | |
| version = "1.0.0" | |
| epistemic_status = "DERIVED" | |
| def _analyze(self, failure_id: str, failure: FailureEvent = None, | |
| seed: int = data_mod.DEFAULT_SEED, | |
| **inputs) -> Tuple[Dict[str, Any], float, str, AgentEscalation]: | |
| assert failure is not None | |
| sequences = {s.sequence_id: (qi, s) for qi, s in | |
| enumerate(scenes_mod.generate_sequences( | |
| n_sequences=data_mod.DEFAULT_N_SEQUENCES, | |
| frames_per_sequence=data_mod.DEFAULT_FRAMES, | |
| seed=seed))} | |
| per_instance: List[Dict] = [] | |
| for inst in failure.instances[:12]: # bounded, deterministic subset | |
| entry = sequences.get(inst.sequence_id) | |
| if entry is None: | |
| per_instance.append({"instance_id": inst.instance_id, | |
| "verdict": "verification_failed", | |
| "reason": "sequence not reproducible"}) | |
| continue | |
| qi, seq = entry | |
| frame = next((f for f in seq.frames if f.frame_id == inst.frame_id), None) | |
| if frame is None: | |
| per_instance.append({"instance_id": inst.instance_id, | |
| "verdict": "verification_failed", | |
| "reason": "frame not reproducible"}) | |
| continue | |
| cam = sensors_mod.simulate_camera(frame, seq, | |
| sensors_mod.camera_rng(seed, qi)) | |
| lid = sensors_mod.simulate_lidar(frame, seq, | |
| sensors_mod.lidar_rng(seed, qi)) | |
| cam_det = next((d for d in cam | |
| if d.source_instance_id == inst.object_instance_id), None) | |
| lid_det = next((d for d in lid | |
| if d.source_instance_id == inst.object_instance_id), None) | |
| gt_dims = scenes_mod.CLASS_DIMS.get(inst.gt_class, (0.8, 0.8, 1.8)) | |
| checks: Dict[str, Any] = {} | |
| camera_supports = lidar_supports = False | |
| conflict = False | |
| if lid_det is not None: | |
| err_gt = _dims_log_error(lid_det.dims, gt_dims) | |
| err_cone = _dims_log_error(lid_det.dims, CONE_DIMS) | |
| lidar_supports = err_gt < err_cone | |
| checks["lidar_shape"] = { | |
| "dims": [round(d, 2) for d in lid_det.dims], | |
| "log_err_vs_gt_template": round(err_gt, 3), | |
| "log_err_vs_cone_template": round(err_cone, 3), | |
| "supports_gt_class": lidar_supports, | |
| } | |
| if cam_det is not None: | |
| camera_supports = cam_det.class_name in VRU | |
| checks["camera_class"] = { | |
| "class": cam_det.class_name, | |
| "confidence": round(cam_det.confidence, 3), | |
| "supports_gt_class": camera_supports, | |
| } | |
| if cam_det is not None and lid_det is not None: | |
| dist = math.hypot(cam_det.x - lid_det.x, cam_det.y - lid_det.y) | |
| checks["position_agreement_m"] = round(dist, 2) | |
| if dist > POSITION_AGREEMENT_M or (camera_supports != lidar_supports): | |
| conflict = True | |
| if conflict: | |
| verdict = "modality_conflict" | |
| elif cam_det is not None and lid_det is not None: | |
| verdict = ("multi_modal_supported" | |
| if camera_supports and lidar_supports | |
| else "modality_conflict") | |
| elif cam_det is not None or lid_det is not None: | |
| verdict = "single_modality_only" | |
| else: | |
| verdict = "single_modality_only" | |
| checks["note"] = ("neither re-simulated modality re-detected " | |
| "the object this frame; tracking continuity " | |
| "is the only supporting evidence") | |
| per_instance.append({"instance_id": inst.instance_id, | |
| "frame_id": inst.frame_id, | |
| "verdict": verdict, "checks": checks}) | |
| counts: Dict[str, int] = {} | |
| for r in per_instance: | |
| counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1 | |
| n = max(len(per_instance), 1) | |
| if counts.get("verification_failed", 0) == n: | |
| overall = "verification_failed" | |
| elif counts.get("modality_conflict", 0) > 0: | |
| overall = "modality_conflict" | |
| elif counts.get("multi_modal_supported", 0) / n >= 0.5: | |
| overall = "multi_modal_supported" | |
| else: | |
| overall = "single_modality_only" | |
| output = { | |
| "overall_verdict": overall, | |
| "verdict_counts": counts, | |
| "instances_verified": len(per_instance), | |
| "per_instance": per_instance, | |
| "method": ("deterministic re-simulation of bevfusion camera/LiDAR " | |
| "sensor models; LiDAR shape template check vs GT class " | |
| "and cone hypothesis; camera class check; BEV position " | |
| "agreement"), | |
| } | |
| confidence = round(min(0.9, 0.4 + 0.5 * counts.get("multi_modal_supported", 0) / n), 3) | |
| if overall == "modality_conflict": | |
| return (output, confidence, | |
| "modality conflict detected — verdict requires human eyes", | |
| escalate(["camera and LiDAR evidence disagree"], | |
| ["modality_disagreement"])) | |
| if overall == "verification_failed": | |
| return (output, 0.0, "verification could not run", | |
| escalate(["fusion verification failed"], ["agent_failure"])) | |
| return (output, confidence, | |
| "share of instances with independent two-modality corroboration", | |
| no_escalation()) | |
| def _llm_prompt(self, failure_id, output, **inputs): | |
| return ("Explain this multi-sensor verification result to a triage " | |
| "engineer in three sentences: " + compact(output)) | |