refactor: image manipulation

robgee86 · robgee86 · commit 20e21a313861 · 2025-11-04T16:51:16.000+01:00
diff --git a/src/arduino/app_bricks/camera_code_detection/__init__.py b/src/arduino/app_bricks/camera_code_detection/__init__.py
@@ -2,7 +2,6 @@
 #
 # SPDX-License-Identifier: MPL-2.0
 
-from .detection import Detection, CameraCodeDetection
-from .utils import draw_bounding_boxes, draw_bounding_box
+from .detection import CameraCodeDetection, Detection
 
-__all__ = ["CameraCodeDetection", "Detection", "draw_bounding_boxes", "draw_bounding_box"]
+__all__ = ["CameraCodeDetection", "Detection"]
diff --git a/src/arduino/app_bricks/camera_code_detection/detection.py b/src/arduino/app_bricks/camera_code_detection/detection.py
@@ -11,7 +11,7 @@
 import numpy as np
 from PIL.Image import Image
 
-from arduino.app_peripherals.usb_camera import USBCamera
+from arduino.app_peripherals.camera import Camera
 from arduino.app_utils import brick, Logger
 
 logger = Logger("CameraCodeDetection")
@@ -55,7 +55,7 @@ class CameraCodeDetection:
 
     def __init__(
         self,
-        camera: USBCamera = None,
+        camera: Camera = None,
         detect_qr: bool = True,
         detect_barcode: bool = True,
     ):
@@ -76,7 +76,7 @@ def __init__(
 
         self.already_seen_codes = set()
 
-        self._camera = camera if camera else USBCamera()
+        self._camera = camera if camera else Camera()
 
     def start(self):
         """Start the detector and begin scanning for codes."""
@@ -147,7 +147,7 @@ def on_error(self, callback: Callable[[Exception], None] | None):
     def loop(self):
         """Main loop to capture frames and detect codes."""
         try:
-            frame = self._camera.capture_bytes()
+            frame = self._camera.capture()
             if frame is None:
                 return
         except Exception as e:
diff --git a/src/arduino/app_bricks/object_detection/README.md b/src/arduino/app_bricks/object_detection/README.md
@@ -23,23 +23,24 @@ The Object Detection Brick allows you to:
 ```python
 import os
 from arduino.app_bricks.object_detection import ObjectDetection
+from arduino.app_utils.image import draw_bounding_boxes
 
 object_detection = ObjectDetection()
 
-# Image frame can be as bytes or PIL image
-frame = os.read("path/to/your/image.jpg")
+# Image can be provided as bytes or PIL.Image
+img = os.read("path/to/your/image.jpg")
 
-out = object_detection.detect(frame)
-# is it possible to customize image type, confidence level and box overlap
-# out = object_detection.detect(frame, image_type = "png", confidence = 0.35, overlap = 0.5)
+out = object_detection.detect(img)
+# You can also provide a confidence level
+# out = object_detection.detect(frame, confidence = 0.35)
 if out and "detection" in out:
     for i, obj_det in enumerate(out["detection"]):
-        # For every object detected, get its details
+        # For every object detected, print its details
         detected_object = obj_det.get("class_name", None)
-        bounding_box = obj_det.get("bounding_box_xyxy", None)
         confidence = obj_det.get("confidence", None)
+        bounding_box = obj_det.get("bounding_box_xyxy", None)
 
-# draw the bounding box and key points on the image
-out_image = object_detection.draw_bounding_boxes(frame, out)
+# Draw the bounding boxes
+out_image = draw_bounding_boxes(img, out)
 ```
 
diff --git a/src/arduino/app_bricks/object_detection/__init__.py b/src/arduino/app_bricks/object_detection/__init__.py
@@ -2,8 +2,7 @@
 #
 # SPDX-License-Identifier: MPL-2.0
 
-from PIL import Image
-from arduino.app_utils import brick, Logger, draw_bounding_boxes
+from arduino.app_utils import brick, Logger
 from arduino.app_internal.core import EdgeImpulseRunnerFacade
 
 logger = Logger("ObjectDetection")
@@ -54,19 +53,6 @@ def detect(self, image_bytes, image_type: str = "jpg", confidence: float = None)
         ret = super().infer_from_image(image_bytes, image_type)
         return self._extract_detection(ret, confidence)
 
-    def draw_bounding_boxes(self, image: Image.Image | bytes, detections: dict) -> Image.Image | None:
-        """Draw bounding boxes on an image enclosing detected objects using PIL.
-
-        Args:
-            image: The input image to annotate. Can be a PIL Image object or raw image bytes.
-            detections: Detection results containing object labels and bounding boxes.
-
-        Returns:
-            Image with bounding boxes and key points drawn.
-            None if no detection or invalid image.
-        """
-        return draw_bounding_boxes(image, detections)
-
     def _extract_detection(self, item, confidence: float = None):
         if not item:
             return None
diff --git a/src/arduino/app_bricks/object_detection/examples/object_detection_example.py b/src/arduino/app_bricks/object_detection/examples/object_detection_example.py
@@ -3,23 +3,24 @@
 # SPDX-License-Identifier: MPL-2.0
 
 # EXAMPLE_NAME = "Object Detection"
+import os
 from arduino.app_bricks.object_detection import ObjectDetection
+from arduino.app_utils.image import draw_bounding_boxes
 
 object_detection = ObjectDetection()
 
-# Image frame can be as bytes or PIL image
-with open("image.png", "rb") as f:
-    frame = f.read()
+# Image can be provided as bytes or PIL.Image
+img = os.read("path/to/your/image.jpg")
 
-out = object_detection.detect(frame)
-# is it possible to customize image type, confidence level and box overlap
-# out = object_detection.detect(frame, image_type = "png", confidence = 0.35, overlap = 0.5)
+out = object_detection.detect(img)
+# You can also provide a confidence level
+# out = object_detection.detect(frame, confidence = 0.35)
 if out and "detection" in out:
     for i, obj_det in enumerate(out["detection"]):
-        # For every object detected, get its details
+        # For every object detected, print its details
         detected_object = obj_det.get("class_name", None)
-        bounding_box = obj_det.get("bounding_box_xyxy", None)
         confidence = obj_det.get("confidence", None)
+        bounding_box = obj_det.get("bounding_box_xyxy", None)
 
-# draw the bounding box and key points on the image
-out_image = object_detection.draw_bounding_boxes(frame, out)
+# Draw the bounding boxes
+out_image = draw_bounding_boxes(img, out)
diff --git a/src/arduino/app_bricks/video_imageclassification/brick_compose.yaml b/src/arduino/app_bricks/video_imageclassification/brick_compose.yaml
@@ -13,7 +13,7 @@ services:
     volumes:
       - "${CUSTOM_MODEL_PATH:-/home/arduino/.arduino-bricks/ei-models/}:${CUSTOM_MODEL_PATH:-/home/arduino/.arduino-bricks/ei-models/}"
       - "/run/udev:/run/udev"
-    command: ["--model-file", "${EI_CLASSIFICATION_MODEL:-/models/ootb/ei/mobilenet-v2-224px.eim}", "--dont-print-predictions", "--mode", "streaming", "--preview-original-resolution", "--camera", "${VIDEO_DEVICE:-/dev/video1}"]
+    command: ["--model-file", "${EI_CLASSIFICATION_MODEL:-/models/ootb/ei/mobilenet-v2-224px.eim}", "--dont-print-predictions", "--mode", "streaming", "--preview-original-resolution", "--gst-launch-args", "tcpserversrc host=0.0.0.0 port=5000 ! jpegdec ! videoconvert ! video/x-raw ! jpegenc"]
     healthcheck:
       test: [ "CMD-SHELL", "wget -q --spider http://ei-video-classification-runner:4912 || exit 1" ]
       interval: 2s
diff --git a/src/arduino/app_bricks/video_objectdetection/brick_compose.yaml b/src/arduino/app_bricks/video_objectdetection/brick_compose.yaml
@@ -13,7 +13,7 @@ services:
     volumes:
       - "${CUSTOM_MODEL_PATH:-/home/arduino/.arduino-bricks/ei-models/}:${CUSTOM_MODEL_PATH:-/home/arduino/.arduino-bricks/ei-models/}"
       - "/run/udev:/run/udev"
-    command: ["--model-file", "${EI_OBJ_DETECTION_MODEL:-/models/ootb/ei/yolo-x-nano.eim}", "--dont-print-predictions", "--mode", "streaming", "--force-target", "--preview-original-resolution", "--camera", "${VIDEO_DEVICE:-/dev/video1}"]
+    command: ["--model-file", "${EI_OBJ_DETECTION_MODEL:-/models/ootb/ei/yolo-x-nano.eim}", "--dont-print-predictions", "--mode", "streaming", "--preview-original-resolution", "--gst-launch-args", "tcpserversrc host=0.0.0.0 port=5000 ! jpegdec ! videoconvert ! video/x-raw ! jpegenc"]
     healthcheck:
       test: [ "CMD-SHELL", "wget -q --spider http://ei-video-obj-detection-runner:4912 || exit 1" ]
       interval: 2s
diff --git a/src/arduino/app_bricks/visual_anomaly_detection/examples/object_detection_example.py b/src/arduino/app_bricks/visual_anomaly_detection/examples/object_detection_example.py
diff --git a/src/arduino/app_bricks/visual_anomaly_detection/examples/visual_anomaly_example.py b/src/arduino/app_bricks/visual_anomaly_detection/examples/visual_anomaly_example.py
@@ -0,0 +1,24 @@
+# SPDX-FileCopyrightText: Copyright (C) 2025 ARDUINO SA <http://www.arduino.cc>
+#
+# SPDX-License-Identifier: MPL-2.0
+
+# EXAMPLE_NAME = "Visual Anomaly Detection"
+import os
+from arduino.app_bricks.visual_anomaly_detection import VisualAnomalyDetection
+from arduino.app_utils.image import draw_anomaly_markers
+
+anomaly_detection = VisualAnomalyDetection()
+
+# Image can be provided as bytes or PIL.Image
+img = os.read("path/to/your/image.jpg")
+
+out = anomaly_detection.detect(img)
+if out and "detection" in out:
+    for i, anomaly in enumerate(out["detection"]):
+        # For every anomaly detected, print its details
+        detected_anomaly = anomaly.get("class_name", None)
+        score = anomaly.get("score", None)
+        bounding_box = anomaly.get("bounding_box_xyxy", None)
+
+# Draw the bounding boxes
+out_image = draw_anomaly_markers(img, out)
diff --git a/src/arduino/app_peripherals/camera/examples/websocket_camera_proxy.py b/src/arduino/app_peripherals/camera/examples/websocket_camera_proxy.py
@@ -9,7 +9,7 @@
 
 This example demonstrates how to use a WebSocketCamera as a proxy/relay.
 It receives frames from clients on one WebSocket server (127.0.0.1:8080) and
-forwards them as raw JPEG binary data to a TCP server (127.0.0.1:5001) at 30fps.
+forwards them as raw JPEG binary data to a TCP server (127.0.0.1:5000) at 30fps.
 
 Usage:
     python websocket_camera_proxy.py [--input-port PORT] [--output-host HOST] [--output-port PORT]
@@ -22,7 +22,6 @@
 import sys
 import time
 
-# Add the parent directory to the path to import from arduino package
 import os
 
 from arduino.app_peripherals.camera import Camera
@@ -173,17 +172,17 @@ async def main():
     global running, camera
     
     parser = argparse.ArgumentParser(description="WebSocket Camera Proxy")
-    parser.add_argument("--input-port", type=int, default=8080, 
+    parser.add_argument("--input-port", type=int, default=8080,
                        help="WebSocketCamera input port (default: 8080)")
-    parser.add_argument("--output-host", default="127.0.0.1", 
-                       help="Output TCP server host (default: 127.0.0.1)")
-    parser.add_argument("--output-port", type=int, default=5001, 
-                       help="Output TCP server port (default: 5001)")
-    parser.add_argument("--fps", type=int, default=30, 
+    parser.add_argument("--output-host", default="0.0.0.0",
+                       help="Output TCP server host (default: 0.0.0.0)")
+    parser.add_argument("--output-port", type=int, default=5000,
+                       help="Output TCP server port (default: 5000)")
+    parser.add_argument("--fps", type=int, default=30,
                        help="Target FPS for forwarding (default: 30)")
-    parser.add_argument("--quality", type=int, default=80, 
+    parser.add_argument("--quality", type=int, default=80,
                        help="JPEG quality 1-100 (default: 80)")
-    parser.add_argument("--verbose", "-v", action="store_true", 
+    parser.add_argument("--verbose", "-v", action="store_true",
                        help="Enable verbose logging")
     
     args = parser.parse_args()
@@ -204,7 +203,8 @@ async def main():
     logger.info(f"Output: TCP server at {args.output_host}:{args.output_port}")
     logger.info(f"Target FPS: {args.fps}")
     
-    camera = Camera("ws://0.0.0.0:5001")
+    from arduino.app_utils.image.image_editor import compressed_to_jpeg
+    camera = Camera("ws://0.0.0.0:5000", transformer=compressed_to_jpeg(80))
     
     try:
         # Start camera input and output connection tasks
diff --git a/src/arduino/app_utils/__init__.py b/src/arduino/app_utils/__init__.py
@@ -8,11 +8,9 @@
 from .bridge import *
 from .folderwatch import *
 from .httprequest import *
-from .image import *
 from .jsonparser import *
 from .logger import *
 from .slidingwindowbuffer import *
-from .userinput import *
 
 __all__ = [
     "App",
@@ -23,12 +21,8 @@
     "provide",
     "FolderWatcher",
     "HttpClient",
-    "draw_bounding_boxes",
-    "get_image_bytes",
-    "get_image_type",
     "JSONParser",
     "Logger",
     "SineGenerator",
     "SlidingWindowBuffer",
-    "UserTextInput",
 ]
diff --git a/src/arduino/app_utils/image/__init__.py b/src/arduino/app_utils/image/__init__.py
@@ -0,0 +1,18 @@
+from .image import *
+from .image_editor import ImageEditor
+from .pipeable import PipeableFunction, pipeable
+
+__all__ = [
+    "get_image_type",
+    "get_image_bytes",
+    "draw_bounding_boxes",
+    "draw_anomaly_markers",
+    "ImageEditor",
+    "pipeable",
+    "letterboxed",
+    "resized",
+    "adjusted",
+    "greyscaled",
+    "compressed_to_jpeg",
+    "compressed_to_png",
+]
diff --git a/src/arduino/app_utils/image/image.py b/src/arduino/app_utils/image/image.py
@@ -35,7 +35,7 @@ def _read(file_path: str) -> bytes:
         with open(file_path, "rb") as f:
             return f.read()
     except Exception as e:
-        logger(f"Error reading image: {e}")
+        logger.error(f"Error reading image: {e}")
         return None
 
 
@@ -78,23 +78,11 @@ def get_image_bytes(image: str | Image.Image | bytes) -> bytes:
         return None
 
 
-def draw_colored_dot(draw, x, y, color, size):
-    """Draws a large colored dot on a PIL Image at the specified coordinate.
-
-    Args:
-        draw: An ImageDraw object from PIL.
-        x: The x-coordinate of the center of the dot.
-        y: The y-coordinate of the center of the dot.
-        color: A color value that PIL understands (e.g., "red", (255, 0, 0), "#FF0000").
-        size: The radius of the dot (in pixels).
-    """
-    # Calculate the bounding box for the circle
-    bounding_box = (x - size, y - size, x + size, y + size)
-    # Draw a filled ellipse (which looks like a circle if the bounding box is a square)
-    draw.ellipse(bounding_box, fill=color)
-
-
-def draw_bounding_boxes(image: Image.Image | bytes, detection: dict, draw: ImageDraw.ImageDraw = None) -> Image.Image | None:
+def draw_bounding_boxes(
+    image: Image.Image | bytes,
+    detection: dict,
+    draw: ImageDraw.ImageDraw = None
+) -> Image.Image | None:
     """Draw bounding boxes on an image using PIL.
 
     The thickness of the box and font size are scaled based on image size.
@@ -181,7 +169,7 @@ def draw_bounding_boxes(image: Image.Image | bytes, detection: dict, draw: Image
 def draw_anomaly_markers(
     image: Image.Image | bytes,
     detection: dict,
-    draw: ImageDraw.ImageDraw = None,
+    draw: ImageDraw.ImageDraw = None
 ) -> Image.Image | None:
     """Draw bounding boxes on an image using PIL.
 
@@ -192,9 +180,6 @@ def draw_anomaly_markers(
         detection (dict): A dictionary containing detection results with keys 'class_name', 'bounding_box_xyxy', and
             'score'.
         draw (ImageDraw.ImageDraw, optional): An existing ImageDraw object to use. If None, a new one is created.
-        label_above_box (bool, optional): If True, labels are drawn above the bounding box. Defaults to False.
-        colours (list, optional): List of colors to use for bounding boxes. Defaults to a predefined palette.
-        text_color (str, optional): Color of the text labels. Defaults to "white".
     """
     if isinstance(image, bytes):
         image_box = Image.open(io.BytesIO(image))
diff --git a/src/arduino/app_utils/image/image_editor.py b/src/arduino/app_utils/image/image_editor.py
diff --git a/src/arduino/app_utils/image/pipeable.py b/src/arduino/app_utils/image/pipeable.py
diff --git a/src/arduino/app_utils/userinput.py b/src/arduino/app_utils/userinput.py
diff --git a/tests/arduino/app_bricks/imageclassification/test_imageclassification.py b/tests/arduino/app_bricks/imageclassification/test_imageclassification.py
@@ -16,7 +16,7 @@ def mock_dependencies(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setattr("arduino.app_internal.core.parse_docker_compose_variable", lambda x: [(None, None), (None, "8200")])
     # make get_image_bytes a no-op for raw bytes
     monkeypatch.setattr(
-        "arduino.app_utils.get_image_bytes",
+        "arduino.app_utils.image.get_image_bytes",
         lambda x: x if isinstance(x, (bytes, bytearray)) else None,
     )
 
diff --git a/tests/arduino/app_bricks/objectdetection/test_objectdetection.py b/tests/arduino/app_bricks/objectdetection/test_objectdetection.py
diff --git a/tests/arduino/app_core/test_edge_impulse.py b/tests/arduino/app_core/test_edge_impulse.py

Original file line number	Diff line number	Diff line change
`@@ -2,7 +2,6 @@`
`2`	`2`	`#`
`3`	`3`	`# SPDX-License-Identifier: MPL-2.0`
`4`	`4`
`5`		`-from .detection import Detection, CameraCodeDetection`
`6`		`-from .utils import draw_bounding_boxes, draw_bounding_box`
	`5`	`+from .detection import CameraCodeDetection, Detection`
`7`	`6`
`8`		`-__all__ = ["CameraCodeDetection", "Detection", "draw_bounding_boxes", "draw_bounding_box"]`
	`7`	`+__all__ = ["CameraCodeDetection", "Detection"]`