diff --git a/csaxs_bec/device_configs/flomni_config.yaml b/csaxs_bec/device_configs/flomni_config.yaml index fd72916..1a3bfd1 100644 --- a/csaxs_bec/device_configs/flomni_config.yaml +++ b/csaxs_bec/device_configs/flomni_config.yaml @@ -393,18 +393,31 @@ cam_flomni_overview: readOnly: false readoutPriority: on_request -# cam_flomni_xeye: -# description: Camera flOMNI Xray eye ID101 -# deviceClass: csaxs_bec.devices.ids_cameras.ids_camera.IDSCamera -# deviceConfig: -# camera_ID: 101 -# bits_per_pixel: 24 -# channels: 3 -# m_n_colormode: 1 -# enabled: true -# onFailure: buffer -# readOnly: false -# readoutPriority: async +cam_xeye_mono: + description: Camera flOMNI Xray eye ID1 + deviceClass: csaxs_bec.devices.ids_cameras.ids_camera.IDSCamera + deviceConfig: + camera_id: 1 + bits_per_pixel: 24 + # channels: 3 + m_n_colormode: 1 + enabled: true + onFailure: buffer + readOnly: false + readoutPriority: async + +cam_xeye_rgb: + description: Camera flOMNI Xray eye ID203 + deviceClass: csaxs_bec.devices.ids_cameras.ids_camera.IDSCamera + deviceConfig: + camera_id: 203 + bits_per_pixel: 24 + # channels: 3 + m_n_colormode: 1 + enabled: true + onFailure: buffer + readOnly: false + readoutPriority: async # ############################################################ diff --git a/csaxs_bec/devices/ids_cameras/__init__.py b/csaxs_bec/devices/ids_cameras/__init__.py index a16d89e..ce0499a 100644 --- a/csaxs_bec/devices/ids_cameras/__init__.py +++ b/csaxs_bec/devices/ids_cameras/__init__.py @@ -1 +1 @@ -from .ids_camera_new import IDSCamera +from .ids_camera import IDSCamera diff --git a/csaxs_bec/devices/ids_cameras/ids_camera.py b/csaxs_bec/devices/ids_cameras/ids_camera.py index 1250927..52c857b 100644 --- a/csaxs_bec/devices/ids_cameras/ids_camera.py +++ b/csaxs_bec/devices/ids_cameras/ids_camera.py @@ -1,403 +1,220 @@ +"""IDS Camera class for cSAXS IDS cameras.""" + +from __future__ import annotations + import threading import time +from typing import TYPE_CHECKING, Literal, Tuple, TypedDict import numpy as np +from bec_lib import messages from bec_lib.logger import bec_logger from ophyd import Component as Cpt -from ophyd import DeviceStatus, Kind, Signal, StatusBase from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase -from ophyd_devices.utils.bec_signals import PreviewSignal +from ophyd_devices.utils.bec_signals import AsyncSignal, PreviewSignal + +from csaxs_bec.devices.ids_cameras.base_integration.camera import Camera + +if TYPE_CHECKING: + from bec_lib.devicemanager import ScanInfo + from pydantic import ValidationInfo + logger = bec_logger.logger -class ROISignal(Signal): - """ - Signal to handle the Region of Interest (ROI) for the IDS camera. - It is a tuple of (x, y, width, height). - """ - - def __init__( - self, - *, - name, - roi: tuple | None = None, - value=0, - dtype=None, - shape=None, - timestamp=None, - parent=None, - labels=None, - kind=Kind.hinted, - tolerance=None, - rtolerance=None, - metadata=None, - cl=None, - attr_name="", - ): - super().__init__( - name=name, - value=value, - dtype=dtype, - shape=shape, - timestamp=timestamp, - parent=parent, - labels=labels, - kind=kind, - tolerance=tolerance, - rtolerance=rtolerance, - metadata=metadata, - cl=cl, - attr_name=attr_name, - ) - self.roi = roi - - def get(self, **kwargs): - image = self.parent.image_data.get().data - if not isinstance(image, np.ndarray): - return -1 # -1 if no valid image is available - - if self.roi is None: - roi = (0, 0, image.shape[1], image.shape[0]) - else: - roi = self.roi - if len(image.shape) > 2: - image = np.sum(image, axis=2) # Convert to grayscale if it's a color image - return np.sum(image[roi[1] : roi[1] + roi[3], roi[0] : roi[0] + roi[2]], (0, 1)) - - class IDSCamera(PSIDeviceBase): - """ " - #--------------------------------------------------------------------------------------------------------------------------------------- + """IDS Camera class for cSAXS. - #Variables - hCam = ueye.HIDS(202) #0: first available camera; 1-254: The camera with the specified camera ID - sInfo = ueye.SENSORINFO() - cInfo = ueye.CAMINFO() - pcImageMemory = ueye.c_mem_p() - MemID = ueye.int() - rectAOI = ueye.IS_RECT() - pitch = ueye.INT() - nBitsPerPixel = ueye.INT(24) #24: bits per pixel for color mode; take 8 bits per pixel for monochrome - channels = 3 #3: channels for color mode(RGB); take 1 channel for monochrome - m_nColorMode = ueye.INT(1) # Y8/RGB16/RGB24/REG32 (1 for our color cameras) - bytes_per_pixel = int(nBitsPerPixel / 8) - - ids_cam - ... + This class inherits from PSIDeviceBase and implements the necessary methods + to interact with the IDS camera using the pyueye library. """ - USER_ACCESS = ["start_live_mode", "stop_live_mode", "set_roi", "width", "height"] + image = Cpt(PreviewSignal, name="image", ndim=2, doc="Preview signal for the camera.") + roi_signal = Cpt( + AsyncSignal, + name="roi_signal", + ndim=0, + max_size=1000, + doc="Signal for the region of interest (ROI).", + async_update={"type": "add", "max_shape": [None]}, + ) - image_data = Cpt(PreviewSignal, ndim=2, kind=Kind.omitted) - # roi_bot_left = Cpt(ROISignal, roi=(400, 525, 118, 105), kind=Kind.normal) - # roi_bot_right = Cpt(ROISignal, roi=(518, 525, 118, 105), kind=Kind.normal) - # roi_top_left = Cpt(ROISignal, roi=(400, 630, 118, 105), kind=Kind.normal) - # roi_top_right = Cpt(ROISignal, roi=(518, 630, 118, 105), kind=Kind.normal) - # roi_signal = Cpt(ROISignal, kind=Kind.normal, doc="Region of Interest signal") + USER_ACCESS = ["live_mode", "mask", "set_rect_roi", "get_last_image"] def __init__( self, - prefix="", *, name: str, - camera_ID: int, - bits_per_pixel: int, - channels: int, - m_n_colormode: int, - kind=None, - device_manager=None, + camera_id: int, + prefix: str = "", + scan_info: ScanInfo | None = None, + m_n_colormode: Literal[0, 1, 2, 3] = 1, + bits_per_pixel: Literal[8, 24] = 24, + live_mode: bool = False, **kwargs, ): + """Initialize the IDS Camera. - super().__init__( - prefix=prefix, name=name, kind=kind, device_manager=device_manager, **kwargs - ) - self.camera_ID = camera_ID - self.bits_per_pixel = bits_per_pixel - self.bytes_per_pixel = None - self.channels = channels - self._m_n_colormode_input = m_n_colormode - self.m_n_colormode = None - self.ueye = ueye - self.h_cam = None - self.s_info = None - self.data_thread = None - self.c_info = None - self.pc_image_memory = None - self.mem_id = None - self.rect_aoi = None - self.pitch = None - self.n_bits_per_pixel = None - self.width = None - self.height = None - self.thread_event = threading.Event() - self.data_thread = None - self._roi: tuple | None = None # x, y, width, height - logger.info( - f"Deprecation warning: The IDSCamera class is deprecated. Use the new IDSCameraNew class instead." + Args: + name (str): Name of the device. + camera_id (int): The ID of the camera device. + prefix (str): Prefix for the device. + scan_info (ScanInfo | None): Scan information for the device. + m_n_colormode (Literal[0, 1, 2, 3]): Color mode for the camera. + bits_per_pixel (Literal[8, 24]): Number of bits per pixel for the camera. + live_mode (bool): Whether to enable live mode for the camera. + """ + super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs) + self._live_mode_thread: threading.Thread | None = None + self._stop_live_mode_event: threading.Event = threading.Event() + self.cam = Camera( + camera_id=camera_id, + m_n_colormode=m_n_colormode, + bits_per_pixel=bits_per_pixel, + connect=False, ) + self._live_mode = False + self._inputs = {"live_mode": live_mode} + self._mask = np.zeros((1, 1), dtype=np.uint8) - def set_roi(self, x: int, y: int, width: int, height: int): - self._roi = (x, y, width, height) + ############## Live Mode Methods ############## - def start_backend(self): - if self.ueye is None: - raise ImportError("The pyueye library is not installed.") - self.h_cam = self.ueye.HIDS( - self.camera_ID - ) # 0: first available camera; 1-254: The camera with the specified camera ID - self.s_info = self.ueye.SENSORINFO() - self.c_info = self.ueye.CAMINFO() - self.pc_image_memory = self.ueye.c_mem_p() - self.mem_id = self.ueye.int() - self.rect_aoi = self.ueye.IS_RECT() - self.pitch = self.ueye.INT() - self.n_bits_per_pixel = self.ueye.INT( - self.bits_per_pixel - ) # 24: bits per pixel for color mode; take 8 bits per pixel for monochrome - self.m_n_colormode = self.ueye.INT( - self._m_n_colormode_input - ) # Y8/RGB16/RGB24/REG32 (1 for our color cameras) - self.bytes_per_pixel = int(self.n_bits_per_pixel / 8) + @property + def mask(self) -> np.ndarray: + """Return the current region of interest (ROI) for the camera.""" + return self._mask - # Starts the driver and establishes the connection to the camera - ret = self.ueye.is_InitCamera(self.h_cam, None) - if ret != self.ueye.IS_SUCCESS: - print("is_InitCamera ERROR") + @mask.setter + def mask(self, value: np.ndarray): + """ + Set the region of interest (ROI) for the camera. - # Reads out the data hard-coded in the non-volatile camera memory and writes it to the data structure that c_info points to - ret = self.ueye.is_GetCameraInfo(self.h_cam, self.c_info) - if ret != self.ueye.IS_SUCCESS: - print("is_GetCameraInfo ERROR") + Args: + value (np.ndarray): The mask to set as the ROI. + """ + if value.ndim != 2: + raise ValueError("ROI mask must be a 2D array.") + img_shape = (self.cam.cam.height.value, self.cam.cam.width.value) + if value.shape[0] != img_shape[0] or value.shape[1] != img_shape[1]: + raise ValueError( + f"ROI mask shape {value.shape} does not match image shape {img_shape}." + ) + self._mask = value - # You can query additional information about the sensor type used in the camera - ret = self.ueye.is_GetSensorInfo(self.h_cam, self.s_info) - if ret != self.ueye.IS_SUCCESS: - print("is_GetSensorInfo ERROR") + @property + def live_mode(self) -> bool: + """Return whether the camera is in live mode.""" + return self._live_mode - ret = self.ueye.is_ResetToDefault(self.h_cam) - if ret != self.ueye.IS_SUCCESS: - print("is_ResetToDefault ERROR") - - # Set display mode to DIB - ret = self.ueye.is_SetDisplayMode(self.h_cam, self.ueye.IS_SET_DM_DIB) - - # Set the right color mode - if ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_BAYER - ): - # setup the color depth to the current windows setting - self.ueye.is_GetColorDepth(self.h_cam, self.n_bits_per_pixel, self.m_n_colormode) - bytes_per_pixel = int(self.n_bits_per_pixel / 8) - print("IS_COLORMODE_BAYER: ") - print("\tm_n_colormode: \t\t", self.m_n_colormode) - print("\tn_bits_per_pixel: \t\t", self.n_bits_per_pixel) - print("\tbytes_per_pixel: \t\t", bytes_per_pixel) - print() - - elif ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_CBYCRY - ): - # for color camera models use RGB32 mode - m_n_colormode = self.ueye.IS_CM_BGRA8_PACKED - n_bits_per_pixel = self.ueye.INT(32) - bytes_per_pixel = int(self.n_bits_per_pixel / 8) - print("IS_COLORMODE_CBYCRY: ") - print("\tm_n_colormode: \t\t", m_n_colormode) - print("\tn_bits_per_pixel: \t\t", n_bits_per_pixel) - print("\tbytes_per_pixel: \t\t", bytes_per_pixel) - print() - - elif ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_MONOCHROME - ): - # for color camera models use RGB32 mode - m_n_colormode = self.ueye.IS_CM_MONO8 - n_bits_per_pixel = self.ueye.INT(8) - bytes_per_pixel = int(n_bits_per_pixel / 8) - print("IS_COLORMODE_MONOCHROME: ") - print("\tm_n_colormode: \t\t", m_n_colormode) - print("\tn_bits_per_pixel: \t\t", n_bits_per_pixel) - print("\tbytes_per_pixel: \t\t", bytes_per_pixel) - print() - - else: - # for monochrome camera models use Y8 mode - m_n_colormode = self.ueye.IS_CM_MONO8 - n_bits_per_pixel = self.ueye.INT(8) - bytes_per_pixel = int(n_bits_per_pixel / 8) - print("else") - - # Can be used to set the size and position of an "area of interest"(AOI) within an image - ret = self.ueye.is_AOI( - self.h_cam, - self.ueye.IS_AOI_IMAGE_GET_AOI, - self.rect_aoi, - self.ueye.sizeof(self.rect_aoi), - ) - if ret != self.ueye.IS_SUCCESS: - print("is_AOI ERROR") - - self.width = self.rect_aoi.s32Width - self.height = self.rect_aoi.s32Height - - # Prints out some information about the camera and the sensor - print("Camera model:\t\t", self.s_info.strSensorName.decode("utf-8")) - print("Camera serial no.:\t", self.c_info.SerNo.decode("utf-8")) - print("Maximum image width:\t", self.width) - print("Maximum image height:\t", self.height) - print() - - # --------------------------------------------------------------------------------------------------------------------------------------- - - # Allocates an image memory for an image having its dimensions defined by width and height and its color depth defined by n_bits_per_pixel - ret = self.ueye.is_AllocImageMem( - self.h_cam, - self.width, - self.height, - self.n_bits_per_pixel, - self.pc_image_memory, - self.mem_id, - ) - if ret != self.ueye.IS_SUCCESS: - print("is_AllocImageMem ERROR") - else: - # Makes the specified image memory the active memory - ret = self.ueye.is_SetImageMem(self.h_cam, self.pc_image_memory, self.mem_id) - if ret != self.ueye.IS_SUCCESS: - print("is_SetImageMem ERROR") + @live_mode.setter + def live_mode(self, value: bool): + """Set the live mode for the camera.""" + if value != self._live_mode: + if self.cam._connected is False: # $ pylint: disable=protected-access + self.cam.on_connect() + self._live_mode = value + if value: + self._start_live() else: - # Set the desired color mode - ret = self.ueye.is_SetColorMode(self.h_cam, self.m_n_colormode) + self._stop_live() - # Activates the camera's live video mode (free run mode) - ret = self.ueye.is_CaptureVideo(self.h_cam, self.ueye.IS_DONT_WAIT) - if ret != self.ueye.IS_SUCCESS: - print("is_CaptureVideo ERROR") + def set_rect_roi(self, x: int, y: int, width: int, height: int): + """Set the rectangular region of interest (ROI) for the camera.""" + if x < 0 or y < 0 or width <= 0 or height <= 0: + raise ValueError("ROI coordinates and dimensions must be positive integers.") + img_shape = (self.cam.cam.height.value, self.cam.cam.width.value) + if x + width > img_shape[1] or y + height > img_shape[0]: + raise ValueError("ROI exceeds camera dimensions.") + mask = np.zeros(img_shape, dtype=np.uint8) + mask[y : y + height, x : x + width] = 1 + self.mask = mask - # Enables the queue mode for existing image memory sequences - ret = self.ueye.is_InquireImageMem( - self.h_cam, - self.pc_image_memory, - self.mem_id, - self.width, - self.height, - self.n_bits_per_pixel, - self.pitch, + def _start_live(self): + """Start the live mode for the camera.""" + if self._live_mode_thread is not None: + logger.info("Live mode thread is already running.") + return + self._stop_live_mode_event.clear() + self._live_mode_thread = threading.Thread( + target=self._live_mode_loop, args=(self._stop_live_mode_event,) ) - if ret != self.ueye.IS_SUCCESS: - print("is_InquireImageMem ERROR") + self._live_mode_thread.start() + + def _stop_live(self): + """Stop the live mode for the camera.""" + if self._live_mode_thread is None: + logger.info("Live mode thread is not running.") + return + self._stop_live_mode_event.set() + self._live_mode_thread.join(timeout=5) + if self._live_mode_thread.is_alive(): + logger.warning("Live mode thread did not stop gracefully.") else: - print("Press q to leave the programm") - # startmeasureframerate = True - # Gain = False + self._live_mode_thread = None + logger.info("Live mode stopped.") - # Start live mode of camera immediately - self.start_live_mode() + def _live_mode_loop(self, stop_event: threading.Event): + """Loop to capture images in live mode.""" + while not stop_event.is_set(): + try: + self.process_data(self.cam.get_image_data()) + except Exception as e: + logger.error(f"Error in live mode loop: {e}") + break + stop_event.wait(0.2) # 5 Hz - def _start_data_thread(self): - self.data_thread = threading.Thread(target=self._receive_data_from_camera, daemon=True) - self.data_thread.start() + def process_data(self, image: np.ndarray | None): + """Process the image data before sending it to the preview signal.""" + if image is None: + return + if len(image.shape)==3 and image.shape[2] == 3: + image = image[:,:,::-1] + self.image.put(image) - def _receive_data_from_camera(self): - while not self.thread_event.is_set(): - if self.ueye is None: - print("pyueye library not available.") + def get_last_image(self) -> np.ndarray: + """Get the last captured image from the camera.""" + image = self.image.get() + if image: + return image.data + + ############## User Interface Methods ############## + + def on_connected(self): + """Connect to the camera.""" + self.cam.on_connect() + self.live_mode = self._inputs.get("live_mode", False) + self.set_rect_roi(0, 0, self.cam.cam.width.value, self.cam.cam.height.value) + + def on_destroy(self): + """Clean up resources when the device is destroyed.""" + self.cam.on_disconnect() + super().on_destroy() + + def on_trigger(self): + """Handle the trigger event.""" + if not self.live_mode: + return + image = self.image.get() + if image is not None: + image: messages.DevicePreviewMessage + if self.mask.shape[0:2] != image.data.shape[0:2]: + logger.info( + f"ROI shape does not match image shape, skipping ROI application for device {self.name}." + ) return - # In order to display the image in an OpenCV window we need to... - # ...extract the data of our image memory - array = self.ueye.get_data( - self.pc_image_memory, - self.width, - self.height, - self.n_bits_per_pixel, - self.pitch, - copy=False, - ) - # ...reshape it in an numpy array... - frame = np.reshape(array, (self.height.value, self.width.value, self.bytes_per_pixel)) - self.image_data.put(frame) - - time.sleep(0.1) - - def wait_for_connection(self, all_signals=False, timeout=10): - if ueye is None: - raise ImportError( - "The pyueye library is not installed or doesn't provide the necessary c libs" - ) - super().wait_for_connection(all_signals, timeout) - - def start_live_mode(self): - if self.data_thread is not None: - self.stop_live_mode() - self._start_data_thread() - - def stop_live_mode(self): - """Stopping the camera live mode.""" - self.thread_event.set() - if self.data_thread is not None: - self.data_thread.join() - self.thread_event.clear() - self.data_thread = None - - ######################################## - # Beamline Specific Implementations # - ######################################## - - def on_init(self) -> None: - """ - Called when the device is initialized. - - No signals are connected at this point. If you like to - set default values on signals, please use on_connected instead. - """ - - def on_connected(self) -> None: - """ - Called after the device is connected and its signals are connected. - Default values for signals should be set here. - """ - self.start_backend() - self.start_live_mode() - - def on_stage(self) -> DeviceStatus | StatusBase | None: - """ - Called while staging the device. - - Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object. - """ - - def on_unstage(self) -> DeviceStatus | StatusBase | None: - """Called while unstaging the device.""" - - def on_pre_scan(self) -> DeviceStatus | StatusBase | None: - """Called right before the scan starts on all devices automatically.""" - - def on_trigger(self) -> DeviceStatus | StatusBase | None: - """Called when the device is triggered.""" - - def on_complete(self) -> DeviceStatus | StatusBase | None: - """Called to inquire if a device has completed a scans.""" - - def on_kickoff(self) -> DeviceStatus | StatusBase | None: - """Called to kickoff a device for a fly scan. Has to be called explicitly.""" - - def on_stop(self) -> None: - """Called when the device is stopped.""" - - def on_destroy(self) -> None: - """Called when the device is destroyed. Cleanup resources here.""" - self.stop_live_mode() + if len(image.data.shape) == 3: + # If the image has multiple channels, apply the mask to each channel + data = image.data * self.mask[:, :, np.newaxis] # Apply mask to the image data + n_channels = 3 + else: + data = image.data * self.mask + n_channels = 1 + self.roi_signal.put(np.sum(data) / (np.sum(self.mask) * n_channels)) if __name__ == "__main__": - # Example usage - camera = IDSCamera(name="camera", camera_ID=201, bits_per_pixel=24, channels=3, m_n_colormode=1) - camera.wait_for_connection() - - camera.on_destroy() + # Example usage of the IDSCamera class + camera = IDSCamera(name="TestCamera", camera_id=201, live_mode=False) + print(f"Camera {camera.name} initialized with ID {camera.cam.camera_id}.") diff --git a/csaxs_bec/devices/ids_cameras/ids_camera_new.py b/csaxs_bec/devices/ids_cameras/ids_camera_new.py deleted file mode 100644 index 3111d61..0000000 --- a/csaxs_bec/devices/ids_cameras/ids_camera_new.py +++ /dev/null @@ -1,218 +0,0 @@ -"""IDS Camera class for cSAXS IDS cameras.""" - -from __future__ import annotations - -import threading -import time -from typing import TYPE_CHECKING, Literal, Tuple, TypedDict - -import numpy as np -from bec_lib import messages -from bec_lib.logger import bec_logger -from ophyd import Component as Cpt -from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase -from ophyd_devices.utils.bec_signals import AsyncSignal, PreviewSignal - -from csaxs_bec.devices.ids_cameras.base_integration.camera import Camera - -if TYPE_CHECKING: - from bec_lib.devicemanager import ScanInfo - from pydantic import ValidationInfo - - -logger = bec_logger.logger - - -class IDSCamera(PSIDeviceBase): - """IDS Camera class for cSAXS. - - This class inherits from PSIDeviceBase and implements the necessary methods - to interact with the IDS camera using the pyueye library. - """ - - image = Cpt(PreviewSignal, name="image", ndim=2, doc="Preview signal for the camera.") - roi_signal = Cpt( - AsyncSignal, - name="roi_signal", - ndim=0, - max_size=1000, - doc="Signal for the region of interest (ROI).", - async_update={"type": "add", "max_shape": [None]}, - ) - - USER_ACCESS = ["live_mode", "mask", "set_rect_roi", "get_last_image"] - - def __init__( - self, - *, - name: str, - camera_id: int, - prefix: str = "", - scan_info: ScanInfo | None = None, - m_n_colormode: Literal[0, 1, 2, 3] = 1, - bits_per_pixel: Literal[8, 24] = 24, - live_mode: bool = False, - **kwargs, - ): - """Initialize the IDS Camera. - - Args: - name (str): Name of the device. - camera_id (int): The ID of the camera device. - prefix (str): Prefix for the device. - scan_info (ScanInfo | None): Scan information for the device. - m_n_colormode (Literal[0, 1, 2, 3]): Color mode for the camera. - bits_per_pixel (Literal[8, 24]): Number of bits per pixel for the camera. - live_mode (bool): Whether to enable live mode for the camera. - """ - super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs) - self._live_mode_thread: threading.Thread | None = None - self._stop_live_mode_event: threading.Event = threading.Event() - self.cam = Camera( - camera_id=camera_id, - m_n_colormode=m_n_colormode, - bits_per_pixel=bits_per_pixel, - connect=False, - ) - self._live_mode = False - self._inputs = {"live_mode": live_mode} - self._mask = np.zeros((1, 1), dtype=np.uint8) - - ############## Live Mode Methods ############## - - @property - def mask(self) -> np.ndarray: - """Return the current region of interest (ROI) for the camera.""" - return self._mask - - @mask.setter - def mask(self, value: np.ndarray): - """ - Set the region of interest (ROI) for the camera. - - Args: - value (np.ndarray): The mask to set as the ROI. - """ - if value.ndim != 2: - raise ValueError("ROI mask must be a 2D array.") - img_shape = (self.cam.cam.height.value, self.cam.cam.width.value) - if value.shape[0] != img_shape[0] or value.shape[1] != img_shape[1]: - raise ValueError( - f"ROI mask shape {value.shape} does not match image shape {img_shape}." - ) - self._mask = value - - @property - def live_mode(self) -> bool: - """Return whether the camera is in live mode.""" - return self._live_mode - - @live_mode.setter - def live_mode(self, value: bool): - """Set the live mode for the camera.""" - if value != self._live_mode: - if self.cam._connected is False: # $ pylint: disable=protected-access - self.cam.on_connect() - self._live_mode = value - if value: - self._start_live() - else: - self._stop_live() - - def set_rect_roi(self, x: int, y: int, width: int, height: int): - """Set the rectangular region of interest (ROI) for the camera.""" - if x < 0 or y < 0 or width <= 0 or height <= 0: - raise ValueError("ROI coordinates and dimensions must be positive integers.") - img_shape = (self.cam.cam.height.value, self.cam.cam.width.value) - if x + width > img_shape[1] or y + height > img_shape[0]: - raise ValueError("ROI exceeds camera dimensions.") - mask = np.zeros(img_shape, dtype=np.uint8) - mask[y : y + height, x : x + width] = 1 - self.mask = mask - - def _start_live(self): - """Start the live mode for the camera.""" - if self._live_mode_thread is not None: - logger.info("Live mode thread is already running.") - return - self._stop_live_mode_event.clear() - self._live_mode_thread = threading.Thread( - target=self._live_mode_loop, args=(self._stop_live_mode_event,) - ) - self._live_mode_thread.start() - - def _stop_live(self): - """Stop the live mode for the camera.""" - if self._live_mode_thread is None: - logger.info("Live mode thread is not running.") - return - self._stop_live_mode_event.set() - self._live_mode_thread.join(timeout=5) - if self._live_mode_thread.is_alive(): - logger.warning("Live mode thread did not stop gracefully.") - else: - self._live_mode_thread = None - logger.info("Live mode stopped.") - - def _live_mode_loop(self, stop_event: threading.Event): - """Loop to capture images in live mode.""" - while not stop_event.is_set(): - try: - self.process_data(self.cam.get_image_data()) - except Exception as e: - logger.error(f"Error in live mode loop: {e}") - break - stop_event.wait(0.2) # 5 Hz - - def process_data(self, image: np.ndarray | None): - """Process the image data before sending it to the preview signal.""" - if image is None: - return - self.image.put(image) - - def get_last_image(self) -> np.ndarray: - """Get the last captured image from the camera.""" - image = self.image.get() - if image: - return image.data - - ############## User Interface Methods ############## - - def on_connected(self): - """Connect to the camera.""" - self.cam.on_connect() - self.live_mode = self._inputs.get("live_mode", False) - self.set_rect_roi(0, 0, self.cam.cam.width.value, self.cam.cam.height.value) - - def on_destroy(self): - """Clean up resources when the device is destroyed.""" - self.cam.on_disconnect() - super().on_destroy() - - def on_trigger(self): - """Handle the trigger event.""" - if not self.live_mode: - return - image = self.image.get() - if image is not None: - image: messages.DevicePreviewMessage - if self.mask.shape[0:2] != image.data.shape[0:2]: - logger.info( - f"ROI shape does not match image shape, skipping ROI application for device {self.name}." - ) - return - - if len(image.data.shape) == 3: - # If the image has multiple channels, apply the mask to each channel - data = image.data * self.mask[:, :, np.newaxis] # Apply mask to the image data - n_channels = 3 - else: - data = image.data * self.mask - n_channels = 1 - self.roi_signal.put(np.sum(data) / (np.sum(self.mask) * n_channels)) - - -if __name__ == "__main__": - # Example usage of the IDSCamera class - camera = IDSCamera(name="TestCamera", camera_id=201, live_mode=False) - print(f"Camera {camera.name} initialized with ID {camera.cam.camera_id}.")