From c2536691e98ca002d476728d8ac57d080cf4e66b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 07:39:16 +0000 Subject: [PATCH 1/5] Split the generic camera plugin from the pylablib one CameraBase (new module camera_base) holds the acquisition thread, the ROI shared with the viewer, the binning, the exposure and the FPS. It does not import pylablib, and opencv is only needed for the colour conversion. CameraBasePyLabLib becomes a thin subclass that adds the serial number setting, and the names the plugins import from camera_base_pylablib (cam_params, CameraCallback, Grab) are kept. CameraDriverBase documents the methods the plugin calls on its controller, those of the pylablib cameras, so that the driver of any other camera can be written against it. The behaviour is unchanged. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Tdiso1u1ZhikSutj2Qs5Cg --- .../hardware/camera_base.py | 488 ++++++++++++++++++ .../hardware/camera_base_pylablib.py | 479 +---------------- .../hardware/camera_driver.py | 102 ++++ 3 files changed, 599 insertions(+), 470 deletions(-) create mode 100644 src/pymodaq_plugins_utils/hardware/camera_base.py create mode 100644 src/pymodaq_plugins_utils/hardware/camera_driver.py diff --git a/src/pymodaq_plugins_utils/hardware/camera_base.py b/src/pymodaq_plugins_utils/hardware/camera_base.py new file mode 100644 index 0000000..c205ba5 --- /dev/null +++ b/src/pymodaq_plugins_utils/hardware/camera_base.py @@ -0,0 +1,488 @@ +"""Base class of the 2D detector plugins driving a camera. + +The plugin talks to its ``controller`` (the camera driver) through the methods listed in +:class:`~pymodaq_plugins_utils.hardware.camera_driver.CameraDriverBase`, which are those of the pylablib cameras. +It provides the acquisition in a thread, the ROI shared with the viewer, the binning, the exposure, the FPS and the +colour conversion; a plugin only has to create the controller in :meth:`CameraBase.ini_detector_custom`. +""" +import dataclasses +from typing import Type, Optional, Tuple + +try: + import cv2 +except ImportError: # only needed for the colour conversion + cv2 = None + +from pymodaq_data import DataToExport +from pymodaq_utils.logger import set_logger, get_module_name +from pymodaq_utils.utils import ThreadCommand +from pymodaq_gui.parameter import Parameter +try: + from pymodaq_gui.plotting.items.roi import RoiInfo # pymodaq > 5.1.x +except ImportError: + from pymodaq_gui.plotting.utils.plot_utils import RoiInfo + +from pymodaq.utils.data import DataFromPlugins, Axis +from pymodaq.control_modules.viewer_utility_classes import DAQ_Viewer_base, comon_parameters, main + +from qtpy import QtWidgets, QtCore +import numpy as np +from time import perf_counter + + +logger = set_logger(get_module_name(__file__)) + + +cam_params = [ + {'title': 'Camera name:', 'name': 'camera_name', 'type': 'str', 'value': ''}, + {'title': 'Color Conversion:', 'name': 'color_conversion', 'type': 'list', + 'limits': ['None', 'RGB2GRAY', 'BAYER_BG2RGB', 'BAYER_BG2GRAY'] if cv2 is not None else ['None']}, + {'title': 'ROI', 'name': 'roi', 'type': 'group', 'children': [ + {'title': 'Update ROI from Viewer', 'name': 'update_roi', 'type': 'led', 'value': False}, + {'title': 'Apply ROI', 'name': 'apply_roi', 'type': 'led', 'value': False}, + {'title': 'Clear ROI+Bin', 'name': 'clear_roi', 'type': 'bool_push', 'value': False}, + {'title': 'ROI:', 'name': 'roi_slices', 'type': 'str', 'value': ''}, + {'title': 'X binning', 'name': 'x_binning', 'type': 'int', 'value': 1}, + {'title': 'Y binning', 'name': 'y_binning', 'type': 'int', 'value': 1}, + ], }, + {'title': 'Image width', 'name': 'hdet', 'type': 'int', 'value': 1, 'readonly': True}, + {'title': 'Image height', 'name': 'vdet', 'type': 'int', 'value': 1, 'readonly': True}, + {'title': 'Timing', 'name': 'timing_opts', 'type': 'group', 'children': + [{'title': 'Exposure Time (ms)', 'name': 'exposure_time', 'type': 'int', 'value': 1}, + {'title': 'Compute FPS', 'name': 'fps_on', 'type': 'bool', 'value': True}, + {'title': 'FPS', 'name': 'fps', 'type': 'float', 'value': 0.0, 'readonly': True}] + }, + {'title': 'Buffer', 'name': 'buffer', 'type': 'group', 'children': [ + {'title': 'Size:', 'name': 'size', 'type': 'int', 'value': 10}, + {'title': 'mode:', 'name': 'mode', 'type': 'list', 'value': 'now', + 'limits': ['now', 'lastread', 'lastwait', 'start']}, + ]}, +] + + +@dataclasses.dataclass +class Grab: + do_acquisition: bool = True + snap: bool = False + since: str = 'now' + nframes: int = 1 + n_average: int = 1 + + +class CameraCallback(QtCore.QObject): + """Callback object """ + data_sig = QtCore.Signal(np.ndarray) + error = QtCore.Signal() + + def __init__(self, controller): + super().__init__() + # Set the wait function + self.controller = controller + self.do_acquisition = True + + def set_do_grab(self, mode: Grab): + self.do_acquisition = mode.do_acquisition + if mode.do_acquisition: + self.wait_for_acquisition(mode) + + def wait_for_acquisition(self, mode: Grab): + while self.do_acquisition: + try: + ind_average = 0 + while ind_average < mode.n_average: + ind_frames = 0 + while ind_frames < mode.nframes: + self.controller.wait_for_frame(since='now') + new_frames, rng = self.controller.read_multiple_images(missing_frame='skip', return_rng=True) + + # in case hardware has a random buffer size (ex: Andor CCD cameras) + if new_frames.shape[0] != mode.nframes: + new_frames = np.expand_dims(new_frames[-mode.nframes:], axis=0) + + if ind_average == 0 and ind_frames == 0: + shape = list(new_frames.shape[1:]) + shape = [mode.n_average, mode.nframes] + shape + frames = np.zeros(shape, dtype=new_frames.dtype) + nacq = rng[1] - rng[0] + frames[ind_average, ind_frames:nacq, ...] = new_frames + ind_frames += nacq + ind_average += 1 + self.data_sig.emit(frames) + QtCore.QThread.msleep(10) + except Exception as e: + logger.exception(str(e)) + self.error.emit() + break + QtWidgets.QApplication.processEvents() + if not self.do_acquisition or mode.snap: + break + + + +class CameraBase(DAQ_Viewer_base): + """Base implementation of a camera plugin + + The camera driver (``self.controller``, created in :meth:`ini_detector_custom`) follows + :class:`~pymodaq_plugins_utils.hardware.camera_driver.CameraDriverBase`. + """ + params = comon_parameters + cam_params + + callback_signal = QtCore.Signal(Grab) + live_mode_available = True + hardware_averaging = True + + def ini_attributes(self): + self.controller = None + self.callback_thread: QtCore.QThread = None + self.is_live: bool = False + self.Naverage: int = 1 + + self.x_axis: Axis = None + self.y_axis: Axis = None + + self.roi_select_info: RoiInfo = None + + self.last_tick = 0.0 # time counter used to compute FPS + self.fps = 0.0 + + self.data_shape: str = '' + + + def roi_select(self, roi_info: RoiInfo, ind_viewer: int = 0): + """ Automatically called when a user use the RoiSelect ROi from a 2D viewer""" + self.roi_select_info = roi_info + self.roi_select_viewer_index = ind_viewer + + if self.settings['roi', 'update_roi']: + self.settings['roi', 'roi_slices'] = str(roi_info.to_slices()) + if self.settings['roi', 'apply_roi']: + self.apply_roi() + + def apply_roi(self): + roi_info = RoiInfo.from_slices(eval(self.settings['roi', 'roi_slices'])) + new_roi = (roi_info.origin[1], roi_info.size[1], self.settings['roi', 'x_binning'], + roi_info.origin[0], roi_info.size[0], self.settings['roi', 'y_binning']) + self.update_rois(new_roi) + + def compute_axes(self): + (hstart, hend, vstart, vend, hbin, vbin) = self.controller.get_roi() + slices = [slice(vstart, vend, vbin), slice(hstart, hend, hbin)] + self.settings.child('roi', 'roi_slices').setValue(str(slices)) + roi_info = RoiInfo.from_slices(slices) + + self.x_axis = Axis('x_axis', offset=roi_info.origin[1], + scaling=self.settings['roi', 'x_binning'], + size=int(roi_info.size[1]), + index=1) + self.y_axis = Axis('y_axis', offset=roi_info.origin[0], + scaling=self.settings['roi', 'y_binning'], + size=int(roi_info.size[0]), + index=0) + + def clear_roi(self): + wdet, hdet = self.controller.get_detector_size() + self.settings.child('roi', 'x_binning').setValue(1) + self.settings.child('roi', 'y_binning').setValue(1) + + new_roi = (0, wdet, 1, 0, hdet, 1) + self.update_rois(new_roi) + + def update_rois(self, new_roi): + # In pylablib, ROIs compare as tuples + (new_x, new_width, new_xbinning, new_y, new_height, new_ybinning) = new_roi + if new_roi != self.controller.get_roi(): + # self.controller.set_attribute_value("ROIs",[new_roi]) + self.controller.set_roi(hstart=new_x, hend=new_x + new_width, vstart=new_y, vend=new_y + new_height, + hbin=new_xbinning, vbin=new_ybinning) + self.emit_status(ThreadCommand('Update_Status', [f'Changed ROI: {new_roi}'])) + self.controller.clear_acquisition() + self.controller.setup_acquisition() + # Finally, prepare view for displaying the new data + self._prepare_view() + self.compute_axes() + + def commit_settings(self, param: Parameter): + """Apply the consequences of a change of value in the detector settings + + Parameters + ---------- + param: Parameter + A given parameter (within detector_settings) whose value has been changed by the user + """ + if param.name() == "exposure_time": + self.controller.set_exposure(param.value()/1000) + + elif param.name() == "fps_on": + self.settings.child('timing_opts', 'fps').setOpts(visible=param.value()) + + elif param.name() == "apply_roi": + if param.value(): # Switching on ROI + self.apply_roi() + else: + self.clear_roi() + + elif param.name() in ['x_binning', 'y_binning']: + # We handle ROI and binning separately for clarity + (x0, w, y0, h, *_) = self.controller.get_roi() # Get current ROI + xbin = self.settings['roi', 'x_binning'] + ybin = self.settings['roi', 'y_binning'] + new_roi = (x0, w, xbin, y0, h, ybin) + self.update_rois(new_roi) + + elif param.name() == "clear_roi": + if param.value(): # Switching on ROI + self.clear_roi() + param.setValue(False) + + def ini_detector_custom(self, controller=None): + """Create (master) or store (slave) the camera driver in ``self.controller``, to reimplement""" + raise NotImplementedError + + def ini_detector(self, controller=None): + """Detector communication initialization + + Parameters + ---------- + controller: (object) + custom object of a PyMoDAQ plugin (Slave case). None if only one actuator/detector by controller + (Master case) + + Returns + ------- + info: str + initialized: bool + False if initialization failed otherwise True + """ + + self.ini_detector_custom(controller) + + self.get_device_info() + self.get_set_main_parameters() + self.setup_callback_thread() + self.controller.set_frame_format("array") + + info = "Initialized camera" + initialized = True + return info, initialized + + def get_device_info(self): + + device_info = self.controller.get_device_info() + + # Get camera name/model + if hasattr(device_info, 'name'): + self.settings.child('camera_name').setValue(device_info.name) + elif hasattr(device_info, 'model'): + self.settings.child('camera_name').setValue(device_info.model) + + def get_set_main_parameters(self): + # Set exposure time + self.controller.set_exposure(self.settings['timing_opts', 'exposure_time']/1000) + + # FPS visibility + self.settings.child('timing_opts', 'fps').setOpts(visible=self.settings['timing_opts', 'fps_on']) + + # get roi limits + self.controller.get_roi_limits() + + # Update image parameters + (hstart, hend, vstart, vend, hbin, vbin) = self.controller.get_roi() + height, width = self.controller.get_data_dimensions() + self.settings.child('roi', 'x_binning').setValue(hbin) + self.settings.child('roi', 'y_binning').setValue(vbin) + self.settings.child('hdet').setValue(width) + self.settings.child('vdet').setValue(height) + slices = [slice(vstart, vend, vbin), slice(hstart, hend, hbin)] + self.settings.child('roi', 'roi_slices').setValue(str(slices)) + self.compute_axes() + + @property + def callback(self) -> Type[CameraCallback]: + """ Return the class handling the wait for acquisition and signal emission + + Should be reimplemented as well as CameraCallback if needed + """ + return CameraCallback + + def setup_callback_thread(self): + callback = self.callback(self.controller) + self.settings.child('buffer', 'mode').setReadonly(True) + + + self.callback_thread = QtCore.QThread(self) # creation of a Qt5 thread + callback.moveToThread(self.callback_thread) # callback object will live within this thread + + callback.data_sig.connect( + self.emit_data) # when the wait for acquisition returns (with data taken), emit_data will be fired + callback.error.connect(self.handle_error) + + self.callback_signal.connect(callback.set_do_grab) + self.callback_thread.callback = callback + self.callback_thread.start() + + self._prepare_view() + + def handle_error(self): + self.stop() + + def _prepare_view(self): + """Preparing a data viewer by emitting temporary data. Typically, needs to be called whenever the + ROIs are changed""" + + height, width = self.controller.get_data_dimensions() + + self.settings.child('hdet').setValue(width) + self.settings.child('vdet').setValue(height) + mock_data = np.zeros((height, width)) + + if width != 1 and height != 1: + data_shape = 'Data2D' + else: + data_shape = 'Data1D' + + if data_shape != self.data_shape: + self.data_shape = data_shape + # init the viewers + self.data_grabed_signal_temp.emit([DataFromPlugins(name='Thorlabs Camera', + data=[np.squeeze(mock_data)], + dim=self.data_shape, + labels=[f'ThorCam_{self.data_shape}'])]) + QtWidgets.QApplication.processEvents() + + def grab_data(self, Naverage=1, **kwargs): + """ + Grabs the data. ASynchronous method (kinda). + ---------- + Naverage: (int) Number of averaging + kwargs: (dict) of others optionals arguments + """ + try: + # Warning, acquisition_in_progress returns 1,0 and not a real bool + self.is_live = kwargs.get('live', False) + self.Naverage = Naverage + + self.n_frames = 1 + + if not self.controller.acquisition_in_progress(): + self.controller.clear_acquisition() + self.controller.start_acquisition(nframes=self.n_frames) + #Then start the acquisition + self.callback_signal.emit(Grab(do_acquisition=True, + snap=not self.is_live, + n_average=Naverage, + nframes=self.n_frames, + since=self.settings['buffer', 'mode'])) + + except Exception as e: + self.emit_status(ThreadCommand('Update_Status', [str(e), "log"])) + + def emit_data(self, frame: np.ndarray=None): + """ Function used to emit data obtained by callback. + + Parameter + --------- + status: bool + If True a frame is available, If False, a Timeout occurred while waiting for the frame + + See Also + -------- + daq_utils.ThreadCommand + """ + try: + # Get data from buffer + if frame is None: + frame = self.controller.read_newest_image() + # Emit the frame. + if frame is not None: + conversion_str = self.settings['color_conversion'] + if conversion_str != "None" and cv2 is not None: + for ind_average in range(frame.shape[0]): + for ind_frame in range(frame.shape[1]): + if ind_frame == 0 and ind_average == 0: + new_frame = cv2.cvtColor(frame[ind_average, ind_frame, ...], + getattr(cv2, f'COLOR_{conversion_str}')) + shape = [frame.shape[0], frame.shape[1]] + list(new_frame.shape) + out_frames = np.zeros(shape, dtype=new_frame.dtype) + out_frames[ind_average, ind_frame, ...] = new_frame + else: + cv2.cvtColor(frame[ind_average, ind_frame, ...], + getattr(cv2, f'COLOR_{conversion_str}'), + out_frames[ind_average, ind_frame, ...]) + else: + out_frames = frame + if self.Naverage > 1: + out_frames = np.sum(out_frames, axis=0) / self.Naverage + else: + out_frames = out_frames[0, ...] + + if self.n_frames > 1: + pass + #todo handle chunks of frames in ND data + else: + out_frames = out_frames[0, ...] + + if out_frames.shape[-1] == 3: + data_arrays = [np.atleast_1d(out_frames[..., ind]) for ind in range(3)] + labels = ['Red', 'Green', 'Blue'] + else: + labels = ['Intensity'] + data_arrays = [out_frames] + + self.dte_signal.emit( + DataToExport('Camera', + data=[DataFromPlugins(name='Camera', + data=data_arrays, + dim=self.data_shape, + labels=labels, + axes=[self.y_axis, self.x_axis])])) + if self.settings.child('timing_opts', 'fps_on').value(): + self.update_fps() + + # To make sure that timed events are executed in continuous grab mode + QtWidgets.QApplication.processEvents() + + except Exception as e: + self.emit_status(ThreadCommand('Update_Status', [str(e), 'log'])) + + def update_fps(self): + current_tick = perf_counter() + frame_time = current_tick-self.last_tick + + if self.last_tick != 0.0 and frame_time != 0.0: + # We don't update FPS for the first frame, and we also avoid divisions by zero + + if self.fps == 0.0: + self.fps = 1 / frame_time + else: + # If we already have an FPS calculated, we smooth its evolution + self.fps = 0.9 * self.fps + 0.1 / frame_time + + self.last_tick = current_tick + + # Update reading + self.settings.child('timing_opts', 'fps').setValue(round(self.fps, 1)) + + def close(self): + """ + Terminate the communication protocol + """ + # Terminate the communication + + self.stop() + if self.callback_thread is not None: + self.callback_thread.quit() + self.callback_thread.wait() + + self.controller.close() + self.settings.child('buffer', 'mode').setReadonly(False) + + def stop(self): + """Stop the acquisition.""" + self.callback_signal.emit(Grab(do_acquisition=False)) + QtWidgets.QApplication.processEvents() + self.controller.clear_acquisition() + return '' + + + + diff --git a/src/pymodaq_plugins_utils/hardware/camera_base_pylablib.py b/src/pymodaq_plugins_utils/hardware/camera_base_pylablib.py index 6ffbab4..bf3e4db 100644 --- a/src/pymodaq_plugins_utils/hardware/camera_base_pylablib.py +++ b/src/pymodaq_plugins_utils/hardware/camera_base_pylablib.py @@ -1,117 +1,15 @@ -import dataclasses -from typing import Type +"""Base class of the camera plugins built on a camera of `pylablib `_. -import cv2 +The generic part (acquisition, ROI, binning, ...) is in :mod:`~pymodaq_plugins_utils.hardware.camera_base`: the +pylablib cameras follow its driver contract as they are. The names below are kept here for the plugins importing them +from this module. +""" +from pymodaq.control_modules.viewer_utility_classes import comon_parameters -from pymodaq_data import DataToExport -from pymodaq_utils.logger import set_logger, get_module_name -from pymodaq_utils.utils import ThreadCommand -from pymodaq_gui.parameter import Parameter -try: - from pymodaq_gui.plotting.items.roi import RoiInfo # pymodaq > 5.1.x -except ImportError: - from pymodaq_gui.plotting.utils.plot_utils import RoiInfo +from pymodaq_plugins_utils.hardware.camera_base import CameraBase, CameraCallback, Grab, cam_params -from pymodaq.utils.data import DataFromPlugins, Axis -from pymodaq.control_modules.viewer_utility_classes import DAQ_Viewer_base, comon_parameters, main -from pylablib.devices.interface.camera import trim_frames - -from qtpy import QtWidgets, QtCore -import numpy as np -from time import perf_counter - - -logger = set_logger(get_module_name(__file__)) - - -cam_params = [ - {'title': 'Camera name:', 'name': 'camera_name', 'type': 'str', 'value': '', 'readonly': True}, - {'title': 'Color Conversion:', 'name': 'color_conversion', 'type': 'list', - 'limits': ['None', 'RGB2GRAY', 'BAYER_BG2RGB', 'BAYER_BG2GRAY']}, - {'title': 'ROI', 'name': 'roi', 'type': 'group', 'children': [ - {'title': 'Update ROI from Viewer', 'name': 'update_roi', 'type': 'led', 'value': False}, - {'title': 'Apply ROI', 'name': 'apply_roi', 'type': 'led', 'value': False}, - {'title': 'Clear ROI+Bin', 'name': 'clear_roi', 'type': 'bool_push', 'value': False}, - {'title': 'ROI:', 'name': 'roi_slices', 'type': 'str', 'value': ''}, - {'title': 'X binning', 'name': 'x_binning', 'type': 'int', 'value': 1}, - {'title': 'Y binning', 'name': 'y_binning', 'type': 'int', 'value': 1}, - ], }, - {'title': 'Image width', 'name': 'hdet', 'type': 'int', 'value': 1, 'readonly': True}, - {'title': 'Image height', 'name': 'vdet', 'type': 'int', 'value': 1, 'readonly': True}, - {'title': 'Timing', 'name': 'timing_opts', 'type': 'group', 'children': - [{'title': 'Exposure Time (ms)', 'name': 'exposure_time', 'type': 'int', 'value': 1}, - {'title': 'Compute FPS', 'name': 'fps_on', 'type': 'bool', 'value': True}, - {'title': 'FPS', 'name': 'fps', 'type': 'float', 'value': 0.0, 'readonly': True}] - }, - {'title': 'Buffer', 'name': 'buffer', 'type': 'group', 'children': [ - {'title': 'Size:', 'name': 'size', 'type': 'int', 'value': 10}, - {'title': 'mode:', 'name': 'mode', 'type': 'list', 'value': 'now', - 'limits': ['now', 'lastread', 'lastwait', 'start']}, - ]}, -] - - -@dataclasses.dataclass -class Grab: - do_acquisition: bool = True - snap: bool = False - since: str = 'now' - nframes: int = 1 - n_average: int = 1 - - -class CameraCallback(QtCore.QObject): - """Callback object """ - data_sig = QtCore.Signal(np.ndarray) - error = QtCore.Signal() - - def __init__(self, controller): - super().__init__() - # Set the wait function - self.controller = controller - self.do_acquisition = True - - def set_do_grab(self, mode: Grab): - self.do_acquisition = mode.do_acquisition - if mode.do_acquisition: - self.wait_for_acquisition(mode) - - def wait_for_acquisition(self, mode: Grab): - while self.do_acquisition: - try: - ind_average = 0 - while ind_average < mode.n_average: - ind_frames = 0 - while ind_frames < mode.nframes: - self.controller.wait_for_frame(since='now') - new_frames, rng = self.controller.read_multiple_images(missing_frame='skip', return_rng=True) - - # in case hardware has a random buffer size (ex: Andor CCD cameras) - if new_frames.shape[0] != mode.nframes: - new_frames = np.expand_dims(new_frames[-mode.nframes:], axis=0) - - if ind_average == 0 and ind_frames == 0: - shape = list(new_frames.shape[1:]) - shape = [mode.n_average, mode.nframes] + shape - frames = np.zeros(shape, dtype=new_frames.dtype) - nacq = rng[1] - rng[0] - frames[ind_average, ind_frames:nacq, ...] = new_frames - ind_frames += nacq - ind_average += 1 - self.data_sig.emit(frames) - QtCore.QThread.msleep(10) - except Exception as e: - logger.exception(str(e)) - self.error.emit() - break - QtWidgets.QApplication.processEvents() - if not self.do_acquisition or mode.snap: - break - - - -class CameraBasePyLabLib(DAQ_Viewer_base): +class CameraBasePyLabLib(CameraBase): """ Base implementation for Camera using pylablib framework. Works for TSI and uc480 thorlabs camera and rpobaly others """ @@ -121,364 +19,5 @@ class CameraBasePyLabLib(DAQ_Viewer_base): params = comon_parameters + serial_params + cam_params - callback_signal = QtCore.Signal(Grab) - live_mode_available = True - hardware_averaging = True - - def ini_attributes(self): - self.controller = None - self.callback_thread: QtCore.QThread = None - self.is_live: bool = False - self.Naverage: int = 1 - - self.x_axis: Axis = None - self.y_axis: Axis = None - - self.roi_select_info: RoiInfo = None - - self.last_tick = 0.0 # time counter used to compute FPS - self.fps = 0.0 - - self.data_shape: str = '' - - - def roi_select(self, roi_info: RoiInfo, ind_viewer: int = 0): - """ Automatically called when a user use the RoiSelect ROi from a 2D viewer""" - self.roi_select_info = roi_info - self.roi_select_viewer_index = ind_viewer - - if self.settings['roi', 'update_roi']: - self.settings['roi', 'roi_slices'] = str(roi_info.to_slices()) - if self.settings['roi', 'apply_roi']: - self.apply_roi() - - def apply_roi(self): - roi_info = RoiInfo.from_slices(eval(self.settings['roi', 'roi_slices'])) - new_roi = (roi_info.origin[1], roi_info.size[1], self.settings['roi', 'x_binning'], - roi_info.origin[0], roi_info.size[0], self.settings['roi', 'y_binning']) - self.update_rois(new_roi) - - def compute_axes(self): - (hstart, hend, vstart, vend, hbin, vbin) = self.controller.get_roi() - slices = [slice(vstart, vend, vbin), slice(hstart, hend, hbin)] - self.settings.child('roi', 'roi_slices').setValue(str(slices)) - roi_info = RoiInfo.from_slices(slices) - - self.x_axis = Axis('x_axis', offset=roi_info.origin[1], - scaling=self.settings['roi', 'x_binning'], - size=int(roi_info.size[1]), - index=1) - self.y_axis = Axis('y_axis', offset=roi_info.origin[0], - scaling=self.settings['roi', 'y_binning'], - size=int(roi_info.size[0]), - index=0) - - def clear_roi(self): - wdet, hdet = self.controller.get_detector_size() - self.settings.child('roi', 'x_binning').setValue(1) - self.settings.child('roi', 'y_binning').setValue(1) - - new_roi = (0, wdet, 1, 0, hdet, 1) - self.update_rois(new_roi) - - def update_rois(self, new_roi): - # In pylablib, ROIs compare as tuples - (new_x, new_width, new_xbinning, new_y, new_height, new_ybinning) = new_roi - if new_roi != self.controller.get_roi(): - # self.controller.set_attribute_value("ROIs",[new_roi]) - self.controller.set_roi(hstart=new_x, hend=new_x + new_width, vstart=new_y, vend=new_y + new_height, - hbin=new_xbinning, vbin=new_ybinning) - self.emit_status(ThreadCommand('Update_Status', [f'Changed ROI: {new_roi}'])) - self.controller.clear_acquisition() - self.controller.setup_acquisition() - # Finally, prepare view for displaying the new data - self._prepare_view() - self.compute_axes() - - def commit_settings(self, param: Parameter): - """Apply the consequences of a change of value in the detector settings - - Parameters - ---------- - param: Parameter - A given parameter (within detector_settings) whose value has been changed by the user - """ - if param.name() == "exposure_time": - self.controller.set_exposure(param.value()/1000) - - elif param.name() == "fps_on": - self.settings.child('timing_opts', 'fps').setOpts(visible=param.value()) - - elif param.name() == "apply_roi": - if param.value(): # Switching on ROI - self.apply_roi() - else: - self.clear_roi() - - elif param.name() in ['x_binning', 'y_binning']: - # We handle ROI and binning separately for clarity - (x0, w, y0, h, *_) = self.controller.get_roi() # Get current ROI - xbin = self.settings['roi', 'x_binning'] - ybin = self.settings['roi', 'y_binning'] - new_roi = (x0, w, xbin, y0, h, ybin) - self.update_rois(new_roi) - - elif param.name() == "clear_roi": - if param.value(): # Switching on ROI - self.clear_roi() - param.setValue(False) - - def ini_detector_custom(self, controller=None): - raise NotImplementedError - - def ini_detector(self, controller=None): - """Detector communication initialization - - Parameters - ---------- - controller: (object) - custom object of a PyMoDAQ plugin (Slave case). None if only one actuator/detector by controller - (Master case) - - Returns - ------- - info: str - initialized: bool - False if initialization failed otherwise True - """ - - self.ini_detector_custom(controller) - - self.get_device_info() - self.get_set_main_parameters() - self.setup_callback_thread() - self.controller.set_frame_format("array") - - info = "Initialized camera" - initialized = True - return info, initialized - - def get_device_info(self): - - device_info = self.controller.get_device_info() - - # Get camera name/model - if hasattr(device_info, 'name'): - self.settings.child('camera_name').setValue(device_info.name) - elif hasattr(device_info, 'model'): - self.settings.child('camera_name').setValue(device_info.model) - - def get_set_main_parameters(self): - # Set exposure time - self.controller.set_exposure(self.settings['timing_opts', 'exposure_time']/1000) - - # FPS visibility - self.settings.child('timing_opts', 'fps').setOpts(visible=self.settings['timing_opts', 'fps_on']) - - # get roi limits - self.controller.get_roi_limits() - - # Update image parameters - (hstart, hend, vstart, vend, hbin, vbin) = self.controller.get_roi() - height, width = self.controller.get_data_dimensions() - self.settings.child('roi', 'x_binning').setValue(hbin) - self.settings.child('roi', 'y_binning').setValue(vbin) - self.settings.child('hdet').setValue(width) - self.settings.child('vdet').setValue(height) - slices = [slice(vstart, vend, vbin), slice(hstart, hend, hbin)] - self.settings.child('roi', 'roi_slices').setValue(str(slices)) - self.compute_axes() - - @property - def callback(self) -> Type[CameraCallback]: - """ Return the class handling the wait for acquisition and signal emission - - Should be reimplement as well as CameraCallback if needed - """ - return CameraCallback - - def setup_callback_thread(self): - # Way to define a wait function with arguments - wait_func = lambda: self.controller.wait_for_frame(since=self.settings['buffer', 'mode'], - nframes=1, timeout=20.0) - callback = CameraCallback(self.controller) - self.settings.child('buffer', 'mode').setReadonly(True) - - - self.callback_thread = QtCore.QThread(self) # creation of a Qt5 thread - callback.moveToThread(self.callback_thread) # callback object will live within this thread - - callback.data_sig.connect( - self.emit_data) # when the wait for acquisition returns (with data taken), emit_data will be fired - callback.error.connect(self.handle_error) - - self.callback_signal.connect(callback.set_do_grab) - self.callback_thread.callback = callback - self.callback_thread.start() - - self._prepare_view() - - def handle_error(self): - self.stop() - - def _prepare_view(self): - """Preparing a data viewer by emitting temporary data. Typically, needs to be called whenever the - ROIs are changed""" - - height, width = self.controller.get_data_dimensions() - - self.settings.child('hdet').setValue(width) - self.settings.child('vdet').setValue(height) - mock_data = np.zeros((height, width)) - - if width != 1 and height != 1: - data_shape = 'Data2D' - else: - data_shape = 'Data1D' - - if data_shape != self.data_shape: - self.data_shape = data_shape - # init the viewers - self.data_grabed_signal_temp.emit([DataFromPlugins(name='Thorlabs Camera', - data=[np.squeeze(mock_data)], - dim=self.data_shape, - labels=[f'ThorCam_{self.data_shape}'])]) - QtWidgets.QApplication.processEvents() - - def grab_data(self, Naverage=1, **kwargs): - """ - Grabs the data. ASynchronous method (kinda). - ---------- - Naverage: (int) Number of averaging - kwargs: (dict) of others optionals arguments - """ - try: - # Warning, acquisition_in_progress returns 1,0 and not a real bool - self.is_live = kwargs.get('live', False) - self.Naverage = Naverage - - self.n_frames = 1 - - if not self.controller.acquisition_in_progress(): - self.controller.clear_acquisition() - self.controller.start_acquisition(nframes=self.n_frames) - #Then start the acquisition - self.callback_signal.emit(Grab(do_acquisition=True, - snap=not self.is_live, - n_average=Naverage, - nframes=self.n_frames, - since=self.settings['buffer', 'mode'])) - - except Exception as e: - self.emit_status(ThreadCommand('Update_Status', [str(e), "log"])) - - def emit_data(self, frame: np.ndarray=None): - """ Function used to emit data obtained by callback. - - Parameter - --------- - status: bool - If True a frame is available, If False, a Timeout occurred while waiting for the frame - - See Also - -------- - daq_utils.ThreadCommand - """ - try: - # Get data from buffer - if frame is None: - frame = self.controller.read_newest_image() - # Emit the frame. - if frame is not None: - conversion_str = self.settings['color_conversion'] - if conversion_str != "None": - for ind_average in range(frame.shape[0]): - for ind_frame in range(frame.shape[1]): - if ind_frame == 0 and ind_average == 0: - new_frame = cv2.cvtColor(frame[ind_average, ind_frame, ...], - getattr(cv2, f'COLOR_{conversion_str}')) - shape = [frame.shape[0], frame.shape[1]] + list(new_frame.shape) - out_frames = np.zeros(shape, dtype=new_frame.dtype) - out_frames[ind_average, ind_frame, ...] = new_frame - else: - cv2.cvtColor(frame[ind_average, ind_frame, ...], - getattr(cv2, f'COLOR_{conversion_str}'), - out_frames[ind_average, ind_frame, ...]) - else: - out_frames = frame - if self.Naverage > 1: - out_frames = np.sum(out_frames, axis=0) / self.Naverage - else: - out_frames = out_frames[0, ...] - - if self.n_frames > 1: - pass - #todo handle chunks of frames in ND data - else: - out_frames = out_frames[0, ...] - - if out_frames.shape[-1] == 3: - data_arrays = [np.atleast_1d(out_frames[..., ind]) for ind in range(3)] - labels = ['Red', 'Green', 'Blue'] - else: - labels = ['Intensity'] - data_arrays = [out_frames] - - self.dte_signal.emit( - DataToExport('Camera', - data=[DataFromPlugins(name='Camera', - data=data_arrays, - dim=self.data_shape, - labels=labels, - axes=[self.y_axis, self.x_axis])])) - if self.settings.child('timing_opts', 'fps_on').value(): - self.update_fps() - - # To make sure that timed events are executed in continuous grab mode - QtWidgets.QApplication.processEvents() - - except Exception as e: - self.emit_status(ThreadCommand('Update_Status', [str(e), 'log'])) - - def update_fps(self): - current_tick = perf_counter() - frame_time = current_tick-self.last_tick - - if self.last_tick != 0.0 and frame_time != 0.0: - # We don't update FPS for the first frame, and we also avoid divisions by zero - - if self.fps == 0.0: - self.fps = 1 / frame_time - else: - # If we already have an FPS calculated, we smooth its evolution - self.fps = 0.9 * self.fps + 0.1 / frame_time - - self.last_tick = current_tick - - # Update reading - self.settings.child('timing_opts', 'fps').setValue(round(self.fps, 1)) - - def close(self): - """ - Terminate the communication protocol - """ - # Terminate the communication - - self.stop() - if self.callback_thread is not None: - self.callback_thread.quit() - self.callback_thread.wait() - - self.controller.close() - self.settings.child('buffer', 'mode').setReadonly(False) - - def stop(self): - """Stop the acquisition.""" - self.callback_signal.emit(Grab(do_acquisition=False)) - QtWidgets.QApplication.processEvents() - self.controller.clear_acquisition() - return '' - - - +__all__ = ['CameraBasePyLabLib', 'CameraCallback', 'Grab', 'cam_params'] diff --git a/src/pymodaq_plugins_utils/hardware/camera_driver.py b/src/pymodaq_plugins_utils/hardware/camera_driver.py new file mode 100644 index 0000000..8e39177 --- /dev/null +++ b/src/pymodaq_plugins_utils/hardware/camera_driver.py @@ -0,0 +1,102 @@ +"""Contract between a camera plugin and its driver. + +:class:`~pymodaq_plugins_utils.hardware.camera_base.CameraBase` talks to its ``controller`` only through the methods +listed in :class:`CameraDriverBase`. They are named and behave as in the camera drivers of +`pylablib `_, so those drivers can be used as they are, +without any adapter. To write the driver of a camera that pylablib does not know, inherit from +:class:`CameraDriverBase` and reimplement the methods that raise ``NotImplementedError``. + +The class has no dependency (neither Qt nor pylablib): it can be tested without hardware. +""" +from typing import Optional, Tuple + +import numpy as np + + +class CameraDriverBase: + """Methods a camera driver provides to :class:`~pymodaq_plugins_utils.hardware.camera_base.CameraBase`. + + The ROI is described as in pylablib: ``(hstart, hend, vstart, vend, hbin, vbin)``, the coordinates being pixels of + the full sensor (``hend`` and ``vend`` excluded) and the binnings integers. + + Images are 2D arrays of shape ``(height, width)``. The driver fills its own buffer while the acquisition runs + (typically in a thread of the driver or of the vendor library) and :meth:`wait_for_frame` blocks until new frames + are available there. + """ + + # --- to reimplement ------------------------------------------------------------------------------------------ + + def get_detector_size(self) -> Tuple[int, int]: + """Size of the full sensor in pixels, as ``(width, height)``.""" + raise NotImplementedError + + def get_data_dimensions(self) -> Tuple[int, int]: + """Shape of one image with the current ROI and binning, as ``(height, width)``.""" + raise NotImplementedError + + def get_roi(self) -> Tuple[int, int, int, int, int, int]: + """Current ROI as ``(hstart, hend, vstart, vend, hbin, vbin)``.""" + raise NotImplementedError + + def set_roi(self, hstart: int = 0, hend: Optional[int] = None, vstart: int = 0, vend: Optional[int] = None, + hbin: int = 1, vbin: int = 1): + """Set the ROI and the binning. ``None`` for ``hend`` or ``vend`` means up to the end of the sensor.""" + raise NotImplementedError + + def set_exposure(self, exposure: float): + """Set the exposure time, in seconds.""" + raise NotImplementedError + + def setup_acquisition(self): + """Allocate the buffer for the current ROI. Called after each change of ROI or binning.""" + raise NotImplementedError + + def clear_acquisition(self): + """Stop the acquisition if it runs and release the buffer.""" + raise NotImplementedError + + def start_acquisition(self, nframes: int = 1): + """Start filling the buffer with new frames.""" + raise NotImplementedError + + def acquisition_in_progress(self) -> bool: + """Whether the acquisition is running.""" + raise NotImplementedError + + def wait_for_frame(self, since: str = 'lastread', nframes: int = 1, timeout: float = 20.0): + """Block until ``nframes`` new frames are available in the buffer. + + ``since`` tells from where the new frames are counted (``'lastread'``, ``'lastwait'``, ``'now'`` or + ``'start'``). Raise an exception on timeout. + """ + raise NotImplementedError + + def read_multiple_images(self, rng=None, peek: bool = False, missing_frame: str = 'skip', + return_rng: bool = False): + """Read the frames available in the buffer. + + Return an array of shape ``(nframes, height, width)``, and the ``(first, last)`` indices of the frames + if ``return_rng`` is ``True``. + """ + raise NotImplementedError + + def read_newest_image(self, peek: bool = False) -> Optional[np.ndarray]: + """Read the last acquired frame, an array of shape ``(height, width)``, or ``None`` if there is none.""" + raise NotImplementedError + + def close(self): + """Release the camera.""" + raise NotImplementedError + + # --- optional, with a harmless default --------------------------------------------------------------------------- + + def get_device_info(self): + """Object with a ``name`` (or ``model``) attribute, shown in the settings.""" + return None + + def get_roi_limits(self): + """Limits of the ROI, not used by :class:`CameraBase` itself but called once at initialization.""" + return None + + def set_frame_format(self, fmt: str): + """Format of the returned frames, ``'array'`` for numpy arrays (the only one used here).""" From 09da13a6f67f51ca92e807fccfc8c54c97908caa Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 07:39:23 +0000 Subject: [PATCH 2/5] Fix the ROI and binning handling of the camera plugin The binning branch of commit_settings read get_roi() as (x, width, y, height) while pylablib returns (hstart, hend, vstart, vend, hbin, vbin), so changing the binning of a ROI that does not start at 0 enlarged it. The ROI is now always passed in pylablib order, which also makes the "unchanged ROI" comparison of update_rois effective, and it is clipped to the sensor. The size of the viewer axes comes from the image shape (it ignored the binning), the ROI string is read-only and no longer evaluated, the first image emitted to set up the viewer is a DataToExport, and the generic plugin does not name it Thorlabs. Tests drive the plugin with a fake driver. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Tdiso1u1ZhikSutj2Qs5Cg --- .../hardware/camera_base.py | 84 ++++---- tests/test_camera_base.py | 186 ++++++++++++++++++ 2 files changed, 232 insertions(+), 38 deletions(-) create mode 100644 tests/test_camera_base.py diff --git a/src/pymodaq_plugins_utils/hardware/camera_base.py b/src/pymodaq_plugins_utils/hardware/camera_base.py index c205ba5..b3a343c 100644 --- a/src/pymodaq_plugins_utils/hardware/camera_base.py +++ b/src/pymodaq_plugins_utils/hardware/camera_base.py @@ -34,14 +34,14 @@ cam_params = [ - {'title': 'Camera name:', 'name': 'camera_name', 'type': 'str', 'value': ''}, + {'title': 'Camera name:', 'name': 'camera_name', 'type': 'str', 'value': '', 'readonly': True}, {'title': 'Color Conversion:', 'name': 'color_conversion', 'type': 'list', 'limits': ['None', 'RGB2GRAY', 'BAYER_BG2RGB', 'BAYER_BG2GRAY'] if cv2 is not None else ['None']}, {'title': 'ROI', 'name': 'roi', 'type': 'group', 'children': [ {'title': 'Update ROI from Viewer', 'name': 'update_roi', 'type': 'led', 'value': False}, {'title': 'Apply ROI', 'name': 'apply_roi', 'type': 'led', 'value': False}, {'title': 'Clear ROI+Bin', 'name': 'clear_roi', 'type': 'bool_push', 'value': False}, - {'title': 'ROI:', 'name': 'roi_slices', 'type': 'str', 'value': ''}, + {'title': 'ROI:', 'name': 'roi_slices', 'type': 'str', 'value': '', 'readonly': True}, {'title': 'X binning', 'name': 'x_binning', 'type': 'int', 'value': 1}, {'title': 'Y binning', 'name': 'y_binning', 'type': 'int', 'value': 1}, ], }, @@ -141,6 +141,7 @@ def ini_attributes(self): self.y_axis: Axis = None self.roi_select_info: RoiInfo = None + self.roi_sensor_slices: Optional[Tuple[slice, slice]] = None # (rows, columns), in pixels of the full sensor self.last_tick = 0.0 # time counter used to compute FPS self.fps = 0.0 @@ -154,52 +155,61 @@ def roi_select(self, roi_info: RoiInfo, ind_viewer: int = 0): self.roi_select_viewer_index = ind_viewer if self.settings['roi', 'update_roi']: - self.settings['roi', 'roi_slices'] = str(roi_info.to_slices()) + rows, columns = roi_info.to_slices() + self.roi_sensor_slices = (rows, columns) + self.settings['roi', 'roi_slices'] = str(list(self.roi_sensor_slices)) if self.settings['roi', 'apply_roi']: self.apply_roi() def apply_roi(self): - roi_info = RoiInfo.from_slices(eval(self.settings['roi', 'roi_slices'])) - new_roi = (roi_info.origin[1], roi_info.size[1], self.settings['roi', 'x_binning'], - roi_info.origin[0], roi_info.size[0], self.settings['roi', 'y_binning']) - self.update_rois(new_roi) + """Set the ROI of the camera from the one last selected in the viewer""" + if self.roi_sensor_slices is None: + return + width, height = self.controller.get_detector_size() + rows, columns = self.roi_sensor_slices + hstart, hend = max(columns.start, 0), min(columns.stop, width) + vstart, vend = max(rows.start, 0), min(rows.stop, height) + if hend <= hstart or vend <= vstart: + self.emit_status(ThreadCommand('Update_Status', ['The ROI is outside of the sensor', 'log'])) + return + self.update_rois((hstart, hend, vstart, vend, + self.settings['roi', 'x_binning'], self.settings['roi', 'y_binning'])) def compute_axes(self): + """Axes of the viewer from the current ROI and binning of the camera""" (hstart, hend, vstart, vend, hbin, vbin) = self.controller.get_roi() - slices = [slice(vstart, vend, vbin), slice(hstart, hend, hbin)] - self.settings.child('roi', 'roi_slices').setValue(str(slices)) - roi_info = RoiInfo.from_slices(slices) - - self.x_axis = Axis('x_axis', offset=roi_info.origin[1], - scaling=self.settings['roi', 'x_binning'], - size=int(roi_info.size[1]), - index=1) - self.y_axis = Axis('y_axis', offset=roi_info.origin[0], - scaling=self.settings['roi', 'y_binning'], - size=int(roi_info.size[0]), - index=0) + height, width = self.controller.get_data_dimensions() + self.settings.child('roi', 'roi_slices').setValue(str([slice(vstart, vend, vbin), slice(hstart, hend, hbin)])) + + self.x_axis = Axis('x_axis', offset=hstart, scaling=hbin, size=int(width), index=1) + self.y_axis = Axis('y_axis', offset=vstart, scaling=vbin, size=int(height), index=0) def clear_roi(self): + """Full sensor, no binning""" wdet, hdet = self.controller.get_detector_size() self.settings.child('roi', 'x_binning').setValue(1) self.settings.child('roi', 'y_binning').setValue(1) - new_roi = (0, wdet, 1, 0, hdet, 1) - self.update_rois(new_roi) + self.update_rois((0, wdet, 0, hdet, 1, 1)) - def update_rois(self, new_roi): - # In pylablib, ROIs compare as tuples - (new_x, new_width, new_xbinning, new_y, new_height, new_ybinning) = new_roi - if new_roi != self.controller.get_roi(): - # self.controller.set_attribute_value("ROIs",[new_roi]) - self.controller.set_roi(hstart=new_x, hend=new_x + new_width, vstart=new_y, vend=new_y + new_height, - hbin=new_xbinning, vbin=new_ybinning) + def update_rois(self, new_roi: Tuple[int, int, int, int, int, int]): + """Set the ROI and binning of the camera if they changed + + Parameters + ---------- + new_roi: tuple of int + As in pylablib: (hstart, hend, vstart, vend, hbin, vbin) + """ + new_roi = tuple(int(value) for value in new_roi) + if new_roi != tuple(self.controller.get_roi()): + (hstart, hend, vstart, vend, hbin, vbin) = new_roi + self.controller.set_roi(hstart=hstart, hend=hend, vstart=vstart, vend=vend, hbin=hbin, vbin=vbin) self.emit_status(ThreadCommand('Update_Status', [f'Changed ROI: {new_roi}'])) self.controller.clear_acquisition() self.controller.setup_acquisition() # Finally, prepare view for displaying the new data - self._prepare_view() self.compute_axes() + self._prepare_view() def commit_settings(self, param: Parameter): """Apply the consequences of a change of value in the detector settings @@ -223,11 +233,10 @@ def commit_settings(self, param: Parameter): elif param.name() in ['x_binning', 'y_binning']: # We handle ROI and binning separately for clarity - (x0, w, y0, h, *_) = self.controller.get_roi() # Get current ROI + (hstart, hend, vstart, vend, *_) = self.controller.get_roi() # Get current ROI xbin = self.settings['roi', 'x_binning'] ybin = self.settings['roi', 'y_binning'] - new_roi = (x0, w, xbin, y0, h, ybin) - self.update_rois(new_roi) + self.update_rois((hstart, hend, vstart, vend, xbin, ybin)) elif param.name() == "clear_roi": if param.value(): # Switching on ROI @@ -292,8 +301,6 @@ def get_set_main_parameters(self): self.settings.child('roi', 'y_binning').setValue(vbin) self.settings.child('hdet').setValue(width) self.settings.child('vdet').setValue(height) - slices = [slice(vstart, vend, vbin), slice(hstart, hend, hbin)] - self.settings.child('roi', 'roi_slices').setValue(str(slices)) self.compute_axes() @property @@ -343,10 +350,11 @@ def _prepare_view(self): if data_shape != self.data_shape: self.data_shape = data_shape # init the viewers - self.data_grabed_signal_temp.emit([DataFromPlugins(name='Thorlabs Camera', - data=[np.squeeze(mock_data)], - dim=self.data_shape, - labels=[f'ThorCam_{self.data_shape}'])]) + self.dte_signal_temp.emit(DataToExport('Camera', + data=[DataFromPlugins(name='Camera', + data=[np.squeeze(mock_data)], + dim=self.data_shape, + labels=[f'Camera_{self.data_shape}'])])) QtWidgets.QApplication.processEvents() def grab_data(self, Naverage=1, **kwargs): diff --git a/tests/test_camera_base.py b/tests/test_camera_base.py new file mode 100644 index 0000000..f402c3a --- /dev/null +++ b/tests/test_camera_base.py @@ -0,0 +1,186 @@ +"""The camera plugin base class, driven by a fake camera driver (no pylablib, no hardware)""" +import importlib.util + +import numpy as np +import pytest + +from pymodaq_plugins_utils.hardware.camera_driver import CameraDriverBase + +try: + available = importlib.util.find_spec('pymodaq.utils.plugin_testing') is not None +except ModuleNotFoundError: # pymodaq is not installed + available = False +if not available: + pytest.skip('The harness needs PyMoDAQ >= 5.3.2', allow_module_level=True) + +from pymodaq.utils.plugin_testing import make_detector, grab_and_wait + +try: + from pymodaq_gui.plotting.items.roi import RoiInfo # pymodaq > 5.1.x +except ImportError: + from pymodaq_gui.plotting.utils.plot_utils import RoiInfo + +from pymodaq_plugins_utils.hardware.camera_base import CameraBase + +WIDTH, HEIGHT = 40, 30 + + +class FakeCamera(CameraDriverBase): + """A sensor whose pixel (row, column) is worth 100 * row + column, with ROI and binning""" + + def __init__(self): + self.sensor = 100. * np.arange(HEIGHT)[:, None] + np.arange(WIDTH)[None, :] + self.roi = (0, WIDTH, 0, HEIGHT, 1, 1) + self.exposure = None + self.running = False + self.closed = False + + def get_detector_size(self): + return WIDTH, HEIGHT + + def get_data_dimensions(self): + hstart, hend, vstart, vend, hbin, vbin = self.roi + return (vend - vstart) // vbin, (hend - hstart) // hbin + + def get_roi(self): + return self.roi + + def set_roi(self, hstart=0, hend=None, vstart=0, vend=None, hbin=1, vbin=1): + self.roi = (hstart, WIDTH if hend is None else hend, vstart, HEIGHT if vend is None else vend, hbin, vbin) + + def set_exposure(self, exposure): + self.exposure = exposure + + def setup_acquisition(self): + pass + + def clear_acquisition(self): + self.running = False + + def start_acquisition(self, nframes=1): + self.running = True + + def acquisition_in_progress(self): + return self.running + + def wait_for_frame(self, since='lastread', nframes=1, timeout=20.0): + pass + + def read_newest_image(self, peek=False): + hstart, hend, vstart, vend, hbin, vbin = self.roi + height, width = self.get_data_dimensions() + crop = self.sensor[vstart:vstart + height * vbin, hstart:hstart + width * hbin] + return crop.reshape(height, vbin, width, hbin).mean(axis=(1, 3)) + + def read_multiple_images(self, rng=None, peek=False, missing_frame='skip', return_rng=False): + frames = self.read_newest_image()[None, ...] + return (frames, (0, 1)) if return_rng else frames + + def close(self): + self.closed = True + + +class DAQ_2DViewer_Fake(CameraBase): + driver_class = FakeCamera + + def ini_detector_custom(self, controller=None): + self.controller = self.driver_class() if self.is_master else controller + + +@pytest.fixture +def camera(qapp): + plugin = make_detector(DAQ_2DViewer_Fake) + yield plugin + plugin.close() + + +def select_roi(plugin, rows: slice, columns: slice): + """What the user does: draw a ROI in the viewer, update it, then apply it""" + plugin.settings.child('roi', 'update_roi').setValue(True) + plugin.roi_select(RoiInfo.from_slices([rows, columns])) + apply_roi = plugin.settings.child('roi', 'apply_roi') + apply_roi.setValue(True) + plugin.commit_settings(apply_roi) + + +def set_binning(plugin, xbin: int, ybin: int): + for name, value in (('x_binning', xbin), ('y_binning', ybin)): + param = plugin.settings.child('roi', name) + param.setValue(value) + plugin.commit_settings(param) + + +class TestDriverContract: + @pytest.mark.parametrize('method, args', [ + ('get_detector_size', ()), ('get_data_dimensions', ()), ('get_roi', ()), ('set_roi', ()), + ('set_exposure', (0.1,)), ('setup_acquisition', ()), ('clear_acquisition', ()), ('start_acquisition', ()), + ('acquisition_in_progress', ()), ('wait_for_frame', ()), ('read_multiple_images', ()), + ('read_newest_image', ()), ('close', ()), + ]) + def test_what_a_driver_has_to_implement(self, method, args): + with pytest.raises(NotImplementedError): + getattr(CameraDriverBase(), method)(*args) + + def test_optional_methods_are_harmless(self): + driver = CameraDriverBase() + assert driver.get_device_info() is None + assert driver.get_roi_limits() is None + driver.set_frame_format('array') + + +class TestCameraBase: + def test_grab_a_full_frame(self, camera): + dte = grab_and_wait(camera) + assert dte[0].shape == (HEIGHT, WIDTH) + assert dte[0][0][3, 7] == pytest.approx(307) + assert (camera.x_axis.size, camera.y_axis.size) == (WIDTH, HEIGHT) + + def test_exposure_is_sent_in_seconds(self, camera): + param = camera.settings.child('timing_opts', 'exposure_time') + param.setValue(50) + camera.commit_settings(param) + assert camera.controller.exposure == pytest.approx(0.05) + + def test_axes_are_in_pixels_when_the_driver_does_not_know_the_pixel_size(self, camera): + assert camera.x_axis.units == '' + assert camera.x_axis.scaling == 1 + assert camera.x_axis.offset == 0 + + def test_roi_selected_in_the_viewer_becomes_the_camera_roi(self, camera): + select_roi(camera, rows=slice(5, 15), columns=slice(10, 30)) + assert camera.controller.get_roi() == (10, 30, 5, 15, 1, 1) + dte = grab_and_wait(camera) + assert dte[0].shape == (10, 20) + assert dte[0][0][0, 0] == pytest.approx(5 * 100 + 10) + assert camera.x_axis.offset == 10 + assert camera.y_axis.offset == 5 + + def test_roi_is_kept_inside_the_sensor(self, camera): + select_roi(camera, rows=slice(-5, 15), columns=slice(10, 100)) + assert camera.controller.get_roi() == (10, WIDTH, 0, 15, 1, 1) + + def test_changing_the_binning_keeps_the_roi(self, camera): + select_roi(camera, rows=slice(5, 15), columns=slice(10, 30)) + set_binning(camera, 2, 5) + assert camera.controller.get_roi() == (10, 30, 5, 15, 2, 5) + dte = grab_and_wait(camera) + assert dte[0].shape == (2, 10) + assert camera.x_axis.size == 10 + assert camera.x_axis.scaling == 2 + assert camera.y_axis.size == 2 + assert camera.y_axis.scaling == 5 + + def test_clear_roi_goes_back_to_the_full_sensor(self, camera): + select_roi(camera, rows=slice(5, 15), columns=slice(10, 30)) + set_binning(camera, 2, 2) + clear = camera.settings.child('roi', 'clear_roi') + clear.setValue(True) + camera.commit_settings(clear) + assert camera.controller.get_roi() == (0, WIDTH, 0, HEIGHT, 1, 1) + assert grab_and_wait(camera)[0].shape == (HEIGHT, WIDTH) + + def test_close_releases_the_camera(self, qapp): + plugin = make_detector(DAQ_2DViewer_Fake) + controller = plugin.controller + plugin.close() + assert controller.closed From 2c529d280aab03f9623a6b333620ae4a098f3ffb Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 07:39:29 +0000 Subject: [PATCH 3/5] Optional pixel size for camera axes in micrometers When the driver has a get_pixel_size method returning (x, y) in micrometers, the axes of the viewer are calibrated in micrometers, scaled by the binning, and a ROI drawn in the viewer is converted back to sensor pixels. Without the method (the pylablib cameras) the axes stay in pixels. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Tdiso1u1ZhikSutj2Qs5Cg --- .../hardware/camera_base.py | 24 ++++++++++-- .../hardware/camera_driver.py | 8 ++++ tests/test_camera_base.py | 37 +++++++++++++++++++ 3 files changed, 65 insertions(+), 4 deletions(-) diff --git a/src/pymodaq_plugins_utils/hardware/camera_base.py b/src/pymodaq_plugins_utils/hardware/camera_base.py index b3a343c..4ef96ea 100644 --- a/src/pymodaq_plugins_utils/hardware/camera_base.py +++ b/src/pymodaq_plugins_utils/hardware/camera_base.py @@ -123,7 +123,8 @@ class CameraBase(DAQ_Viewer_base): """Base implementation of a camera plugin The camera driver (``self.controller``, created in :meth:`ini_detector_custom`) follows - :class:`~pymodaq_plugins_utils.hardware.camera_driver.CameraDriverBase`. + :class:`~pymodaq_plugins_utils.hardware.camera_driver.CameraDriverBase`. If its ``get_pixel_size`` method returns + a size, the axes of the viewer are in micrometers, otherwise in pixels. """ params = comon_parameters + cam_params @@ -149,13 +150,23 @@ def ini_attributes(self): self.data_shape: str = '' + @property + def pixel_size(self) -> Optional[Tuple[float, float]]: + """Size (x, y) of a sensor pixel in micrometers if the driver knows it, else None""" + get_pixel_size = getattr(self.controller, 'get_pixel_size', None) + return get_pixel_size() if get_pixel_size is not None else None + def roi_select(self, roi_info: RoiInfo, ind_viewer: int = 0): """ Automatically called when a user use the RoiSelect ROi from a 2D viewer""" self.roi_select_info = roi_info self.roi_select_viewer_index = ind_viewer if self.settings['roi', 'update_roi']: - rows, columns = roi_info.to_slices() + rows, columns = roi_info.to_slices() # in the units of the viewer axes + pixel_size = self.pixel_size + if pixel_size is not None: # axes in um, back to pixels of the sensor + rows = slice(int(round(rows.start / pixel_size[1])), int(round(rows.stop / pixel_size[1]))) + columns = slice(int(round(columns.start / pixel_size[0])), int(round(columns.stop / pixel_size[0]))) self.roi_sensor_slices = (rows, columns) self.settings['roi', 'roi_slices'] = str(list(self.roi_sensor_slices)) if self.settings['roi', 'apply_roi']: @@ -181,8 +192,13 @@ def compute_axes(self): height, width = self.controller.get_data_dimensions() self.settings.child('roi', 'roi_slices').setValue(str([slice(vstart, vend, vbin), slice(hstart, hend, hbin)])) - self.x_axis = Axis('x_axis', offset=hstart, scaling=hbin, size=int(width), index=1) - self.y_axis = Axis('y_axis', offset=vstart, scaling=vbin, size=int(height), index=0) + pixel_size = self.pixel_size + x_size, y_size = pixel_size if pixel_size is not None else (1, 1) + units = 'µm' if pixel_size is not None else '' + self.x_axis = Axis('x_axis', units=units, offset=hstart * x_size, scaling=hbin * x_size, size=int(width), + index=1) + self.y_axis = Axis('y_axis', units=units, offset=vstart * y_size, scaling=vbin * y_size, size=int(height), + index=0) def clear_roi(self): """Full sensor, no binning""" diff --git a/src/pymodaq_plugins_utils/hardware/camera_driver.py b/src/pymodaq_plugins_utils/hardware/camera_driver.py index 8e39177..ac12ccf 100644 --- a/src/pymodaq_plugins_utils/hardware/camera_driver.py +++ b/src/pymodaq_plugins_utils/hardware/camera_driver.py @@ -100,3 +100,11 @@ def get_roi_limits(self): def set_frame_format(self, fmt: str): """Format of the returned frames, ``'array'`` for numpy arrays (the only one used here).""" + + def get_pixel_size(self) -> Optional[Tuple[float, float]]: + """Size of a sensor pixel in micrometers as ``(x, y)``, or ``None`` if unknown. + + When given, the axes of the 2D viewer are calibrated in micrometers instead of pixels. pylablib cameras + do not have this method, they keep axes in pixels. + """ + return None diff --git a/tests/test_camera_base.py b/tests/test_camera_base.py index f402c3a..557d0aa 100644 --- a/tests/test_camera_base.py +++ b/tests/test_camera_base.py @@ -80,6 +80,11 @@ def close(self): self.closed = True +class FakeCameraWithPixelSize(FakeCamera): + def get_pixel_size(self): + return 5.0, 10.0 + + class DAQ_2DViewer_Fake(CameraBase): driver_class = FakeCamera @@ -87,6 +92,10 @@ def ini_detector_custom(self, controller=None): self.controller = self.driver_class() if self.is_master else controller +class DAQ_2DViewer_FakeWithPixelSize(DAQ_2DViewer_Fake): + driver_class = FakeCameraWithPixelSize + + @pytest.fixture def camera(qapp): plugin = make_detector(DAQ_2DViewer_Fake) @@ -94,6 +103,13 @@ def camera(qapp): plugin.close() +@pytest.fixture +def calibrated_camera(qapp): + plugin = make_detector(DAQ_2DViewer_FakeWithPixelSize) + yield plugin + plugin.close() + + def select_roi(plugin, rows: slice, columns: slice): """What the user does: draw a ROI in the viewer, update it, then apply it""" plugin.settings.child('roi', 'update_roi').setValue(True) @@ -125,6 +141,7 @@ def test_optional_methods_are_harmless(self): driver = CameraDriverBase() assert driver.get_device_info() is None assert driver.get_roi_limits() is None + assert driver.get_pixel_size() is None driver.set_frame_format('array') @@ -184,3 +201,23 @@ def test_close_releases_the_camera(self, qapp): controller = plugin.controller plugin.close() assert controller.closed + + +class TestPixelSize: + def test_axes_are_calibrated(self, calibrated_camera): + dte = grab_and_wait(calibrated_camera) + assert calibrated_camera.x_axis.units == 'µm' + assert calibrated_camera.x_axis.scaling == pytest.approx(5.0) + assert calibrated_camera.y_axis.scaling == pytest.approx(10.0) + assert dte[0].shape == (HEIGHT, WIDTH) + + def test_roi_drawn_on_calibrated_axes_is_converted_to_pixels(self, calibrated_camera): + select_roi(calibrated_camera, rows=slice(50, 150), columns=slice(50, 150)) # um: 5 to 15 rows, 10 to 30 cols + assert calibrated_camera.controller.get_roi() == (10, 30, 5, 15, 1, 1) + assert calibrated_camera.x_axis.offset == pytest.approx(50.0) + assert calibrated_camera.y_axis.offset == pytest.approx(50.0) + + def test_binning_multiplies_the_scaling(self, calibrated_camera): + set_binning(calibrated_camera, 2, 1) + assert calibrated_camera.x_axis.scaling == pytest.approx(10.0) + assert calibrated_camera.x_axis.size == WIDTH // 2 From de2e788d1eeab35a8f5a0892f127297ab86633ac Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 07:39:44 +0000 Subject: [PATCH 4/5] Describe CameraBase and the driver contract in the README Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Tdiso1u1ZhikSutj2Qs5Cg --- README.rst | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/README.rst b/README.rst index e76457c..203a72b 100644 --- a/README.rst +++ b/README.rst @@ -43,5 +43,8 @@ Utilities ========= * pysvisa stuff -* CameraBasePyLabLib: Base plugin class inheriting from DAQ_Viewer_base but implementing ROI management, binning, - the use of buffers through the pylablib package implementation. Mostly for (S)CMOS camera +* CameraBase: Base plugin class inheriting from DAQ_Viewer_base and implementing the acquisition in a thread, ROI + management (shared with the viewer), binning, exposure and FPS for any camera. The camera driver follows + CameraDriverBase, whose methods are those of the pylablib cameras: write the driver of your camera against it, + or use a pylablib one as it is. +* CameraBasePyLabLib: CameraBase for the cameras of the pylablib package. Mostly for (S)CMOS camera From 582545e9ca57333986e91320ecc2653133d8a0b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 07:47:18 +0000 Subject: [PATCH 5/5] Do not hang when the ROI changes during a live acquisition Changing the ROI or the binning while live cleared the acquisition under the thread that waits for the frames. pylablib then returns from wait_for_frame at once and read_multiple_images gives no frame, so the loop of CameraCallback spun without handling the event that stops it, and closing the plugin blocked on the thread. The loop now ends when a read gives no frame and the acquisition is stopped. update_rois stops the acquisition before it sets the ROI (most cameras refuse to change it while a buffer exists) and restarts the live afterwards. stop() now resets is_live. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Tdiso1u1ZhikSutj2Qs5Cg --- .../hardware/camera_base.py | 11 +++++++- tests/test_camera_base.py | 28 ++++++++++++++++--- 2 files changed, 34 insertions(+), 5 deletions(-) diff --git a/src/pymodaq_plugins_utils/hardware/camera_base.py b/src/pymodaq_plugins_utils/hardware/camera_base.py index 4ef96ea..02ca5a9 100644 --- a/src/pymodaq_plugins_utils/hardware/camera_base.py +++ b/src/pymodaq_plugins_utils/hardware/camera_base.py @@ -94,6 +94,11 @@ def wait_for_acquisition(self, mode: Grab): while ind_frames < mode.nframes: self.controller.wait_for_frame(since='now') new_frames, rng = self.controller.read_multiple_images(missing_frame='skip', return_rng=True) + if rng[1] == rng[0]: # no new frame: the acquisition was stopped while waiting for it + QtWidgets.QApplication.processEvents() # let the signal that stops this loop be handled + if not self.do_acquisition or not self.controller.acquisition_in_progress(): + return + continue # in case hardware has a random buffer size (ex: Andor CCD cameras) if new_frames.shape[0] != mode.nframes: @@ -219,13 +224,16 @@ def update_rois(self, new_roi: Tuple[int, int, int, int, int, int]): new_roi = tuple(int(value) for value in new_roi) if new_roi != tuple(self.controller.get_roi()): (hstart, hend, vstart, vend, hbin, vbin) = new_roi + was_live = self.is_live + self.stop() # most cameras do not change the ROI while they acquire self.controller.set_roi(hstart=hstart, hend=hend, vstart=vstart, vend=vend, hbin=hbin, vbin=vbin) self.emit_status(ThreadCommand('Update_Status', [f'Changed ROI: {new_roi}'])) - self.controller.clear_acquisition() self.controller.setup_acquisition() # Finally, prepare view for displaying the new data self.compute_axes() self._prepare_view() + if was_live: + self.grab_data(self.Naverage, live=True) def commit_settings(self, param: Parameter): """Apply the consequences of a change of value in the detector settings @@ -502,6 +510,7 @@ def close(self): def stop(self): """Stop the acquisition.""" + self.is_live = False self.callback_signal.emit(Grab(do_acquisition=False)) QtWidgets.QApplication.processEvents() self.controller.clear_acquisition() diff --git a/tests/test_camera_base.py b/tests/test_camera_base.py index 557d0aa..a43710f 100644 --- a/tests/test_camera_base.py +++ b/tests/test_camera_base.py @@ -1,5 +1,6 @@ """The camera plugin base class, driven by a fake camera driver (no pylablib, no hardware)""" import importlib.util +import time import numpy as np import pytest @@ -13,7 +14,7 @@ if not available: pytest.skip('The harness needs PyMoDAQ >= 5.3.2', allow_module_level=True) -from pymodaq.utils.plugin_testing import make_detector, grab_and_wait +from pymodaq.utils.plugin_testing import make_detector, grab_and_wait, wait_for_signal try: from pymodaq_gui.plotting.items.roi import RoiInfo # pymodaq > 5.1.x @@ -46,6 +47,8 @@ def get_roi(self): return self.roi def set_roi(self, hstart=0, hend=None, vstart=0, vend=None, hbin=1, vbin=1): + if self.running: + raise RuntimeError('The ROI cannot change during the acquisition') # as most cameras self.roi = (hstart, WIDTH if hend is None else hend, vstart, HEIGHT if vend is None else vend, hbin, vbin) def set_exposure(self, exposure): @@ -64,7 +67,9 @@ def acquisition_in_progress(self): return self.running def wait_for_frame(self, since='lastread', nframes=1, timeout=20.0): - pass + if self.running: + time.sleep(0.005) # the exposure + return self.running # as pylablib, False if the acquisition is stopped def read_newest_image(self, peek=False): hstart, hend, vstart, vend, hbin, vbin = self.roi @@ -73,8 +78,11 @@ def read_newest_image(self, peek=False): return crop.reshape(height, vbin, width, hbin).mean(axis=(1, 3)) def read_multiple_images(self, rng=None, peek=False, missing_frame='skip', return_rng=False): - frames = self.read_newest_image()[None, ...] - return (frames, (0, 1)) if return_rng else frames + if self.running: + frames, rng = self.read_newest_image()[None, ...], (0, 1) + else: + frames, rng = np.zeros((0,) + self.get_data_dimensions()), (0, 0) + return (frames, rng) if return_rng else frames def close(self): self.closed = True @@ -196,6 +204,18 @@ def test_clear_roi_goes_back_to_the_full_sensor(self, camera): assert camera.controller.get_roi() == (0, WIDTH, 0, HEIGHT, 1, 1) assert grab_and_wait(camera)[0].shape == (HEIGHT, WIDTH) + def test_roi_change_during_live_restarts_it_with_the_new_roi(self, camera): + wait_for_signal(camera.dte_signal, lambda: camera.grab_data(1, live=True), timeout_ms=3000) + assert camera.is_live + select_roi(camera, rows=slice(5, 15), columns=slice(10, 30)) + assert camera.controller.get_roi() == (10, 30, 5, 15, 1, 1) + dte = wait_for_signal(camera.dte_signal, lambda: None, timeout_ms=3000) + while dte[0].shape != (10, 20): # images of the old ROI may still be on their way + dte = wait_for_signal(camera.dte_signal, lambda: None, timeout_ms=3000) + camera.stop() + assert not camera.is_live + assert not camera.controller.acquisition_in_progress() + def test_close_releases_the_camera(self, qapp): plugin = make_detector(DAQ_2DViewer_Fake) controller = plugin.controller