[libcamera-devel] [PATCH v4 14/16] py: examples: Add simple-continuous-capture.py

[Tomi Valkeinen]

Add a slightly more complex, and I think a more realistic, example,
where the script reacts to events and re-queues the buffers.

Signed-off-by: Tomi Valkeinen <tomi.valkeinen at ideasonboard.com>
---
 src/py/examples/simple-continuous-capture.py | 189 +++++++++++++++++++
 1 file changed, 189 insertions(+)
 create mode 100755 src/py/examples/simple-continuous-capture.py

diff --git a/src/py/examples/simple-continuous-capture.py b/src/py/examples/simple-continuous-capture.py
new file mode 100755
index 00000000..d0f8a7e9
--- /dev/null
+++ b/src/py/examples/simple-continuous-capture.py
@@ -0,0 +1,189 @@
+#!/usr/bin/env python3
+
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (C) 2022, Tomi Valkeinen <tomi.valkeinen at ideasonboard.com>
+
+# A simple capture example extending the simple-capture.py example:
+# - Capture frames using events from multiple cameras
+# - Listening events from stdin to exit the application
+# - Memory mapping the frames and calculating CRC
+
+import binascii
+import libcamera as libcam
+import libcamera.utils
+import selectors
+import sys
+
+
+# A container class for our state per camera
+class CameraCaptureContext:
+    idx: int
+    cam: libcam.Camera
+    reqs: list[libcam.Request]
+    mfbs: dict[libcam.FrameBuffer, libcamera.utils.MappedFrameBuffer]
+
+    def __init__(self, cam, idx):
+        # Acquire the camera for our use
+
+        ret = cam.acquire()
+        assert ret == 0
+
+        # Configure the camera
+
+        cam_config = cam.generate_configuration([libcam.StreamRole.Viewfinder])
+
+        stream_config = cam_config.at(0)
+
+        ret = cam.configure(cam_config)
+        assert ret == 0
+
+        stream = stream_config.stream
+
+        # Allocate the buffers for capture
+
+        allocator = libcam.FrameBufferAllocator(cam)
+        ret = allocator.allocate(stream)
+        assert ret > 0
+
+        num_bufs = len(allocator.buffers(stream))
+
+        print(f'cam{idx} ({cam.id}): capturing {num_bufs} buffers with {stream_config}')
+
+        # Create the requests and assign a buffer for each request
+
+        reqs = []
+        for i in range(num_bufs):
+            # Use the buffer index as the "cookie"
+            req = cam.create_request(idx)
+
+            buffer = allocator.buffers(stream)[i]
+            ret = req.add_buffer(stream, buffer)
+            assert ret == 0
+
+            reqs.append(req)
+
+        self.idx = idx
+        self.cam = cam
+        self.reqs = reqs
+        self.mfbs = dict([(fb, libcamera.utils.MappedFrameBuffer(fb).mmap()) for fb in allocator.buffers(stream)])
+
+    def uninit_camera(self):
+        # Stop the camera
+
+        ret = self.cam.stop()
+        assert ret == 0
+
+        # Release the camera
+
+        ret = self.cam.release()
+        assert ret == 0
+
+
+# A container class for our state
+class CaptureContext:
+    cm: libcam.CameraManager
+    camera_contexts: list[CameraCaptureContext] = []
+
+    def handle_camera_event(self):
+        # cm.get_ready_requests() will not block here, as we know there is an event
+        # to read.
+
+        reqs = self.cm.get_ready_requests()
+
+        # Process the captured frames
+
+        for req in reqs:
+            self.handle_request(req)
+
+        return True
+
+    def handle_request(self, req: libcam.Request):
+        cam_ctx = self.camera_contexts[req.cookie]
+
+        buffers = req.buffers
+
+        assert len(buffers) == 1
+
+        # A ready Request could contain multiple buffers if multiple streams
+        # were being used. Here we know we only have a single stream,
+        # and we use next(iter()) to get the first and only buffer.
+
+        stream, fb = next(iter(buffers.items()))
+
+        # Use the MappedFrameBuffer to access the pixel data with CPU. We calculate
+        # the crc for each plane.
+
+        mfb = cam_ctx.mfbs[fb]
+        crcs = [binascii.crc32(p) for p in mfb.planes]
+
+        meta = fb.metadata
+
+        print('cam{:<6} seq {:<6} bytes {:10} CRCs {}'
+              .format(cam_ctx.idx,
+                      meta.sequence,
+                      '/'.join([str(p.bytes_used) for p in meta.planes]),
+                      crcs))
+
+        # We want to re-queue the buffer we just handled. Instead of creating
+        # a new Request, we re-use the old one. We need to call req.reuse()
+        # to re-initialize the Request before queuing.
+
+        req.reuse()
+        cam_ctx.cam.queue_request(req)
+
+    def capture(self):
+        # Queue the requests to the camera
+
+        for cam_ctx in self.camera_contexts:
+            for req in cam_ctx.reqs:
+                ret = cam_ctx.cam.queue_request(req)
+                assert ret == 0
+
+        # Use Selector to wait for events from the camera and from the keyboard
+
+        sel = selectors.DefaultSelector()
+        sel.register(sys.stdin, selectors.EVENT_READ, handle_key_event)
+        sel.register(self.cm.event_fd, selectors.EVENT_READ, lambda: self.handle_camera_event())
+
+        running = True
+
+        while running:
+            events = sel.select()
+            for key, mask in events:
+                # If the handler return False, we should exit
+                if not key.data():
+                    running = False
+
+
+def handle_key_event():
+    sys.stdin.readline()
+    print('Exiting...')
+    return False
+
+
+def main():
+    cm = libcam.CameraManager.singleton()
+
+    ctx = CaptureContext()
+    ctx.cm = cm
+
+    for idx, cam in enumerate(cm.cameras):
+        cam_ctx = CameraCaptureContext(cam, idx)
+        ctx.camera_contexts.append(cam_ctx)
+
+    # Start the cameras
+
+    for cam_ctx in ctx.camera_contexts:
+        ret = cam_ctx.cam.start()
+        assert ret == 0
+
+    ctx.capture()
+
+    for cam_ctx in ctx.camera_contexts:
+        cam_ctx.uninit_camera()
+
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())
-- 
2.34.1