""" metrics.py — Vision scoring (replaces the text substring-grounding metric). Every category carries two UNIVERSAL metrics that encode the project's thesis — robust, schema-valid JSON — plus a category-specific accuracy scorer: schema_valid : did the output validate against the category's Pydantic model (after the never-raises recovery walk)? json_robust : did it parse WITHOUT repair (clean bare JSON)? This, measured in json_mode, is the native-capability signal driving the no-finetune decision. The headline `labeler_score` MULTIPLIES accuracy by validity and robustness, so a model that is accurate but emits fragile JSON scores worse than a slightly less accurate model that emits clean JSON — exactly what you want when pointing a labeler at a million images with no human in the loop. """ from __future__ import annotations import math import re from dataclasses import asdict, dataclass from typing import Callable, Optional from ..evaluator import parse_against from .coords import XYWH, XYXY, BBox, CoordSpace, to_canonical from .tasks_vision import VisionTaskSpec, model_for # ────────────────────────────────────────────────────────────────────────────── # Result types # ────────────────────────────────────────────────────────────────────────────── @dataclass class MetricResult: category: str image_id: str mode: str parse_ok: bool # a JSON object was recovered + decoded (maybe invalid schema) schema_valid: bool # validated against the category model needed_repair: bool # recovery had to strip fences / skip prose / trim junk grammar_conformant: bool # constrained decoding actually applied (backend == xgrammar) primary_score: Optional[float] # task accuracy 0..1; None if no GT / invalid metrics: dict needed_structural_repair: bool = False # repair beyond a benign fence strip n_output_tokens: int = 0 gen_seconds: float = 0.0 error: Optional[str] = None notes: str = "" @property def json_robust(self) -> bool: """Valid JSON that needed no STRUCTURAL repair — the native-capability signal. A benign markdown-fence wrap is tolerated (fence-stripping is deterministic); prose/runaway/malformed is not.""" return self.schema_valid and not self.needed_structural_repair def to_dict(self) -> dict: d = asdict(self) d["json_robust"] = self.json_robust return d @dataclass class VisionRunMetrics: category: str model: str reasoning: str mode: str n: int schema_valid_rate: float json_robustness: float has_task_score: bool primary_score_mean: Optional[float] metrics_mean: dict mean_output_tokens: float total_gen_seconds: float tokens_per_sec: float labeler_score: Optional[float] def __str__(self) -> str: acc = "n/a" if self.primary_score_mean is None else f"{self.primary_score_mean:.3f}" lab = "n/a" if self.labeler_score is None else f"{self.labeler_score:.3f}" return (f"[{self.model}/{self.reasoning}/{self.category}/{self.mode}] n={self.n} " f"valid={self.schema_valid_rate:.1%} robust={self.json_robustness:.1%} " f"acc={acc} labeler={lab} tok/s={self.tokens_per_sec:.0f}") # ────────────────────────────────────────────────────────────────────────────── # The labeler-selection composite (the verdict core) # ────────────────────────────────────────────────────────────────────────────── def labeler_score(accuracy: Optional[float], schema_valid_rate: float, json_robustness: float) -> Optional[float]: """Multiplicative composite: accuracy × validity-gate × robustness-penalty. Invalid JSON is unusable (hard-ish cap via the 0.5+0.5·valid term); fragile but repairable JSON is penalized, not killed (0.7+0.3·robust term). """ if accuracy is None: return None return accuracy * (0.5 + 0.5 * schema_valid_rate) * (0.7 + 0.3 * json_robustness) # ────────────────────────────────────────────────────────────────────────────── # Small pure helpers (no external deps — editdistance/jiwer are optional accel) # ────────────────────────────────────────────────────────────────────────────── def _norm_text(s: str) -> str: return re.sub(r"\s+", " ", (s or "").strip().lower()) def _levenshtein(a: list, b: list) -> int: if a == b: return 0 if not a: return len(b) if not b: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a, 1): cur = [i] for j, cb in enumerate(b, 1): cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb))) prev = cur return prev[-1] def _cer(pred: str, gt: str) -> float: g = _norm_text(gt) p = _norm_text(pred) if not g: return 0.0 if not p else 1.0 return _levenshtein(list(p), list(g)) / len(g) def _wer(pred: str, gt: str) -> float: g = _norm_text(gt).split() p = _norm_text(pred).split() if not g: return 0.0 if not p else 1.0 return _levenshtein(p, g) / len(g) # ────────────────────────────────────────────────────────────────────────────── # Tolerant label matching — VLMs use richer / synonymous labels than dataset # vocabularies (e.g. "television" vs COCO's "tv"). Without this, correct boxes are # discarded on a string mismatch (observed: Qwen3-VL localizes COCO near-perfectly # but exact-match F1 read ~0.25). Synonym groups + substring + plural fallback. # ────────────────────────────────────────────────────────────────────────────── _SYNONYM_GROUPS = [ {"tv", "television", "televisions", "telly", "monitor", "screen"}, {"couch", "sofa", "settee", "loveseat"}, {"motorcycle", "motorbike", "moped", "scooter"}, {"airplane", "aeroplane", "plane", "aircraft", "jet"}, {"cell phone", "cellphone", "mobile phone", "mobile", "phone", "smartphone"}, {"potted plant", "houseplant", "plant", "pot plant", "flowerpot", "flower pot"}, {"dining table", "table", "desk"}, {"car", "automobile", "sedan", "vehicle"}, {"bicycle", "bike", "cycle"}, {"person", "people", "man", "men", "woman", "women", "human", "boy", "girl", "child", "kid", "pedestrian", "player", "lady", "guy", "skier", "surfer", "rider", "athlete", "batter", "pitcher", "catcher"}, {"hot dog", "hotdog"}, {"donut", "doughnut"}, {"remote", "remote control"}, {"sports ball", "ball"}, {"wine glass", "wineglass", "glass"}, {"tie", "necktie"}, ] _SYN_GROUP: dict[str, int] = {} for _gi, _grp in enumerate(_SYNONYM_GROUPS): for _w in _grp: _SYN_GROUP[_w] = _gi def _depluralize(t: str) -> str: if len(t) <= 3: return t if t.endswith("ies"): return t[:-3] + "y" if t.endswith("es"): return t[:-2] if t.endswith("s"): return t[:-1] return t def labels_match(a: str, b: str) -> bool: """True if two object labels refer to the same thing (tolerant).""" a, b = _norm_text(a), _norm_text(b) if not a or not b: return False if a == b: return True if _depluralize(a) == _depluralize(b): return True ga, gb = _SYN_GROUP.get(a), _SYN_GROUP.get(b) if ga is not None and ga == gb: return True # word-level containment: "dining table" vs "table", "red car" vs "car" aw, bw = set(a.split()), set(b.split()) if aw and bw and (aw <= bw or bw <= aw): return True return False # ────────────────────────────────────────────────────────────────────────────── # Per-category scorers: (pred_dict, gt, ctx) -> (primary_score|None, metrics_dict) # ctx carries {"size": (W,H), "coord_space": CoordSpace} # ────────────────────────────────────────────────────────────────────────────── def _acceptable_labels(gt) -> set[str]: if isinstance(gt, dict): if "labels" in gt: return {_norm_text(x) for x in gt["labels"]} if "label" in gt: return {_norm_text(gt["label"])} if isinstance(gt, str): return {_norm_text(gt)} return set() def score_classification(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: accept = _acceptable_labels(gt) pred_label = _norm_text(str(pred.get("label", ""))) # tolerant: "spaghetti" credits "spaghetti bolognese", "tv" credits "television" top1 = 1.0 if any(labels_match(pred_label, a) for a in accept) else 0.0 top5_labels = {_norm_text(str(d.get("label", ""))) for d in (pred.get("top5") or [])} top5_labels.add(pred_label) top5 = 1.0 if any(labels_match(p, a) for p in top5_labels for a in accept) else 0.0 return top1, {"top1": top1, "top5": top5} def _gt_boxes_to_canonical(gt, size) -> list[tuple[str, BBox]]: out = [] for b in (gt.get("boxes") if isinstance(gt, dict) else []) or []: label = _norm_text(str(b.get("label", ""))) fmt = b.get("fmt", XYWH) box = to_canonical(b["bbox"], CoordSpace.PIXEL_ABS, size, fmt=fmt) out.append((label, box)) return out def _greedy_match_f1(preds, gts, iou_thr, require_label) -> tuple[float, float, float, int]: """Greedy IoU matching (preds pre-sorted by score). Returns (precision, recall, f1, tp). `require_label` toggles labeled vs class-agnostic matching.""" matched: set[int] = set() tp = 0 for plabel, _score, pbox in preds: best_gi, best_iou = -1, iou_thr for gi, (glabel, gbox) in enumerate(gts): if gi in matched: continue if require_label and not labels_match(plabel, glabel): continue iou = pbox.iou(gbox) if iou >= best_iou: best_gi, best_iou = gi, iou if best_gi >= 0: matched.add(best_gi) tp += 1 fp = len(preds) - tp fn = len(gts) - len(matched) precision = tp / (tp + fp) if (tp + fp) else (1.0 if not gts else 0.0) recall = tp / (tp + fn) if (tp + fn) else 1.0 f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) else 0.0 return precision, recall, f1, tp def score_detection(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: size = ctx["size"] space = ctx["coord_space"] gts = _gt_boxes_to_canonical(gt, size) preds = [] for d in (pred.get("detections") or []): box_raw = d.get("box") if not (isinstance(box_raw, (list, tuple)) and len(box_raw) == 4): continue try: box = to_canonical(box_raw, space, size, fmt=XYXY) except (ValueError, TypeError): continue preds.append((_norm_text(str(d.get("label", ""))), float(d.get("score", 1.0) or 1.0), box)) preds.sort(key=lambda t: t[1], reverse=True) iou_thr = 0.5 # labeled (tolerant) match — the headline accuracy precision, recall, f1, _ = _greedy_match_f1(preds, gts, iou_thr, require_label=True) # class-agnostic localization — "can it find/box objects" regardless of naming loc_p, loc_r, loc_f1, _ = _greedy_match_f1(preds, gts, iou_thr, require_label=False) pred_count = pred.get("count") count_err = abs(int(pred_count) - len(gts)) if isinstance(pred_count, (int, float)) else len(preds) - len(gts) return f1, {"precision": precision, "recall": recall, "f1": f1, "localization_f1": loc_f1, "localization_recall": loc_r, "iou_thr": iou_thr, "count_abs_err": float(abs(count_err)), "n_pred": float(len(preds)), "n_gt": float(len(gts))} def score_ocr(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: gt_text = gt.get("text") if isinstance(gt, dict) else str(gt) pred_text = str(pred.get("full_text", "")) cer = _cer(pred_text, gt_text) wer = _wer(pred_text, gt_text) exact = 1.0 if _norm_text(pred_text) == _norm_text(gt_text) else 0.0 # answer-containment credit (TextVQA-style: GT is the answer phrase) contains = 1.0 if _norm_text(gt_text) and _norm_text(gt_text) in _norm_text(pred_text) else 0.0 primary = max(exact, contains, max(0.0, 1.0 - cer)) return primary, {"cer": cer, "wer": wer, "exact": exact, "contains": contains} def score_datatype_diff(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Did the model identify the rendered data format (json/yaml/md/...)?""" gt_type = _norm_text(gt.get("data_type") if isinstance(gt, dict) else str(gt)) pred_type = _norm_text(str(pred.get("data_type", ""))) ok = 1.0 if pred_type == gt_type else 0.0 return ok, {"type_acc": ok} def _flatten_kv(obj, prefix="") -> set[str]: """Flatten a parsed JSON-ish object to a set of 'path=value' leaf strings.""" out: set[str] = set() if isinstance(obj, dict): for k, v in obj.items(): out |= _flatten_kv(v, f"{prefix}{k}.") elif isinstance(obj, list): for i, v in enumerate(obj): out |= _flatten_kv(v, f"{prefix}{i}.") else: out.add(f"{prefix.rstrip('.')}={_norm_text(str(obj))}") return out def score_datatype_util(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Did the model re-emit the rendered data as normalized JSON matching the GT? Scored by leaf 'path=value' F1 (order-independent), plus a type-correct bonus.""" import json as _json gt_obj = gt.get("content") if isinstance(gt, dict) else gt if isinstance(gt_obj, str): try: gt_obj = _json.loads(gt_obj) except (ValueError, TypeError): pass pred_content = pred.get("content") if isinstance(pred_content, str): try: pred_content = _json.loads(pred_content) except (ValueError, TypeError): pass # leave as string → flatten will treat as a single leaf g = _flatten_kv(gt_obj) p = _flatten_kv(pred_content) if not g: return (1.0 if not p else 0.0), {"kv_f1": 0.0} tp = len(g & p) prec = tp / len(p) if p else 0.0 rec = tp / len(g) f1 = (2 * prec * rec / (prec + rec)) if (prec + rec) else 0.0 type_ok = 1.0 if _norm_text(str(pred.get("data_type", ""))) == _norm_text( gt.get("data_type", "") if isinstance(gt, dict) else "") else 0.0 return f1, {"kv_f1": f1, "kv_precision": prec, "kv_recall": rec, "type_acc": type_ok} _SPATIAL_PRED_NORM = { "left of": "left_of", "to the left of": "left_of", "left": "left_of", "right of": "right_of", "to the right of": "right_of", "right": "right_of", "in front of": "in_front_of", "front of": "in_front_of", } def _norm_pred(p: str) -> str: n = _norm_text(p) return _SPATIAL_PRED_NORM.get(n, n.replace(" ", "_")) def _pred_triples(pred: dict) -> list[tuple]: """Extract (subject, predicate, object) triples from either the spatial ('relations': [{subject,predicate,object}]) or semantic ('associations': [{a,relation,b}]) shape.""" out = [] for r in (pred.get("relations") or []): if isinstance(r, dict): out.append((_norm_text(str(r.get("subject", ""))), _norm_pred(str(r.get("predicate", ""))), _norm_text(str(r.get("object", ""))))) for r in (pred.get("associations") or []): if isinstance(r, dict): out.append((_norm_text(str(r.get("a", ""))), _norm_pred(str(r.get("relation", ""))), _norm_text(str(r.get("b", ""))))) return out def score_triples(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Triple-F1 with tolerant subject/object matching + exact (normalized) predicate.""" gts = [(_norm_text(s), _norm_pred(p), _norm_text(o)) for s, p, o in (gt.get("triples", []) if isinstance(gt, dict) else [])] preds = _pred_triples(pred) matched = set() tp = 0 for ps, pp, po in preds: for gi, (gs, gp, go) in enumerate(gts): if gi in matched or pp != gp: continue if labels_match(ps, gs) and labels_match(po, go): matched.add(gi) tp += 1 break prec = tp / len(preds) if preds else (1.0 if not gts else 0.0) rec = tp / len(gts) if gts else 1.0 f1 = (2 * prec * rec / (prec + rec)) if (prec + rec) else 0.0 return f1, {"triple_f1": f1, "precision": prec, "recall": rec, "n_pred": float(len(preds)), "n_gt": float(len(gts))} def score_depth_order(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Pairwise nearer/farther ordering accuracy over the GT pairs.""" gpairs = gt.get("pairs", []) if isinstance(gt, dict) else [] preds = pred.get("relative_depth") or [] def _find(a, b): for r in preds: if not isinstance(r, dict): continue ra, rb = _norm_text(str(r.get("a", ""))), _norm_text(str(r.get("b", ""))) if labels_match(ra, a) and labels_match(rb, b): return _norm_text(str(r.get("a_is", ""))) if labels_match(ra, b) and labels_match(rb, a): # reversed → flip v = _norm_text(str(r.get("a_is", ""))) return {"nearer": "farther", "farther": "nearer"}.get(v, v) return None correct = 0 for p in gpairs: got = _find(_norm_text(p["a"]), _norm_text(p["b"])) if got == _norm_text(p["a_is"]): correct += 1 acc = correct / len(gpairs) if gpairs else (1.0 if not preds else 0.0) return acc, {"order_acc": acc, "n_gt_pairs": float(len(gpairs))} def score_subject_fixation(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """IoU of the predicted primary-subject box vs the GT salient box (+ label).""" size = ctx["size"] space = ctx["coord_space"] ps = pred.get("primary_subject") or {} raw = ps.get("box") if isinstance(ps, dict) else None if not (isinstance(raw, (list, tuple)) and len(raw) == 4): return 0.0, {"iou": 0.0, "label_ok": 0.0} try: pbox = to_canonical(raw, space, size, fmt=XYXY) except (ValueError, TypeError): return 0.0, {"iou": 0.0, "label_ok": 0.0} gbox = to_canonical(gt["box"], CoordSpace.PIXEL_ABS, size, fmt=gt.get("fmt", XYXY)) iou = pbox.iou(gbox) label_ok = 1.0 if labels_match(str(ps.get("label", "")), str(gt.get("label", ""))) else 0.0 primary = 1.0 if (iou >= 0.5 and label_ok) else (0.5 if iou >= 0.5 else 0.0) return primary, {"iou": iou, "label_ok": label_ok} def _seg_poly_points(flat, sx=1.0, sy=1.0): """Flat [x1,y1,x2,y2,...] -> list of (x,y) tuples scaled by (sx,sy). Tolerant: ignores a trailing odd value and skips non-numeric entries.""" pts = [] if not isinstance(flat, (list, tuple)): return pts n = (len(flat) // 2) * 2 for i in range(0, n, 2): try: x = float(flat[i]) * sx y = float(flat[i + 1]) * sy except (TypeError, ValueError): continue pts.append((x, y)) # Tolerate a 4-number bbox-as-polygon: expand [x1,y1,x2,y2] -> rectangle corners. if len(pts) == 2: (x1, y1), (x2, y2) = pts pts = [(x1, y1), (x2, y1), (x2, y2), (x1, y2)] return pts def _seg_rasterize(points, grid, gsx, gsy): """Fill a polygon (pixel-coord points) onto a grid×grid boolean mask. Maps pixel->grid via (px*gsx, py*gsy). Returns a bool ndarray or None.""" import numpy as np from PIL import Image, ImageDraw if len(points) < 3: return None mapped = [(px * gsx, py * gsy) for (px, py) in points] img = Image.new("L", (grid, grid), 0) ImageDraw.Draw(img).polygon(mapped, fill=1) return np.asarray(img, dtype=bool) def score_segmentation(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Instance-segmentation mIoU. For each GT mask, greedily match a predicted mask of the same (tolerant) label by polygon IoU, computed by rasterizing both polygons onto a shared grid. mIoU is averaged over GT masks. GT polygons (gt['masks'][i]['polygon_pixels']) are absolute pixels. Predicted polygons (pred['masks'][i]['polygon']) are in ctx['coord_space'] (NORM_0_1000 by default) and are scaled to pixels before rasterizing. Never raises: missing / short / malformed polygons score 0 for that mask.""" import numpy as np size = ctx.get("size", (1, 1)) space = ctx.get("coord_space", CoordSpace.NORM_0_1000) if isinstance(size, (list, tuple)) and len(size) == 2: W, H = size else: W, H = (1, 1) W = max(int(W or 1), 1) H = max(int(H or 1), 1) # GT masks (pixel coords). Accept 'polygon_pixels' or 'polygon' as a fallback. gts = [] for m in (gt.get("masks") if isinstance(gt, dict) else []) or []: if not isinstance(m, dict): continue poly = m.get("polygon_pixels") if poly is None: poly = m.get("polygon") or [] pts = _seg_poly_points(poly) if len(pts) >= 3: gts.append((_norm_text(str(m.get("label", ""))), pts)) # Predicted masks: scale ctx-space polygons to pixels. if space == CoordSpace.NORM_0_1: sx, sy = float(W), float(H) elif space == CoordSpace.NORM_0_1000: sx, sy = float(W) / 1000.0, float(H) / 1000.0 else: # PIXEL_ABS or unknown -> treat as pixels sx, sy = 1.0, 1.0 preds = [] for m in (pred.get("masks") if isinstance(pred, dict) else []) or []: if not isinstance(m, dict): continue pts = _seg_poly_points(m.get("polygon") or [], sx, sy) if len(pts) >= 3: preds.append((_norm_text(str(m.get("label", ""))), pts)) if not gts: ok = 1.0 if not preds else 0.0 return ok, {"miou": ok, "n_pred": float(len(preds)), "n_gt": 0.0, "matched": 0.0} # Shared raster grid. Per-axis scale preserves the image aspect ratio so a # non-square image doesn't distort IoU. 128 keeps cost trivial. GRID = 128 gsx = GRID / float(W) gsy = GRID / float(H) gt_masks = [(lbl, _seg_rasterize(pts, GRID, gsx, gsy)) for (lbl, pts) in gts] pred_masks = [(lbl, _seg_rasterize(pts, GRID, gsx, gsy)) for (lbl, pts) in preds] used: set = set() ious = [] matched = 0 for glabel, garr in gt_masks: if garr is None or not garr.any(): ious.append(0.0) continue best_iou, best_j = 0.0, -1 for j, (plabel, parr) in enumerate(pred_masks): if j in used or parr is None: continue if not labels_match(plabel, glabel): continue inter = int(np.logical_and(garr, parr).sum()) if inter == 0: continue union = int(np.logical_or(garr, parr).sum()) iou = inter / union if union else 0.0 if iou > best_iou: best_iou, best_j = iou, j if best_j >= 0: used.add(best_j) matched += 1 ious.append(best_iou) miou = sum(ious) / len(ious) if ious else 0.0 return miou, {"miou": miou, "n_pred": float(len(preds)), "n_gt": float(len(gts)), "matched": float(matched)} def _outline_points(flat, space, size): """Parse a flat [x1,y1,x2,y2,...] list into pixel (x,y) vertices. Each vertex is run through to_canonical as a degenerate 1px box so the documented space/clip handling is reused. Robust to odd length / junk; returns [] (no polygon) if fewer than 3 usable vertices.""" import math as _math if not isinstance(flat, (list, tuple)): return [] pts = [] n = len(flat) // 2 for k in range(n): try: x = float(flat[2 * k]); y = float(flat[2 * k + 1]) except (TypeError, ValueError, IndexError): continue if _math.isnan(x) or _math.isnan(y) or _math.isinf(x) or _math.isinf(y): continue b = to_canonical([x, y, x, y], space, size, fmt=XYXY) # scales space + clips to image pts.append((b.x1, b.y1)) # Tolerate a 4-number bbox-as-outline: expand [x1,y1,x2,y2] -> rectangle corners. if len(pts) == 2: (x1, y1), (x2, y2) = pts pts = [(x1, y1), (x2, y1), (x2, y2), (x1, y2)] return pts if len(pts) >= 3 else [] def _outline_raster_iou(poly_a, poly_b, size, max_dim=200): """Polygon IoU by scanline rasterization on a downscaled grid (<= max_dim on the long side). Even-odd fill; handles convex + concave outlines. Never raises.""" import math as _math import numpy as np W, H = size if len(poly_a) < 3 or len(poly_b) < 3: return 0.0 scale = min(1.0, float(max_dim) / max(1, max(W, H))) rw = max(1, int(round(W * scale))); rh = max(1, int(round(H * scale))) def _fill(poly): grid = np.zeros((rh, rw), dtype=bool) xs = [p[0] * scale for p in poly] ys = [p[1] * scale for p in poly] m = len(poly) y0 = max(0, int(_math.floor(min(ys)))); y1 = min(rh - 1, int(_math.ceil(max(ys)))) for yy in range(y0, y1 + 1): yc = yy + 0.5 xint = [] for i in range(m): xi, yi = xs[i], ys[i] xj, yj = xs[(i + 1) % m], ys[(i + 1) % m] if (yi <= yc < yj) or (yj <= yc < yi): xint.append(xi + (yc - yi) / (yj - yi) * (xj - xi)) xint.sort() for c in range(0, len(xint) - 1, 2): xa = max(0, int(_math.ceil(xint[c] - 0.5))) xb = min(rw - 1, int(_math.floor(xint[c + 1] - 0.5))) if xb >= xa: grid[yy, xa:xb + 1] = True return grid a = _fill(poly_a); b = _fill(poly_b) inter = int(np.logical_and(a, b).sum()) union = int(np.logical_or(a, b).sum()) return inter / union if union else 0.0 def score_outline_iou(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Polygon IoU (rasterized) of the predicted main-object outline vs the GT outline, gated by a tolerant label match. primary = 0.0 if IoU < 0.5 = 1.0 if label matches else 0.5 if IoU >= 0.5 Pred outline is in ctx['coord_space']; GT outline is pixel-abs. Never raises.""" size = ctx["size"] space = ctx["coord_space"] pred_poly = _outline_points(pred.get("outline"), space, size) if isinstance(gt, dict): gt_flat = gt.get("outline") or [] gt_label = str(gt.get("label", "")) else: gt_flat, gt_label = [], "" gt_poly = _outline_points(gt_flat, CoordSpace.PIXEL_ABS, size) iou = _outline_raster_iou(pred_poly, gt_poly, size) label_ok = 1.0 if labels_match(str(pred.get("label", "")), gt_label) else 0.0 if iou >= 0.5: primary = 1.0 if label_ok else 0.5 else: primary = 0.0 return primary, {"poly_iou": iou, "label_ok": label_ok, "n_pred_pts": float(len(pred_poly)), "n_gt_pts": float(len(gt_poly))} def _as_xyzwhl_yaw(b): """Coerce a bbox3d list to 7 floats [x,y,z,w,h,l,yaw]; pad missing yaw. Returns None if fewer than 6 usable numbers (need at least center+size).""" if not isinstance(b, (list, tuple)): return None vals = [] for v in b[:7]: try: vals.append(float(v)) except (TypeError, ValueError): vals.append(0.0) if len(vals) < 6: return None while len(vals) < 7: vals.append(0.0) # missing yaw -> 0 return vals def score_iou3d(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """3D-center matching for the simplified ground-plane proxy. A predicted object matches a GT object when (a) their 3D-center L2 distance is below `dist_thr` (normalized units) and (b) the size ratio is reasonable (each of w,h,l within [1/size_tol, size_tol]); class is checked separately for credit. Headline = (0.5*matched + 0.5*class_correct) / n_gt, so a box in the right place but mislabeled earns half credit. Reports center distance + class/recall metrics. Never raises: short/missing bbox3d entries are dropped, not fatal. """ import math dist_thr = 0.3 size_tol = 3.0 gts = gt.get("objects", []) if isinstance(gt, dict) else [] raw_preds = pred.get("objects") if isinstance(pred, dict) else None preds = [] for d in (raw_preds or []): if not isinstance(d, dict): continue vec = _as_xyzwhl_yaw(d.get("bbox3d")) if vec is None: continue cls = _norm_text(str(d.get("class", d.get("label", "")))) try: sc = float(d.get("score", 1.0)) except (TypeError, ValueError): sc = 1.0 preds.append((cls, sc, vec)) preds.sort(key=lambda t: t[1], reverse=True) if not gts: return (1.0 if not preds else 0.0), {"matched_frac": 0.0, "precision": 0.0, "class_acc": 0.0, "center_dist": float(dist_thr), "n_pred": float(len(preds)), "n_gt": 0.0} matched_gt: set = set() n_class_ok = 0 dist_sum = 0.0 dist_n = 0 for pcls, _sc, pvec in preds: best_gi, best_d = -1, dist_thr for gi, g in enumerate(gts): if gi in matched_gt: continue gvec = _as_xyzwhl_yaw(g.get("bbox3d")) if gvec is None: continue dx, dy, dz = pvec[0] - gvec[0], pvec[1] - gvec[1], pvec[2] - gvec[2] d3 = math.sqrt(dx * dx + dy * dy + dz * dz) if d3 > best_d: continue ok_size = True for idx in (3, 4, 5): # w, h, l ps, gs = abs(pvec[idx]), abs(gvec[idx]) if gs <= 1e-6: continue r = (ps / gs) if ps > 1e-6 else 0.0 if r < (1.0 / size_tol) or r > size_tol: ok_size = False break if not ok_size: continue best_gi, best_d = gi, d3 if best_gi >= 0: matched_gt.add(best_gi) dist_sum += best_d dist_n += 1 gcls = _norm_text(str(gts[best_gi].get("class", gts[best_gi].get("label", "")))) if labels_match(pcls, gcls): n_class_ok += 1 n_gt = len(gts) n_pred = len(preds) matched = len(matched_gt) recall = matched / n_gt precision = matched / n_pred if n_pred else 0.0 class_acc = n_class_ok / matched if matched else 0.0 mean_center_dist = (dist_sum / dist_n) if dist_n else float(dist_thr) primary = (0.5 * matched + 0.5 * n_class_ok) / n_gt return primary, {"matched_frac": recall, "precision": precision, "class_acc": class_acc, "center_dist": mean_center_dist, "n_pred": float(n_pred), "n_gt": float(n_gt)} def _wrap_deg(a: float) -> float: """Wrap an angle (degrees) into [-180, 180).""" return (float(a) + 180.0) % 360.0 - 180.0 def score_camera_rotation(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Camera rotation: mean absolute angular error across yaw/pitch/roll, each wrapped to [-180,180). primary = acc@30deg = fraction of the 3 axes within 30 deg of GT. Robust to missing / short / non-numeric rotation lists (never raises).""" gt_rot = (gt or {}).get("rotation") if isinstance(gt, dict) else None if not isinstance(gt_rot, (list, tuple)) or len(gt_rot) < 3: return None, {} raw = pred.get("rotation") p = list(raw) if isinstance(raw, (list, tuple)) else [] p = (p + [0.0, 0.0, 0.0])[:3] # pad missing axes with 0 so a short list scores, not crashes errs, within = [], 0 for i in range(3): try: d = abs(_wrap_deg(float(p[i]) - float(gt_rot[i]))) except (TypeError, ValueError): d = 180.0 errs.append(d) if d <= 30.0: within += 1 mean_abs_err = sum(errs) / 3.0 acc30 = within / 3.0 return acc30, {"acc@30deg": acc30, "mean_abs_err": mean_abs_err, "yaw_err": errs[0], "pitch_err": errs[1], "roll_err": errs[2]} def score_vqa(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Grounded-VQA answer accuracy (OCR-style containment, reused here). Credit if the predicted answer EQUALS any gold answer, CONTAINS a gold answer, or is CONTAINED IN a gold answer (all after _norm_text). Reports `exact` and `contains` separately. The optional grounded_region box is NOT scored — VQA datasets carry no per-question GT box, so answer text is the only signal. Never raises: missing/short fields collapse to a 0.0 score. """ golds = _vqa_gold_answers(gt) pred_ans = _norm_text(str(pred.get("answer", ""))) if not golds: # no GT answers -> only an empty prediction can be "correct" return (1.0 if not pred_ans else 0.0), {"exact": 0.0, "contains": 0.0} exact = 1.0 if pred_ans in golds else 0.0 contains = 0.0 if pred_ans: for g in golds: if g and (g in pred_ans or pred_ans in g): contains = 1.0 break primary = max(exact, contains) return primary, {"exact": exact, "contains": contains} def _vqa_gold_answers(gt) -> list[str]: """Normalize GT into a list of acceptable (normalized) gold answer strings. Accepts {"answers": [...]}, {"answers": "x"}, {"answer": "x"}, a bare list, or a bare string. Anything else -> [].""" if isinstance(gt, dict): a = gt.get("answers") if isinstance(a, (list, tuple)): return [_norm_text(str(x)) for x in a if str(x).strip()] if a is not None: return [_norm_text(str(a))] if "answer" in gt: return [_norm_text(str(gt["answer"]))] if isinstance(gt, (list, tuple)): return [_norm_text(str(x)) for x in gt if str(x).strip()] if isinstance(gt, str): return [_norm_text(gt)] return [] def score_style(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Style accuracy = 1.0 if pred style matches gt style (tolerant via labels_match), else 0.0. Layout/symmetry accuracy reported as side metrics when the GT carries them. Robust: never raises; guards missing/short fields and non-dict GT.""" def _acc(pred_val, gt_val): if gt_val is None: return None gv = _norm_text(str(gt_val)) if not gv: return None pv = _norm_text(str(pred_val if pred_val is not None else "")) if not pv: return 0.0 return 1.0 if (pv == gv or labels_match(pv, gv)) else 0.0 if not isinstance(gt, dict): gt = {"style": gt} style_acc = _acc(pred.get("style"), gt.get("style")) metrics: dict = {} if style_acc is not None: metrics["style_acc"] = style_acc layout_acc = _acc(pred.get("layout"), gt.get("layout")) if layout_acc is not None: metrics["layout_acc"] = layout_acc symmetry_acc = _acc(pred.get("symmetry"), gt.get("symmetry")) if symmetry_acc is not None: metrics["symmetry_acc"] = symmetry_acc # Headline accuracy is style accuracy (the category's defining axis). primary = style_acc if style_acc is not None else None return primary, metrics def score_schema_only(pred: dict, gt, ctx) -> tuple[Optional[float], dict]: """Stub categories: no GT wired yet → no task accuracy, only universal metrics.""" return None, {} _SCORERS: dict[str, Callable] = { "classification": score_classification, "detection": score_detection, "ocr": score_ocr, "datatype_diff": score_datatype_diff, "datatype_util": score_datatype_util, "triples": score_triples, "depth_order": score_depth_order, "subject_fixation": score_subject_fixation, "segmentation": score_segmentation, "outline_iou": score_outline_iou, "iou3d": score_iou3d, "angular_error": score_camera_rotation, "vqa": score_vqa, "style": score_style, "schema_only": score_schema_only, } # ────────────────────────────────────────────────────────────────────────────── # Sample + run scoring # ────────────────────────────────────────────────────────────────────────────── def score_vision_sample( spec: VisionTaskSpec, raw_output: str, gt, *, mode: str, image_id: str, image_size: tuple[int, int], grammar_conformant: bool = False, n_output_tokens: int = 0, gen_seconds: float = 0.0, ) -> MetricResult: parse = parse_against(raw_output, model_for(spec)) parse_ok = parse.schema_valid or (parse.error or "").startswith("schema:") if not parse.schema_valid or parse.parsed is None: return MetricResult( category=spec.category, image_id=image_id, mode=mode, parse_ok=parse_ok, schema_valid=False, needed_repair=parse.needed_repair, needed_structural_repair=parse.needed_structural_repair, grammar_conformant=grammar_conformant, primary_score=None, metrics={}, n_output_tokens=n_output_tokens, gen_seconds=gen_seconds, error=parse.error, ) pred = parse.parsed.model_dump() ctx = {"size": image_size, "coord_space": spec.coord_space} scorer = _SCORERS.get(spec.metric, score_schema_only) try: primary, m = scorer(pred, gt, ctx) except Exception as e: # a scorer bug must never crash a long run primary, m = None, {} return MetricResult( category=spec.category, image_id=image_id, mode=mode, parse_ok=True, schema_valid=True, needed_repair=parse.needed_repair, needed_structural_repair=parse.needed_structural_repair, grammar_conformant=grammar_conformant, primary_score=None, metrics={}, n_output_tokens=n_output_tokens, gen_seconds=gen_seconds, error=f"scorer error: {e}", ) return MetricResult( category=spec.category, image_id=image_id, mode=mode, parse_ok=True, schema_valid=True, needed_repair=parse.needed_repair, needed_structural_repair=parse.needed_structural_repair, grammar_conformant=grammar_conformant, primary_score=primary, metrics=m, n_output_tokens=n_output_tokens, gen_seconds=gen_seconds, ) def score_vision_run(results: list[MetricResult], *, model: str = "", reasoning: str = "", category: str = "", mode: str = "") -> VisionRunMetrics: n = len(results) if n == 0: return VisionRunMetrics(category, model, reasoning, mode, 0, 0.0, 0.0, False, None, {}, 0.0, 0.0, 0.0, None) valid = [r for r in results if r.schema_valid] schema_valid_rate = len(valid) / n json_robustness = sum(1 for r in results if r.json_robust) / n scored = [r for r in valid if r.primary_score is not None] has_task = bool(scored) primary_mean = (sum(r.primary_score for r in scored) / len(scored)) if scored else None # average the per-sample metric dicts metrics_mean: dict = {} keys = set() for r in scored: keys |= set(r.metrics.keys()) for k in keys: vals = [r.metrics[k] for r in scored if k in r.metrics and not math.isnan(r.metrics[k])] if vals: metrics_mean[k] = sum(vals) / len(vals) total_tokens = sum(r.n_output_tokens for r in results) total_secs = sum(r.gen_seconds for r in results) mean_tokens = total_tokens / n tok_per_sec = (total_tokens / total_secs) if total_secs > 0 else 0.0 return VisionRunMetrics( category=category or results[0].category, model=model, reasoning=reasoning, mode=mode or results[0].mode, n=n, schema_valid_rate=schema_valid_rate, json_robustness=json_robustness, has_task_score=has_task, primary_score_mean=primary_mean, metrics_mean=metrics_mean, mean_output_tokens=mean_tokens, total_gen_seconds=total_secs, tokens_per_sec=tok_per_sec, labeler_score=labeler_score(primary_mean, schema_valid_rate, json_robustness), )