Usage

Here is a small usage example of setting exposure time, and printing what value actually got set (this might not always be exactly the value we requested), then start the stream and get a single frame and frame metadata

import qamlib

cam = qamlib.Camera() # Opens /dev/qtec/video0

# Try to set exposure time (us)
cam.set_control("Exposure Time, Absolute", 1000)

exp = cam.get_control("Exposure Time, Absolute")

# Print the exposure time that we ended up with
print(f"Got exposure time {exp}us")

# Start and stop streaming with context manager
with cam:
    meta, frame = cam.get_frame()

    print(meta.sequence) # Frame number since start of streaming

Bigger example

"""Demonstrates various Qamlib features."""

import sys

import qamlib


def needs_byteswap(frame):
    """Check if a byte-swap is needed for the PNM format."""
    # 8-bit
    if frame.itemsize == 1:
        return False
    # Big Endian
    if frame.dtype.byteorder == ">":
        return False
    # Big Endian native byte order
    if frame.dtype.byteorder == "=" and sys.byteorder == "big":
        return False
    return True


def write_ppm(name, frame):
    """Write frame as PNM file."""
    with open(name, "wb") as f:
        height, width, channels = frame.shape

        if channels == 1:
            type = "P5"
        else:
            type = "P6"

        if frame.nbytes == width * height * channels:
            if frame.dtype == "uint8":
                max_val = 255
            else:
                max_val = 65535
        else:
            max_val = 65535

        ppm_header = f"{type} {width} {height} {max_val}\n"
        f.write(bytearray(ppm_header, "ascii"))

        if needs_byteswap(frame):
            # swap data in memory in order to properly write PNM
            # since PPM is Big Endian by definition
            # Note that if the operation is done in place
            # it will mess up the endianess when reading single values out
            # so in this case use: frame = frame.byteswap(True).newbyteorder()
            # in order to keep the correct byte order
            frame.byteswap().tofile(f)
        else:
            frame.tofile(f)


if __name__ == "__main__":
    if len(sys.argv) > 1:
        device = sys.argv[1]
    else:
        device = "/dev/qtec/video0"

    # Open video capture device
    try:
        cam = qamlib.Camera(device)
    except Exception as e:
        print(e)
        exit(-1)

    print("List of formats:")
    fmts = cam.list_formats()
    for name in fmts:
        print(fmts[name])
    print("\n")

    # V4l2 "Settings"
    fps = 20.0
    width = 800
    height = 600

    cam.set_format("Y16_BE")
    # cam.set_format("Y16")
    # cam.set_format("GREY")
    cam.set_framerate(fps)
    cam.set_resolution(width, height)

    # crop to center
    bounds = cam.get_crop_bounds()
    def_rect = cam.get_crop_default()
    left = int(((bounds.width - bounds.left) - width) / 2) + bounds.left
    top = int(((bounds.height - bounds.top) - height) / 2) + bounds.top
    cam.set_crop(left, top, width, height)

    img_format = cam.get_format()
    px_format = img_format.pixelformat

    print(f"FPS: {cam.get_framerate()}")
    print(f"Frame Size: {cam.get_resolution()}")
    print(f"Crop: {cam.get_crop()}")
    print(f"Pixel format: {img_format}")
    print("\n")

    # V4l2 Controls
    ctrls = cam.list_controls()
    print("Found ", len(ctrls), " controls")
    for name in ctrls:
        print(ctrls[name])
    print("\n")

    cam.set_control("Exposure Time Absolute", 5000)
    print(f"Exposure: {cam.get_control('Exposure Time Absolute')}")

    # Frame capture
    print("Starting Streaming")
    with cam:
        for i in range(10):
            metadata, frame = cam.get_frame()

            print(metadata)
            print(frame.size, " ", frame.shape)

            write_ppm(f"/tmp/img{i}.pnm", frame)
    print("Done Streaming")

ExtendedControl usage

"""Demonstrates how to use extended V4L2 controls."""

import numpy as np

import qamlib


cam = qamlib.Camera()  # /dev/qtec/video0

ctrls = cam.list_controls()

ctrl_lut = ctrls["lut red"]

assert ctrl_lut.type == qamlib.ControlType.INTEGER

# If the control is INTEGER and has a payload it is an ArrayControl
assert ctrl_lut.flags.has_payload

arr = np.ones([ctrl_lut.elements], dtype=np.int32)

arr[500:] = 12**2

lut_red = qamlib.ArrayControlValue(arr)

lut_green = cam.get_ext_control("lut green")

exposure = qamlib.IntegerControlValue(9992)

new_values = {
    "lut red": lut_red,
    "lut blue": lut_green,
    "exposure time absolute": exposure,
}

cam.set_controls(new_values)

print(cam.get_controls(["lut red", "lut blue", "exposure time absolute"]))

HDR example

"""Example of doing "HDR" with a trigger sequence.

This uses features from Qtechnology kernel patches
"""

import cv2

import qamlib


exposures = [5000, 15000, 29000, 41000, 50000]

# Defaults to /dev/qtec/video0
cam = qamlib.Camera()

# External trigger sequence
cam.set_control("trigger_mode", 5)

trig_seq = qamlib.TriggerSequenceValue()

# Used to calculate minimal frame delay
fot = cam.get_control("frame_overhead_time")
rot = cam.get_control("read_out_time")

# Create trigger sequence with minimal delay
for i in range(len(exposures)):
    exp = exposures[i]
    if i + 1 >= len(exposures):
        delay = max(exp + fot, exp + fot + rot - exposures[0])
    else:
        delay = max(exp + fot, exp + fot + rot - exposures[i + 1])

    trig_seq.add_exposure(exp, exp, delay, 0)

# Set trigger sequence
cam.set_ext_control("trigger sequence", trig_seq)

# Set white balance, values from qtec-camera-gwt
cam.set_control("Red Balance", 32604)
cam.set_control("Green Balance", 16384)
cam.set_control("Blue Balance", 30802)

# Use BGR to avoid having to make OpenCV do a RGB -> BGR conversion
cam.set_format("BGR3")

# HDR Fusion merger
merge = cv2.createMergeMertens()

# Start streaming
with cam:
    images = []

    # Trigger capture, since we are in external trigger mode
    cam.set_control("Manual Trigger", 1)

    for exposure in exposures:
        name = f"exposure_{exposure}.png"

        _, img = cam.get_frame(timeout=1, buffered=True)

        cv2.imwrite(name, img)

        images.append(img)

    fusion = merge.process(images)
    cv2.imwrite("fusion.png", fusion * 255)  # Values after merge are in [0, 1]

Events example

"""Demonstrates how to use Qamlib events."""

import qamlib


def _cb(event):
    global num_events
    if event.type == qamlib.EventType.CTRL:
        print("Got control event")
    else:
        raise Exception("Unkown event type")


evs = qamlib.EventDevice()

evs.set_callback(_cb)

cam = qamlib.Camera()
ctrls = cam.list_controls()

# Subscribe to events for all controls
for c in ctrls:
    evs.subscribe(qamlib.EventType.CTRL, ctrls[c].id)

# Start listening
evs.start()

# Do stuff

# Stop listening
evs.stop()

Dual Head example

"""Dual camera example.

This is an example of using a Qtec dual head camera setup, in which the the
secondary head (/dev/qtec/video1) is triggered by the main head, in order to
have the frames from the two sensors be syncronized. The example is based on
having one RGB sensor and one SWIR sensor.
"""

import cv2

import qamlib


SAVE_DIR = "/home/root/test"


# Frame capture
def stream_on(rgb_cam, swir_cam):
    """Start both camera streams."""
    i1 = 0
    i2 = 0

    print("Starting Streaming")
    try:
        rgb_cam.start()
        swir_cam.start()
        while True:
            # Using 'buffered=True' will return the next frame in the queue
            # and throw a 'DroppedFrameException' if the queue gets filled up
            # set 'overflow_exception=False' in the Camera constructor
            # if you want to disable this exception, but then it will be necessary
            # to check for dropped frames using the sequence nr from the metadata
            rgb_meta, rgb_frame = rgb_cam.get_frame(timeout=1, buffered=True)
            swir_meta, swir_frame = swir_cam.get_frame(timeout=1, buffered=True)

            # all this sequence nr checks shouldn't be necessary
            # since we will get a 'DroppedFrameException' if frames are dropped
            # but are present to illustrate how to check for dropped frames
            # under other circumstances.

            n1 = rgb_meta.sequence - i1
            if n1 > 0:
                print(f"rgb_cam skipped {n1} frames")
                break
            i1 = rgb_meta.sequence + 1

            n2 = swir_meta.sequence - i2
            if n2 > 0:
                print(f"swir_cam skipped {n2} frames")
                break
            i2 = swir_meta.sequence + 1

            if rgb_meta.sequence != swir_meta.sequence:
                print(
                    f"Error, heads are out of sync: {rgb_meta.sequence} != "
                    f"{swir_meta.sequence}"
                )
                break

            if rgb_meta.sequence % fps == 0:
                print(f"rgb_cam: {rgb_meta.sequence}")

            if swir_meta.sequence % fps == 0:
                print(f"swir_cam: {swir_meta.sequence}")

            cv2.imwrite(f"{SAVE_DIR}/rgb_{rgb_meta.sequence:09d}.png", rgb_frame)
            cv2.imwrite(f"{SAVE_DIR}/swir_{swir_meta.sequence:09d}.png", swir_frame)
    except Exception as e:
        template = "An exception of type {0} occurred. Arguments:\n{1!r}"
        message = template.format(type(e).__name__, e.args)
        print(message)
    finally:
        rgb_cam.stop()
        swir_cam.stop()

    print("Done Streaming")


if __name__ == "__main__":
    # Open video capture device
    try:
        rgb_cam = qamlib.Camera("/dev/qtec/video0")
        swir_cam = qamlib.Camera("/dev/qtec/video1")
    except Exception as e:
        print(e)
        exit(-1)

    # V4l2 "Settings"
    fps = 5.0

    # Use BGR so we don't have to do the conversion to save with OpenCV
    rgb_cam.set_format("BGR3")
    rgb_cam.set_framerate(fps)

    swir_cam.set_format("GREY")
    # set swir head for "external trigger" so the other head drives it (fps)
    swir_cam.set_control("trigger mode", 1)

    img_format1 = rgb_cam.get_format()
    px_format1 = img_format1.pixelformat

    img_format2 = swir_cam.get_format()
    px_format2 = img_format2.pixelformat

    print(f"FPS: {rgb_cam.get_framerate()}")
    print(f"Frame Size: {rgb_cam.get_resolution()} , {swir_cam.get_resolution()}")
    print(f"Crop: {rgb_cam.get_crop()} , {swir_cam.get_crop()}")
    print(f"Pixel format: {img_format1} , {img_format2}")
    print("\n")

    # adjust exposure time
    rgb_cam.set_control("Exposure Time, Absolute", 5000)
    swir_cam.set_control("Exposure Time, Absolute", 5000)
    print(
        f"Exposure: {rgb_cam.get_control('Exposure Time, Absolute')} , "
        f"{swir_cam.get_control('Exposure Time, Absolute')}"
    )

    # restart streaming on errors
    while True:
        stream_on(rgb_cam, swir_cam)