scorevision: push artifact

miner.py

@@ -1,5 +1,6 @@
-"""ScoreVision crime detector v24 — YOLOv11s trained on rival-consensus labels.
-Per-class conf thresholds tuned vs validator-PGT proxy; flip TTA; cross-class NMS."""
+"""ScoreVision crime detector v27 — YOLOv11s trained on 4-voter consensus
+(new vmodel0, hermes_sv, pmodel_9, SAM3-refined; >=2 agree, IoU>=0.5).
+Per-class confidence + per-class rescue (bonus) logic, flip TTA, cross-class NMS."""
 from pathlib import Path
 import math
 
@@ -26,17 +27,23 @@ class TVFrameResult(BaseModel):
 
 
 class Miner:
-    """ONNX Runtime miner with horizontal-flip TTA and per-class confidence."""
+    """YOLOv11s + per-class threshold + per-class rescue + flip TTA + cross-class dedup."""
 
+    # validator output order
     class_names = ["balaclava", "hoodie", "glove", "bat", "spray paint", "graffiti"]
     input_size = 1280
-    iou_thres = 0.45
-    cross_iou_thresh = 0.50
+    iou_thres = 0.40
+    cross_iou_thresh = 0.70
     max_aspect_ratio = 10.0
     max_det = 150
-    # per-class thresholds — tuned on rival-consensus GT (proxy for validator PGT)
+    # tuned on consensus-GT, healthy distribution: [balaclava, hoodie, glove, bat, spray, graffiti]
     _conf_thres_array = np.array(
-        [0.30, 0.70, 0.60, 0.50, 0.40, 0.40], dtype=np.float32
+        [0.60, 0.70, 0.70, 0.80, 0.50, 0.20], dtype=np.float32
+    )
+    # per-class rescue gap: if a class has no box above threshold, admit top-1
+    # if its score >= (threshold - bonus). Inspired by new-vmodel0's _bonus_array.
+    _bonus_array = np.array(
+        [0.20, 0.25, 0.20, 0.30, 0.20, 0.15], dtype=np.float32
     )
 
     def __init__(self, path_hf_repo: Path) -> None:
@@ -47,25 +54,22 @@ class Miner:
             print("preload_dlls success")
         except Exception as e:
             print(f"preload_dlls failed: {e}")
-        print("ORT available providers:", ort.get_available_providers())
+        print("ORT providers:", ort.get_available_providers())
 
-        sess_options = ort.SessionOptions()
-        sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
+        opts = ort.SessionOptions()
+        opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
         try:
             self.session = ort.InferenceSession(
-                str(model_path),
-                sess_options=sess_options,
+                str(model_path), sess_options=opts,
                 providers=["CUDAExecutionProvider", "CPUExecutionProvider"],
             )
-            print("Created ORT session with preferred CUDA provider list")
+            print("CUDA session created")
         except Exception as e:
-            print(f"CUDA session creation failed, falling back to CPU: {e}")
+            print(f"CUDA failed, CPU fallback: {e}")
             self.session = ort.InferenceSession(
-                str(model_path),
-                sess_options=sess_options,
-                providers=["CPUExecutionProvider"],
+                str(model_path), sess_options=opts, providers=["CPUExecutionProvider"]
             )
-        print("ORT session providers:", self.session.get_providers())
+        print("session providers:", self.session.get_providers())
 
         for inp in self.session.get_inputs():
             print("INPUT:", inp.name, inp.shape, inp.type)
@@ -74,50 +78,43 @@ class Miner:
 
         self.input_name = self.session.get_inputs()[0].name
         self.output_names = [o.name for o in self.session.get_outputs()]
-        self.input_shape = self.session.get_inputs()[0].shape
-        self.input_height = self._safe_dim(self.input_shape[2], self.input_size)
-        self.input_width = self._safe_dim(self.input_shape[3], self.input_size)
-        print(f"ONNX loaded: {model_path}  input={self.input_shape}")
+        sh = self.session.get_inputs()[0].shape
+        self.input_height = self._safe_dim(sh[2], self.input_size)
+        self.input_width = self._safe_dim(sh[3], self.input_size)
+        print(f"ONNX loaded: {model_path}")
         print(
             "per-class conf: "
-            + ", ".join(
-                f"{n}={t:.3f}" for n, t in zip(self.class_names, self._conf_thres_array.tolist())
-            )
+            + ", ".join(f"{n}={t:.2f}" for n, t in zip(self.class_names, self._conf_thres_array.tolist()))
+        )
+        print(
+            "per-class rescue bonus: "
+            + ", ".join(f"{n}={t:.2f}" for n, t in zip(self.class_names, self._bonus_array.tolist()))
         )
-
-    def __repr__(self) -> str:
-        return f"ONNXRuntime(providers={self.session.get_providers()})"
 
     @staticmethod
-    def _safe_dim(value, default: int) -> int:
-        return value if isinstance(value, int) and value > 0 else default
+    def _safe_dim(v, d):
+        return v if isinstance(v, int) and v > 0 else d
 
-    def _letterbox(self, image: ndarray, new_shape: tuple[int, int],
-                   color=(114, 114, 114)
-                   ) -> tuple[ndarray, float, tuple[float, float]]:
+    def _letterbox(self, image, new_shape, color=(114, 114, 114)):
         h, w = image.shape[:2]
-        new_w, new_h = new_shape
-        ratio = min(new_w / w, new_h / h)
-        rw, rh = int(round(w * ratio)), int(round(h * ratio))
+        nw, nh = new_shape
+        r = min(nw / w, nh / h)
+        rw, rh = int(round(w * r)), int(round(h * r))
         if (rw, rh) != (w, h):
-            interp = cv2.INTER_CUBIC if ratio > 1.0 else cv2.INTER_LINEAR
+            interp = cv2.INTER_CUBIC if r > 1.0 else cv2.INTER_LINEAR
             image = cv2.resize(image, (rw, rh), interpolation=interp)
-        dw, dh = (new_w - rw) / 2.0, (new_h - rh) / 2.0
+        dw, dh = (nw - rw) / 2.0, (nh - rh) / 2.0
         top, bot = int(round(dh - 0.1)), int(round(dh + 0.1))
         lt, rt = int(round(dw - 0.1)), int(round(dw + 0.1))
-        padded = cv2.copyMakeBorder(
-            image, top, bot, lt, rt, cv2.BORDER_CONSTANT, value=color
-        )
-        return padded, ratio, (dw, dh)
+        padded = cv2.copyMakeBorder(image, top, bot, lt, rt, cv2.BORDER_CONSTANT, value=color)
+        return padded, r, (dw, dh)
 
-    def _preprocess(self, image: ndarray):
-        orig_h, orig_w = image.shape[:2]
+    def _preprocess(self, image):
+        h, w = image.shape[:2]
         img, ratio, pad = self._letterbox(image, (self.input_width, self.input_height))
-        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
-        img = img.astype(np.float32) / 255.0
+        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
         img = np.transpose(img, (2, 0, 1))[None, ...]
-        img = np.ascontiguousarray(img, dtype=np.float32)
-        return img, ratio, pad, (orig_w, orig_h)
+        return np.ascontiguousarray(img, dtype=np.float32), ratio, pad, (w, h)
 
     @staticmethod
     def _clip(boxes, size):
@@ -131,33 +128,27 @@ class Miner:
     @staticmethod
     def _hard_nms(boxes, scores, iou_thr):
         n = len(boxes)
-        if n == 0:
-            return np.array([], dtype=np.intp)
+        if n == 0: return np.array([], dtype=np.intp)
         order = np.argsort(-scores)
         keep = []
         while len(order) > 0:
-            i = int(order[0])
-            keep.append(i)
-            if len(order) == 1:
-                break
+            i = int(order[0]); keep.append(i)
+            if len(order) == 1: break
             rest = order[1:]
             xx1 = np.maximum(boxes[i, 0], boxes[rest, 0])
             yy1 = np.maximum(boxes[i, 1], boxes[rest, 1])
             xx2 = np.minimum(boxes[i, 2], boxes[rest, 2])
             yy2 = np.minimum(boxes[i, 3], boxes[rest, 3])
-            inter = np.maximum(0.0, xx2 - xx1) * np.maximum(0.0, yy2 - yy1)
-            a_i = max(0.0, boxes[i, 2] - boxes[i, 0]) * max(0.0, boxes[i, 3] - boxes[i, 1])
-            a_r = (
-                np.maximum(0.0, boxes[rest, 2] - boxes[rest, 0])
-                * np.maximum(0.0, boxes[rest, 3] - boxes[rest, 1])
-            )
-            iou = inter / (a_i + a_r - inter + 1e-7)
+            inter = np.maximum(0.0, xx2-xx1)*np.maximum(0.0, yy2-yy1)
+            a_i = max(0.0, boxes[i, 2]-boxes[i, 0])*max(0.0, boxes[i, 3]-boxes[i, 1])
+            a_r = (np.maximum(0.0, boxes[rest, 2]-boxes[rest, 0])
+                   * np.maximum(0.0, boxes[rest, 3]-boxes[rest, 1]))
+            iou = inter / (a_i+a_r-inter+1e-7)
             order = rest[iou <= iou_thr]
         return np.array(keep, dtype=np.intp)
 
     def _per_class_hard_nms(self, boxes, scores, cls_ids, iou_thr):
-        if len(boxes) == 0:
-            return np.array([], dtype=np.intp)
+        if len(boxes) == 0: return np.array([], dtype=np.intp)
         keep_all = []
         for c in np.unique(cls_ids):
             mask = cls_ids == c
@@ -169,68 +160,73 @@ class Miner:
 
     def _cross_class_dedup(self, boxes, scores, cls_ids, iou_thr):
         n = len(boxes)
-        if n == 0:
-            return np.array([], dtype=np.intp)
+        if n == 0: return np.array([], dtype=np.intp)
         order = np.argsort(-scores)
-        keep = []
-        suppressed = np.zeros(n, dtype=bool)
+        keep = []; suppressed = np.zeros(n, dtype=bool)
         for i in order:
-            if suppressed[i]:
-                continue
+            if suppressed[i]: continue
             keep.append(int(i))
             ix1 = np.maximum(boxes[i, 0], boxes[:, 0])
             iy1 = np.maximum(boxes[i, 1], boxes[:, 1])
             ix2 = np.minimum(boxes[i, 2], boxes[:, 2])
             iy2 = np.minimum(boxes[i, 3], boxes[:, 3])
-            inter = np.maximum(0.0, ix2 - ix1) * np.maximum(0.0, iy2 - iy1)
-            a_i = max(0.0, boxes[i, 2] - boxes[i, 0]) * max(0.0, boxes[i, 3] - boxes[i, 1])
-            a_r = (
-                np.maximum(0.0, boxes[:, 2] - boxes[:, 0])
-                * np.maximum(0.0, boxes[:, 3] - boxes[:, 1])
-            )
-            iou = inter / (a_i + a_r - inter + 1e-7)
+            inter = np.maximum(0.0, ix2-ix1)*np.maximum(0.0, iy2-iy1)
+            a_i = max(0.0, boxes[i, 2]-boxes[i, 0])*max(0.0, boxes[i, 3]-boxes[i, 1])
+            a_r = (np.maximum(0.0, boxes[:, 2]-boxes[:, 0])
+                   * np.maximum(0.0, boxes[:, 3]-boxes[:, 1]))
+            iou = inter / (a_i+a_r-inter+1e-7)
             iou[i] = 0.0
             suppressed |= iou >= iou_thr
         return np.array(keep, dtype=np.intp)
 
     def _filter_sane(self, boxes, scores, cls_ids, orig_size):
-        if len(boxes) == 0:
-            return boxes, scores, cls_ids
+        if len(boxes) == 0: return boxes, scores, cls_ids
         w, h = orig_size
         area_img = float(w * h)
-        bw = np.maximum(0.0, boxes[:, 2] - boxes[:, 0])
-        bh = np.maximum(0.0, boxes[:, 3] - boxes[:, 1])
+        bw = np.maximum(0.0, boxes[:, 2]-boxes[:, 0])
+        bh = np.maximum(0.0, boxes[:, 3]-boxes[:, 1])
         area = bw * bh
-        ar = np.where(
-            (bw > 0) & (bh > 0),
-            np.maximum(bw / np.maximum(bh, 1e-6), bh / np.maximum(bw, 1e-6)),
-            np.inf,
-        )
+        ar = np.where((bw > 0) & (bh > 0),
+                      np.maximum(bw / np.maximum(bh, 1e-6), bh / np.maximum(bw, 1e-6)),
+                      np.inf)
         keep = (area <= 0.95 * area_img) & (ar <= self.max_aspect_ratio)
         return boxes[keep], scores[keep], cls_ids[keep]
 
-    def _decode(self, preds: ndarray, ratio: float, pad: tuple[float, float],
-                orig_size: tuple[int, int]) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
+    def _conf_filter_with_rescue(self, boxes, scores, cls_ids):
+        """Per-class threshold + per-class rescue (admit top-1 if score >= thr - bonus
+        and no box of that class passed the normal threshold)."""
+        if len(scores) == 0:
+            return np.zeros(0, dtype=bool)
+        thr = self._conf_thres_array[cls_ids]
+        keep = scores >= thr
+        for c in np.unique(cls_ids):
+            bonus = float(self._bonus_array[c])
+            if bonus <= 0.0: continue
+            cm = cls_ids == c
+            if keep[cm].any(): continue
+            idx = np.where(cm)[0]
+            top = int(idx[int(np.argmax(scores[idx]))])
+            if scores[top] >= self._conf_thres_array[c] - bonus:
+                keep[top] = True
+        return keep
+
+    def _decode(self, preds, ratio, pad, orig_size):
         if preds.ndim == 3 and preds.shape[0] == 1:
             preds = preds[0]
         if preds.ndim != 2 or preds.shape[1] < 6:
-            return (
-                np.empty((0, 4), dtype=np.float32),
-                np.empty((0,), dtype=np.float32),
-                np.empty((0,), dtype=np.int32),
-            )
+            return (np.empty((0, 4), dtype=np.float32),
+                    np.empty((0,), dtype=np.float32),
+                    np.empty((0,), dtype=np.int32))
         boxes = preds[:, :4].astype(np.float32)
         scores = preds[:, 4].astype(np.float32)
         cls_ids = preds[:, 5].astype(np.int32)
         keep = (scores > 0) & (cls_ids >= 0) & (cls_ids < len(self.class_names))
         boxes, scores, cls_ids = boxes[keep], scores[keep], cls_ids[keep]
-        if len(boxes) == 0:
-            return boxes, scores, cls_ids
-        thr = self._conf_thres_array[cls_ids]
-        keep = scores >= thr
+        if len(boxes) == 0: return boxes, scores, cls_ids
+        # per-class threshold WITH rescue
+        keep = self._conf_filter_with_rescue(boxes, scores, cls_ids)
         boxes, scores, cls_ids = boxes[keep], scores[keep], cls_ids[keep]
-        if len(boxes) == 0:
-            return boxes, scores, cls_ids
+        if len(boxes) == 0: return boxes, scores, cls_ids
         pad_w, pad_h = pad
         boxes[:, [0, 2]] -= pad_w
         boxes[:, [1, 3]] -= pad_h
@@ -238,12 +234,12 @@ class Miner:
         boxes = self._clip(boxes, orig_size)
         return boxes, scores, cls_ids
 
-    def _predict_single(self, image: ndarray):
+    def _predict_single(self, image):
         x, ratio, pad, orig_size = self._preprocess(image)
         out = self.session.run(self.output_names, {self.input_name: x})[0]
         return self._decode(out, ratio, pad, orig_size)
 
-    def _predict_tta(self, image: ndarray) -> list[BoundingBox]:
+    def _predict_tta(self, image):
         b0, s0, c0 = self._predict_single(image)
         flipped = cv2.flip(image, 1)
         bf, sf, cf = self._predict_single(flipped)
@@ -256,48 +252,39 @@ class Miner:
         boxes = np.concatenate([b0, bf], axis=0) if len(b0) or len(bf) else b0
         scores = np.concatenate([s0, sf], axis=0) if len(b0) or len(bf) else s0
         cls_ids = np.concatenate([c0, cf], axis=0) if len(b0) or len(bf) else c0
-        if len(boxes) == 0:
-            return []
-        boxes, scores, cls_ids = self._filter_sane(
-            boxes, scores, cls_ids, (image.shape[1], image.shape[0])
-        )
-        if len(boxes) == 0:
-            return []
+        if len(boxes) == 0: return []
+        boxes, scores, cls_ids = self._filter_sane(boxes, scores, cls_ids,
+                                                    (image.shape[1], image.shape[0]))
+        if len(boxes) == 0: return []
         keep = self._per_class_hard_nms(boxes, scores, cls_ids, self.iou_thres)
-        if len(keep) == 0:
-            return []
+        if len(keep) == 0: return []
         boxes, scores, cls_ids = boxes[keep], scores[keep], cls_ids[keep]
         keep = self._cross_class_dedup(boxes, scores, cls_ids, self.cross_iou_thresh)
         boxes, scores, cls_ids = boxes[keep], scores[keep], cls_ids[keep]
         if len(scores) > self.max_det:
-            top = np.argsort(-scores)[: self.max_det]
+            top = np.argsort(-scores)[:self.max_det]
             boxes, scores, cls_ids = boxes[top], scores[top], cls_ids[top]
         return [
             BoundingBox(
-                x1=int(math.floor(b[0])),
-                y1=int(math.floor(b[1])),
-                x2=int(math.ceil(b[2])),
-                y2=int(math.ceil(b[3])),
-                cls_id=int(c),
-                conf=float(s),
+                x1=int(math.floor(b[0])), y1=int(math.floor(b[1])),
+                x2=int(math.ceil(b[2])), y2=int(math.ceil(b[3])),
+                cls_id=int(c), conf=float(s),
             )
             for b, s, c in zip(boxes, scores, cls_ids)
             if b[2] > b[0] and b[3] > b[1]
         ]
 
-    def predict_batch(self, batch_images: list[ndarray], offset: int,
-                      n_keypoints: int) -> list[TVFrameResult]:
-        results: list[TVFrameResult] = []
-        for j, image in enumerate(batch_images):
+    def predict_batch(self, batch_images, offset, n_keypoints):
+        results = []
+        for j, img in enumerate(batch_images):
             try:
-                boxes = self._predict_tta(image)
+                boxes = self._predict_tta(img)
             except Exception as e:
                 print(f"Inference failed for frame {offset + j}: {e}")
                 boxes = []
             results.append(
                 TVFrameResult(
-                    frame_id=offset + j,
-                    boxes=boxes,
+                    frame_id=offset + j, boxes=boxes,
                     keypoints=[(0, 0) for _ in range(max(0, int(n_keypoints)))],
                 )
             )