add predict_batch interface for chute template

miner.py

@@ -148,74 +148,73 @@ class Miner:
         keep.sort()
         return np.array(keep, dtype=np.intp)
 
-    def run(self, frames: list[ndarray]) -> list[TVFrameResult]:
-        results: list[TVFrameResult] = []
-        for frame_id, frame in enumerate(frames):
-            if frame is None:
-                results.append(TVFrameResult(frame_id=frame_id, boxes=[], keypoints=[]))
+    def _predict_one(self, frame: ndarray) -> list[BoundingBox]:
+        x, scale, (dx, dy), (W, H) = self._preprocess(frame)
+        out = self.session.run(None, {self.input_name: x})[0]
+        # output shape: [1, 300, 6] — (x1, y1, x2, y2, conf, cls_id)
+        raw = out[0]
+        if raw.shape[0] == 0:
+            return []
+
+        # Apply conf filter (keep raw for fallback)
+        primary = raw[raw[:, 4] >= self.conf_thresh]
+
+        # Per-class dedup on the conf-filtered set
+        final_dets = []
+        if len(primary) > 0:
+            xyxy = primary[:, :4].astype(np.float32)
+            scores = primary[:, 4].astype(np.float32)
+            cls_ids = primary[:, 5].astype(np.int32)
+            keep_idx = self._dedup(xyxy, scores, cls_ids)
+            primary = primary[keep_idx]
+            for det in primary:
+                final_dets.append(det)
+
+        # Fallback: nothing left → return single highest-conf raw box (above floor)
+        if not final_dets and raw.shape[0] > 0:
+            top = raw[np.argmax(raw[:, 4])]
+            if top[4] >= self.fallback_min_conf:
+                final_dets.append(top)
+
+        # Build BoundingBox list with un-letterbox + cls remap
+        boxes_out: list[BoundingBox] = []
+        for det in final_dets:
+            x1, y1, x2, y2, conf, model_cls_id = det
+            x1 = (x1 - dx) / scale; x2 = (x2 - dx) / scale
+            y1 = (y1 - dy) / scale; y2 = (y2 - dy) / scale
+            x1 = max(0.0, min(W - 1.0, x1)); x2 = max(0.0, min(W - 1.0, x2))
+            y1 = max(0.0, min(H - 1.0, y1)); y2 = max(0.0, min(H - 1.0, y2))
+            if x2 <= x1 or y2 <= y1:
                 continue
-
-            x, scale, (dx, dy), (W, H) = self._preprocess(frame)
+            mapped_cls = int(self.cls_remap[int(model_cls_id)])
+            boxes_out.append(BoundingBox(
+                x1=int(x1), y1=int(y1), x2=int(x2), y2=int(y2),
+                cls_id=mapped_cls, conf=float(conf),
+            ))
+        return boxes_out
+
+    def predict_batch(
+        self,
+        batch_images: list[ndarray],
+        offset: int,
+        n_keypoints: int,
+    ) -> list[TVFrameResult]:
+        """Required interface for chute template (sv_chutes_*.py)."""
+        results: list[TVFrameResult] = []
+        for frame_number_in_batch, image in enumerate(batch_images):
             try:
-                out = self.session.run(None, {self.input_name: x})[0]
+                boxes = self._predict_one(image)
             except Exception as e:
-                print(f"frame {frame_id} ONNX error: {e}")
-                results.append(TVFrameResult(frame_id=frame_id, boxes=[], keypoints=[]))
-                continue
-
-            # output shape: [1, 300, 6] — (x1, y1, x2, y2, conf, cls_id)
-            raw = out[0]
-            if raw.shape[0] == 0:
-                results.append(TVFrameResult(frame_id=frame_id, boxes=[], keypoints=[]))
-                continue
-
-            # Apply conf filter (keep raw for fallback)
-            mask_conf = raw[:, 4] >= self.conf_thresh
-            primary = raw[mask_conf]
-
-            # Per-class dedup on the conf-filtered set
-            final_boxes = []
-            if len(primary) > 0:
-                xyxy = primary[:, :4].astype(np.float32)
-                scores = primary[:, 4].astype(np.float32)
-                cls_ids = primary[:, 5].astype(np.int32)
-                keep_idx = self._dedup(xyxy, scores, cls_ids)
-                primary = primary[keep_idx]
-
-                for det in primary:
-                    final_boxes.append(det)
-
-            # Fallback: nothing left → return the single highest-conf raw box (if above floor)
-            if not final_boxes:
-                if raw.shape[0] > 0:
-                    top = raw[np.argmax(raw[:, 4])]
-                    if top[4] >= self.fallback_min_conf:
-                        final_boxes.append(top)
-
-            # Build BoundingBox list with un-letterbox + remap cls_id
-            boxes_out: list[BoundingBox] = []
-            for det in final_boxes:
-                x1, y1, x2, y2, conf, model_cls_id = det
-                # un-letterbox: subtract padding then scale back
-                x1 = (x1 - dx) / scale
-                x2 = (x2 - dx) / scale
-                y1 = (y1 - dy) / scale
-                y2 = (y2 - dy) / scale
-                # clip
-                x1 = max(0.0, min(W - 1.0, x1))
-                x2 = max(0.0, min(W - 1.0, x2))
-                y1 = max(0.0, min(H - 1.0, y1))
-                y2 = max(0.0, min(H - 1.0, y2))
-                if x2 <= x1 or y2 <= y1:
-                    continue
-                # cls remap: v8 model order → validator-visible order
-                mapped_cls = int(self.cls_remap[int(model_cls_id)])
-                boxes_out.append(BoundingBox(
-                    x1=int(x1), y1=int(y1), x2=int(x2), y2=int(y2),
-                    cls_id=mapped_cls, conf=float(conf),
-                ))
-
+                print(f"⚠️ Inference failed for frame "
+                      f"{offset + frame_number_in_batch}: {e}")
+                boxes = []
             results.append(TVFrameResult(
-                frame_id=frame_id, boxes=boxes_out, keypoints=[],
+                frame_id=offset + frame_number_in_batch,
+                boxes=boxes,
+                keypoints=[(0, 0) for _ in range(max(0, int(n_keypoints)))],
             ))
         return results
+
+    # Back-compat alias for local sanity testing
+    def run(self, frames: list[ndarray]) -> list[TVFrameResult]:
+        return self.predict_batch(frames, offset=0, n_keypoints=0)