[libcamera-devel] [PATCH v4 14/16] py: examples: Add simple-continuous-capture.py
Jacopo Mondi
jacopo at jmondi.org
Mon Jun 6 11:19:01 CEST 2022
Hi Tomi
On Mon, Jun 06, 2022 at 11:56:11AM +0300, Tomi Valkeinen wrote:
> On 05/06/2022 15:31, Jacopo Mondi wrote:
> > Hi Tomi,
> >
> > On Mon, May 30, 2022 at 05:27:20PM +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.
> > >
> >
> > My first question is if such similar examples are useful or they will
> > become a maintainership burden.
> >
> > From my side, the more examples the better, but I think the question
> > if it's woth it stays..
>
> We can always change and drop these later.
>
> But my reasoning was summarized in the intro letter. I think this and
> simple-capture.py are quite different. Maybe this could be renamed.
> simple-capture-app.py?
>
> > > 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)])
> >
> > Not that relevant, but you could populate self.reqs directly and
> > append a {buffer, libcamera.utils.MappedFrameBuffer(fb).mmap()} pair
> > to the self.mfbs dictionary in the request creation loop ?
>
> Yes, you're right. It makes the code easier to read.
>
> > > +
> > > + 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
> >
> > Nit: why is this a global function handle_camera_event() a class
> > member ?
>
> Hmm, yes, I think I can move this inside CaptureContext too.
>
> > To be honest I would get rid of the CaptureContext class and open-code
> > CaptureContext::capture() in your main function, with all the handlers
> > being global functions. Not a big deal though
>
> I wanted to structure this example to look a bit more like a bigger
> application, even if it's not really needed in this example.
>
> > > +
> > > +
> > > +def main():
> > > + cm = libcam.CameraManager.singleton()
> > > +
> > > + ctx = CaptureContext()
> > > + ctx.cm = cm
> > > +
> > > + for idx, cam in enumerate(cm.cameras):
> > > + cam_ctx = CameraCaptureContext(cam, idx)
> >
> > Can't you start the camera here ?
>
> I could. My thinking was that you'd first collect the cameras and configure
> them, and if everything is still good, then you go and start them. Is there
> a reason you think it's better to start here?
Not particularly, but seeing an additional loop had me wonder if there
was a reason to do so or not, as the CameraCaptureContext() is created
unconditionally. Even more, the configuration correctness is actually
asserted, which I assume will raise an exception that will terminate
the program and will not simply skip the faulty camera..
Thanks
j
>
>
> > > + 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())
> >
> > Nits apart
> > Reviewed-by: Jacopo Mondi <jacopo at jmondi.org>
>
> Thanks!
>
> Tomi
More information about the libcamera-devel
mailing list