MLBOX: debugging of mlbox prediction

src/aare/daq/mlbox.py

@@ -41,12 +41,22 @@ class MlBox:
 
     def get_response(self, image):
         # cv2.imwrite("/sls/mx/applications/logs/image.png", image)
-        # ok, buf = cv2.imencode(".png", image)
-        # if not ok:
-        #     raise RuntimeError("Failed to encode image")
-        # files={"file": ("image.png", buf.tobytes(), "image/png")}
-        files = {"file": ("image.raw", image.tobytes(), "application/octet-stream")}
+        ok, buf = cv2.imencode(".png", image)
+        if not ok:
+            raise RuntimeError("Failed to encode image")
+        files={"file": ("image.png", buf.tobytes(), "image/png")}
         response = requests.post(self.__url, files=files, timeout=10)
+        #height, width = image.shape[:2]
+        # files = {"file": ("image.raw", image.tobytes(), "application/octet-stream")}
+        # params = {
+        #     "raw_format": "binary",
+        #     "raw_width": 2040,
+        #     "raw_height": 2044,
+        #     "raw_channels": 3,
+        #     "raw_dtype": "uint8",
+        # }
+        #params = {"raw_width": width, "raw_height": height, "raw_format": "binary"}
+        #response = requests.post(self.__url, files=files, params=params, timeout=10)
         response.raise_for_status()
         logger.debug(response.text)
         return response.json()
@@ -333,4 +343,59 @@ class MlBox:
             key = f"{MLOutputModel.get_class_str(cls)}_{ordinal}"
             out.setdefault(key, []).append(m)
 
-        return out
+        return out
+
+    def predict_best_n_frames(self,
+                              get_image_func,
+                              n_frames: int = 5,
+                              preferred_class=None,
+                              overlap_with_pin: float | None = None,
+                              confidence_min: float | None = None) -> Optional[MLBoxModel]:
+        """
+        Request bounding boxes for the next N frames and return the first non-None box.
+
+        This is useful for filtering out detection noise by sampling multiple frames
+        and using the first successful detection.
+
+        Args:
+            get_image_func: Callable that returns the next image (e.g., camera.get_image)
+            n_frames: Number of frames to try. Default is 5.
+            preferred_class: Preferred class(es) for detection
+            overlap_with_pin: Overlap parameter for filtering
+            confidence_min: Minimum confidence threshold
+
+        Returns:
+            First non-None MLBoxModel found, or None if all frames failed.
+        """
+        boxes_found = []
+
+        for frame_idx in range(n_frames):
+            try:
+                image = get_image_func()
+                if image is None:
+                    logger.debug(f"Frame {frame_idx + 1}/{n_frames}: no image available")
+                    continue
+
+                box = self.predict(
+                    image,
+                    preferred_class=preferred_class,
+                    overlap_with_pin=overlap_with_pin,
+                    confidence_min=confidence_min
+                )
+
+                if box is not None:
+                    logger.info(f"ML detection successful at frame {frame_idx + 1}/{n_frames}: {box}")
+                    boxes_found.append(box)
+                    # Return immediately on first successful detection
+                    return box
+                else:
+                    logger.debug(f"Frame {frame_idx + 1}/{n_frames}: no detection")
+
+            except Exception as e:
+                logger.warning(f"Frame {frame_idx + 1}/{n_frames}: prediction error: {e}")
+                continue
+
+        # No successful detections in any frame
+        if not boxes_found:
+            logger.warning(f"No detections in any of {n_frames} frames")
+        return None