sensorflow/studio2/funnel.py

"""Unified observability funnel over the real stores.

One view of the whole system: the evaluation funnel (raw → selected →
simulated → evaluated → failed → HITL → regression suite), the safety panel
(SCR, VRU miss rate, TTC distribution), model comparison, drift, and compute.

Everything is aggregated best-effort per source package with an availability
flag. Absent sources report available=False with the reason — numbers are
NEVER fabricated. Reads go through the packages' persisted stores (plain
JSON/filesystem) so an unimportable in-flight package does not break the
funnel for the landed ones.
"""

from __future__ import annotations

import json
import os
from typing import Dict, List, Optional

from sensorflow.studio2 import store
from sensorflow.studio2.registry import Registry, get_registry


def _panel(available: bool, source: str, data: Optional[Dict] = None,
           reason: str = "") -> Dict:
    out = {"available": available, "source": source}
    if available:
        out["data"] = data or {}
    else:
        out["status"] = "UNAVAILABLE"
        out["reason"] = reason
    return out


def _safe_json(path: str):
    try:
        with open(path) as f:
            return json.load(f)
    except Exception:
        return None


def _mega_states(repo_root: str) -> List[Dict]:
    base = os.path.join(repo_root, "runs", "megaeval", "runs")
    out = []
    if os.path.isdir(base):
        for rid in sorted(os.listdir(base)):
            d = _safe_json(os.path.join(base, rid, "run.json"))
            state = (d or {}).get("state")
            if state:
                out.append(state)
    return out


# ------------------------------------------------------------------ stages


def _stage_raw(repo_root: str) -> Dict:
    base = os.path.join(repo_root, "runs", "megaeval", "populations")
    total, pops = 0, 0
    if os.path.isdir(base):
        for pid in os.listdir(base):
            meta = _safe_json(os.path.join(base, pid, "meta.json"))
            if meta:
                pops += 1
                total += int(meta.get("num_objects") or 0)
    if pops == 0:
        return _panel(False, "megaeval.populations", reason="no populations built")
    return _panel(True, "megaeval.populations",
                  {"populations": pops, "objects": total})


def _stage_selected(repo_root: str) -> Dict:
    base = os.path.join(repo_root, "runs", "seqeval", "runs")
    allocated, runs = 0, 0
    if os.path.isdir(base):
        for rid in sorted(os.listdir(base)):
            st = _safe_json(os.path.join(base, rid, "run.json"))
            if not st:
                continue
            plan = st.get("plan") or {}
            if plan.get("total_allocated"):
                runs += 1
                allocated += int(plan["total_allocated"])
    if runs == 0:
        return _panel(False, "seqeval.plans", reason="no sequential plans frozen")
    return _panel(True, "seqeval.plans",
                  {"plans": runs, "objects_allocated": allocated})


def _stage_simulated(repo_root: str) -> Dict:
    # nextgen counterfactual scenarios (in-flight package; read its store)
    candidates = [os.path.join(repo_root, "runs", "nextgen", "counterfactuals"),
                  os.path.join(repo_root, "runs", "nextgen", "scenarios")]
    n = 0
    found_dir = None
    for base in candidates:
        if os.path.isdir(base):
            found_dir = base
            n += sum(1 for f in os.listdir(base) if f.endswith(".json"))
    if found_dir is None:
        try:
            from sensorflow.nextgen import counterfactual as cf
            n = len(cf.list_scenarios())
            found_dir = "nextgen.counterfactual.list_scenarios"
        except Exception:
            return _panel(False, "nextgen.counterfactuals",
                          reason="nextgen counterfactual store absent "
                                 "(in-flight package)")
    return _panel(True, str(found_dir), {"counterfactual_scenarios": n})


def _stage_evaluated(repo_root: str) -> Dict:
    states = _mega_states(repo_root)
    published = [s for s in states if s.get("status") == "published"]
    if not published:
        return _panel(False, "megaeval.runs", reason="no published evaluation runs")
    return _panel(True, "megaeval.runs", {
        "runs": len(published),
        "objects_evaluated": sum(int(s.get("objects_processed") or 0)
                                 for s in published)})


def _stage_failed(repo_root: str) -> Dict:
    states = _mega_states(repo_root)
    published = [s for s in states if s.get("status") == "published"]
    fp = sum(int((s.get("headline") or {}).get("fp") or 0) for s in published)
    fn = sum(int((s.get("headline") or {}).get("fn") or 0) for s in published)
    agentic_dir = os.path.join(repo_root, "runs", "agentic", "failures")
    agentic_failures = (len([f for f in os.listdir(agentic_dir) if f.endswith(".json")])
                        if os.path.isdir(agentic_dir) else None)
    if not published and agentic_failures is None:
        return _panel(False, "megaeval.errors + agentic.failures",
                      reason="no evaluation errors recorded")
    return _panel(True, "megaeval.errors + agentic.failures",
                  {"false_positives": fp, "false_negatives": fn,
                   "agentic_failure_events": agentic_failures})


def _stage_hitl(repo_root: str) -> Dict:
    try:
        from sensorflow.evaluation.records import get_store
        st = get_store()
        tasks = st.all("review_tasks")
        reviews = st.all("human_reviews")
        return _panel(True, "evaluation.review_queue",
                      {"review_tasks": len(tasks),
                       "completed_reviews": len(reviews),
                       "open_tasks": sum(1 for t in tasks
                                         if getattr(t, "status", "") not in
                                         ("resolved", "completed"))})
    except Exception as e:
        return _panel(False, "evaluation.review_queue",
                      reason=f"labeleval store unavailable: {type(e).__name__}")


def _stage_regression(repo_root: str, registry: Registry) -> Dict:
    suites_dir = os.path.join(repo_root, "runs", "agentic", "suites")
    suites = (len([f for f in os.listdir(suites_dir) if f.endswith(".json")])
              if os.path.isdir(suites_dir) else 0)
    reg_datasets = [d for d in registry.list("datasets")
                    if d.get("role") == "REGRESSION"]
    if suites == 0 and not reg_datasets:
        return _panel(False, "agentic.suites + studio2.registry",
                      reason="no regression suites registered yet")
    return _panel(True, "agentic.suites + studio2.registry",
                  {"agentic_evaluation_suites": suites,
                   "regression_role_datasets": len(reg_datasets)})


# ------------------------------------------------------------------ panels


def _safety_panel(repo_root: str) -> Dict:
    data: Dict = {}
    sources: List[str] = []

    # SCR / risk-weighted metrics need a nextgen safety report (in-flight)
    try:
        from sensorflow.nextgen.safety_metrics import divergence_demo
        rep = divergence_demo()
        base = ((rep.get("baseline") or {}).get("safety_informed") or {})
        cand = ((rep.get("candidate") or {}).get("safety_informed") or {})
        data["scr"] = {
            "safety_critical_recall": {
                "baseline": base.get("safety_critical_recall"),
                "candidate": cand.get("safety_critical_recall")},
            "risk_weighted_recall": {
                "baseline": base.get("risk_weighted_recall"),
                "candidate": cand.get("risk_weighted_recall")},
            "deltas": rep.get("deltas"),
            "note": "nextgen safety-region analysis (deterministic demo scene, "
                    "data_label SIMULATED)"}
        sources.append("nextgen.safety_metrics")
    except Exception:
        data["scr"] = {"status": "UNAVAILABLE",
                       "reason": "nextgen safety metrics not importable (in-flight)"}

    # VRU miss rate from the real megaeval per-class recall
    states = [s for s in _mega_states(repo_root) if s.get("status") == "published"]
    if states:
        latest = max(states, key=lambda s: s.get("published_at") or "")
        vru = {}
        for cls in ("pedestrian", "cyclist"):
            pc = (latest.get("per_class") or {}).get(cls) or {}
            if pc.get("recall") is not None:
                vru[cls] = {"miss_rate": round(1.0 - pc["recall"], 4),
                            "n": pc.get("n")}
        if vru:
            data["vru_miss_rate"] = {"run_id": latest.get("run_id"), **vru}
            sources.append("megaeval.per_class")
    if "vru_miss_rate" not in data:
        data["vru_miss_rate"] = {"status": "UNAVAILABLE",
                                 "reason": "no published megaeval run with per-class stats"}

    # TTC distribution from the safety SSAM store
    ssam_dir = os.path.join(repo_root, "runs", "safety", "ssam")
    ttcs: List[float] = []
    if os.path.isdir(ssam_dir):
        for name in os.listdir(ssam_dir):
            doc = _safe_json(os.path.join(ssam_dir, name))
            if not doc:
                continue
            for sc in doc.get("scenarios") or []:
                v = sc.get("min_ttc_s")
                if isinstance(v, (int, float)):
                    ttcs.append(float(v))
    if ttcs:
        ttcs.sort()
        data["ttc"] = {"n_scenarios": len(ttcs), "min": round(ttcs[0], 3),
                       "p50": round(ttcs[len(ttcs) // 2], 3),
                       "max": round(ttcs[-1], 3), "values": ttcs[:50]}
        sources.append("safety.ssam")
    else:
        data["ttc"] = {"status": "UNAVAILABLE",
                       "reason": "no SSAM analyses persisted under runs/safety/ssam"}

    available = any(k for k in ("scr", "vru_miss_rate", "ttc")
                    if "status" not in data[k])
    return _panel(available, " + ".join(sources) or "safety sources", data,
                  reason="no safety metric source available")


def _comparison_panel(repo_root: str) -> Dict:
    bev = _safe_json(os.path.join(repo_root, "runs", "bevfusion", "latest.json"))
    if bev:
        return _panel(True, "bevfusion.latest", {
            "baseline": (bev.get("engines") or {}).get("baseline"),
            "candidate": (bev.get("engines") or {}).get("candidate"),
            "headline_deltas": bev.get("headline_deltas"),
            "recommendation": bev.get("recommendation"),
            "blockers": bev.get("blockers")})
    states = [s for s in _mega_states(repo_root) if s.get("status") == "published"]
    if len(states) >= 2:
        states.sort(key=lambda s: s.get("published_at") or "")
        b, c = states[-2], states[-1]
        deltas = []
        for m in ("precision", "recall", "safety_recall", "mean_iou"):
            bv, cv = (b.get("headline") or {}).get(m), (c.get("headline") or {}).get(m)
            if bv is not None and cv is not None:
                deltas.append({"metric": m, "baseline": round(bv, 4),
                               "candidate": round(cv, 4),
                               "delta": round(cv - bv, 4)})
        return _panel(True, "megaeval.headlines",
                      {"baseline": b.get("model_version"),
                       "candidate": c.get("model_version"),
                       "headline_deltas": deltas})
    return _panel(False, "bevfusion + megaeval",
                  reason="fewer than two comparable runs")


def _drift_panel(repo_root: str) -> Dict:
    base = os.path.join(repo_root, "runs", "megaeval", "runs")
    if os.path.isdir(base):
        for rid in sorted(os.listdir(base), reverse=True):
            shift = _safe_json(os.path.join(base, rid, "shift.json"))
            if shift:
                worst = (shift.get("shifts") or [])[:5]
                return _panel(True, f"megaeval.shift ({rid})",
                              {"run_id": rid, "method": shift.get("method"),
                               "worst_shifts": worst,
                               "n_shifts": len(shift.get("shifts") or [])})
    return _panel(False, "megaeval.shift",
                  reason="no cached distribution-shift report; run "
                         "megaeval distribution_shift for a published run")


def _compute_panel(repo_root: str) -> Dict:
    data: Dict = {}
    sources: List[str] = []
    comp = _safe_json(os.path.join(repo_root, "runs", "nextgen", "compute",
                                   "latest.json"))
    if comp:
        data["dedup"] = {k: comp.get(k) for k in
                         ("cache_hit_rate", "hit_rate", "saved_units",
                          "baseline_units", "dedup_units", "speedup")
                         if comp.get(k) is not None} or comp
        sources.append("nextgen.compute")
    try:
        from sensorflow.nextgen.cache import get_feature_cache
        stats = get_feature_cache().stats
        data["feature_cache"] = {"hits": stats.hits, "misses": stats.misses,
                                 "hit_rate": round(stats.hits / max(1, stats.hits + stats.misses), 4)}
        sources.append("nextgen.cache")
    except Exception:
        pass
    gaunt = _safe_json(os.path.join(repo_root, "runs", "nextgen", "gauntlets",
                                    "latest.json"))
    if gaunt:
        elapsed = gaunt.get("elapsed_s") or gaunt.get("wall_s")
        units = gaunt.get("units_evaluated") or gaunt.get("samples_used")
        data["gauntlet"] = {"run_id": gaunt.get("run_id"),
                            "units_evaluated": units,
                            "elapsed_s": elapsed,
                            "scenarios_per_sec": (round(units / elapsed, 2)
                                                  if units and elapsed else None),
                            "halted": gaunt.get("halted")}
        sources.append("nextgen.gauntlets")
    # megaeval throughput is always real when runs exist
    states = [s for s in _mega_states(repo_root) if s.get("status") == "published"]
    if states:
        latest = max(states, key=lambda s: s.get("published_at") or "")
        data["megaeval_throughput"] = {
            "run_id": latest.get("run_id"),
            "objects_per_sec": latest.get("throughput_objs_per_s")}
        sources.append("megaeval.runs")
    if not data:
        return _panel(False, "nextgen + megaeval",
                      reason="no compute telemetry persisted")
    return _panel(True, " + ".join(sources), data)


# ------------------------------------------------------------------ assembly


def build_funnel(repo_root: str = ".", registry: Optional[Registry] = None) -> Dict:
    registry = registry or get_registry()
    stages = [
        {"stage": "raw", "label": "Raw population objects",
         **_stage_raw(repo_root)},
        {"stage": "selected", "label": "Selected for evaluation (frozen plans)",
         **_stage_selected(repo_root)},
        {"stage": "simulated", "label": "Counterfactual / simulated scenarios",
         **_stage_simulated(repo_root)},
        {"stage": "evaluated", "label": "Objects evaluated",
         **_stage_evaluated(repo_root)},
        {"stage": "failed", "label": "Failures detected",
         **_stage_failed(repo_root)},
        {"stage": "hitl", "label": "Human-in-the-loop review",
         **_stage_hitl(repo_root)},
        {"stage": "regression", "label": "Regression suite entries",
         **_stage_regression(repo_root, registry)},
    ]
    funnel = {
        "generated_at": store.now_iso(),
        "stages": stages,
        "safety": _safety_panel(repo_root),
        "model_comparison": _comparison_panel(repo_root),
        "drift": _drift_panel(repo_root),
        "compute": _compute_panel(repo_root),
        "availability": {s["stage"]: s["available"] for s in stages},
    }
    return funnel