scorevision: push artifact

miner.py

@@ -0,0 +1,185 @@
+"""TurboVision miner for Detect-petrol-station-1-0.
+
+YOLOv11s ONNX FP16 + NMS baked in, with horizontal-flip TTA to boost recall.
+4 classes: 0=petrol hose, 1=petrol pump, 2=price board, 3=roof canopy.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import List, Tuple
+
+import cv2
+import numpy as np
+import onnxruntime as ort
+from pydantic import BaseModel
+
+
+class BoundingBox(BaseModel):
+    x1: int
+    y1: int
+    x2: int
+    y2: int
+    cls_id: int
+    conf: float
+
+
+class TVFrameResult(BaseModel):
+    frame_id: int
+    boxes: list[BoundingBox]
+    keypoints: list[tuple[int, int]]
+
+
+class Miner:
+    IMGSZ = 1280
+    # Per-class conf thresholds: 0=petrol hose, 1=petrol pump, 2=price board, 3=roof canopy.
+    # Tuned via greedy grid search on 100 fresh challenges vs real SAM3 pseudo-GT.
+    CLASS_CONF_THRES = (0.43, 0.63, 0.37, 0.41)
+    CONF_THRES = 0.37  # fallback / pre-filter at lowest per-class threshold
+    IOU_THRES = 0.45
+    NUM_CLASSES = 4
+    MIN_BOX_FRAC = 0.005
+    USE_TTA = True
+
+    def __init__(self, path_hf_repo: Path) -> None:
+        self.onnx_path = path_hf_repo / 'weights.onnx'
+        if not self.onnx_path.exists():
+            raise FileNotFoundError(f'Model not found at {self.onnx_path}')
+
+        providers: list = []
+        try:
+            ort.preload_dlls()
+        except Exception:
+            pass
+        available = ort.get_available_providers()
+        if 'CUDAExecutionProvider' in available:
+            providers.append(('CUDAExecutionProvider', {'device_id': 0}))
+        providers.append('CPUExecutionProvider')
+        so = ort.SessionOptions()
+        so.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
+        self.session = ort.InferenceSession(str(self.onnx_path), sess_options=so, providers=providers)
+        self.input_name = self.session.get_inputs()[0].name
+        inp = self.session.get_inputs()[0]
+        self.input_shape = inp.shape
+        self.input_dtype = np.float16 if inp.type == 'tensor(float16)' else np.float32
+        print(f'[Miner] Loaded {self.onnx_path.name} | providers={self.session.get_providers()} | dtype={self.input_dtype}')
+
+    def __repr__(self) -> str:
+        return f'PetrolMiner(yolo11s-onnx-fp16-nms, tta={self.USE_TTA}, conf={self.CONF_THRES})'
+
+    @staticmethod
+    def _letterbox(img, new_size=1280, color=(114, 114, 114)):
+        h, w = img.shape[:2]
+        r = min(new_size / h, new_size / w)
+        nh, nw = int(round(h * r)), int(round(w * r))
+        resized = cv2.resize(img, (nw, nh), interpolation=cv2.INTER_LINEAR)
+        top = (new_size - nh) // 2
+        bottom = new_size - nh - top
+        left = (new_size - nw) // 2
+        right = new_size - nw - left
+        padded = cv2.copyMakeBorder(resized, top, bottom, left, right, cv2.BORDER_CONSTANT, value=color)
+        return padded, r, (left, top)
+
+    def _preprocess(self, img):
+        h, w = img.shape[:2]
+        img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
+        padded, r, (lx, ty) = self._letterbox(img_rgb, self.IMGSZ)
+        x = padded.astype(self.input_dtype) / 255.0
+        x = x.transpose(2, 0, 1)[None, ...]
+        return np.ascontiguousarray(x), r, (lx, ty), (w, h)
+
+    def _run_onnx(self, img):
+        x, r, (lx, ty), (W, H) = self._preprocess(img)
+        outputs = self.session.run(None, {self.input_name: x})
+        det = outputs[0]
+        if det.ndim == 3: det = det[0]
+        if det.size == 0: return [], [], [], W, H
+        det = np.asarray(det, dtype=np.float32)
+        if det.shape[-1] < 6: return [], [], [], W, H
+        xyxy = det[:, :4].copy()
+        conf = det[:, 4].copy()
+        cls_id = det[:, 5].astype(int)
+        keep = conf >= self.CONF_THRES
+        xyxy, conf, cls_id = xyxy[keep], conf[keep], cls_id[keep]
+        if len(xyxy) == 0: return [], [], [], W, H
+        xyxy[:, [0, 2]] = (xyxy[:, [0, 2]] - lx) / r
+        xyxy[:, [1, 3]] = (xyxy[:, [1, 3]] - ty) / r
+        xyxy[:, 0::2] = np.clip(xyxy[:, 0::2], 0, W - 1)
+        xyxy[:, 1::2] = np.clip(xyxy[:, 1::2], 0, H - 1)
+        min_side = self.MIN_BOX_FRAC * min(W, H)
+        mask = (
+            (cls_id >= 0) & (cls_id < self.NUM_CLASSES)
+            & ((xyxy[:, 2] - xyxy[:, 0]) >= min_side)
+            & ((xyxy[:, 3] - xyxy[:, 1]) >= min_side)
+        )
+        return xyxy[mask], conf[mask], cls_id[mask], W, H
+
+    @staticmethod
+    def _hard_nms_per_class(xyxy, conf, cls_id, iou_thres=0.5):
+        if len(xyxy) == 0: return np.empty((0,), dtype=int)
+        keep = []
+        for c in np.unique(cls_id):
+            idx = np.where(cls_id == c)[0]
+            b = xyxy[idx]; s = conf[idx]
+            order = np.argsort(-s)
+            b = b[order]; s = s[order]; idx = idx[order]
+            areas = (b[:, 2] - b[:, 0]) * (b[:, 3] - b[:, 1])
+            suppressed = np.zeros(len(b), dtype=bool)
+            for i in range(len(b)):
+                if suppressed[i]: continue
+                keep.append(idx[i])
+                xx1 = np.maximum(b[i, 0], b[i+1:, 0])
+                yy1 = np.maximum(b[i, 1], b[i+1:, 1])
+                xx2 = np.minimum(b[i, 2], b[i+1:, 2])
+                yy2 = np.minimum(b[i, 3], b[i+1:, 3])
+                inter = np.maximum(0, xx2 - xx1) * np.maximum(0, yy2 - yy1)
+                iou = inter / (areas[i] + areas[i+1:] - inter + 1e-9)
+                suppressed[i+1:][iou > iou_thres] = True
+        return np.array(keep, dtype=int)
+
+    def _predict_single(self, img):
+        xyxy1, conf1, cls1, W, H = self._run_onnx(img)
+        if not self.USE_TTA:
+            xyxy, conf, cls_id = xyxy1, conf1, cls1
+        else:
+            img_f = cv2.flip(img, 1)
+            xyxy2, conf2, cls2, _, _ = self._run_onnx(img_f)
+            if len(xyxy2) > 0:
+                tmp = xyxy2.copy()
+                tmp[:, 0] = W - xyxy2[:, 2]
+                tmp[:, 2] = W - xyxy2[:, 0]
+                xyxy2 = tmp
+            pieces_xyxy = [a for a in (xyxy1, xyxy2) if len(a) > 0]
+            pieces_conf = [a for a in (conf1, conf2) if len(a) > 0]
+            pieces_cls  = [a for a in (cls1,  cls2)  if len(a) > 0]
+            xyxy   = np.vstack(pieces_xyxy) if pieces_xyxy else np.empty((0, 4))
+            conf   = np.concatenate(pieces_conf) if pieces_conf else np.empty((0,))
+            cls_id = np.concatenate(pieces_cls) if pieces_cls else np.empty((0,))
+            if len(xyxy) > 0:
+                keep = self._hard_nms_per_class(xyxy, conf, cls_id, iou_thres=self.IOU_THRES)
+                xyxy, conf, cls_id = xyxy[keep], conf[keep], cls_id[keep]
+        boxes = []
+        for i in range(len(xyxy)):
+            ci = int(cls_id[i])
+            if 0 <= ci < self.NUM_CLASSES and float(conf[i]) < self.CLASS_CONF_THRES[ci]:
+                continue
+            boxes.append(BoundingBox(
+                x1=int(round(float(xyxy[i, 0]))),
+                y1=int(round(float(xyxy[i, 1]))),
+                x2=int(round(float(xyxy[i, 2]))),
+                y2=int(round(float(xyxy[i, 3]))),
+                cls_id=ci,
+                conf=float(conf[i]),
+            ))
+        return boxes
+
+    def predict_batch(self, batch_images, offset, n_keypoints):
+        results = []
+        for i, img in enumerate(batch_images):
+            try:
+                boxes = self._predict_single(img)
+            except Exception as e:
+                print(f'[Miner] predict error on frame {offset + i}: {e}')
+                boxes = []
+            kps = [(0, 0) for _ in range(n_keypoints)]
+            results.append(TVFrameResult(frame_id=offset + i, boxes=boxes, keypoints=kps))
+        return results