sensorflow/safety/discrepancy.py

"""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