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 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..02ca5a9 --- /dev/null +++ b/src/pymodaq_plugins_utils/hardware/camera_base.py @@ -0,0 +1,521 @@ +"""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': '', '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': '', 'readonly': True}, + {'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) + 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: + 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`. 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 + + 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.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 + + 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() # 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']: + self.apply_roi() + + def apply_roi(self): + """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() + height, width = self.controller.get_data_dimensions() + self.settings.child('roi', 'roi_slices').setValue(str([slice(vstart, vend, vbin), slice(hstart, hend, hbin)])) + + 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""" + wdet, hdet = self.controller.get_detector_size() + self.settings.child('roi', 'x_binning').setValue(1) + self.settings.child('roi', 'y_binning').setValue(1) + + self.update_rois((0, wdet, 0, hdet, 1, 1)) + + 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 + 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.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 + + 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 + (hstart, hend, vstart, vend, *_) = self.controller.get_roi() # Get current ROI + xbin = self.settings['roi', 'x_binning'] + ybin = self.settings['roi', 'y_binning'] + self.update_rois((hstart, hend, vstart, vend, xbin, ybin)) + + 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) + 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.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): + """ + 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.is_live = False + 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..ac12ccf --- /dev/null +++ b/src/pymodaq_plugins_utils/hardware/camera_driver.py @@ -0,0 +1,110 @@ +"""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).""" + + 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 new file mode 100644 index 0000000..a43710f --- /dev/null +++ b/tests/test_camera_base.py @@ -0,0 +1,243 @@ +"""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 + +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, wait_for_signal + +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): + 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): + 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): + 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 + 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): + 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 + + +class FakeCameraWithPixelSize(FakeCamera): + def get_pixel_size(self): + return 5.0, 10.0 + + +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 + + +class DAQ_2DViewer_FakeWithPixelSize(DAQ_2DViewer_Fake): + driver_class = FakeCameraWithPixelSize + + +@pytest.fixture +def camera(qapp): + plugin = make_detector(DAQ_2DViewer_Fake) + yield plugin + 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) + 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 + assert driver.get_pixel_size() 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_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 + 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