Spaces:
Paused
Paused
File size: 4,743 Bytes
73d9b6f | 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 108 109 110 111 112 113 114 115 116 117 118 119 120 | """Quantitative validation: for candidates with model outputs, compute the
OBSERVED difficulty from actual baseline behavior and compare it with the
miner's PREDICTED difficulty (agreement matrix).
Observed metrics per track candidate: mean baseline confidence, per-example
correctness, localization quality (IoU), miss rate, class confusion, track
stability. Predicted difficulty comes from the miner and is never edited here
— this stage only measures.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from sensorflow.raremine.models import RareMineStore, TrackCandidate
DIFF_ORDER = ["EASY", "MODERATE", "HARD", "EXTREME"]
def _track_object_ids(store: RareMineStore, tc: TrackCandidate) -> List[tuple]:
"""(scene, object) pairs across the track's frames."""
out = []
for scene in store.where("scenes", sequence_id=tc.sequence_id):
for obj in scene.objects:
if obj.track_id == tc.track_id:
out.append((scene, obj))
return out
def observed_metrics(store: RareMineStore, tc: TrackCandidate) -> Optional[Dict[str, Any]]:
"""None when no baseline predictions exist anywhere on the track."""
pairs = [(s, o) for s, o in _track_object_ids(store, tc)
if s.modalities.get("baseline_predictions")]
if not pairs:
return None
confidences: List[float] = []
ious: List[float] = []
misses = 0
confusion: Dict[str, int] = {}
unstable = 0
correct = 0
for scene, obj in pairs:
pred = next((p for p in scene.baseline_predictions if p.object_id == obj.object_id), None)
if pred is None:
misses += 1
continue
confidences.append(pred.confidence)
if pred.iou_with_gt is not None:
ious.append(pred.iou_with_gt)
if pred.predicted_class != "pedestrian":
confusion[pred.predicted_class] = confusion.get(pred.predicted_class, 0) + 1
else:
correct += 1
if not pred.track_stable:
unstable += 1
n = len(pairs)
mean_conf = round(sum(confidences) / len(confidences), 3) if confidences else 0.0
mean_iou = round(sum(ious) / len(ious), 3) if ious else None
miss_rate = round(misses / n, 3)
confused_rate = round(sum(confusion.values()) / n, 3)
stability = round(1.0 - unstable / max(n - misses, 1), 3)
# observed difficulty grade from measured behavior
if miss_rate > 0.5 or confused_rate > 0.5:
grade = "EXTREME"
elif miss_rate > 0.0 or confusion or unstable:
grade = "HARD"
elif mean_conf < 0.4 or (mean_iou is not None and mean_iou < 0.5):
grade = "MODERATE" if mean_conf >= 0.25 else "HARD"
else:
grade = "EASY"
return {
"frames_with_predictions": n,
"mean_baseline_confidence": mean_conf,
"mean_iou": mean_iou,
"miss_rate": miss_rate,
"class_confusion": confusion,
"confused_rate": confused_rate,
"track_stability": stability,
"per_example_correct": correct,
"observed_difficulty": grade,
"exposes_model_failure": miss_rate > 0 or bool(confusion) or unstable > 0
or mean_conf < 0.35 or (mean_iou is not None and mean_iou < 0.5),
}
def quantitative_report(store: RareMineStore, bank_id: str) -> Dict[str, Any]:
tcs = [t for t in store.where("track_candidates", bank_id=bank_id)
if t.duplicate_of is None and t.representative.edge_case_detected]
rows = []
matrix: Dict[str, Dict[str, int]] = {d: {d2: 0 for d2 in DIFF_ORDER} for d in DIFF_ORDER}
agree = 0
with_obs = 0
for tc in sorted(tcs, key=lambda t: t.track_candidate_id):
obs = observed_metrics(store, tc)
predicted = tc.representative.perception_difficulty
row = {
"track_candidate_id": tc.track_candidate_id,
"costume_type": tc.representative.costume_type,
"predicted_difficulty": predicted,
"observed": obs,
}
if obs is not None:
with_obs += 1
matrix[predicted][obs["observed_difficulty"]] += 1
if abs(DIFF_ORDER.index(predicted) - DIFF_ORDER.index(obs["observed_difficulty"])) <= 1:
agree += 1
rows.append(row)
return {
"candidates_evaluated": len(rows),
"with_model_outputs": with_obs,
"agreement_matrix": matrix,
"within_one_level_agreement": round(agree / with_obs, 3) if with_obs else None,
"rows": rows,
"note": "predicted difficulty is the miner's forecast; observed difficulty is "
"measured from actual baseline model behavior — they are compared, "
"never merged",
}
|