Continuous High-Speed Capture (GbE-2 / Python)

Steps for creating a script that uses the common module puclib.py created in HelloWorld together with an inter-thread queue (queue.Queue) to capture and decode frame data continuously streamed from the GbE-2 at high speed, without dropping frames.

Create the script

In the same folder as puclib.py, create gbe2_continuous_capture.py and write the following code.

import ctypes
from ctypes import c_uint32, c_ushort, c_ubyte, c_void_p, POINTER, WINFUNCTYPE, byref
import threading
import queue
import time
import numpy as np

# Import from the common module puclib.py
from puclib import (
    load_sdk, PUC_SUCCEEDED, PUC_DETECT_TARGET_GBE2,
    PUC_Q_COUNT, PUC_DETECT_FILTER, PUC_DETECT_INFO, PUC_XFER_DATA_INFO
)

def main():
    print("Starting INFINICAM GbE-2 continuous high-speed capture...")

    # 1. Load and initialize the SDK
    puclib = load_sdk()
    if puclib.PUC_Initialize() != PUC_SUCCEEDED:
        print("Error: Failed to initialize the SDK.")
        return

    # 2. Additional type definitions for the continuous-transfer control API
    CALLBACK_FUNC = WINFUNCTYPE(None, POINTER(PUC_XFER_DATA_INFO), c_void_p)

    puclib.PUC_SetRingBufferCount.argtypes = [c_void_p, c_uint32]
    puclib.PUC_SetRingBufferCount.restype = c_uint32

    puclib.PUC_BeginXferData.argtypes = [c_void_p, CALLBACK_FUNC, c_void_p]
    puclib.PUC_BeginXferData.restype = c_uint32

    puclib.PUC_EndXferData.argtypes = [c_void_p]
    puclib.PUC_EndXferData.restype = c_uint32

    # 3. Search for and open the device
    detect_filter = PUC_DETECT_FILTER(nDeviceTypeFlags=PUC_DETECT_TARGET_GBE2)
    detect_info = PUC_DETECT_INFO()

    if puclib.PUC_DetectDevice2(byref(detect_filter), byref(detect_info)) != PUC_SUCCEEDED or detect_info.nDeviceCount == 0:
        print("Error: No GbE-2 camera was found.")
        return

    h_device = c_void_p()
    if puclib.PUC_OpenDevice2(0, byref(h_device)) != PUC_SUCCEEDED:
        print("Error: Failed to open the device.")
        return

    # 4. Retrieve camera parameters
    width, height = c_uint32(), c_uint32()
    puclib.PUC_GetResolution(h_device, byref(width), byref(height))
    w, h = width.value, height.value

    q_vals = (c_ushort * PUC_Q_COUNT)()
    puclib.PUC_GetQuantization2(h_device, q_vals)

    print(f"Resolution: {w} x {h} pixels")

    # 5. Prepare the data queue and processing state
    line_bytes = (w + 3) & ~3
    frame_queue = queue.Queue(maxsize=100)  # Receive buffer queue
    captured_count = 0
    target_frames = 100                     # Target number of frames to capture
    stop_event = threading.Event()

    # 6. Image-receive callback function (called from an internal SDK thread)
    @CALLBACK_FUNC
    def image_callback(p_xfer_data, p_user_param):
        nonlocal captured_count
        if not p_xfer_data or p_xfer_data.contents.isDamaged:
            return

        # Inside the callback, quickly copy the raw compressed data and push it to the queue
        raw_data = bytes(
            ctypes.cast(p_xfer_data.contents.pData, POINTER(c_ubyte * p_xfer_data.contents.nDataSize)).contents
        )
        seq_no = p_xfer_data.contents.nSequenceNo

        try:
            frame_queue.put_nowait((seq_no, raw_data))
            captured_count += 1
        except queue.Full:
            print(f"Warning: Dropped frame {seq_no} because the queue was full.")

    # 7. Background decode/processing thread
    processed_count = 0
    def processing_worker():
        nonlocal processed_count
        # Decode buffer
        decoded_buf = (c_ubyte * (line_bytes * h))()

        while not stop_event.is_set() or not frame_queue.empty():
            try:
                seq_no, raw_bytes = frame_queue.get(timeout=0.1)
            except queue.Empty:
                continue

            # Convert the compressed data into a ctypes buffer and decode it
            comp_buf = (c_ubyte * len(raw_bytes)).from_buffer_copy(raw_bytes)
            ret = puclib.PUC_DecodeData(
                decoded_buf, 0, 0, w, h, line_bytes, comp_buf, q_vals
            )

            if ret == PUC_SUCCEEDED:
                processed_count += 1
                # Perform NumPy conversion, analysis, or file saving here as needed
                if processed_count % 20 == 0:
                    print(f"Progress: {processed_count}/{target_frames} frames completed (SequenceNo: {seq_no})")

            frame_queue.task_done()

    # 8. Start the worker thread
    worker_thread = threading.Thread(target=processing_worker, daemon=True)
    worker_thread.start()

    # 9. Start the continuous transfer
    puclib.PUC_SetRingBufferCount(h_device, 64)
    ret = puclib.PUC_BeginXferData(h_device, image_callback, None)
    if ret != PUC_SUCCEEDED:
        print(f"Error: Failed to start the continuous transfer. ErrorCode: {ret}")
        puclib.PUC_CloseDevice(h_device)
        return

    start_time = time.time()
    print(f"Capturing continuously (waiting to reach {target_frames} frames)...")

    # 10. Wait until the target frame count is reached
    while captured_count < target_frames:
        time.sleep(0.01)

    elapsed_time = time.time() - start_time

    # 11. Stop the transfer and perform cleanup
    puclib.PUC_EndXferData(h_device)
    stop_event.set()
    worker_thread.join()

    fps = target_frames / elapsed_time if elapsed_time > 0 else 0
    print(f"\n--- Capture complete ---")
    print(f"Frames captured : {captured_count} frames")
    print(f"Frames decoded  : {processed_count} frames")
    print(f"Elapsed time    : {elapsed_time:.2f} sec")
    print(f"Average throughput: {fps:.1f} fps")

    # 12. Close
    puclib.PUC_CloseDevice(h_device)


if __name__ == "__main__":
    main()

Run the script
Terminal window
```
python gbe2_continuous_capture.py
```
When run, the target number of frames (100 frames) is quickly captured and decoded from the camera via the queue, and once processing finishes the capture time and effective fps are displayed.

High-Speed Capture Implementation Highlights

1. Minimizing work inside the callback (data hand-off only)

If heavy image decoding (PUC_DecodeData) or disk saving is performed inside the image_callback function, the SDK’s internal receive processing is delayed, causing frame drops. Inside the callback, only copying the compressed data (byte array) and pushing it to the queue is done, as quickly as possible.

2. Producer-consumer architecture (`queue.Queue`)

Data reception (producer): The callback function pushes the received raw data into the queue.
Data decoding (consumer): A separate thread (processing_worker) pulls data from the queue and decodes/analyzes it asynchronously.

Adopting this multithreaded structure makes efficient use of CPU cores and enables stable, high-speed continuous capture without frame drops.

Continuous High-Speed Capture (GbE-2 / Python)

High-Speed Capture Implementation Highlights

1. Minimizing work inside the callback (data hand-off only)

2. Producer-consumer architecture (queue.Queue)

2. Producer-consumer architecture (`queue.Queue`)