Clarity support with pipeline for data processing

microscope/devices.py

@@ -256,7 +256,10 @@ def initialize(self):
 
     def shutdown(self):
         """Shutdown the device for a prolonged period of inactivity."""
-        self.disable()
+        try:
+            self.disable()
+        except Exception as e:
+            _logger.warning("Exception in disable() during shutdown: %s" % e)
         _logger.info("Shutting down ... ... ...")
         self._on_shutdown()
         _logger.info("... ... ... ... shut down completed.")
@@ -304,11 +307,15 @@ def get_setting(self, name):
 
     def get_all_settings(self):
         """Return ordered settings as a list of dicts."""
-        try:
-            return {k: v.get() for k, v in self._settings.items()}
-        except Exception as err:
-            _logger.error("in get_all_settings:", exc_info=err)
-            raise
+        # Fetching some settings may fail depending on device state.
+        # Report these values as 'None' and continue fetching other settings.
+        def catch(f):
+            try:
+                return f()
+            except Exception as err:
+                _logger.error("getting %s: %s" % (f.__self__.name, err))
+                return None
+        return {k: catch(v.get) for k, v in self._settings.items()}
 
     def set_setting(self, name, value):
         """Set a setting."""
@@ -411,6 +418,8 @@ def __init__(self, buffer_length=0, **kwargs):
         self._acquiring = False
         # A condition to signal arrival of a new data and unblock grab_next_data
         self._new_data_condition = threading.Condition()
+        # A data processing pipeline: a list of f(data) -> data.
+        self.pipeline = []
 
     def __del__(self):
         self.disable()
@@ -431,18 +440,6 @@ def enable(self):
         Implement device-specific code in _on_enable .
         """
         _logger.debug("Enabling ...")
-        if self._using_callback:
-            if self._fetch_thread:
-                self._fetch_thread_run = False
-        else:
-            if not self._fetch_thread or not self._fetch_thread.is_alive():
-                self._fetch_thread = Thread(target=self._fetch_loop)
-                self._fetch_thread.daemon = True
-                self._fetch_thread.start()
-        if not self._dispatch_thread or not self._dispatch_thread.is_alive():
-            self._dispatch_thread = Thread(target=self._dispatch_loop)
-            self._dispatch_thread.daemon = True
-            self._dispatch_thread.start()
         # Call device-specific code.
         try:
             result = self._on_enable()
@@ -454,7 +451,20 @@ def enable(self):
             self.enabled = False
         else:
             self.enabled = True
-        _logger.debug("... enabled.")
+            # Set up data fetching
+            if self._using_callback:
+                if self._fetch_thread:
+                    self._fetch_thread_run = False
+            else:
+                if not self._fetch_thread or not self._fetch_thread.is_alive():
+                    self._fetch_thread = Thread(target=self._fetch_loop)
+                    self._fetch_thread.daemon = True
+                    self._fetch_thread.start()
+            if not self._dispatch_thread or not self._dispatch_thread.is_alive():
+                self._dispatch_thread = Thread(target=self._dispatch_loop)
+                self._dispatch_thread.daemon = True
+                self._dispatch_thread.start()
+            _logger.debug("... enabled.")
         return self.enabled
 
 
@@ -483,8 +493,14 @@ def _fetch_data(self):
         return None
 
     def _process_data(self, data):
-        """Do any data processing and return data."""
-        return data
+        """Do any data processing and return data.
+
+        Subclasses should add their processing to self.pipeline in preference
+        to overriding this method. Anything that overrides this method must
+        call super()_process_data(data) either before or after its own
+        processing."""
+        import functools
+        return functools.reduce(lambda x, f: f(x), self.pipeline, data)
 
     def _send_data(self, client, data, timestamp):
         """Dispatch data to the client."""
@@ -611,6 +627,8 @@ def grab_next_data(self, soft_trigger=True):
         :param soft_trigger: calls soft_trigger if True,
                                waits for hardware trigger if False.
         """
+        if not self.enabled:
+            raise Exception("Camera not enabled.")
         self._new_data_condition.acquire()
         # Push self onto client stack.
         self.set_client(self)
@@ -652,6 +670,8 @@ def __init__(self, **kwargs):
         self._client_transform = (False, False, False)
         # Result of combining client and readout transforms
         self._transform = (False, False, False)
+        # Add _apply_transform to pipeline.
+        self.pipeline.insert(0, self._apply_transform)
         # A transform provided by the client.
         self.add_setting('transform', 'enum',
                          lambda: CameraDevice.ALLOWED_TRANSFORMS.index(self._transform),
@@ -669,18 +689,22 @@ def __init__(self, **kwargs):
                          self.get_roi,
                          self.set_roi,
                          None)
-    def _process_data(self, data):
+
+    def _apply_transform(self, data):
         """Apply self._transform to data."""
         flips = (self._transform[0], self._transform[1])
         rot = self._transform[2]
 
         # Choose appropriate transform based on (flips, rot).
-        return {(0, 0): numpy.rot90(data, rot),
-                (0, 1): numpy.flipud(numpy.rot90(data, rot)),
-                (1, 0): numpy.fliplr(numpy.rot90(data, rot)),
-                (1, 1): numpy.fliplr(numpy.flipud(numpy.rot90(data, rot)))
-                }[flips]
-
+        # Do rotation
+        data = numpy.rot90(data, rot)
+        # Flip
+        data = {(0, 0): lambda d: d,
+               (0, 1): numpy.flipud,
+               (1, 0): numpy.fliplr,
+               (1, 1): lambda d: numpy.fliplr(numpy.flipud(d))
+               }[flips](data)
+        return data
 
     def set_readout_mode(self, description):
         """Set the readout mode and _readout_transform."""

microscope/filterwheels/aurox.py

@@ -18,11 +18,38 @@
 
 """Adds support for Aurox devices
 
-Requires package hidapi."""
+Requires package hidapi.
 
+Config sample:
+
+device(microscope.filterwheels.aurox.Clarity,
+       {'camera': 'microscope.Cameras.cameramodule.SomeCamera',
+        'camera.someSetting': value})
+
+Deconvolving data requires:
+ * availability of clarity_process and cv2
+ * successful completion of a calibration step
+   + set_mode(Modes.calibrate)
+   + trigger the camera to generate an image
+   + when the camera returns the image, calibration is complete
+"""
 import hid
+import logging
 import microscope.devices
-from enum import Enum
+from typing import Mapping
+from enum import IntEnum
+
+_logger = logging.getLogger(__name__)
+
+try:
+    # Currently, clarity_process is a module that is not packaged, so needs
+    # to be put on the python path somewhere manually.
+    from clarity_process import ClarityProcessor
+except:
+    _logger.warning("Could not import clarity_process module: no processing available.")
+
+
+Mode = IntEnum("Mode", "difference, raw, calibrate")
 
 ## Clarity constants. These may differ across products, so mangle names.
 # USB IDs
@@ -72,7 +99,11 @@
 _Clarity__SETSVCMODE1 = 0xe0 #1 byte for service mode. SLEEP activates service mode. RUN returns to normal mode.
 
 
-class Clarity(microscope.devices.FilterWheelBase):
+class Clarity(microscope.devices.ControllerDevice, microscope.devices.FilterWheelBase):
+    """Adds support for Aurox Clarity
+
+    Acts as a ControllerDevice providing the camera attached to the Clarity.
+    """
     _slide_to_sectioning = {__SLDPOS0: 'bypass',
                             __SLDPOS1: 'low',
                             __SLDPOS2: 'mid',
@@ -86,20 +117,94 @@ class Clarity(microscope.devices.FilterWheelBase):
                   __GETSERIAL: 4,
                   __FULLSTAT: 10}
 
-    def __init__(self, **kwargs):
+
+    def __init__(self, camera=None, **kwargs) -> None:
+        """Create a Clarity instance controlling an optional Camera device.
+
+           :param camera: a class to control the connected camera
+           :param kwargs: Provide camera parameters as keyword arguments:
+                          'camera.some_parameter'
+                          The 'camera.' prefix will be stripped, and
+                          'some_parameter' passed to the camera's constructor.
+        """
+        # Extract kwargs for camera device.
+        cam_kw_keys = [k for k in kwargs if k.startswith("camera.")]
+        cam_kwargs = {}
+        for key in cam_kw_keys:
+            cam_kwargs[key.replace("camera.", "")] = kwargs[key]
+            del kwargs[key]
         super().__init__(**kwargs)
         from threading import Lock
+        # A comms lock.
         self._lock = Lock()
+        # The hid connection object
         self._hid = None
+        if camera is None:
+            self._cam = None
+            _logger.warning("No camera specified.")
+        else:
+            self._cam = camera(**cam_kwargs)
+            self._cam.pipeline.append(self._c_process_data)
+        # Is processing available?
+        self._can_process = 'ClarityProcessor' in globals()
+        # Data processor object, created after calibration
+        self._processor = None
+        # Acquisition mode
+        self._mode = Mode.raw
+        # Add device settings
         self.add_setting("sectioning", "enum",
                          self.get_slide_position,
                          lambda val: self.set_slide_position(val),
                          self._slide_to_sectioning)
+        self.add_setting("mode", "enum",
+                         lambda: self._mode.name,
+                         self.set_mode,
+                         Mode)
+
+    def _c_process_data(self, data):
+        """A function to insert into the camera's processing pipeline.
+
+        Depending on the mode, this function will pass through, perform
+        calibration, or deconvolve the camera data."""
+        if self._mode == Mode.raw:
+            return data
+        elif self._mode == Mode.difference:
+            if self._processor is None:
+                raise Exception("Not calibrated yet - can not process image")
+            return self._processor.process(data).get() # get converts UMat to ndarray.
+        elif self._mode == Mode.calibrate:
+            # This will introduce a significant delay, but returning the
+            # image indicates that the calibration step is complete.
+            self._processor = ClarityProcessor(data)
+            return data
+        else:
+            raise Exception("Unrecognised mode: %s", self._mode)
+
+    @property
+    def devices(self) -> Mapping[str, microscope.devices.Device]:
+        """Devices property, required by ControllerDevice interface."""
+        if self._cam:
+            return {'camera': self._cam}
+        else:
+            return {}
+
+    def set_mode(self, mode: Mode) -> None:
+        """Set the operation mode"""
+        if mode in [Mode.calibrate, Mode.difference] and not self._can_process:
+            raise Exception ("Processing not available")
+        if mode == Mode.calibrate:
+            self._set_calibration(True)
+        else:
+            self._set_calibration(False)
+        self._mode = mode
 
     def _send_command(self, command, param=0, max_length=16, timeout_ms=100):
         """Send a command to the Clarity and return its response"""
         if not self._hid:
-            self.open()
+            try:
+                self.open()
+            except:
+                raise Exception("Connection error")
         with self._lock:
             # The device expects a list of 16 integers
             buffer = [0x00] * max_length # The 0th element must be 0.
@@ -153,15 +258,13 @@ def get_id(self):
         return self._send_command(__GETSERIAL)
 
     def _on_enable(self):
-        if not self.is_connected:
-            self.open()
         self._send_command(__SETONOFF, __RUN)
         return self._send_command(__GETONOFF) == __RUN
 
     def _on_disable(self):
         self._send_command(__SETONOFF, __SLEEP)
 
-    def set_calibration(self, state):
+    def _set_calibration(self, state):
         if state:
             result = self._send_command(__SETCAL, __CALON)
         else:
@@ -188,12 +291,17 @@ def get_slides(self):
         return (self._slide_to_sectioning)
 
     def get_status(self):
-        # Fetch 10 bytes VERSION[3],ONOFF,SHUTTER,SLIDE,FILT,CAL,??,??
-        result = self._send_command(__FULLSTAT)
-        if result is None:
-            return
         # A status dict to populate and return
-        status = {}
+        status = dict.fromkeys(['connected','on','door open','slide',
+                                'filter','calibration','busy', 'mode'])
+        status['mode'] = self._mode.name
+        # Fetch 10 bytes VERSION[3],ONOFF,SHUTTER,SLIDE,FILT,CAL,??,??
+        try:
+            result = self._send_command(__FULLSTAT)
+            status['connected'] = True
+        except:
+            status['connected'] = False
+            return status
         # A list to track states, any one of which mean the device is busy.
         busy = []
         # Disk running
@@ -263,8 +371,5 @@ def set_position(self, pos, blocking=True):
             pass
         return result
 
-    def _on_shutdown(self):
-        pass
-
     def initialize(self):
-        pass
+        super().initialize()