diff --git a/README.rst b/README.rst index 18f028d..1231907 100644 --- a/README.rst +++ b/README.rst @@ -4,7 +4,7 @@ pymodaq_plugins_opencv .. the following must be adapted to your developped package, links to pypi, github description... .. image:: https://img.shields.io/pypi/v/pymodaq_plugins_opencv.svg - :target: https://pypi.org/project/pymodaq_plugins_thorlabs/ + :target: https://pypi.org/project/pymodaq_plugins_opencv/ :alt: Latest Version .. image:: https://readthedocs.org/projects/pymodaq/badge/?version=latest @@ -34,6 +34,22 @@ Viewer2D * **opencv**: control of cameras using the opencv library (mainly webcams) + It is built on the camera base class of ``pymodaq_plugins_utils`` (``CameraBase``), with the driver + ``hardware/opencv_camera.py`` that follows the camera driver interface of that package. So it has the same + settings as the other camera plugins: live and snap, averaging, the ROI drawn in the viewer (*Update ROI from + Viewer*, *Apply ROI*, *Clear ROI+Bin*), binning and FPS. A webcam has neither a hardware ROI nor binning: they are + applied on each image (binning is the mean of the pixels). + + * *Camera index*: the camera to open (0 is usually the one built in the laptop). + * *Color Conversion*: OpenCV gives colour images, converted by default to gray (``RGB2GRAY``). With ``None`` + the three channels (red, green, blue) are given. + * *Cam. Settings*: the properties that OpenCV can read on the camera (brightness, gain, exposure...), the unit + of some depends on the backend. Changing the frame width or height there resets the ROI. *Open Settings* shows + the settings window of the driver of the camera (DirectShow, Windows). + + The exposure time and buffer settings of the camera base class are hidden: OpenCV does not set them in a portable + way, use *Cam. Settings*. + .. if needed use this field Infos diff --git a/pyproject.toml b/pyproject.toml index 521e8ac..a6e2d74 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "pymodaq_utils>=5.3.1", "pymodaq_gui>=5.3.0", "pymodaq_data>=5.3.0", + "pymodaq_plugins_utils>=5.3.1", 'opencv-python', ] diff --git a/src/pymodaq_plugins_opencv/daq_viewer_plugins/plugins_2D/daq_2Dviewer_OpenCV.py b/src/pymodaq_plugins_opencv/daq_viewer_plugins/plugins_2D/daq_2Dviewer_OpenCV.py index 0cd05cd..14877ae 100644 --- a/src/pymodaq_plugins_opencv/daq_viewer_plugins/plugins_2D/daq_2Dviewer_OpenCV.py +++ b/src/pymodaq_plugins_opencv/daq_viewer_plugins/plugins_2D/daq_2Dviewer_OpenCV.py @@ -1,56 +1,64 @@ -from sys import platform +from copy import deepcopy -import numpy as np -import cv2 - -from pymodaq_utils.utils import ThreadCommand -from pymodaq.utils.data import DataFromPlugins, Axis, DataToExport -from pymodaq.control_modules.viewer_utility_classes import DAQ_Viewer_base, comon_parameters, main +from pymodaq_utils.logger import set_logger, get_module_name from pymodaq_gui.parameter import Parameter -from pymodaq_plugins_opencv.hardware.opencv import OpenCVProp +from pymodaq.control_modules.viewer_utility_classes import comon_parameters, main +from pymodaq_plugins_utils.hardware.camera_base import CameraBase, cam_params -class DAQ_2DViewer_OpenCV(DAQ_Viewer_base): - """ +from pymodaq_plugins_opencv.hardware.opencv_camera import OpenCVCamera + +logger = set_logger(get_module_name(__file__)) + + +def webcam_params() -> list: + """The settings of ``CameraBase`` that suit a webcam: colour image converted to gray by default, no exposure + time nor buffer (OpenCV sets neither, see ``Cam. Settings``)""" + params = deepcopy(cam_params) + for param in params: + if param['name'] == 'color_conversion': + param['value'] = 'RGB2GRAY' + elif param['name'] == 'buffer': + param['visible'] = False + elif param['name'] == 'timing_opts': + for child in param['children']: + if child['name'] == 'exposure_time': + child['visible'] = False + return params + + +class DAQ_2DViewer_OpenCV(CameraBase): + """Cameras read with OpenCV (webcams, any device handled by ``cv2.VideoCapture``) + + The acquisition, the live mode, the averaging, the ROI drawn in the viewer, the binning (both done on the image + since a webcam has no ROI) and the FPS are those of ``CameraBase``. The images are RGB: select a colour + conversion (default RGB2GRAY) or None to get the three channels. + + ``Cam. Settings`` lists the properties of the camera that OpenCV can read (brightness, gain, exposure...), the + unit of some of them depends on the backend of OpenCV. Changing the width or the height there resets the ROI. """ params = comon_parameters + \ [{'title': 'Camera index:', 'name': 'camera_index', 'type': 'int', 'value': 0, 'default': 0, 'min': 0}, - {'title': 'Colors:', 'name': 'colors', 'type': 'list', 'value': 'gray', 'limits': ['gray', 'RGB']}, {'title': 'Open Settings:', 'name': 'open_settings', 'type': 'bool', 'value': False}, - {'title': 'Cam. Settings:', 'name': 'cam_settings', 'type': 'group', 'children': [ - # {'title': 'Autoexposure:', 'name': 'autoexposure', 'type': 'bool', 'value': False}, - # {'title': 'Exposure:', 'name': 'exposure', 'type': 'int', 'value': 0}, - ]}, - ] - - def ini_attributes(self): - # TODO declare the type of the wrapper (and assign it to self.controller) you're going to use for easy - # autocompletion - self.controller: cv2.VideoCapture = None - - # TODO declare here attributes you want/need to init with a default value - - self.x_axis = None - self.y_axis = None - - def get_active_properties(self): - props = OpenCVProp.names() - self.additional_params = [] - for prop in props: - try: - ret = int(self.controller.get(OpenCVProp[prop].value)) - if ret != -1: - try: - ret_set = self.controller.set(OpenCVProp[prop].value, ret) - except: - ret_set = False - self.additional_params.append( - {'title': prop[7:], 'name': prop[7:], 'type': 'float', 'value': ret, 'readonly': not ret_set}) - except: - pass - self.settings.child('cam_settings').addChildren(self.additional_params) - pass + {'title': 'Cam. Settings:', 'name': 'cam_settings', 'type': 'group', 'children': []}, + ] + webcam_params() + + def ini_detector_custom(self, controller=None): + """Open the camera (master) or take the one of the master (slave)""" + if self.is_master: + self.controller = OpenCVCamera(self.settings['camera_index']) + else: + self.controller = controller + self.add_camera_properties() + + def add_camera_properties(self): + """Make a setting of each property that this camera supports""" + group = self.settings.child('cam_settings') + group.clearChildren() + group.addChildren([ + {'title': name, 'name': name, 'type': 'float', 'value': value, 'readonly': not writable} + for name, (value, writable) in self.controller.get_properties().items()]) def commit_settings(self, param: Parameter): """Apply the consequences of a change of value in the detector settings @@ -62,116 +70,24 @@ def commit_settings(self, param: Parameter): """ if param.name() == 'open_settings': if param.value(): - self.controller.set(OpenCVProp['CV_CAP_PROP_SETTINGS'].value, 1) - # param.setValue(False) - elif param.name() == 'colors': - pass + self.controller.show_settings() + elif param.parent() is not None and param.parent().name() == 'cam_settings': + self.set_camera_property(param) else: - self.controller.set(OpenCVProp['CV_CAP_' + param.name()].value, param.value()) - val = self.controller.get(OpenCVProp['CV_CAP_' + param.name()].value) - param.setValue(val) - - 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 - """ - if self.is_master: - if 'win' in platform: - self.controller = cv2.VideoCapture(self.settings['camera_index'], cv2.CAP_DSHOW) - else: - self.controller = cv2.VideoCapture(self.settings['camera_index']) - # to add settable settings to the param list (but driver builtin settings window is prefered (OpenCVProp['CV_CAP_PROP_SETTINGS']) - self.get_active_properties() - - else: - self.controller = controller - - - self.x_axis = self.get_xaxis() - self.y_axis = self.get_yaxis() - - self.dte_signal_temp.emit(DataToExport('myplugin', - data=[DataFromPlugins(name='Mock1', - data=[np.zeros((len(self.y_axis), len(self.x_axis)))], - dim='Data2D', labels=['opencv'], - axes=[self.x_axis, self.y_axis]), ])) - - info = "Whatever info you want to log" - initialized = True - return info, initialized - - def close(self): - """Terminate the communication protocol""" - self.controller.release() - - def get_xaxis(self) -> Axis: - """ Get the current x_axis from the controller - - Returns - ------- - Axis - """ - Nx = int(self.controller.get(cv2.CAP_PROP_FRAME_WIDTH)) # property index corresponding to width - return Axis('xaxis', data=np.linspace(0, Nx - 1, Nx), index=1) - - def get_yaxis(self) -> Axis: - """ Get the current y_axis from the controller - - Returns - ------- - Axis - """ - Ny = int(self.controller.get(cv2.CAP_PROP_FRAME_HEIGHT)) # property index corresponding to width - return Axis('yaxis', data=np.linspace(0, Ny - 1, Ny), index=0) - - def grab_data(self, Naverage=1, **kwargs): - """Start a grab from the detector - - Parameters - ---------- - Naverage: int - Number of hardware averaging (if hardware averaging is possible, self.hardware_averaging should be set to - True in class preamble and you should code this implementation) - kwargs: dict - others optionals arguments - """ - if not self.controller.isOpened(): - self.controller.open(self.settings['camera_index']) - - ret, frame = self.controller.read() - - if ret: - if self.settings['colors'] == 'gray': - data_cam = [cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)] - data_cam[0] = data_cam[0].astype(np.float32) - else: - if len(frame.shape) == 3: - data_cam = [frame[:, :, ind] for ind in range(frame.shape[2])] - else: - data_cam = [cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)] - - else: - data_cam = [np.zeros((len(self.y_axis), len(self.x_axis)))] - self.emit_status(ThreadCommand('Update_Status', ['no return from the controller', 'log'])) - - self.dte_signal.emit(DataToExport('OpenCV data', - data=[ - DataFromPlugins(name='OpenCV', data=data_cam, dim='Data2D')])) - - def stop(self): - """Stop the current grab hardware wise if necessary""" - pass + super().commit_settings(param) + + def set_camera_property(self, param: Parameter): + """Set a property of the camera and show the value that it took""" + resize = param.name() in self.controller.size_properties + was_live = self.is_live + if resize: # the images get another shape + self.stop() + param.setValue(self.controller.set_property(param.name(), param.value())) + if resize: + self.get_set_main_parameters() + self._prepare_view() + if was_live: + self.grab_data(self.Naverage, live=True) if __name__ == '__main__': diff --git a/src/pymodaq_plugins_opencv/hardware/opencv_camera.py b/src/pymodaq_plugins_opencv/hardware/opencv_camera.py new file mode 100644 index 0000000..9a97c59 --- /dev/null +++ b/src/pymodaq_plugins_opencv/hardware/opencv_camera.py @@ -0,0 +1,232 @@ +"""Driver of the cameras (webcams, video files) read with OpenCV, for ``CameraBase`` of ``pymodaq_plugins_utils``. + +It follows ``pymodaq_plugins_utils.hardware.camera_driver.CameraDriverBase``, the methods being those of the pylablib +cameras. OpenCV cameras are simpler than most: + +- there is no buffer to manage: ``VideoCapture.read`` blocks until the next frame, so ``wait_for_frame`` reads it and + ``read_multiple_images`` gives what was read; +- the ROI and the binning are not done by the camera, the driver applies them to each frame (the sensor is the full + frame, so it can change during the acquisition); +- the exposure is not set from here: its unit depends on the backend, change it with the properties of the camera + (see :meth:`OpenCVCamera.get_properties`). + +Frames are given as RGB, the order of OpenCV being BGR. +""" +import sys +import threading +import time +from collections import deque +from types import SimpleNamespace +from typing import Dict, Optional, Tuple, Union + +import cv2 +import numpy as np + +from pymodaq_plugins_utils.hardware.camera_driver import CameraDriverBase + +from pymodaq_plugins_opencv.hardware.opencv import OpenCVProp + + +class OpenCVCamera(CameraDriverBase): + """A camera read with ``cv2.VideoCapture`` + + Parameters + ---------- + source: int or str + Index of the camera, or file (or url) opened by ``cv2.VideoCapture`` + """ + # the properties that change the size of the frame + size_properties = ('PROP_FRAME_WIDTH', 'PROP_FRAME_HEIGHT') + + def __init__(self, source: Union[int, str] = 0): + super().__init__() + self.source = source + self._capture_lock = threading.RLock() # VideoCapture is not thread safe: reads and properties are exclusive + self._lock = threading.Lock() # protects the ROI and the frames + self._capture = self._open() + self._width, self._height = self._read_size() + self._roi = (0, self._width, 0, self._height, 1, 1) + self._running = threading.Event() + self._frames: deque = deque() + self._n_read = 0 # frames made since the start + self._last_frame: Optional[np.ndarray] = None + + def _open(self) -> cv2.VideoCapture: + if isinstance(self.source, int) and sys.platform.startswith('win'): + capture = cv2.VideoCapture(self.source, cv2.CAP_DSHOW) + else: + capture = cv2.VideoCapture(self.source) + if not capture.isOpened(): + capture.release() + raise IOError(f'The OpenCV camera {self.source!r} cannot be opened') + try: + capture.set(cv2.CAP_PROP_BUFFERSIZE, 1) # the oldest frames of a buffer are late: keep only the last + except cv2.error: + pass + return capture + + def _read_size(self) -> Tuple[int, int]: + return (int(self._capture.get(cv2.CAP_PROP_FRAME_WIDTH)), int(self._capture.get(cv2.CAP_PROP_FRAME_HEIGHT))) + + # --- sensor ---------------------------------------------------------------------------------------------------- + + def get_device_info(self): + with self._capture_lock: + try: + backend = self._capture.getBackendName() + except (cv2.error, AttributeError): + backend = 'unknown backend' + return SimpleNamespace(name=f'OpenCV camera {self.source} ({backend})') + + def get_detector_size(self) -> Tuple[int, int]: + return self._width, self._height + + def get_data_dimensions(self) -> Tuple[int, int]: + hstart, hend, vstart, vend, hbin, vbin = self._roi + return (vend - vstart) // vbin, (hend - hstart) // hbin + + def get_roi(self) -> Tuple[int, int, int, int, int, int]: + return self._roi + + def set_roi(self, hstart: int = 0, hend: Optional[int] = None, vstart: int = 0, vend: Optional[int] = None, + hbin: int = 1, vbin: int = 1): + hend = self._width if hend is None else hend + vend = self._height if vend is None else vend + if not (0 <= hstart < hend <= self._width and 0 <= vstart < vend <= self._height): + raise ValueError(f'The ROI {(hstart, hend, vstart, vend)} is not inside the {self._width}x{self._height} ' + f'frame') + if hbin < 1 or vbin < 1 or hend - hstart < hbin or vend - vstart < vbin: + raise ValueError(f'The binning {(hbin, vbin)} does not fit in the ROI') + with self._lock: + self._roi = (hstart, hend, vstart, vend, hbin, vbin) + self._frames.clear() # they have the previous shape + self._last_frame = None + + def set_exposure(self, exposure: float): + """Not used: the unit of the exposure of OpenCV depends on the backend, see :meth:`set_property`""" + + # --- properties of the camera (brightness, gain, exposure...) --------------------------------------------------- + + def get_properties(self) -> Dict[str, Tuple[float, bool]]: + """The properties of ``OpenCVProp`` that this camera supports: name -> (value, writable) + + Names are those of ``OpenCVProp`` without the ``CV_CAP_`` prefix, for instance ``PROP_BRIGHTNESS``. + """ + properties = {} + with self._capture_lock: + for name in OpenCVProp.names(): + if OpenCVProp[name].value < 0: # modes of the DC1394 registers, not properties + continue + try: + value = int(self._capture.get(OpenCVProp[name].value)) + if value == -1: # not supported by the backend or the camera + continue + try: + writable = bool(self._capture.set(OpenCVProp[name].value, value)) + except cv2.error: + writable = False + except cv2.error: + continue + properties[name[len('CV_CAP_'):]] = (value, writable) + return properties + + def set_property(self, name: str, value: float) -> float: + """Set a property (a name of :meth:`get_properties`) and return the value the camera took""" + prop = OpenCVProp[f'CV_CAP_{name}'].value + with self._capture_lock: + self._capture.set(prop, value) + taken = self._capture.get(prop) + if name in self.size_properties: + self._width, self._height = self._read_size() + with self._lock: + self._roi = (0, self._width, 0, self._height, 1, 1) + self._frames.clear() + self._last_frame = None + return taken + + def show_settings(self): + """Open the window of settings of the driver of the camera (DirectShow on Windows)""" + with self._capture_lock: + self._capture.set(OpenCVProp['CV_CAP_PROP_SETTINGS'].value, 1) + + # --- acquisition ----------------------------------------------------------------------------------------------- + + def setup_acquisition(self): + with self._lock: + self._frames.clear() + + def clear_acquisition(self): + self._running.clear() + self.setup_acquisition() + + def start_acquisition(self, nframes: int = 1): + with self._capture_lock: + if not self._capture.isOpened(): + self._capture.open(self.source) + self._running.set() + + def acquisition_in_progress(self) -> bool: + return self._running.is_set() + + def wait_for_frame(self, since: str = 'lastread', nframes: int = 1, timeout: float = 20.0): + """Read ``nframes`` new frames, which blocks until the camera gives them + + Return ``False`` at once if the acquisition is stopped, as pylablib does, and raise ``TimeoutError`` if the + camera gave nothing after ``timeout`` seconds. + """ + deadline = time.perf_counter() + timeout + n_new = 0 + while n_new < nframes: + if not self._running.is_set(): + return False + with self._capture_lock: + ok, frame = self._capture.read() + if ok: + self._add_frame(frame) + n_new += 1 + elif time.perf_counter() > deadline: + raise TimeoutError('Timeout while waiting for a frame of the OpenCV camera') + else: + time.sleep(0.005) + return True + + def _add_frame(self, frame: np.ndarray): + if frame.ndim == 3 and frame.shape[2] == 3: + frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) + with self._lock: + hstart, hend, vstart, vend, hbin, vbin = self._roi + frame = frame[vstart:vend, hstart:hend] + if hbin > 1 or vbin > 1: # mean of the pixels of a bin, the last incomplete ones are dropped + height, width = (vend - vstart) // vbin, (hend - hstart) // hbin + frame = frame[:height * vbin, :width * hbin] + frame = frame.reshape((height, vbin, width, hbin) + frame.shape[2:]).mean(axis=(1, 3), + dtype=np.float32) + self._frames.append(frame) + self._last_frame = frame + self._n_read += 1 + + def read_multiple_images(self, rng=None, peek: bool = False, missing_frame: str = 'skip', + return_rng: bool = False): + """The frames read and not given yet, as an array (nframes, height, width[, 3])""" + with self._lock: + frames = list(self._frames) + last = self._n_read + first = last - len(frames) + if not peek: + self._frames.clear() + if frames: + array = np.array(frames) + else: + array = np.zeros((0,) + self.get_data_dimensions()) + return (array, (first, last)) if return_rng else array + + def read_newest_image(self, peek: bool = False) -> Optional[np.ndarray]: + with self._lock: + if not peek: + self._frames.clear() + return self._last_frame + + def close(self): + self.clear_acquisition() + with self._capture_lock: + self._capture.release() diff --git a/tests/test_opencv_camera.py b/tests/test_opencv_camera.py new file mode 100644 index 0000000..931a62c --- /dev/null +++ b/tests/test_opencv_camera.py @@ -0,0 +1,292 @@ +"""The OpenCV camera driver and its plugin, without a webcam: a video file read by the real OpenCV, and a fake of +``cv2.VideoCapture`` for what a file cannot do (properties, failing reads, change of the frame size)""" +import importlib.util +import time + +import cv2 +import numpy as np +import pytest + +try: + available = (importlib.util.find_spec('pymodaq.utils.plugin_testing') is not None and + importlib.util.find_spec('pymodaq_plugins_utils.hardware.camera_base') is not None) +except ModuleNotFoundError: + available = False +if not available: + pytest.skip('Needs PyMoDAQ >= 5.3.2 and pymodaq_plugins_utils >= 5.3.1', 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 +except ImportError: + from pymodaq_gui.plotting.utils.plot_utils import RoiInfo + +from pymodaq_plugins_utils.hardware.camera_driver import CameraDriverBase + +from pymodaq_plugins_opencv.hardware.opencv_camera import OpenCVCamera +from pymodaq_plugins_opencv.daq_viewer_plugins.plugins_2D.daq_2Dviewer_OpenCV import DAQ_2DViewer_OpenCV + +WIDTH, HEIGHT = 64, 48 +LEFT_RGB, RIGHT_RGB = (200, 100, 50), (20, 60, 220) # colours of the two halves of the images + + +def make_frame(width=WIDTH, height=HEIGHT) -> np.ndarray: + """What OpenCV gives: BGR""" + frame = np.zeros((height, width, 3), np.uint8) + frame[:, :width // 2] = LEFT_RGB[::-1] + frame[:, width // 2:] = RIGHT_RGB[::-1] + return frame + + +class FakeCapture: + """Stands for ``cv2.VideoCapture``, with a few properties""" + fail = False # the read gives nothing, as an unplugged camera + + def __init__(self, source=0, *args): + self.opened = source != 'missing' + self.values = {cv2.CAP_PROP_FRAME_WIDTH: WIDTH, cv2.CAP_PROP_FRAME_HEIGHT: HEIGHT, + cv2.CAP_PROP_BRIGHTNESS: 128., cv2.CAP_PROP_GAIN: 10.} + + def isOpened(self): + return self.opened + + def open(self, source): + self.opened = True + return True + + def read(self): + time.sleep(0.005) # a webcam gives a frame every few ms at most + if self.fail: + return False, None + return True, make_frame(int(self.values[cv2.CAP_PROP_FRAME_WIDTH]), int(self.values[cv2.CAP_PROP_FRAME_HEIGHT])) + + def get(self, prop): + return self.values.get(prop, -1.) + + def set(self, prop, value): + if prop in self.values: + self.values[prop] = value + return True + return False + + def getBackendName(self): + return 'FAKE' + + def release(self): + self.opened = False + + +@pytest.fixture +def fake_capture(monkeypatch): + monkeypatch.setattr(cv2, 'VideoCapture', FakeCapture) + + +@pytest.fixture +def video(tmp_path): + """A video file of the same images, read by the real OpenCV""" + path = str(tmp_path / 'camera.avi') + writer = cv2.VideoWriter(path, cv2.VideoWriter_fourcc(*'MJPG'), 30., (WIDTH, HEIGHT)) + if not writer.isOpened(): + pytest.skip('This OpenCV cannot write video files') + for _ in range(100): + writer.write(make_frame()) + writer.release() + return path + + +@pytest.fixture +def driver(fake_capture): + camera = OpenCVCamera(0) + yield camera + camera.close() + + +@pytest.fixture +def camera(qapp, fake_capture): + plugin = make_detector(DAQ_2DViewer_OpenCV) + yield plugin + plugin.close() + + +def acquire(driver: OpenCVCamera, nframes=1) -> np.ndarray: + driver.setup_acquisition() + driver.start_acquisition() + assert driver.wait_for_frame(nframes=nframes, timeout=2) + return driver.read_multiple_images() + + +def apply_roi(plugin, rows: slice, columns: slice): + plugin.settings.child('roi', 'update_roi').setValue(True) + plugin.roi_select(RoiInfo.from_slices([rows, columns])) + param = plugin.settings.child('roi', 'apply_roi') + param.setValue(True) + plugin.commit_settings(param) + + +class TestVideoFile: + """The driver with the real OpenCV""" + def test_frames_are_rgb(self, video): + camera = OpenCVCamera(video) + try: + assert camera.get_detector_size() == (WIDTH, HEIGHT) + frames = acquire(camera, 2) + assert frames.shape == (2, HEIGHT, WIDTH, 3) and frames.dtype == np.uint8 + assert frames[0, :, :WIDTH // 2].mean(axis=(0, 1)) == pytest.approx(LEFT_RGB, abs=8) # JPEG compression + assert frames[0, :, WIDTH // 2:].mean(axis=(0, 1)) == pytest.approx(RIGHT_RGB, abs=8) + finally: + camera.close() + + def test_a_file_that_cannot_be_opened_is_an_error(self, tmp_path): + with pytest.raises(IOError): + OpenCVCamera(str(tmp_path / 'nothing.avi')) + + +class TestDriver: + def test_follows_the_driver_contract(self, driver): + assert isinstance(driver, CameraDriverBase) + not_implemented = [name for name in dir(CameraDriverBase) + if not name.startswith('_') and getattr(OpenCVCamera, name) is getattr(CameraDriverBase, name) + and name not in ('get_roi_limits', 'set_frame_format', 'get_pixel_size')] + assert not_implemented == [] + + def test_device_info(self, driver): + assert driver.get_device_info().name == 'OpenCV camera 0 (FAKE)' + + def test_frames_are_given_once(self, driver): + assert acquire(driver, 3).shape == (3, HEIGHT, WIDTH, 3) + assert driver.read_multiple_images().shape[0] == 0 + frames, (first, last) = driver.read_multiple_images(return_rng=True) + assert last == first + + def test_wait_returns_at_once_when_stopped(self, driver): + driver.start_acquisition() + driver.clear_acquisition() + assert not driver.acquisition_in_progress() + assert driver.wait_for_frame(timeout=1) is False + + def test_wait_times_out_when_the_camera_gives_nothing(self, driver): + driver._capture.fail = True + driver.start_acquisition() + with pytest.raises(TimeoutError): + driver.wait_for_frame(timeout=0.05) + + def test_roi(self, driver): + driver.set_roi(hstart=40, hend=60, vstart=10, vend=30) + assert driver.get_data_dimensions() == (20, 20) + frames = acquire(driver) + assert frames.shape == (1, 20, 20, 3) + assert frames[0].mean(axis=(0, 1)) == pytest.approx(RIGHT_RGB) + + def test_binning_averages_the_pixels(self, driver): + driver.set_roi(hstart=0, hend=WIDTH, vstart=0, vend=HEIGHT, hbin=2, vbin=4) + assert driver.get_data_dimensions() == (12, 32) + frames = acquire(driver) + assert frames.shape == (1, 12, 32, 3) and frames.dtype == np.float32 + assert frames[0, 0, 0] == pytest.approx(LEFT_RGB) + + def test_wrong_roi_is_refused(self, driver): + with pytest.raises(ValueError): + driver.set_roi(hstart=0, hend=WIDTH + 1) + with pytest.raises(ValueError): + driver.set_roi(hstart=0, hend=3, hbin=4) + + def test_roi_change_forgets_the_frames_of_the_previous_one(self, driver): + acquire(driver) + driver.set_roi(hstart=0, hend=10, vstart=0, vend=10) + assert driver.read_multiple_images().shape[0] == 0 + assert driver.read_newest_image() is None + + def test_properties_are_those_the_camera_supports(self, driver): + properties = driver.get_properties() + assert properties['PROP_BRIGHTNESS'] == (128, True) + assert properties['PROP_FRAME_WIDTH'] == (WIDTH, True) + assert 'PROP_GAIN' in properties + assert 'PROP_ZOOM' not in properties + assert not any(name.startswith('PROP_DC1394') for name in properties) + + def test_size_change_resets_the_roi(self, driver): + driver.set_roi(hstart=10, hend=20, vstart=10, vend=20) + assert driver.set_property('PROP_FRAME_WIDTH', 32) == 32 + assert driver.get_detector_size() == (32, HEIGHT) + assert driver.get_roi() == (0, 32, 0, HEIGHT, 1, 1) + assert acquire(driver).shape == (1, HEIGHT, 32, 3) + + +class TestPlugin: + def test_snap_gives_a_gray_image_in_pixels(self, camera): + dte = grab_and_wait(camera) + assert dte[0].shape == (HEIGHT, WIDTH) + assert dte[0][0][0, 0] == pytest.approx(0.299 * LEFT_RGB[0] + 0.587 * LEFT_RGB[1] + 0.114 * LEFT_RGB[2], abs=1) + assert camera.x_axis.units == '' and camera.x_axis.size == WIDTH + assert camera.settings['camera_name'] == 'OpenCV camera 0 (FAKE)' + + def test_without_conversion_the_channels_are_given(self, camera): + camera.settings.child('color_conversion').setValue('None') + dte = grab_and_wait(camera) + assert dte[0].labels == ['Red', 'Green', 'Blue'] + assert [data[0, 0] for data in dte[0].data] == list(LEFT_RGB) + + def test_average(self, camera): + assert grab_and_wait(camera, naverage=3)[0].shape == (HEIGHT, WIDTH) + + def test_settings_of_the_camera_are_listed(self, camera): + cam_settings = camera.settings.child('cam_settings') + assert {'PROP_FRAME_WIDTH', 'PROP_BRIGHTNESS', 'PROP_GAIN'} <= set(cam_settings.names) + + def test_exposure_and_buffer_are_not_shown(self, camera): + assert not camera.settings.child('timing_opts', 'exposure_time').opts['visible'] + assert not camera.settings.child('buffer').opts['visible'] + + def test_changing_a_property_is_sent_to_the_camera(self, camera): + param = camera.settings.child('cam_settings', 'PROP_BRIGHTNESS') + param.setValue(50) + camera.commit_settings(param) + assert camera.controller._capture.values[cv2.CAP_PROP_BRIGHTNESS] == 50 + + def test_changing_the_frame_size_updates_the_image_and_the_axes(self, camera): + param = camera.settings.child('cam_settings', 'PROP_FRAME_WIDTH') + param.setValue(32) + camera.commit_settings(param) + dte = grab_and_wait(camera) + assert dte[0].shape == (HEIGHT, 32) + assert camera.x_axis.size == 32 + assert camera.settings['hdet'] == 32 + + def test_roi_drawn_in_the_viewer_is_applied(self, camera): + apply_roi(camera, rows=slice(10, 30), columns=slice(40, 60)) + assert camera.controller.get_roi() == (40, 60, 10, 30, 1, 1) + dte = grab_and_wait(camera) + assert dte[0].shape == (20, 20) + assert camera.x_axis.offset == 40 + + def test_binning(self, camera): + for name in ('x_binning', 'y_binning'): + camera.settings.child('roi', name).setValue(2) + camera.commit_settings(camera.settings.child('roi', name)) + assert grab_and_wait(camera)[0].shape == (HEIGHT // 2, WIDTH // 2) + assert camera.x_axis.scaling == 2 + + def test_live_gives_images_until_stopped(self, camera): + received = [] + camera.dte_signal.connect(received.append) + wait_for_signal(camera.dte_signal, lambda: camera.grab_data(1, live=True), timeout_ms=3000) + wait_for_signal(camera.dte_signal, lambda: None, timeout_ms=3000) + camera.stop() + assert len(received) >= 2 + assert not camera.controller.acquisition_in_progress() + + def test_roi_can_change_during_the_live_and_the_plugin_still_closes(self, qapp, fake_capture): + plugin = make_detector(DAQ_2DViewer_OpenCV) + try: + wait_for_signal(plugin.dte_signal, lambda: plugin.grab_data(1, live=True), timeout_ms=3000) + apply_roi(plugin, rows=slice(10, 30), columns=slice(40, 60)) + assert plugin.controller.get_roi() == (40, 60, 10, 30, 1, 1) + finally: + plugin.close() + + def test_close_releases_the_camera(self, qapp, fake_capture): + plugin = make_detector(DAQ_2DViewer_OpenCV) + capture = plugin.controller._capture + plugin.close() + assert not capture.isOpened()