Give a Linux video frame the shape the other cameras give

.basedpyright/baseline.json

@@ -3330,38 +3330,6 @@
                     "endColumn": 29,
                     "lineCount": 1
                 }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 46,
-                    "endColumn": 50,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportAttributeAccessIssue",
-                "range": {
-                    "startColumn": 43,
-                    "endColumn": 49,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalSubscript",
-                "range": {
-                    "startColumn": 32,
-                    "endColumn": 38,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportReturnType",
-                "range": {
-                    "startColumn": 18,
-                    "endColumn": 30,
-                    "lineCount": 1
-                }
             }
         ],
         "./positronic/drivers/camera/luxonis.py": [
@@ -6977,4 +6945,4 @@
             }
         ]
     }
-}
+}

pimm/logging.py

@@ -39,6 +39,7 @@
     'boto3',
     's3transfer',  # per part of a multipart upload
     'asyncio',  # per selector event, under its debug mode
+    'linuxpy',  # per ioctl, which is several times a frame for every camera
 )
 
 

positronic/drivers/camera/linux_video.py

@@ -1,3 +1,4 @@
+import logging
 from collections.abc import Iterator
 
 import av
@@ -10,29 +11,69 @@
 with vendor_import('linuxpy', 'Linux video capture', platforms=('linux',)):
     from linuxpy.video.device import Device, PixelFormat
 
+logger = logging.getLogger(__name__)
+
 
 class LinuxVideo(pimm.ControlSystem):
+    # The formats a camera may compress in, and what decodes each of them.
+    _CODECS = {
+        PixelFormat.H264: 'h264',
+        PixelFormat.HEVC: 'hevc',
+        PixelFormat.VP8: 'vp8',
+        PixelFormat.VP9: 'vp9',
+        PixelFormat.MPEG4: 'mpeg4',
+        PixelFormat.MJPEG: 'mjpeg',
+    }
+
     def __init__(self, device_path: str, width: int, height: int, fps: int, pixel_format: str):
         self.device_path = device_path
         self.width = width
         self.height = height
         self.fps = fps
         self.pixel_format = pixel_format
         self.fps_counter = pimm.utils.RateCounter(f'LinuxVideo {device_path}')
-        self.frame: pimm.SignalEmitter = pimm.ControlSystemEmitter(self)
-
-    def run(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[pimm.Sleep]:  # noqa: C901
-        codec_mapping = {
-            PixelFormat.H264: 'h264',
-            PixelFormat.HEVC: 'hevc',
-            PixelFormat.VP8: 'vp8',
-            PixelFormat.VP9: 'vp9',
-            PixelFormat.MPEG4: 'mpeg4',
-            PixelFormat.MJPEG: 'mjpeg',
-        }
+        self.frame = pimm.ControlSystemEmitter[pimm.shared_memory.NumpySMAdapter](self)
+        self._frame_adapter = None  # Lazy init
+
+    @staticmethod
+    def _framed(data: np.ndarray, frame, bytes_per_pixel: int) -> np.ndarray | None:
+        """``data`` shaped as the frame it belongs to, or ``None`` where it is not one frame's worth."""
+        if data.size != frame.height * frame.width * bytes_per_pixel:
+            return None
+        return data.reshape((frame.height, frame.width, bytes_per_pixel))
+
+    def _images(self, frame, codec_context) -> list[np.ndarray] | None:
+        """Every image the buffer ``frame`` carries, as RGB, or ``None`` where it is not one frame's worth.
+
+        A compressed buffer whole enough to read may still carry no image: the parser and the decoder both
+        hold data back until they have a frame to give, so an empty list is a wait, not a loss.
+        """
+        data = np.frombuffer(frame.data, dtype=np.uint8)
+        match frame.pixel_format:
+            case PixelFormat.YUYV:
+                raw = self._framed(data, frame, 2)
+                return None if raw is None else [cv2.cvtColor(raw, cv2.COLOR_YUV2RGB_YUYV)]
+            case PixelFormat.UYVY:
+                raw = self._framed(data, frame, 2)
+                return None if raw is None else [cv2.cvtColor(raw, cv2.COLOR_YUV2RGB_UYVY)]
+            case _ if frame.pixel_format in self._CODECS:
+                codec_ctx = codec_context(self._CODECS[frame.pixel_format])
+                # `av` types what it parses as `bytes` and carries `decode` on the subclasses of
+                # `CodecContext`, so the buffer the device hands over and the base class both read wrong.
+                packets = codec_ctx.parse(data)  # pyright: ignore[reportArgumentType]
+                return [
+                    decoded.to_ndarray(format='rgb24')
+                    for packet in packets
+                    for decoded in codec_ctx.decode(packet)  # pyright: ignore[reportAttributeAccessIssue]
+                ]
+            case _:
+                raw = self._framed(data, frame, 3)  # assume 3 bytes per pixel (RGB/BGR)
+                return None if raw is None else [raw]
+
+    def run(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[pimm.Command]:
         codec_contexts = {}
 
-        def get_codec_context(codec_name: str) -> av.CodecContext:
+        def codec_context(codec_name: str) -> av.CodecContext:
             """Lazily initialize and return codec context for given codec"""
             if codec_name not in codec_contexts:
                 codec_contexts[codec_name] = av.CodecContext.create(codec_name, 'r')
@@ -44,40 +85,31 @@ def get_codec_context(codec_name: str) -> av.CodecContext:
         device.set_format(device.info.buffers[0], self.width, self.height, self.pixel_format)
         device.set_fps(device.info.buffers[0], self.fps)
 
+        misframed = overtaken = 0
         for frame in device:
             if should_stop.value:
                 break
 
-            data = np.frombuffer(frame.data, dtype=np.uint8)
-            result = None
-
-            match frame.pixel_format:
-                case PixelFormat.YUYV:
-                    data = data.reshape((frame.height, frame.width, 2))
-                    result = {'image': cv2.cvtColor(data, cv2.COLOR_YUV2RGB_YUYV)}
-                case PixelFormat.UYVY:
-                    data = data.reshape((frame.height, frame.width, 2))
-                    result = {'image': cv2.cvtColor(data, cv2.COLOR_YUV2RGB_UYVY)}
-                case _ if frame.pixel_format in codec_mapping:
-                    codec_name = codec_mapping[frame.pixel_format]
-                    codec_ctx = get_codec_context(codec_name)
-                    packets = codec_ctx.parse(data)
-                    for packet in packets:
-                        frames = codec_ctx.decode(packet)
-                        if len(frames) == 1:
-                            result = {'image': frames[0].to_ndarray(format='rgb24')}
-                        else:
-                            for i, decoded_frame in enumerate(frames):
-                                result[f'image_{i}'] = decoded_frame.to_ndarray(format='rgb24')
-                case _:
-                    # Assume 3 bytes per pixel (RGB/BGR)
-                    rgb_data = data.reshape((frame.height, frame.width, 3))
-                    result = {'image': rgb_data}
-
-            if result is not None:
-                self.frame.emit(result)
+            images = self._images(frame, codec_context)
+            # rules-allow: swallowed-error — a busy bus hands over a buffer that is not a whole frame as
+            # ordinary traffic; the device keeps streaming, the buffers after it read whole, and the count
+            # says how many went
+            if images is None:
+                misframed += 1
+                if misframed == 1:
+                    logger.warning('%s handed over a buffer that is not one frame in size', self.device_path)
+            elif images:
+                # The port holds one image and nothing runs between two emissions of the same tick, so a
+                # buffer decoding to several has only its newest to give; the rest are counted, not sent.
+                overtaken += len(images) - 1
+                self._frame_adapter = pimm.shared_memory.NumpySMAdapter.lazy_init(images[-1], self._frame_adapter)
+                self.frame.emit(self._frame_adapter)
                 self.fps_counter.tick()
 
             yield pimm.Yield()  # Give control back to the world
 
+        if misframed:
+            logger.warning('%s handed over %d buffers that are not one frame in size', self.device_path, misframed)
+        if overtaken:
+            logger.warning('%s decoded %d images a newer one of the same buffer overtook', self.device_path, overtaken)
         device.close()

positronic/drivers/camera/tests/conftest.py

@@ -0,0 +1,31 @@
+"""A stand-in for the vendor package the Linux video driver imports.
+
+``linuxpy`` ships only in the ``hardware`` extra, so the driver module cannot be imported from a default
+sync. The tests drive a device of their own, so a module carrying the names the driver binds at import is
+enough. Installed here, before any test module imports the driver, and only where the real package is
+absent.
+"""
+
+import importlib.util
+import sys
+import types
+from enum import Enum
+
+VENDOR = 'linuxpy'
+VIDEO_MODULE = f'{VENDOR}.video'
+DEVICE_MODULE = f'{VIDEO_MODULE}.device'
+
+if importlib.util.find_spec(VENDOR) is None:
+    # The formats the driver names. The values are the V4L2 four-character codes, as `linuxpy` reports them.
+    PixelFormat = Enum('PixelFormat', ['YUYV', 'UYVY', 'RGB24', 'H264', 'HEVC', 'VP8', 'VP9', 'MPEG4', 'MJPEG'])
+
+    device = types.ModuleType(DEVICE_MODULE)
+    device.__dict__.update(Device=object, PixelFormat=PixelFormat)
+
+    video = types.ModuleType(VIDEO_MODULE)
+    video.__dict__.update(device=device)
+
+    package = types.ModuleType(VENDOR)
+    package.__dict__.update(video=video)
+
+    sys.modules.update({VENDOR: package, VIDEO_MODULE: video, DEVICE_MODULE: device})