Spaces:
Paused
Paused
File size: 4,703 Bytes
a2bbd94 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """Error index: fast multi-criteria search over evaluation errors.
The index is a columnar table (one row per error) built during the run:
error_id, object_id, container_id, error_type, severity, confidence,
risk_score, safety_critical, sensor_disagree, + the 9 cube dimensions.
Searches return the *worst-N containers* (aggregated), never raw record dumps —
this backs the Investigation UI. Individual exemplar errors are capped.
"""
from __future__ import annotations
from typing import Dict, List, Optional
import numpy as np
import pandas as pd
from sensorflow.megaeval.population import DIMENSIONS, DIM_NAMES
from sensorflow.megaeval.runs import ERROR_TYPES
def search_errors(
errors: pd.DataFrame,
containers: pd.DataFrame,
error_types: Optional[List[str]] = None,
filters: Optional[Dict[str, List[str]]] = None,
confidence_max: Optional[float] = None,
confidence_min: Optional[float] = None,
risk_min: Optional[float] = None,
severity_min: Optional[float] = None,
safety_only: bool = False,
limit_containers: int = 25,
limit_examples: int = 50,
) -> Dict:
if errors is None or not len(errors):
return {"matched_errors": 0, "worst_containers": [], "examples": [],
"by_type": {}}
mask = np.ones(len(errors), dtype=bool)
if error_types:
codes = [ERROR_TYPES.index(t) for t in error_types if t in ERROR_TYPES]
mask &= errors["error_type"].isin(codes).to_numpy()
for dim, values in (filters or {}).items():
if dim not in DIM_NAMES:
continue
vocab = DIMENSIONS[dim]
codes = [vocab.index(v) for v in (values if isinstance(values, list) else [values])
if v in vocab]
mask &= errors[dim].isin(codes).to_numpy()
if confidence_max is not None:
mask &= errors["confidence"].to_numpy() <= confidence_max
if confidence_min is not None:
mask &= errors["confidence"].to_numpy() >= confidence_min
if risk_min is not None:
mask &= errors["risk_score"].to_numpy() >= risk_min
if severity_min is not None:
mask &= errors["severity"].to_numpy() >= severity_min
if safety_only:
mask &= errors["safety_critical"].to_numpy().astype(bool)
sub = errors[mask]
by_type = {ERROR_TYPES[int(c)]: int(n)
for c, n in sub["error_type"].value_counts().items()}
# aggregate to worst containers: rank by count x mean risk
grp = sub.groupby("container_id", as_index=False).agg(
error_count=("error_id", "size"), mean_risk=("risk_score", "mean"),
max_severity=("severity", "max"), safety_hits=("safety_critical", "sum"))
grp["rank_score"] = grp["error_count"] * grp["mean_risk"]
grp = grp.sort_values("rank_score", ascending=False).head(limit_containers)
cmeta = containers.set_index("container_id") if containers is not None else None
worst = []
for _, r in grp.iterrows():
cid = int(r["container_id"])
item = {"container_id": cid, "error_count": int(r["error_count"]),
"mean_risk": round(float(r["mean_risk"]), 4),
"max_severity": round(float(r["max_severity"]), 4),
"safety_hits": int(r["safety_hits"])}
if cmeta is not None and cid in cmeta.index:
row = cmeta.loc[cid]
for dim in ("weather", "lighting", "road_type", "scenario"):
item[dim] = DIMENSIONS[dim][int(row[dim])]
item["n_objects"] = int(row["n_objects"])
item["risk_score"] = round(float(row["risk_score"]), 4)
worst.append(item)
ex = sub.sort_values("risk_score", ascending=False).head(limit_examples)
examples = []
for _, r in ex.iterrows():
examples.append({
"error_id": int(r["error_id"]),
"annotation_id": (f"obj-{int(r['object_id'])}" if r["object_id"] >= 0
else f"fp-{int(r['error_id'])}"),
"container_id": int(r["container_id"]),
"error_type": ERROR_TYPES[int(r["error_type"])],
"class": DIMENSIONS["class"][int(r["class"])],
"severity": round(float(r["severity"]), 4),
"confidence": round(float(r["confidence"]), 4),
"risk_score": round(float(r["risk_score"]), 4),
"safety_critical": bool(r["safety_critical"]),
"scenario": DIMENSIONS["scenario"][int(r["scenario"])],
"lighting": DIMENSIONS["lighting"][int(r["lighting"])],
"weather": DIMENSIONS["weather"][int(r["weather"])],
})
return {"matched_errors": int(len(sub)), "by_type": by_type,
"worst_containers": worst, "examples": examples}
|