OpenCV LiveView (Monochrome) Project (GbE-2 / Python)

Steps for creating a script that uses the common module puclib.py created in HelloWorld together with OpenCV (cv2) to receive and decode, via a C callback function, compressed data continuously streamed from the GbE-2 (monochrome model), and display it in real time (live view) in a window.

Create the script

In the same folder as puclib.py, create gbe2_opencv_live.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 numpy as np
import cv2

# 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 OpenCV LiveView (monochrome)...")

    # 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 (monochrome)")

    # 5. Prepare the buffer and the mutual-exclusion mechanism
    line_bytes = (w + 3) & ~3
    decoded_buffer = (c_ubyte * (line_bytes * h))()

    latest_frame = None
    frame_lock = threading.Lock()

    # 6. Define the callback function
    @CALLBACK_FUNC
    def image_callback(p_xfer_data, p_user_param):
        nonlocal latest_frame
        if not p_xfer_data or p_xfer_data.contents.isDamaged:
            return

        # Decode the received compressed data
        ret = puclib.PUC_DecodeData(
            decoded_buffer, 0, 0, w, h, line_bytes,
            p_xfer_data.contents.pData, q_vals
        )

        if ret == PUC_SUCCEEDED:
            # Convert to a NumPy array (copy it so it can be safely shared between threads)
            img = np.ctypeslib.as_array(decoded_buffer).reshape((h, line_bytes))[:, :w].copy()
            with frame_lock:
                latest_frame = img

    # 7. Start the continuous transfer
    puclib.PUC_SetRingBufferCount(h_device, 32)
    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

    window_name = "INFINICAM GbE-2 LiveView (press ESC or Q to exit)"
    cv2.namedWindow(window_name, cv2.WINDOW_AUTOSIZE)

    print("Displaying the live view... Press 'ESC' or 'Q' in the window to exit.")

    # 8. Main rendering loop
    try:
        while True:
            img_to_show = None
            with frame_lock:
                if latest_frame is not None:
                    img_to_show = latest_frame

            if img_to_show is not None:
                cv2.imshow(window_name, img_to_show)

            # Wait 1 ms and check for key input
            key = cv2.waitKey(1) & 0xFF
            if key in (27, ord('q'), ord('Q')):  # ESC (27) or 'q' / 'Q'
                print("Exiting due to user action.")
                break
    finally:
        # 9. Shutdown processing
        puclib.PUC_EndXferData(h_device)
        cv2.destroyAllWindows()
        puclib.PUC_CloseDevice(h_device)


if __name__ == "__main__":
    main()

Run the script
Terminal window
```
python gbe2_opencv_live.py
```
When run, an OpenCV display window opens and the video from the camera is streamed and rendered in real time. Pressing the ESC or Q key stops the transfer and exits safely.

Real-Time Processing Implementation Highlights

1. The C callback function (`ctypes.WINFUNCTYPE`)

PUC_BeginXferData automatically calls the Python-side function from an internal SDK thread whenever a new data frame is received from the camera. WINFUNCTYPE is used to define the type for the interface with C.

CALLBACK_FUNC = WINFUNCTYPE(None, POINTER(PUC_XFER_DATA_INFO), c_void_p)

2. Thread-safe frame updates (`threading.Lock`)

threading.Lock() is used to arbitrate access, preventing memory contention (screen flickering or inconsistency) that could occur if the SDK’s receive/decode thread (inside the callback) and the OpenCV rendering thread (the main loop) accessed the frame data at the same time.