feat(web_ui): add expose_camera to expose an MJPEG stream

Author: robgee86

src/arduino/app_bricks/web_ui/web_ui.py

@@ -14,6 +14,7 @@
 from fastapi.responses import FileResponse
 from fastapi_socketio import SocketManager
 
+from arduino.app_peripherals.camera.base_camera import BaseCamera
 from arduino.app_utils import brick, Logger
 
 logger = Logger("WebUI")
@@ -175,6 +176,44 @@ def expose_api(self, method: str, path: str, function: Callable):
         """
         self.app.add_api_route(self._api_path_prefix + path, function, methods=[method])
 
+    def expose_camera(self, path: str, camera: BaseCamera, quality: int = 80):
+        """
+        Expose a camera stream at the specified URL path in MJPEG format.
+
+        This can be consumed for example using `<img>` tags in a web application.
+
+        Args:
+            path (str): URL path for the MJPEG stream endpoint.
+            camera (BaseCamera): A camera instance, will be started if not already running.
+            quality (int, optional): JPEG compression quality (0-100). Default: 80.
+        """
+        from fastapi.responses import StreamingResponse
+        from arduino.app_utils.image import compress_to_jpeg
+
+        if not camera.is_started:
+            camera.start()
+
+        def generate_frames():
+            try:
+                while True:
+                    frame = camera.capture()
+                    if frame is None:
+                        continue
+                    jpeg = compress_to_jpeg(frame, quality=quality)
+                    if jpeg is None:
+                        continue
+                    yield b"--frame\r\nContent-Type: image/jpeg\r\n\r\n" + jpeg.tobytes() + b"\r\n"
+            except Exception as e:
+                logger.error(f"Terminating stream on camera error: {e}")
+
+        def stream_route():
+            return StreamingResponse(
+                generate_frames(),
+                media_type="multipart/x-mixed-replace; boundary=frame",
+            )
+
+        self.expose_api("GET", path, stream_route)
+
     def on_connect(self, callback: Callable[[str], None]):
         """Register a callback for WebSocket connection events.
 

tests/arduino/app_bricks/web_ui/test_web_ui.py

@@ -79,3 +79,59 @@ def test_stop_sets_should_exit():
     ui._server = dummy_server
     ui.stop()
     assert dummy_server.should_exit is True
+
+
+def test_expose_camera_starts_camera_if_not_started():
+    from unittest.mock import Mock, patch
+    import numpy as np
+
+    ui = WebUI()
+    mock_camera = Mock()
+    mock_camera.is_started = False
+    mock_camera.capture = Mock(side_effect=[np.zeros((2, 2, 3), dtype=np.uint8), RuntimeError("end")])
+
+    with patch("arduino.app_utils.image.compress_to_jpeg", return_value=np.array([0], dtype=np.uint8)):
+        ui.expose_camera("/stream", mock_camera)
+        TestClient(ui.app, raise_server_exceptions=False).get("/stream")
+
+    mock_camera.start.assert_called_once()
+
+
+def test_expose_camera_streams_mjpeg_response():
+    from unittest.mock import Mock, patch
+    import numpy as np
+
+    ui = WebUI()
+    mock_camera = Mock()
+    mock_camera.is_started = True
+
+    fake_frame = np.zeros((2, 2, 3), dtype=np.uint8)
+    mock_camera.capture = Mock(side_effect=[fake_frame, RuntimeError("end")])
+
+    fake_jpeg = b"\xff\xd8jpeg"
+
+    with patch("arduino.app_utils.image.compress_to_jpeg", return_value=np.frombuffer(fake_jpeg, dtype=np.uint8)):
+        ui.expose_camera("/stream", mock_camera)
+        response = TestClient(ui.app).get("/stream")
+
+    assert response.status_code == 200
+    assert "multipart/x-mixed-replace" in response.headers["content-type"]
+    assert fake_jpeg in response.content
+
+
+def test_expose_camera_passes_quality_to_compress():
+    from unittest.mock import Mock, patch
+    import numpy as np
+
+    ui = WebUI()
+    mock_camera = Mock()
+    mock_camera.is_started = True
+
+    fake_frame = np.zeros((2, 2, 3), dtype=np.uint8)
+    mock_camera.capture = Mock(side_effect=[fake_frame, RuntimeError("end")])
+
+    with patch("arduino.app_utils.image.compress_to_jpeg", return_value=np.array([0], dtype=np.uint8)) as mock_compress:
+        ui.expose_camera("/stream", mock_camera, quality=95)
+        TestClient(ui.app).get("/stream")
+
+    mock_compress.assert_called_with(fake_frame, quality=95)