perf(csi): request YUV420 main stream when supported (-28% to -39% CPU)

docs/adr/002_main_stream_pixel_format.md

@@ -0,0 +1,50 @@
+# 002 - Main Stream Pixel Format for CSI Cameras
+
+## Date
+
+2026-05-31
+
+## Status
+
+Decision
+
+## Category
+
+Performance
+
+## Authors
+
+@antoinecellerier
+
+## References
+
+[picamera2 V4L2 encoder source](https://github.com/raspberrypi/picamera2/blob/main/picamera2/encoders/v4l2_encoder.py)
+
+## Context
+
+`picamera2.create_video_configuration()` defaults the main stream to `XBGR8888`
+(32 bits/pixel). Both the V4L2 hardware encoders (`MJPEGEncoder`,
+`H264Encoder`) and the `JpegEncoder` software fallback accept `YUV420`
+(12 bits/pixel) as a native input. Forcing `YUV420` upstream of the encoders
+skips an implicit colour-space conversion and reduces DMA bandwidth by ~2.7×.
+
+## Options
+
+1. Keep the picamera2 default (`XBGR8888`).
+2. Request `YUV420` for the main stream, falling back to the picamera2 default
+   when the camera does not advertise it.
+
+## Decision
+
+We request `YUV420` for the main stream when libcamera reports it as supported,
+selected from a preference-ordered list of encoder-compatible formats with
+`YUV420` first. If none are supported, we omit `format` from the config and
+defer to the picamera2 default.
+
+## Consequences
+
+* Measured CPU on a Pi 4B with Camera Module 3 at 1920×1080 @ 30 fps drops by
+  ~28–39 % (idle and live WebRTC respectively) versus the `XBGR8888` default.
+* No expected image-quality impact: JPEG and H.264 both encode in YUV
+  internally, so this change moves the colour-space conversion earlier in the
+  pipeline rather than introducing a new one.

spyglass/camera/camera.py

@@ -56,12 +56,18 @@ def configure(
             vflip=int(flip_vertical or upsidedown),
         )
 
+        main_cfg = self._main_stream_config(width, height)
         self.picam2.configure(
             self.picam2.create_video_configuration(
-                main={"size": (width, height)}, controls=controls, transform=transform
+                main=main_cfg, controls=controls, transform=transform
             )
         )
 
+    def _main_stream_config(self, width: int, height: int) -> dict:
+        """Picamera2 main-stream config dict. Subclasses override to pick the
+        most efficient pixel format supported by their camera and encoders."""
+        return {"size": (width, height)}
+
     def _run_server(
         self,
         bind_address,

spyglass/camera/csi.py

@@ -1,14 +1,66 @@
 import io
 from threading import Condition
 
+import libcamera
 from picamera2.encoders import _hw_encoder_available
 from picamera2.outputs import FileOutput
 
-from spyglass import WEBRTC_ENABLED, camera
+from spyglass import WEBRTC_ENABLED, camera, logger
 from spyglass.server.http_server import StreamingHandler
 
+# Preference ordered pixel formats accepted by the picamera2 V4L2 HW and
+# JpegEncoder SW encoders.
+_PREFERRED_MAIN_STREAM_FORMATS = (
+    "YUV420",
+    "BGR888",
+    "RGB888",
+    "XBGR8888",
+    "XRGB8888",
+)
+
 
 class CSI(camera.Camera):
+    def _main_stream_config(self, width: int, height: int) -> dict:
+        cfg = super()._main_stream_config(width, height)
+        chosen = self._pick_main_stream_format()
+        if chosen is not None:
+            cfg["format"] = chosen
+        return cfg
+
+    def _pick_main_stream_format(self) -> str | None:
+        """Return the highest-priority encoder-compatible format the camera
+        actually supports, or ``None`` to defer to the picamera2 default."""
+        supported = self._enumerate_supported_main_stream_formats()
+        if not supported:
+            return None
+
+        preferred_fmts = (f for f in _PREFERRED_MAIN_STREAM_FORMATS if f in supported)
+        fmt = next(preferred_fmts, None)
+
+        logger.info("Supported formats: %s", sorted(supported))
+        if fmt is None:
+            logger.warning(
+                "Camera reports no encoder-compatible main-stream formats; using picamera2 default.",
+            )
+            return None
+
+        logger.info(f"Main stream using %r.", fmt)
+        return fmt
+
+    def _enumerate_supported_main_stream_formats(self) -> set[str]:
+        try:
+            libcamera_cfg = self.picam2.camera.generate_configuration(
+                [libcamera.StreamRole.VideoRecording]
+            )
+            return {str(pf) for pf in libcamera_cfg.at(0).formats.pixel_formats}
+        except Exception as exc:
+            logger.warning(
+                "Could not enumerate supported main-stream formats from libcamera "
+                "(%s); using picamera2 default.",
+                exc,
+            )
+            return set()
+
     def start_and_run_server(
         self,
         bind_address,

tests/test_camera_configure.py

@@ -0,0 +1,119 @@
+import sys
+from unittest.mock import MagicMock
+
+AF_MODE_ENUM_MANUAL = 3
+AF_SPEED_ENUM_NORMAL = 1
+
+mock_libcamera = MagicMock()
+mock_picamera2 = MagicMock()
+mock_picamera2.encoders._hw_encoder_available = False
+mock_picamera2.outputs.Output = MagicMock
+sys.modules.update(
+    {
+        "libcamera": mock_libcamera,
+        "picamera2": mock_picamera2,
+        "picamera2.encoders": mock_picamera2.encoders,
+        "picamera2.outputs": mock_picamera2.outputs,
+    }
+)
+mock_libcamera.controls.AfModeEnum.Manual = AF_MODE_ENUM_MANUAL
+mock_libcamera.controls.AfSpeedEnum.Normal = AF_SPEED_ENUM_NORMAL
+mock_libcamera.StreamRole.VideoRecording = object()
+
+
+def _make_picam2(supported_formats=[], enumerate_raises=None):
+    """Build a mocked Picamera2 whose libcamera-level
+    ``camera.generate_configuration(...)`` reports the given pixel formats."""
+    picam2 = MagicMock()
+    picam2.camera_controls = {}
+    picam2.create_video_configuration.side_effect = lambda **kw: dict(kw)
+
+    if enumerate_raises is not None:
+        picam2.camera.generate_configuration.side_effect = enumerate_raises
+    else:
+        formats = MagicMock()
+        formats.pixel_formats = supported_formats
+        stream_cfg = MagicMock()
+        stream_cfg.formats = formats
+        libcam_cfg = MagicMock()
+        libcam_cfg.at.return_value = stream_cfg
+        picam2.camera.generate_configuration.return_value = libcam_cfg
+    return picam2
+
+
+def _run_configure(cam):
+    cam.configure(
+        width=1920,
+        height=1080,
+        fps=30,
+        autofocus=AF_MODE_ENUM_MANUAL,
+        lens_position=0.0,
+        autofocus_speed=AF_SPEED_ENUM_NORMAL,
+    )
+
+
+def test_csi_picks_yuv420_when_supported():
+    from spyglass.camera.csi import CSI
+
+    picam2 = _make_picam2(
+        supported_formats=["YUV420", "XBGR8888", "BGR888", "NV12", "RGB565"]
+    )
+    cam = CSI(picam2)
+    _run_configure(cam)
+
+    main = picam2.create_video_configuration.call_args.kwargs["main"]
+    assert main == {"size": (1920, 1080), "format": "YUV420"}
+
+
+def test_csi_falls_through_preference_when_yuv420_missing():
+    from spyglass.camera.csi import CSI
+
+    picam2 = _make_picam2(supported_formats=["XBGR8888", "BGR888", "RGB565"])
+    cam = CSI(picam2)
+    _run_configure(cam)
+
+    main = picam2.create_video_configuration.call_args.kwargs["main"]
+    assert main == {"size": (1920, 1080), "format": "BGR888"}
+
+
+def test_csi_omits_format_when_no_encoder_compatible_supported():
+    from spyglass.camera.csi import CSI
+
+    picam2 = _make_picam2(supported_formats=["NV12", "NV21", "RGB565", "YUYV"])
+    cam = CSI(picam2)
+    _run_configure(cam)
+
+    main = picam2.create_video_configuration.call_args.kwargs["main"]
+    assert "format" not in main
+    assert main == {"size": (1920, 1080)}
+
+
+def test_csi_omits_format_when_enumeration_fails():
+    from spyglass.camera.csi import CSI
+
+    picam2 = _make_picam2(enumerate_raises=RuntimeError("camera not yet acquired"))
+    cam = CSI(picam2)
+    _run_configure(cam)
+
+    main = picam2.create_video_configuration.call_args.kwargs["main"]
+    assert "format" not in main
+
+
+def test_base_camera_does_not_query_libcamera_formats():
+    """Subclasses other than CSI (e.g. USB) should not auto-pick a format."""
+    from spyglass.camera.camera import Camera
+
+    class _StubCamera(Camera):
+        def start_and_run_server(self, *args, **kwargs):
+            raise NotImplementedError
+
+        def stop(self):
+            raise NotImplementedError
+
+    picam2 = _make_picam2(supported_formats=["YUV420"])
+    cam = _StubCamera(picam2)
+    _run_configure(cam)
+
+    picam2.camera.generate_configuration.assert_not_called()
+    main = picam2.create_video_configuration.call_args.kwargs["main"]
+    assert "format" not in main