Asynchronous acquisition session (pymanip.video.session)
This module provides a high-level class, VideoSession, to be used in acquisition scripts.
Users should subclass VideoSession, and add two methods: prepare_camera(self, cam) in which
additionnal camera setup may be done, and process_image(self, img) in which possible image post-processing
can be done.
The prepare_camera method, if defined, is called before starting camera acquisition.
The process_image method, if defined, is called before saving the image.
The implementation allows simultaneous acquisition of several cameras, trigged by a function generator.
Concurrent tasks are set up to grab images from the cameras to RAM memory, while a background task saves
the images to the disk. The cameras, and the trigger, are released as soon as acquisition is finished, even if the images are
still being written to the disk, which allows several scripts to be executed concurrently (if the host computer
has enough RAM). The trigger_gbf object must implement configure_square() and configure_burst() methods
to configure square waves and bursts, as well as trigger() method for software trigger. An exemple is
fluidlab.instruments.funcgen.agilent_33220a.Agilent33220a.
A context manager must be used to ensure proper saving of the metadata to the database.
Example
import numpy as np
import cv2
from pymanip.video.session import VideoSession
from pymanip.video.ximea import Ximea_Camera
from pymanip.instruments import Agilent33220a
class TLCSession(VideoSession):
def prepare_camera(self, cam):
# Set up camera (keep values as custom instance attributes)
self.exposure_time = 50e-3
self.decimation_factor = 2
cam.set_exposure_time(self.exposure_time)
cam.set_auto_white_balance(False)
cam.set_limit_bandwidth(False)
cam.set_vertical_skipping(self.decimation_factor)
cam.set_roi(1298, 1833, 2961, 2304)
# Save some metadata to the AsyncSession underlying database
self.save_parameter(
exposure_time=self.exposure_time,
decimation_factor=self.decimation_factor,
)
def process_image(self, img):
# On decimate manuellement dans la direction horizontale
img = img[:, ::self.decimation_factor, :]
# On redivise encore tout par 2
# image_size = (img.shape[0]//2, img.shape[1]//2)
# img = cv2.resize(img, image_size)
# Correction (fixe) de la balance des blancs
kR = 1.75
kG = 1.0
kB = 2.25
b, g, r = cv2.split(img)
img = np.array(
cv2.merge([kB*b, kG*g, kR*r]),
dtype=img.dtype,
)
# Rotation de 180°
img = cv2.rotate(img, cv2.ROTATE_180)
return img
with TLCSession(
Ximea_Camera(),
trigger_gbf=Agilent33220a("USB0::0x0957::0x0407::SG43000299::INSTR"),
framerate=24,
nframes=1000,
output_format="png",
) as sesn:
# ROI helper (remove cam.set_roi in prepare_camera for correct usage)
# sesn.roi_finder()
# Single picture test
# sesn.show_one_image()
# Live preview
# ts, count = sesn.live()
# Run actual acquisition
ts, count = sesn.run(additionnal_trig=1)
- class pymanip.video.session.VideoSession(camera_or_camera_list, trigger_gbf, framerate, nframes, output_format, output_format_params=None, output_path=None, exist_ok=False, timeout=None, burst_mode=True)[source]
Bases:
AsyncSessionThis class represents a video acquisition session.
- Parameters
camera_or_camera_list (
Cameraor list ofCamera) – Camera(s) to be acquiredtrigger_gbf (
Driver) – function generator to be used as triggerframerate (float) – desired framerate
nframes (int) – desired number of frames
output_format (str) – desired output image format, “bmp”, “png”, tif”, or video format “mp4”
output_format_params (list) – additionnal params to be passed to
cv2.imwrite()exist_ok (bool) – allows to override existing output folder
- async _acquire_images(cam_no, live=False)[source]
Private instance method: image acquisition task. This task asynchronously iterates over the given camera frames, and puts the obtained images in a simple FIFO queue.
- _convert_for_ffmpeg(cam_no, img, fmin, fmax, gain)[source]
Private instance method: image conversion for ffmpeg process. This method prepares the input image to bytes to be sent to the ffmpeg pipe.
- async _fast_acquisition_to_ram(cam_no, total_timeout_s)[source]
Private instance method: fast acquisition to ram task
- async _live_preview(unprocessed=False)[source]
Private instance method: live preview task. This task checks the FIFO queue, drains it and shows the last frame of each camera using cv2. If more than one frame is in the queues, the older frames are dropped.
- Parameters
unprocessed (bool) – do not call
process_image()method.
- async _save_images(keep_in_RAM=False, unprocessed=False, no_save=False)[source]
Private instance method: image saving task. This task checks the image FIFO queue. If an image is available, it is taken out of the queue and saved to the disk.
- async _save_video(cam_no, gain=1.0, unprocessed=False)[source]
Private instance method: video saving task. This task waits for images in the FIFO queue, and sends them to ffmpeg via a pipe.
- async _start_clock()[source]
Private instance method: clock starting task. This task waits for all the cameras to be ready for trigger, and then sends a software trig to the function generator.
- get_one_image(additionnal_trig=0, unprocessed=False, unpack_solo_cam=True)[source]
Get one image from the camera(s).
- Parameters
- Returns
image(s) from the camera(s)
- Return type
numpy.ndarrayor list ofnumpy.ndarray(if multiple cameras, orunpack_solo_cam=False).
- async main(keep_in_RAM=False, additionnal_trig=0, live=False, unprocessed=False, delay_save=False, no_save=False)[source]
Main entry point for acquisition tasks. This asynchronous task can be called with
asyncio.run(), or combined with other user-defined tasks.- Parameters
- Returns
camera_timestamps, camera_counter
- Return type
numpy.ndarray,numpy.ndarray
- roi_finder(additionnal_trig=0)[source]
Helper to determine the ROI. This method grabs one unprocessed image from the camera(s), and allows interactive selection of the region of interest. Attention: it is assumed that the
prepare_camera()method did not already set the ROI.