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

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:

crappy/src/crappy/tool/camera_config/camera_config.py

Lines 854 to 856 in e069fd4

	# First, convert BGR to RGB
	if len(img.shape) == 3:
	img = img[:, :, ::-1]

crappy/src/crappy/tool/camera_config/camera_config.py

Lines 968 to 970 in e069fd4

	# The histogram is calculated on a grey level image
	if len(self._original_img.shape) == 3:
	hist_img = hist_img.convert('L')

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