[libcamera-devel] [PATCH v2 11/19] py: examples: Add simple-continuous-capture.py
Tomi Valkeinen
tomi.valkeinen at ideasonboard.com
Tue May 24 13:46:02 CEST 2022
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 | 179 +++++++++++++++++++
1 file changed, 179 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..05fa2ada
--- /dev/null
+++ b/src/py/examples/simple-continuous-capture.py
@@ -0,0 +1,179 @@
+#!/usr/bin/env python3
+
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (C) 2022, Tomi Valkeinen <tomi.valkeinen at ideasonboard.com>
+
+# A simple minimal capture example showing:
+# - How to setup the camera
+# - Capture frames using events
+# - How to requeue requests
+# - Memory map the frames
+# - How to stop the camera
+
+import argparse
+import binascii
+import libcamera as libcam
+import libcamera.utils
+import selectors
+import sys
+
+
+def handle_camera_event(cm, cam):
+ # cm.read_event() will not block here, as we know there is an event to read.
+ # We have to read the event to clear it.
+
+ cm.read_event()
+
+ reqs = cm.get_ready_requests()
+
+ # Process the captured frames
+
+ for req in reqs:
+ 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 MappedFrameBuffer to access the pixel data with CPU. We calculate
+ # the crc for each plane.
+
+ with libcamera.utils.MappedFrameBuffer(fb) as mfb:
+ crcs = [binascii.crc32(p) for p in mfb.planes]
+
+ meta = fb.metadata
+
+ print('buf {}, seq {}, bytes {}, CRCs {}'
+ .format(req.cookie, meta.sequence, meta.bytesused, 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.queue_request(req)
+
+ return True
+
+
+def handle_key_envet(fileob):
+ sys.stdin.readline()
+ print("Exiting...")
+ return False
+
+
+def capture(cm, cam, reqs):
+ # Queue the requests to the camera
+
+ for req in reqs:
+ ret = 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(cm.efd, selectors.EVENT_READ, lambda fd: handle_camera_event(cm, cam))
+ sel.register(sys.stdin, selectors.EVENT_READ, handle_key_envet)
+
+ reqs = []
+ running = True
+
+ while running:
+ events = sel.select()
+ for key, mask in events:
+ running = key.data(key.fileobj)
+
+
+def main():
+ parser = argparse.ArgumentParser()
+ parser.add_argument('-c', '--camera', type=str, default='1',
+ help='Camera index number (starting from 1) or part of the name')
+ parser.add_argument('-f', '--format', type=str, help='Pixel format')
+ parser.add_argument('-s', '--size', type=str, help='Size ("WxH")')
+ args = parser.parse_args()
+
+ cm = libcam.CameraManager.singleton()
+
+ try:
+ if args.camera.isnumeric():
+ cam_idx = int(args.camera)
+ cam = next((cam for i, cam in enumerate(cm.cameras) if i + 1 == cam_idx))
+ else:
+ cam = next((cam for cam in cm.cameras if args.camera in cam.id))
+ except Exception:
+ print(f'Failed to find camera "{args.camera}"')
+ return -1
+
+ # Acquire the camera for our use
+
+ ret = cam.acquire()
+ assert ret == 0
+
+ # Configure the camera
+
+ camconfig = cam.generate_configuration([libcam.StreamRole.StillCapture])
+
+ streamconfig = camconfig.at(0)
+
+ if args.format:
+ fmt = libcam.PixelFormat(args.format)
+ streamconfig.pixel_format = fmt
+
+ if args.size:
+ w, h = [int(v) for v in args.size.split('x')]
+ streamconfig.size = libcam.Size(w, h)
+
+ ret = cam.configure(camconfig)
+ assert ret == 0
+
+ stream = streamconfig.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'Capturing {streamconfig} with {num_bufs} buffers from {cam.id}')
+
+ # 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(i)
+
+ buffer = allocator.buffers(stream)[i]
+ ret = req.add_buffer(stream, buffer)
+ assert ret == 0
+
+ reqs.append(req)
+
+ # Start the camera
+
+ ret = cam.start()
+ assert ret == 0
+
+ capture(cm, cam, reqs)
+
+ # Stop the camera
+
+ ret = cam.stop()
+ assert ret == 0
+
+ # Release the camera
+
+ ret = cam.release()
+ assert ret == 0
+
+ return 0
+
+
+if __name__ == '__main__':
+ sys.exit(main())
--
2.34.1
More information about the libcamera-devel
mailing list