bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(ids_camera): old ids_camera deleted, ids_camera_new is now ids_camera

Browse Source
This commit is contained in:
x01dc
2025-10-23 11:12:37 +02:00
committed by wyzula_j
parent 4723f6768b
commit 53294d95c0
4 changed files with 202 additions and 590 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/device_configs/flomni_config.yaml
+25 -12
View File
@@ -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
# ############################################################
csaxs_bec/devices/ids_cameras/__init__.py
+1 -1
View File
@@ -1 +1 @@
from .ids_camera_new import IDSCamera
from .ids_camera import IDSCamera
csaxs_bec/devices/ids_cameras/ids_camera.py
+176 -359
View File
@@ -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}.")
csaxs_bec/devices/ids_cameras/ids_camera_new.py
-218
View File
@@ -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}.")