[libcamera-devel] [PATCH v2 11/19] py: examples: Add simple-continuous-capture.py
Laurent Pinchart
laurent.pinchart at ideasonboard.com
Thu May 26 18:14:26 CEST 2022
Hi Tomi,
On Tue, May 24, 2022 at 02:46:02PM +0300, Tomi Valkeinen wrote:
> Add a slightly more complex, and I think a more realistic, example,
> where the script reacts to events and re-queues the buffers.
Ah, I should have read this before reviewing 10/19 and asking you to
implement the features of this application in simple-capture.py :-)
I think simple-capture.py is a bit too simple, I'm tempted to keep this
one only and rename it simple-capture.py, or maybe better simple-cam.py
to match the C++ application. Having the exact same features in both the
C++ and Python examples would be very nice I think.
> 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()))
Could this be made simpler, for instance
stream, fb = buffers[0]
> +
> + # 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]
Same comment as in 10/19, buffers should be pre-mapped.
> +
> + 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):
s/envet/event/
> + sys.stdin.readline()
> + print("Exiting...")
s/"/'/g
Muscle memory ? :-)
> + 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)
If we happen to have two event, first a key press and then a frame
capture, processed in the same iteration of the while loop, then running
will be overridden. You can make running a global variable and set it in
handle_key_event, or possibly make handle_key_event a closure that
captures a running variable local to this function.
> +
> +
> +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)
Same comments as for patch 10/19.
> +
> + 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())
--
Regards,
Laurent Pinchart
More information about the libcamera-devel
mailing list