Bug in cameraconfig

[jeffwitz]

I try to implement RGB harvester camera using USB3 driver.

Here is a basic code example that allows one to take image in RGB and configurate the number of channels in the cameraconfig block.

I see that for that same framerate in the cameraconfig, and the same framerate in the image acquisition, if you apply 3 channels, then the speed of the cameraconfig interface can be divided by 3 ...

It seems that is causes the buffer to get corrupted, I can try anything I want for copy, deepcopy, reshape inside or outside of the with ... Nothing seems to work.

It is surely a bug that comes from how color images are managed in the cameraconfig, as when I close the interface everything is ok, in the OpenCV displayer ...

Available if you want to discuss how to debug it.

Regards

Details

```python import platform import numpy as np from crappy.camera.meta_camera import Camera import time from typing import Optional,Tuple, List, Dict, Any from crappy._global import OptionalModule import cv2 import copy try : from harvesters.core import Harvester from harvesters.util.pfnc import mono_location_formats,rgb_formats from harvesters.util.pfnc import bgr_formats from harvesters.util.pfnc import rgba_formats, bgra_formats except (ModuleNotFoundError, ImportError): harverster_exception = OptionalModule( "Harvester", "To use Genicam cameras, please install the " "harvester module : python -m pip intstall harvesters" ) try: import cv2 except (ModuleNotFoundError, ImportError): opencv = OptionalModule("OpenCv", "to use genicam" "camera OpenCV is requirerd official official OpenCV module :" "python -m pip install opencv-python") class GenicamHarvesterCamera(Camera): def __init__(self) -> None: super().__init__() Camera.__init__(self) self.camera = None if platform.system() == 'Windows': # Windows self.cti_file_path = "C:\\ath\\to\\cti\\file" if platform.system() == 'Linux': # Linux self.cti_file_path = ('/opt/mvIMPACT_Acquire/lib/x86_64/' 'mvGenTLProducer.cti') self.num_image = 0 self.h = None self.add_choice_setting(name="channels", choices=('1', '3'), default='1') def open(self, cti_file_path = None, model: Optional[str] = None, serial_number: Optional[str] = None, data_format: Optional[str] = None, # exposure_time_us: Optional[int] = None, **kwargs: any) -> None: self.h = Harvester() if cti_file_path is not None: self.cti_file_path = cti_file_path self.h.add_file(self.cti_file_path) self.h.update() list_dict_devices = self.h.device_info_list if len(list_dict_devices) == 0: raise IOError("---------------------------------------\n" "No camera found \n" "Check: Your GenTL Producer \n" "Your network : you must be in the same network in the " "same subnet mask\n" "the camera parameters you enter could be wrong\n" "Maybe just the wire 🔌" ) if model is None: self.camera = self.h.create() else: open_dict={} if serial_number is none: open_dict['model']=model self.camera = self.h.create(open_dict) else: open_dict['serial_number']=serial_number self.camera = self.h.create(open_dict) self.camera.start() # et = self.camera.remote_device.node_map.ExposureTime # self.add_scale_setting(name = 'exposure_time_us', # highest =float(et.max), # lowest = float(et.min), # getter = self._expo_time_get, # setter = self._expo_time_set, # default = 15000, # step = 100) self.set_all(**kwargs) def get_image(self) -> Tuple[Dict[str, Any], np.ndarray]: if not self.camera: raise RuntimeError("Cameras(s) detected but " "No camera opened, check your model, or serial_number") with self.camera.fetch() as buffer: component = buffer.payload.components[0] width = component.width height = component.height num_comp=component.num_components_per_pixel data_format = component.data_format # img= copy.deepcopy( # component.data.reshape((int(height),int(width)))) img_temp = np.array(component.data) img_data = copy.deepcopy(img_temp) if data_format == 'BayerRG8': img_data = cv2.cvtColor(img_data.reshape( (int(height),int(width)) ), cv2.COLOR_BAYER_BG2BGR ); if self.channels == '1': img_data = cv2.cvtColor(img_data,cv2.COLOR_BGR2GRAY) if self.channels == '3': img_data = img_data.copy() #if data_format == 'BayerRG8': # lots of cases metadata = { 't(s)': time.time(), 'ImageUniqueID': self.num_image, } # exptime = self.camera.remote_device.node_map.ExposureTime self.num_image += 1 return metadata, img_data def close(self): if self.camera: self.camera.stop() self.camera.destroy() self.h.reset() else: print("Camera is not opened or already closed.") def _expo_time_get(self): return self.camera.remote_device.node_map.ExposureTime.value def _expo_time_set(self,expo_time): self.camera.remote_device.node_map.ExposureTime.value = \ expo_time if __name__ == '__main__': import crappy cam1 = crappy.blocks.Camera( 'GenicamHarvesterCamera', config = True, # Before the test starts, displays a configuration window # for configuring the camera display_images = True, # During the test, the acquired images are # displayed in a dedicated window save_images = False, # Here, we don't want the images to be recorded # Sticking to default for the other arguments freq=20 ) # This Block allows the user to properly exit the script stop = crappy.blocks.StopButton( # No specific argument to give for this Block ) crappy.start() ```

[WeisLeDocto]

I see that for that same framerate in the cameraconfig, and the same framerate in the image acquisition, if you apply 3 channels, then the speed of the cameraconfig interface can be divided by 3 ...

Well it's only normal that the speed decreases when switching from 1 to 3 channels, I can observe the same behavior with my webcam (drop from 30fps to 18fps when switching to color image in 720p). I wouldn't be too concerned about this aspect, although it's indeed strange that your performance is affected that much. What image size are the images you're acquiring ?

It seems that is causes the buffer to get corrupted, I can try anything I want for copy, deepcopy, reshape inside or outside of the with ... Nothing seems to work.

Can't really help without more information. Seems more related to the way data is acquired from the camera than to the way Crappy handles the images.

if self.channels == '1':
    img_data = cv2.cvtColor(img_data,cv2.COLOR_BGR2GRAY)
if self.channels == '3':
    img_data = img_data.copy()

Interestingly, in the case when you have only 1 channel, a new object is for sure created. In the case when you have 3 channels, although .copy() should work, maybe try something basic like img_data = (img_data + 1) - 1 to make sure that img_data stores a fresh independent copy of the original data.

It is surely a bug that comes from how color images are managed in the cameraconfig, as when I close the interface everything is ok, in the OpenCV displayer ...

The only two differences between color and grey scale images in the camera configuration window are:

https://github.com/LaboratoireMecaniqueLille/crappy/blob/e069fd4c6cf7209c56833290a24f3f4235c9684c/src/crappy/tool/camera_config/camera_config.py#L854-L856

https://github.com/LaboratoireMecaniqueLille/crappy/blob/e069fd4c6cf7209c56833290a24f3f4235c9684c/src/crappy/tool/camera_config/camera_config.py#L968-L970

If there's any difference in the way the color and grey scale images are handled that would be relevant to your problem, it would be further inside tkinter or Pillow.

[jeffwitz]

well finally the bug was not really in camera_config. In fact the bug comes from Harvester that does not allow to wait to long for taking 2 buffers as I report in the issue #452 in Harvester github.

for those interested in, you can find a code example with a acquisition thread that made in order to test the draft of GenTL camera in crappy

import platform
import numpy as np
from crappy.camera.meta_camera import Camera
import time
from typing import Optional, Tuple, List, Dict, Any
from threading import Thread, RLock
import cv2
import copy

try :
    from harvesters.core import Harvester
    from harvesters.util.pfnc import mono_location_formats,rgb_formats
    from harvesters.util.pfnc import bgr_formats
    from harvesters.util.pfnc import  rgba_formats, bgra_formats
except (ModuleNotFoundError, ImportError):
    harverster_exception = OptionalModule(
    "Harvester", "To use Genicam cameras, please install the "
    "harvester module : python -m pip intstall harvesters"
    )

try:
    import cv2
except (ModuleNotFoundError, ImportError):
    opencv = OptionalModule("OpenCv", "to use genicam"
    "camera OpenCV is requirerd official OpenCV module :"
    "python -m pip install opencv-python")

class GenicamHarvesterCamera(Camera):
    def __init__(self) -> None:
        super().__init__()
        Camera.__init__(self)
        self.camera = None
        self._frame_grabber = None
        self._lock = RLock()
        self._frame = None
        self._stop = False
        self.h = None
        self.model = None
        self.serial_number = None
        self.num_image = 0
        if platform.system() == 'Windows':  # Windows
            self.cti_file_path = "C:\\ath\\to\\cti\\file"
        if platform.system() == 'Linux': # Linux
            self.cti_file_path = ('/opt/mvIMPACT_Acquire/lib/x86_64/'
              'mvGenTLProducer.cti')
        self.add_choice_setting(name="channels", choices=('1', '3'), default='1')

    def open(self,
             cti_file_path=None,
             model =None,
             serial_number = None,
             data_format: Optional[str]=None,
             **kwargs: any) -> None:
        self.model = model
        self.serial_number = serial_number
        self.h = Harvester()
        self.cti_file_path = cti_file_path or self.cti_file_path
        self.h.add_file(self.cti_file_path)
        self.h.update()
        self.devices = self.h.device_info_list
        device_info = self.find_device_info()
        self.camera = self.h.create(device_info) if device_info \
          else self.h.create()
        self.camera.start()
        self._stop = False
        self._frame_grabber = Thread(target=self._grab_frame)
        self._frame_grabber.start()
        self.set_all(**kwargs)

    def _grab_frame(self):
        while not self._stop:
            with self.camera.fetch() as buffer:
                component = buffer.payload.components[0]
                frame_data = np.array(component.data).reshape(
                  (component.height, component.width))
                if component.data_format == 'BayerRG8':
                    frame_data = cv2.cvtColor(
                      frame_data, cv2.COLOR_BAYER_BG2BGR)
                    if self.channels == '1':
                        frame_data = cv2.cvtColor(
                          frame_data, cv2.COLOR_BGR2GRAY)
                with self._lock:
                    self._frame = frame_data

    def get_image(self) -> Tuple[Dict[str, Any], np.ndarray]:
        with self._lock:
            if self._frame is None:
                raise RuntimeError("No frame available")
            metadata = {'t(s)': time.time(),
                        'ImageUniqueID': self.num_image}
            self.num_image += 1
            return metadata, self._frame.copy()

    def close(self):
        self._stop = True
        if self._frame_grabber is not None:
            self._frame_grabber.join()
        self.camera.stop()
        self.camera.destroy()
        self.h.reset()

    def find_device_info(self):
        # Print the available devices for debugging.
        print("List of available devices:", self.devices)

        # Handle cases where both model and serial_number might be None.
        if self.model is None and self.serial_number is None:
            return None

        # Iterate over each device in the list.
        for device in self.devices:
            print("Examining device:", device)

            # Check if model matches (if model is specified).
            model_matches = (self.model is None or
                            device.property_dict['model'] == self.model)

            # Check if serial_number matches (if serial_number is specified).
            serial_matches = (self.serial_number is None or
                              device.property_dict.get('serial_number') == self.serial_number)

            # If both model and serial_number are provided, both must match.
            if model_matches and serial_matches:
                if self.model is not None and self.serial_number is not None:
                    print("Both model and serial_number match found.")
                    return {'model': self.model, 'serial_number': self.serial_number}
                elif self.model is not None:
                    print("Model match found.")
                    return {'model': self.model}
                elif self.serial_number is not None:
                    print("Serial_number match found.")
                    return {'serial_number': self.serial_number}

            else:
                if not model_matches:
                    print("Device model does not match:", device.property_dict['model'])
                if self.serial_number and not serial_matches:
                    print("Device serial number does not match:",
                          device.property_dict.get('serial_number'))

        # If no matching device is found, raise an exception.
        raise ValueError("No device found matching the provided model and/or serial_number")

if __name__ == '__main__':
    import crappy
    cam1 = crappy.blocks.Camera(
        'GenicamHarvesterCamera',
        config = True,
        model = "MV-CS060-10GC",
        serial_number='K36941209',
        display_images = True,  # During the test, the acquired images are
        )
    cam2 = crappy.blocks.Camera(
        'GenicamHarvesterCamera',
        config = True,
        model = "DFK 37AUX250",
        serial_number = '48910488',
        display_images = True,  # During the test, the acquired images are
        )

 # This Block allows the user to properly exit the script
    stop = crappy.blocks.StopButton(
        # No specific argument to give for this Block
    )
    crappy.start()

Still a long way to go before having a proper GenTL genicam camera in crappy ...

I will now closed this thread

[jeffwitz]

closed : Harvester bug Workaround : create an acquisition thread