"""Auto-labeling discrepancy mining (continuous-learning loop). Industry concept: production AV stacks run a weaker real-time ("online") perception pass in the vehicle and a stronger non-causal ("offline") auto-label pass over the same logs. Diffing the two is one of the highest-yield scenario mining signals — anything the online stack missed but the offline stack found is a candidate training example / regression scenario. This module simulates the online pass deterministically over an existing labeleval dataset (the stored auto-labels play the role of the strong offline pass), diffs the two, and converts discrepancies into queryable scenario-mining records: - MISSED_ONLINE: offline labeled the object, online did not detect it - CLASS_MISMATCH: both detected it, classes differ - GEOMETRY_DELTA: both detected it, geometry disagrees (center offset / IoU) - OFFLINE_MISS: online detected an object the offline pass dropped (rare) High-severity discrepancies feed the existing rare-event store; all of them feed the scenario database (source="discrepancy") and cohort summaries. SIMULATED: the online pass is a seeded degradation model over ground truth (higher miss rates, class confusion, geometry noise), not a real detector. """ from __future__ import annotations import math from datetime import datetime, timezone from typing import Dict, List, Optional import numpy as np from sensorflow.evaluation.records import ( SAFETY_CRITICAL_CLASSES, EvalStore, RareEvent, new_id, ) from sensorflow.metrics.perception_3d import bev_iou from sensorflow.safety.store import read_json, write_json DEFAULT_ONLINE_PROFILE = { "extra_miss_rate": 0.12, # online misses this fraction on top of offline "miss_rate_by_class": {"pedestrian": 0.08, "cyclist": 0.08}, # additional VRU misses "class_confusion_rate": 0.05, "position_noise_m": 0.55, "yaw_noise_deg": 6.0, "night_miss_multiplier": 1.6, # online degrades harder at night/rain "rain_miss_multiplier": 1.4, } GEOMETRY_CENTER_THRESHOLD_M = 0.75 GEOMETRY_IOU_THRESHOLD = 0.5 _CONFUSABLE = { "vehicle": "truck", "truck": "vehicle", "pedestrian": "cyclist", "cyclist": "pedestrian", "motorcycle": "cyclist", } def _now() -> str: return datetime.now(timezone.utc).isoformat() def simulate_online_pass(store: EvalStore, dataset_id: str, profile: Optional[Dict] = None, seed: int = 101) -> Dict[str, Dict]: """Deterministic weaker 'online' perception pass over the dataset's GT. Returns {gt_id: detection dict} for every ground-truth object the online stack detected. SIMULATED degradation model, clearly marked. """ p = {**DEFAULT_ONLINE_PROFILE, **(profile or {})} rng = np.random.default_rng(seed) detections: Dict[str, Dict] = {} frames = sorted(store.where("frames", dataset_id=dataset_id), key=lambda f: f.index) for frame in frames: cond_mult = 1.0 if frame.time_of_day == "night": cond_mult *= p["night_miss_multiplier"] if frame.weather not in ("clear", ""): cond_mult *= p["rain_miss_multiplier"] for gt in frame.gt_boxes: miss = p["extra_miss_rate"] + p["miss_rate_by_class"].get(gt.class_name, 0.0) miss = min(1.0, miss * cond_mult) if miss < 1.0 else 1.0 if rng.random() < miss: continue # online miss x, y, z, l, w, h, yaw = gt.bbox_3d x += float(rng.normal(0, p["position_noise_m"])) y += float(rng.normal(0, p["position_noise_m"])) yaw += float(rng.normal(0, math.radians(p["yaw_noise_deg"]))) cls = gt.class_name if rng.random() < p["class_confusion_rate"]: cls = _CONFUSABLE.get(cls, cls) detections[gt.gt_id] = { "gt_id": gt.gt_id, "frame_id": frame.frame_id, "class_name": cls, "bbox_3d": [round(v, 4) for v in (x, y, z, l, w, h, yaw)], "confidence": round(float(rng.uniform(0.35, 0.85)), 3), } return detections def mine(store: EvalStore, dataset_id: str, profile: Optional[Dict] = None, seed: int = 101, persist: bool = True) -> Dict: """Diff the simulated online pass against the stored offline auto-labels.""" dataset = store.get("datasets", dataset_id) if dataset is None: raise KeyError(f"Unknown dataset {dataset_id}") online = simulate_online_pass(store, dataset_id, profile, seed) frames = {f.frame_id: f for f in store.where("frames", dataset_id=dataset_id)} offline_by_gt = {a.matched_gt_id: a for a in store.where("annotations", dataset_id=dataset_id) if a.matched_gt_id} discrepancies: List[Dict] = [] cohort_totals: Dict[str, int] = {} cohort_hits: Dict[str, int] = {} for frame in frames.values(): for gt in frame.gt_boxes: cohort = f"{gt.class_name}/{frame.weather}/{frame.time_of_day}" cohort_totals[cohort] = cohort_totals.get(cohort, 0) + 1 off = offline_by_gt.get(gt.gt_id) on = online.get(gt.gt_id) dtype = details = None if off is not None and on is None: dtype = "MISSED_ONLINE" details = {"offline_confidence": off.confidence} elif off is not None and on is not None: if on["class_name"] != off.class_name: dtype = "CLASS_MISMATCH" details = {"offline_class": off.class_name, "online_class": on["class_name"]} elif off.bbox_3d: center_d = math.hypot(on["bbox_3d"][0] - off.bbox_3d[0], on["bbox_3d"][1] - off.bbox_3d[1]) iou = bev_iou(on["bbox_3d"], off.bbox_3d) if center_d > GEOMETRY_CENTER_THRESHOLD_M or iou < GEOMETRY_IOU_THRESHOLD: dtype = "GEOMETRY_DELTA" details = {"center_delta_m": round(center_d, 3), "bev_iou": round(iou, 4)} elif off is None and on is not None: dtype = "OFFLINE_MISS" details = {"online_class": on["class_name"]} if dtype is None: continue cohort_hits[cohort] = cohort_hits.get(cohort, 0) + 1 safety = gt.class_name in SAFETY_CRITICAL_CLASSES severity = ("critical" if dtype == "MISSED_ONLINE" and safety else "high" if dtype == "MISSED_ONLINE" or safety else "medium") discrepancies.append({ "discrepancy_id": new_id("disc"), "dataset_id": dataset_id, "frame_id": frame.frame_id, "gt_id": gt.gt_id, "type": dtype, "class_name": gt.class_name, "weather": frame.weather, "time_of_day": frame.time_of_day, "safety_critical": safety, "severity": severity, "details": details, }) by_type: Dict[str, int] = {} for d in discrepancies: by_type[d["type"]] = by_type.get(d["type"], 0) + 1 cohorts = [] for cohort, total in sorted(cohort_totals.items()): hits = cohort_hits.get(cohort, 0) cls, weather, tod = cohort.split("/") cohorts.append({"cohort": cohort, "class": cls, "weather": weather, "time_of_day": tod, "objects": total, "discrepancies": hits, "discrepancy_rate": round(hits / total, 4)}) cohorts.sort(key=lambda c: -c["discrepancy_rate"]) total_objects = sum(cohort_totals.values()) report = { "dataset_id": dataset_id, "seed": seed, "created_at": _now(), "online_profile": {**DEFAULT_ONLINE_PROFILE, **(profile or {})}, "totals": { "objects": total_objects, "online_detections": len(online), "offline_annotations": len(offline_by_gt), "discrepancies": len(discrepancies), "discrepancy_rate": round(len(discrepancies) / max(total_objects, 1), 4), }, "by_type": by_type, "cohorts": cohorts, "discrepancies": discrepancies, "simulated": True, "method": "deterministic simulated online pass vs stored offline auto-labels; " "diff over shared ground-truth ids", } if persist: write_json(report, "discrepancy", f"{dataset_id}.json") _feed_stores(store, report) return report def _feed_stores(store: EvalStore, report: Dict) -> None: """Feed high-severity discrepancies into the rare-event store, all of them into the scenario database, and record process units (best-effort).""" critical = [d for d in report["discrepancies"] if d["severity"] == "critical"] for d in critical[:40]: store.put("rare_events", RareEvent( event_id=new_id("evt"), dataset_id=d["dataset_id"], scenario_type="online_perception_miss", severity="critical", rarity_score=0.85, anomaly_score=0.0, confidence=0.9, evidence_frames=[d["frame_id"]], sensor_evidence={"discrepancy": f"{d['type']} on {d['class_name']} " f"({d['weather']}/{d['time_of_day']})"}, description=f"Online stack missed a {d['class_name']} the offline " f"auto-label pass found (simulated online pass)", )) try: from sensorflow.evaluation.process_units import ProcessMeter ProcessMeter(store, run_id=report["dataset_id"]).record( "discrepancy_mining", report["totals"]["objects"], factor=0.3) except Exception: pass store.save() try: from sensorflow.safety import scenario_db scenario_db.get_db().add_from_discrepancies(report) except Exception: pass def latest_summary(dataset_id: str) -> Optional[Dict]: """Persisted mining report without the raw per-object rows.""" report = read_json("discrepancy", f"{dataset_id}.json") if report is None: return None slim = {k: v for k, v in report.items() if k != "discrepancies"} slim["sample_discrepancies"] = report["discrepancies"][:25] return slim