To verify the synchronized operation of the dual camera, I performed two acquisitions of a running light arrangement, that I built specifically for this purpose. The first acquisition is in the planned operation mode with external triggers while for the second acquisition the external trigger signal is switched off to show the effect.

The running light arrangement consists of ten white LEDs that are connected to GPIO pins of a raspberry pi. The pigpio library is used to drive them with precise timing. The LEDs are positioned in a straight line and are switched on sequentially, with only one active at a time. The same raspberry pi is used to create the external trigger signals for the two cameras. Video 1 shows the course of the measurement.

Video 1: measurement procedure

For the measurement I chose a nominal framerate of 50 frames per second (fps) for both cameras, because I want to use the dual cam of highrate_longexp as a replacement for the single camera running at 100 fps in my other project lalelu_drums. I set the frequency of the external trigger to 51 Hz to show the difference between the two cases with and without external trigger.

To generate the trigger signals and control the running light, the following script was used. Since the trigger signals and the running light are generated from the same clock, it is expected that the acquired images are phase locked to the pattern of the running light.

import pigpio
import time


pi = pigpio.pi()

pins = [26, 16, 20, 21, 19, 13, 12, 6, 5, 7]

triggerPinA = 2
triggerPinB = 3

for pin in pins + [triggerPinA, triggerPinB]:
    pi.set_mode(pin, pigpio.OUTPUT)

# camera fps: 50
# exposure = 20ms
# period of complete pattern = 40ms
# 10 LEDs but 20 'pulses' to support trigger at 2.5 LED steps
# pulse-delay = 40ms / 20 = 2ms
# detune from 50fps to 51fps:
# 2ms * 50 / 51 = 1.96ms
delay = 1960

pulses = []
for i in range(2 * len(pins)):
    onMask = 0
    offMask = 0

    if i % 2 == 0:
        j = i // 2
        p0 = pins[j]
        p1 = pins[(j+1) % len(pins)]
        onMask = onMask | 1<<p1
        offMask = offMask | 1<<p0
    if i in [0, 10]:
        onMask = onMask | (1<<triggerPinA)
    if i in [1, 11]:
        offMask = offMask | (1<<triggerPinA)
    if i in [5, 15]:
        onMask = onMask | (1<<triggerPinB)
    if i in [6, 16]:
        offMask = offMask | (1<<triggerPinB)

    pulses.append(pigpio.pulse(onMask, offMask, delay))

pi.wave_clear()

pi.wave_add_generic(pulses)
wave = pi.wave_create()

pi.wave_send_repeat(wave)

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print('stopping')

pi.wave_tx_stop()
pi.wave_clear()

For the acquisition of the camera images, the following script was used. It makes use of a gstreamer appsink that allows to retrieve the image data in a python callback function (onNewSample). The script records 500 frames and corresponding timestamps for each camera, yielding a recording duration of ~10s. The data is stored in a hdf5 file.

import gi
from time import sleep, monotonic
import tables
from datetime import datetime
import subprocess

gi.require_version('Gst', '1.0')
gi.require_version('GstVideo', '1.0')

from gi.repository import Gst, GstVideo
import numpy as N


Gst.init(None)


def setExposureTime(deviceIndex, exposureTime):
    subprocess.run(['v4l2-ctl', '-d' f'/dev/video{deviceIndex}', '-c', 'auto_exposure=1'])
    value = int(N.round(exposureTime / 100e-6))
    subprocess.run(['v4l2-ctl', '-d' f'/dev/video{deviceIndex}', '-c', f'exposure={value}'])

def setGain(deviceIndex, gain):
    subprocess.run(['v4l2-ctl', '-d' f'/dev/video{deviceIndex}', '-c', 'gain_automatic=0'])
    subprocess.run(['v4l2-ctl', '-d' f'/dev/video{deviceIndex}', '-c', f'gain={gain}'])


class PipelineAppSink:
    def __init__(self, deviceIndex, width, height):
        self.height = height
        self.width = width
        pipelineDescription = f'v4l2src device=/dev/video{deviceIndex}'
        pipelineDescription += f' ! video/x-raw,width={self.width},height={self.height},framerate=50/1'
        pipelineDescription += ' ! queue max-size-buffers=1 leaky=downstream'
        pipelineDescription += ' ! videoconvert ! video/x-raw,format=RGB'
        pipelineDescription += ' ! appsink name=sink emit-signals=true drop=true max-buffers=1 sync=false'

        self.pipeline = Gst.parse_launch(pipelineDescription)
        
        self.sink = self.pipeline.get_by_name('sink')
        self.sink.connect('new-sample', self.onNewSample)

        self.times = []
        self.frames = []
    
    def onNewSample(self, sink):
        self.sample = sink.emit('pull-sample')

        if len(self.times) < 500:
            self.times.append(monotonic())
            self.frames.append(self.getFrame())

        self.sample = None

        return Gst.FlowReturn.OK

    def getFrame(self):
        buffer = self.sample.get_buffer()
        success, mapInfo = buffer.map(Gst.MapFlags.READ)
        frame = N.ndarray(shape=(self.height, self.width, 3),
                          dtype=N.uint8,
                          buffer=mapInfo.data).copy()
        
        buffer.unmap(mapInfo)
        
        return frame
    
    def play(self, play=True):
        if play:
            self.pipeline.set_state(Gst.State.PLAYING)
        else:
            self.pipeline.set_state(Gst.State.NULL)


for deviceIndex in [0, 1]:
    setExposureTime(deviceIndex, 20e-3)
    setGain(deviceIndex, 10)

pipelines = [PipelineAppSink(d, 320, 240) for d in [0, 1]]

for p in pipelines:
    p.play()

sleep(15)

for p in pipelines:
    p.play(False)


filename = f'dual_cam_recording_{datetime.now()}.h5'
filename = filename.replace(':', '-')

h5file = tables.open_file(filename,
                          mode='w',
                          title='dual_cam_recording')

for i, p in enumerate(pipelines):
    framesGroup = h5file.create_group(h5file.root, f'frames{i}')

    for j, f in enumerate(p.frames):
        name = f'frame_{j:04}'
        h5file.create_array(framesGroup, name, f)
    
    h5file.create_array(h5file.root, f'times{i}', N.array(p.times))

h5file.close()

 I will present the analysis of the recorded data in the next log entry.