diff --git a/.basedpyright/baseline.json b/.basedpyright/baseline.json index 06a39d376..896f01e0d 100644 --- a/.basedpyright/baseline.json +++ b/.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 @@ } ] } -} +} \ No newline at end of file diff --git a/pimm/logging.py b/pimm/logging.py index 99d628a69..748337053 100644 --- a/pimm/logging.py +++ b/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 ) diff --git a/positronic/drivers/camera/linux_video.py b/positronic/drivers/camera/linux_video.py index c982f10bc..3d04544b7 100644 --- a/positronic/drivers/camera/linux_video.py +++ b/positronic/drivers/camera/linux_video.py @@ -1,3 +1,4 @@ +import logging from collections.abc import Iterator import av @@ -10,8 +11,20 @@ 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 @@ -19,20 +32,48 @@ def __init__(self, device_path: str, width: int, height: int, fps: int, pixel_fo 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() diff --git a/positronic/drivers/camera/tests/conftest.py b/positronic/drivers/camera/tests/conftest.py new file mode 100644 index 000000000..7cb3a48ef --- /dev/null +++ b/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}) diff --git a/positronic/drivers/camera/tests/test_linux_video.py b/positronic/drivers/camera/tests/test_linux_video.py new file mode 100644 index 000000000..16e730b91 --- /dev/null +++ b/positronic/drivers/camera/tests/test_linux_video.py @@ -0,0 +1,179 @@ +"""What the Linux video driver puts on its frame port, from the buffers a device hands it.""" + +import io + +import av +import numpy as np +import pytest + +import pimm +from positronic.drivers.camera import linux_video +from positronic.tests.testing_coutils import RecordingEmitter + +WIDTH, HEIGHT = 4, 2 + + +class StopFlag(pimm.SignalReceiver[bool]): + """``should_stop`` under the test's control.""" + + def __init__(self): + self.stopped = False + + def read(self) -> pimm.Message[bool]: + return pimm.Message(self.stopped) + + +class FakeFrame: + def __init__(self, data: bytes, pixel_format): + self.data = data + self.pixel_format = pixel_format + self.width, self.height = WIDTH, HEIGHT + + +class FakeDevice: + """A device that hands over the frames a test gives it, and records what it was set to.""" + + to_serve: list['FakeFrame'] = [] + opened: 'FakeDevice | None' = None + + def __init__(self, path: str): + self.path = path + self.info = type('Info', (), {'buffers': ['capture']})() + self.frames = list(FakeDevice.to_serve) + self.format = None + self.fps = None + self.closed = False + + def open(self) -> None: + FakeDevice.opened = self + + def set_format(self, buffer, width, height, pixel_format) -> None: + self.format = (buffer, width, height, pixel_format) + + def set_fps(self, buffer, fps) -> None: + self.fps = (buffer, fps) + + def __iter__(self): + return iter(self.frames) + + def close(self) -> None: + self.closed = True + + +@pytest.fixture +def device(monkeypatch): + """The class the driver opens, replaced by one a test hands frames to. Pass it to ``_driven``.""" + monkeypatch.setattr(linux_video, 'Device', FakeDevice) + FakeDevice.to_serve, FakeDevice.opened = [], None + return FakeDevice + + +def _driven(device, frames): + """A driver over ``device`` carrying ``frames``, with its port recorded, run to exhaustion.""" + camera = linux_video.LinuxVideo(device_path='/dev/null', width=WIDTH, height=HEIGHT, fps=30, pixel_format='YUYV') + emitted = RecordingEmitter() + camera.frame._bind(emitted) + device.to_serve = frames + list(camera.run(StopFlag(), pimm.world.SystemClock())) + opened = device.opened + assert opened is not None, 'the driver opened no device' + return emitted, opened + + +def _yuyv(luma: int) -> bytes: + """One YUYV buffer of a flat grey, which converts to a flat grey RGB image.""" + return bytes([luma, 128] * (WIDTH * HEIGHT)) + + +def test_a_yuyv_buffer_reaches_the_port_as_an_image(device): + emitted, _ = _driven(device, [FakeFrame(_yuyv(200), linux_video.PixelFormat.YUYV)]) + + assert len(emitted.emitted) == 1 + _, adapter = emitted.emitted[0] + assert adapter.array.shape == (HEIGHT, WIDTH, 3) + assert adapter.array.dtype == np.uint8 + assert adapter.array.min() > 150 # the grey survives the conversion + + +def test_every_whole_yuyv_buffer_is_one_frame(device): + frames = [FakeFrame(_yuyv(v), linux_video.PixelFormat.YUYV) for v in (50, 120, 200)] + + emitted, _ = _driven(device, frames) + + assert len(emitted.emitted) == 3 + + +def test_the_device_is_set_to_what_the_driver_was_asked_for(device): + _, opened = _driven(device, [FakeFrame(_yuyv(100), linux_video.PixelFormat.YUYV)]) + + assert opened.format == ('capture', WIDTH, HEIGHT, 'YUYV') + assert opened.fps == ('capture', 30) + + +def test_the_device_is_closed_when_the_frames_run_out(device): + _, opened = _driven(device, [FakeFrame(_yuyv(100), linux_video.PixelFormat.YUYV)]) + + assert opened.closed + + +def test_a_buffer_that_is_not_one_frame_in_size_is_dropped(device, caplog): + """A busy bus hands over a buffer that is not a whole frame, and a run outlives it.""" + misframed = FakeFrame(_yuyv(200)[: WIDTH * HEIGHT], linux_video.PixelFormat.YUYV) + whole = FakeFrame(_yuyv(200), linux_video.PixelFormat.YUYV) + + emitted, _ = _driven(device, [misframed, whole, misframed]) + + assert len(emitted.emitted) == 1 + assert 'not one frame in size' in caplog.text + assert 'handed over 2 buffers' in caplog.text + + +def test_a_buffer_of_three_bytes_a_pixel_is_taken_as_it_is(device): + raw = bytes(range(WIDTH * HEIGHT * 3)) + + emitted, _ = _driven(device, [FakeFrame(raw, linux_video.PixelFormat.RGB24)]) + + _, adapter = emitted.emitted[0] + np.testing.assert_array_equal(adapter.array, np.frombuffer(raw, dtype=np.uint8).reshape(HEIGHT, WIDTH, 3)) + + +def _h264(count: int) -> bytes: + """An H.264 stream of ``count`` frames, as a camera that compresses would hand it over.""" + stream = io.BytesIO() + container = av.open(stream, 'w', format='h264') + encoded = container.add_stream('libx264', rate=30) + encoded.width, encoded.height, encoded.pix_fmt = WIDTH * 16, HEIGHT * 16, 'yuv420p' + encoded.options = {'tune': 'zerolatency'} # a camera streams frames in order, and so must the fixture + for i in range(count): + image = np.full((HEIGHT * 16, WIDTH * 16, 3), i * 40, np.uint8) + for packet in encoded.encode(av.VideoFrame.from_ndarray(image, format='rgb24')): + container.mux(packet) + for packet in encoded.encode(None): + container.mux(packet) + container.close() + return stream.getvalue() + + +def test_a_compressed_buffer_the_decoder_holds_back_is_not_counted_misframed(device, caplog): + """The decoder gives no image until it has one, and a whole buffer is not a truncated one.""" + head = FakeFrame(_h264(4)[:64], linux_video.PixelFormat.H264) + + emitted, _ = _driven(device, [head]) + + assert emitted.emitted == [] + assert 'not one frame in size' not in caplog.text + + +def test_a_buffer_of_several_images_gives_the_newest(device, caplog): + """The frame port holds one image, so the older images of a buffer have nowhere to go.""" + buffer = FakeFrame(_h264(4), linux_video.PixelFormat.H264) + driver = linux_video.LinuxVideo(device_path='/dev/null', width=WIDTH, height=HEIGHT, fps=30, pixel_format='YUYV') + decoded = driver._images(buffer, lambda name: av.CodecContext.create(name, 'r')) + assert decoded is not None and len(decoded) > 1, 'the buffer decoded to a single image' + + emitted, _ = _driven(device, [FakeFrame(_h264(4), linux_video.PixelFormat.H264)]) + + assert len(emitted.emitted) == 1 + _, adapter = emitted.emitted[0] + assert adapter.array.mean() == pytest.approx(decoded[-1].mean()) + assert f'{len(decoded) - 1} images a newer one' in caplog.text