Genicam Harvesters (genicam/harvesters)

Genicam Harvesters

https://github.com/genicam/harvesters
Admin
Harvesters is a Python library designed to simplify the acquisition and management of images from...

Tokens:13,065
Snippets:109
Trust Score:6.3
Update:1 week ago
Show doc for...
Context Summary (auto-generated)
Raw
# Harvesters

Harvesters is a Python library for acquiring images from GenICam-compliant machine vision cameras through GenTL Producers. The library provides a minimalistic front-end that enables developers to prepare images for computer vision applications with ease, hiding the complexities of transport layer dependent technologies. Harvesters supports multiple transport layers (GigE Vision, USB3 Vision, etc.) and allows loading multiple GenTL Producers in a single Python script.

The core functionality revolves around the `Harvester` class, which manages GenTL Producers and device enumeration, and the `ImageAcquirer` class, which handles the actual image acquisition process from connected cameras. Harvesters delivers acquired images as NumPy arrays, making it seamless to integrate with image processing libraries like OpenCV, PIL, or Matplotlib. The library follows the GenICam GenApi standard for device feature manipulation, allowing dynamic configuration of camera parameters such as resolution, pixel format, and exposure.

## Installation

Install Harvesters via pip along with its dependencies (numpy and genicam).

```bash
pip install harvesters

# For upgrading to the latest version
pip install -U --no-cache-dir harvesters
```

## Harvester Class - Initialize and Manage GenTL Producers

The `Harvester` class is the entry point for working with machine vision cameras. It manages the loading of CTI files (GenTL Producers), enumerates available devices, and creates ImageAcquirer objects for image acquisition.

```python
from harvesters.core import Harvester

# Create a Harvester instance
h = Harvester()

# Add a CTI file (GenTL Producer)
h.add_file('/path/to/your/gentl_producer.cti')

# You can add multiple CTI files
h.add_file('/path/to/another_producer.cti')

# List loaded CTI files
print(h.files)
# Output: ['/path/to/your/gentl_producer.cti', '/path/to/another_producer.cti']

# Update device list (enumerate available cameras)
h.update()

# Check available devices
print(len(h.device_info_list))
# Output: 4

# View device information
for device in h.device_info_list:
    print(device)
# Output: {'display_name': 'TLSimuMono (SN_InterfaceA_0)', 'id_': 'TLSimuMono',
#          'model': 'TLSimuMono', 'serial_number': 'SN_InterfaceA_0',
#          'tl_type': 'Custom', 'user_defined_name': 'Center',
#          'vendor': 'EMVA_D', 'version': '1.2.3'}

# Remove a CTI file
h.remove_file('/path/to/your/gentl_producer.cti')

# Reset and release all resources when done
h.reset()
```

## Create ImageAcquirer - Connect to a Camera

The `create()` method creates an `ImageAcquirer` object that connects to a specific camera device. You can specify the target device by index, dictionary of properties, or DeviceInfo object.

```python
from harvesters.core import Harvester

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()

# Method 1: Create by index (first available device)
ia = h.create(0)

# Method 2: Create by device properties
ia = h.create({'serial_number': 'SN_InterfaceA_0'})

# Method 3: Create with multiple search criteria
ia = h.create({'vendor': 'EMVA_D', 'model': 'TLSimuMono'})

# Method 4: Create without arguments (uses first device)
ia = h.create()

# Check available search keys for device selection
from harvesters.core import DeviceInfo
print(DeviceInfo.search_keys)
# Output: ['access_status', 'display_name', 'id_', 'model', 'parent',
#          'serial_number', 'tl_type', 'user_defined_name', 'vendor', 'version']

# Always destroy the acquirer when done
ia.destroy()
h.reset()
```

## ImageAcquirer - Configure Camera Parameters

The `ImageAcquirer` provides access to the remote device's GenICam node map for configuring camera parameters like resolution, pixel format, exposure time, and gain.

```python
from harvesters.core import Harvester

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()
ia = h.create(0)

# Access the remote device node map
node_map = ia.remote_device.node_map

# Set image dimensions
node_map.Width.value = 640
node_map.Height.value = 480

# Set pixel format (Mono8, RGB8, BayerRG8, etc.)
node_map.PixelFormat.value = 'Mono8'

# Set acquisition mode
node_map.AcquisitionMode.value = 'Continuous'  # or 'SingleFrame', 'MultiFrame'

# For MultiFrame mode, set frame count
# node_map.AcquisitionFrameCount.value = 10

# Set exposure time (if supported by camera)
# node_map.ExposureTime.value = 10000  # microseconds

# Set gain (if supported)
# node_map.Gain.value = 1.0

# List all available feature nodes
print(dir(node_map))

# Read current values
print(f"Width: {node_map.Width.value}")
print(f"Height: {node_map.Height.value}")
print(f"PixelFormat: {node_map.PixelFormat.value}")

# Execute a command node (e.g., software trigger)
# node_map.TriggerSoftware.execute()

ia.destroy()
h.reset()
```

## Start and Stop Image Acquisition

The `start()` and `stop()` methods control the image acquisition process. You can run acquisition in the foreground or as a background thread.

```python
from harvesters.core import Harvester

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()
ia = h.create(0)

# Configure acquisition parameters
ia.num_buffers = 5  # Number of buffers for image acquisition
print(f"Min buffers required: {ia.min_num_buffers}")

# Start acquisition in foreground mode
ia.start()

# Check if acquisition is running
print(ia.is_acquiring())  # Output: True

# Stop acquisition
ia.stop()

# Start acquisition in background thread mode
ia.start(run_as_thread=True)

# Configure buffer holding for background mode
ia.num_filled_buffers_to_hold = 3

# Check number of available buffers
print(ia.num_holding_filled_buffers)

ia.stop()
ia.destroy()
h.reset()
```

## Fetch Images - Acquire Image Data

The `fetch()` method retrieves acquired image buffers. Images are delivered as NumPy arrays through the buffer's payload components.

```python
from harvesters.core import Harvester
import numpy as np

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()
ia = h.create(0)

# Configure camera
ia.remote_device.node_map.Width.value = 640
ia.remote_device.node_map.Height.value = 480
ia.remote_device.node_map.PixelFormat.value = 'Mono8'

ia.start()

# Method 1: Fetch with context manager (auto-queues buffer)
with ia.fetch() as buffer:
    # Access image component
    component = buffer.payload.components[0]

    # Get raw 1D data as NumPy array
    data_1d = component.data
    print(f"1D array shape: {data_1d.shape}")

    # Reshape to 2D image
    image_2d = data_1d.reshape(component.height, component.width)
    print(f"2D image shape: {image_2d.shape}")

    # Get image properties
    print(f"Width: {component.width}")
    print(f"Height: {component.height}")
    print(f"Data format: {component.data_format}")

    # Perform calculations
    print(f"Mean: {np.mean(image_2d):.2f}")
    print(f"Min: {image_2d.min()}, Max: {image_2d.max()}")

    # Get buffer timestamp
    print(f"Timestamp: {buffer.timestamp_ns} ns")

# Method 2: Manual fetch and queue
buffer = ia.fetch()
component = buffer.payload.components[0]
image = component.data.reshape(component.height, component.width)
# Process image...
buffer.queue()  # Don't forget to queue the buffer back!

ia.stop()
ia.destroy()
h.reset()
```

## Fetch with Timeout - Handle Acquisition Delays

Use `fetch()` with timeout or `try_fetch()` for non-blocking acquisition with timeout handling.

```python
from harvesters.core import Harvester
from genicam.gentl import TimeoutException

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()
ia = h.create(0)

ia.start()

# Method 1: fetch() with timeout (raises TimeoutException)
try:
    buffer = ia.fetch(timeout=3.0)  # 3 second timeout
    with buffer:
        component = buffer.payload.components[0]
        print(f"Acquired image: {component.width}x{component.height}")
except TimeoutException:
    print("No image received within timeout period")

# Method 2: try_fetch() returns None on timeout (no exception)
buffer = ia.try_fetch(timeout=3.0)
if buffer:
    component = buffer.payload.components[0]
    print(f"Acquired image: {component.width}x{component.height}")
    buffer.queue()
else:
    print("No image available")

ia.stop()
ia.destroy()
h.reset()
```

## Handle Multi-Part Payloads

Some cameras transmit multi-part data (e.g., image + 3D coordinates). Access multiple components from the payload.

```python
from harvesters.core import Harvester

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()
ia = h.create(0)

ia.start()

with ia.fetch() as buffer:
    payload = buffer.payload

    # Check payload type
    print(f"Payload type: {payload.payload_type}")

    # Iterate through all components
    for i, component in enumerate(payload.components):
        print(f"Component {i}:")
        print(f"  Data format: {component.data_format}")
        print(f"  Dimensions: {component.width}x{component.height}")
        print(f"  Data size: {component.data.size}")

        # Each component may have different pixel format
        # Handle accordingly based on data_format

ia.stop()
ia.destroy()
h.reset()
```

## Handle RGB and Color Formats

Process color images by reshaping based on pixel format and number of components per pixel.

```python
from harvesters.core import Harvester
from harvesters.util.pfnc import mono_location_formats, rgb_formats, bgr_formats, rgba_formats, bgra_formats

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()
ia = h.create(0)

# Configure for color output
ia.remote_device.node_map.PixelFormat.value = 'RGB8'

ia.start()

with ia.fetch() as buffer:
    component = buffer.payload.components[0]
    data_format = component.data_format
    width = component.width
    height = component.height

    # Reshape based on pixel format
    if data_format in mono_location_formats:
        # Monochrome image
        image = component.data.reshape(height, width)
    elif data_format in rgb_formats or data_format in rgba_formats:
        # RGB or RGBA image
        num_channels = int(component.num_components_per_pixel)
        image = component.data.reshape(height, width, num_channels)
    elif data_format in bgr_formats or data_format in bgra_formats:
        # BGR or BGRA image - swap R and B channels for RGB
        num_channels = int(component.num_components_per_pixel)
        image = component.data.reshape(height, width, num_channels)
        image = image[:, :, ::-1]  # Convert BGR to RGB

    print(f"Image shape: {image.shape}")
    print(f"Pixel format: {data_format}")

ia.stop()
ia.destroy()
h.reset()
```

## Register Callbacks for Events

Register callbacks to receive notifications for acquisition events like new buffer availability.

```python
from harvesters.core import Harvester, Callback, ImageAcquirer

class NewBufferCallback(Callback):
    def emit(self, context=None):
        if isinstance(context, ImageAcquirer):
            print("New buffer available!")
            # Process or signal other threads

class IncompleteBufferCallback(Callback):
    def emit(self, context=None):
        print("Warning: Incomplete buffer received")

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()
ia = h.create(0)

# Add callbacks for events
ia.add_callback(ImageAcquirer.Events.NEW_BUFFER_AVAILABLE, NewBufferCallback())
ia.add_callback(ImageAcquirer.Events.INCOMPLETE_BUFFER, IncompleteBufferCallback())

# List supported events
print(ia.supported_events)
# [Events.TURNED_OBSOLETE, Events.RETURN_ALL_BORROWED_BUFFERS,
#  Events.READY_TO_STOP_ACQUISITION, Events.NEW_BUFFER_AVAILABLE,
#  Events.INCOMPLETE_BUFFER, Events.ON_CHUNK_DATA_UPDATED,
#  Events.ON_EVENT_DATA_UPDATED]

ia.start(run_as_thread=True)

# ... acquisition runs with callbacks ...

# Remove callbacks when done
ia.remove_callback(ImageAcquirer.Events.NEW_BUFFER_AVAILABLE)
ia.remove_callbacks()  # Remove all

ia.stop()
ia.destroy()
h.reset()
```

## Context Manager Pattern - Safe Resource Management

Use the `with` statement for automatic resource cleanup to prevent resource leaks.

```python
from harvesters.core import Harvester
import numpy as np

# Using context managers for automatic cleanup
with Harvester() as h:
    h.add_file('/path/to/gentl_producer.cti')
    h.update()

    with h.create(0) as ia:
        # Configure camera
        ia.remote_device.node_map.Width.value = 640
        ia.remote_device.node_map.Height.value = 480
        ia.remote_device.node_map.PixelFormat.value = 'Mono8'

        ia.start()

        # Acquire multiple images
        for i in range(10):
            with ia.fetch() as buffer:
                component = buffer.payload.components[0]
                image = component.data.reshape(
                    component.height, component.width
                )
                print(f"Frame {i}: mean={np.mean(image):.2f}")

        ia.stop()
        # ia.destroy() called automatically

    # h.reset() called automatically

# All resources properly released
```

## Pixel Format Utilities

Use the PFNC (Pixel Format Naming Convention) utilities to work with different pixel formats.

```python
from harvesters.util.pfnc import (
    get_effective_pixel_size,
    get_bits_per_pixel,
    is_custom,
    mono_location_formats,
    rgb_formats,
    bayer_location_formats,
    uint8_formats,
    uint16_formats,
    dict_by_names,
    dict_by_ints
)

# Get pixel size information
pixel_format_value = dict_by_names['Mono8']
print(f"Mono8 value: {pixel_format_value}")
print(f"Effective pixel size: {get_effective_pixel_size(pixel_format_value)} bits")
print(f"Bits per pixel: {get_bits_per_pixel('Mono8')}")

# Convert between symbolic name and integer value
mono8_int = dict_by_names['Mono8']
mono8_name = dict_by_ints[mono8_int]
print(f"Mono8: {mono8_int} -> {mono8_name}")

# Check format categories
print(f"Mono formats: {mono_location_formats[:5]}")
print(f"RGB formats: {rgb_formats}")
print(f"Bayer formats: {bayer_location_formats[:4]}")

# Check if format is in a category
pixel_format = 'RGB8'
if pixel_format in rgb_formats:
    print(f"{pixel_format} is an RGB format")
if pixel_format in uint8_formats:
    print(f"{pixel_format} uses 8-bit unsigned integers")
```

## Acquisition Statistics

Monitor acquisition performance using the built-in statistics.

```python
from harvesters.core import Harvester
import time

h = Harvester()
h.add_file('/path/to/gentl_producer.cti')
h.update()
ia = h.create(0)

ia.start()

# Acquire some images
for _ in range(100):
    with ia.fetch() as buffer:
        pass

# Access statistics
stats = ia.statistics
print(f"Total images acquired: {stats.num_images}")
print(f"Acquisition started: {stats.has_acquired_1st_image}")

ia.stop()
ia.destroy()
h.reset()
```

## Summary

Harvesters provides a streamlined interface for machine vision image acquisition in Python, making it ideal for computer vision applications, quality inspection systems, scientific imaging, and industrial automation. The library's key strengths include support for multiple transport layers through GenTL Producers, dynamic camera configuration via GenICam, and seamless NumPy integration for image processing. Common use cases include real-time image capture for object detection, high-speed inspection on production lines, multi-camera synchronization, and research imaging applications.

Integration with other libraries is straightforward: acquired images can be directly passed to OpenCV for processing (`cv2.imshow(image)`), converted to PIL Images (`Image.fromarray(image)`), or visualized with Matplotlib (`plt.imshow(image)`). For GUI applications, Harvesters has a companion project called Harvester GUI that provides visualization capabilities. The library supports threading models for background acquisition, making it suitable for applications that require continuous image streaming while performing other tasks. For production deployments, use the context manager pattern to ensure proper resource cleanup and implement proper error handling with timeout mechanisms.