enh: fetch QuickView image and trace data in EventGetterThread

Author: paulmueller

CHANGELOG

@@ -1,3 +1,6 @@
+2.26.3
+ - enh: fetch QuickView image and trace data in `EventGetterThread`
+ - ref: new `qv_event_getter` and `qv_image_vis` submodules
 2.26.2
  - reg: filters not applied in final plots since 2.25.1
  - fix: prevent `pipeline.lock` deadlocks via Qt signal cluttering

dcscope/gui/main.py

@@ -390,11 +390,18 @@ def add_plot_window(self, plot_id):
             self.subwindows_plots[plot_id] = sub
         sub.show()
 
+    def close(self):
+        if self.widget_quick_view is not None:
+            self.widget_quick_view.close()
+        return super(DCscope, self).close()
+
     @QtCore.pyqtSlot(QtCore.QEvent)
     def closeEvent(self, event):
         """Determine what happens when the user wants to quit"""
         if self.pipeline.slots or self.pipeline.filters:
             if self.on_action_clear():
+                if self.widget_quick_view is not None:
+                    self.widget_quick_view.close()
                 event.accept()
             else:
                 event.ignore()

dcscope/gui/quick_view/qv_event_getter.py

@@ -0,0 +1,103 @@
+import logging
+import threading
+import time
+import traceback
+
+import dclab
+from dclab.rtdc_dataset import RTDCBase
+import numpy as np
+from PyQt6 import QtCore
+
+
+class EventGetterThread(QtCore.QThread):
+    new_event_data = QtCore.pyqtSignal(dict)
+
+    def __init__(self, parent):
+        super(EventGetterThread, self).__init__(parent)
+        self.worker_lock = threading.Lock()
+        self.event_abort = threading.Event()
+        self.request = (None, None)
+        self.prev_request = (None, None)
+        self.logger = logging.getLogger(__name__)
+
+    def close(self):
+        self.event_abort.set()
+        while self.isRunning():
+            time.sleep(0.1)
+
+    @QtCore.pyqtSlot(RTDCBase, int)
+    def request_event_data(self, ds: RTDCBase, event_index: int):
+        with self.worker_lock:
+            self.request = (ds, event_index)
+
+    def run(self):
+        while not self.event_abort.is_set():
+            with self.worker_lock:
+                ds, event_index = self.request
+
+            if self.prev_request == (ds, event_index):
+                time.sleep(0.05)
+                continue
+
+            self.prev_request = (ds, event_index)
+
+            if ds is not None and event_index is not None:
+                try:
+                    event_data = self.get_event_data(ds, event_index)
+                    self.new_event_data.emit(event_data)
+                except BaseException:
+                    self.logger.error(traceback.format_exc())
+            else:
+                time.sleep(0.01)
+
+    def get_event_data(self, ds: RTDCBase, event_index: int):
+        """Return all event data relevant for QuickView visualization"""
+        data = {}
+        data["index"] = event_index
+        try:
+            # Image data
+            for feat in ["image", "image_bg", "mask", "qpi_amp", "qpi_pha"]:
+                if feat in ds:
+                    data[feat] = ds[feat][event_index]
+
+            # Trace data
+            if "trace" in ds:
+                try:
+                    data["traces"] = self.get_event_traces(ds, event_index)
+                except BaseException:
+                    self.logger.error(traceback.format_exc())
+        except IndexError:
+            if event_index != 0:
+                data = self.get_event_data(ds, 0)
+            else:
+                self.logger.error(traceback.format_exc())
+        return data
+
+    def get_event_traces(self,
+                         ds: RTDCBase,
+                         event_index: int,
+                         ):
+        """Return all trace data"""
+        tdata = {}
+        # time axis
+        flsamples = ds.config["fluorescence"]["samples per event"]
+        flrate = ds.config["fluorescence"]["sample rate"]
+        fltime = np.arange(flsamples) / flrate * 1e6
+        tdata["time"] = fltime
+
+        # fluorescence traces and pos/width/max features
+        range_fl = [0, 0]
+        for name in dclab.dfn.FLUOR_TRACES:
+            flid = name.split("_")[0]
+            if name in ds["trace"]:
+                # show the trace information
+                tracey = ds["trace"][name][event_index]  # trace data
+                range_fl[0] = min(range_fl[0], tracey.min())
+                range_fl[1] = max(range_fl[1], tracey.max())
+                tdata[name] = tracey
+                for which in ["pos", "width", "max"]:
+                    feat = f"{flid}_{which}"
+                    if feat not in tdata:
+                        tdata[feat] = ds[feat][event_index]
+
+        return tdata

dcscope/gui/quick_view/qv_image_vis.py

@@ -0,0 +1,222 @@
+from typing import Literal
+
+import numpy as np
+import pyqtgraph as pg
+from scipy.ndimage import binary_erosion
+
+
+cmap_pha: pg.ColorMap = pg.colormap.get('CET-D1A', skipCache=True)
+cmap_pha_with_black: pg.ColorMap = pg.colormap.get('CET-D1A', skipCache=True)
+cmap_pha_with_black.color[0] = [0, 0, 0, 1]
+
+
+@staticmethod
+def convert_to_rgb(cell_img):
+    """Add a third axis of length 3 with copies"""
+    cell_img = cell_img.reshape(
+        cell_img.shape[0], cell_img.shape[1], 1)
+    return np.repeat(cell_img, 3, axis=2)
+
+
+def get_rgb_image(data: dict,
+                  feat: str,
+                  zoom: bool = False,
+                  draw_contour: bool = False,
+                  auto_contrast: bool = False,
+                  subtract_background: bool = False,
+                  ) -> tuple[np.ndarray, float, float, pg.ColorMap | None]:
+    """Return a pretty visualization of image data"""
+    if feat == "image":
+        cmap = None
+        cell_img, vmin, vmax = prepare_event_image_image(
+            data,
+            zoom=zoom,
+            draw_contour=draw_contour,
+            auto_contrast=auto_contrast,
+            subtract_background=subtract_background,
+        )
+    elif feat == "qpi_amp":
+        cmap = None
+        cell_img, vmin, vmax = prepare_event_image_qpi_amp(
+            data,
+            zoom=zoom,
+            draw_contour=draw_contour,
+            auto_contrast=auto_contrast,
+        )
+    elif feat == "qpi_pha":
+        cell_img, vmin, vmax, cmap = prepare_event_image_qpi_pha(
+            data,
+            zoom=zoom,
+            draw_contour=draw_contour,
+            auto_contrast=auto_contrast,
+        )
+    else:
+        raise KeyError(f"Unknown image feature '{feat}'")
+
+    return cell_img, vmin, vmax, cmap
+
+
+def image_insert_contour(cell_img: np.ndarray,
+                         mask: np.ndarray,
+                         cmap_levels: tuple[float, float],
+                         contour_style: Literal["red", "lowest-level"],
+                         ):
+    """Insert contour data in an image"""
+    # Compute contour image from mask. If you are wondering
+    # whether this is kosher, please take a look at issue #76:
+    # https://github.com/DC-analysis/dclab/issues/76
+    cont = mask ^ binary_erosion(mask)
+    if contour_style == "red":
+        vmin, vmax = cmap_levels
+        # draw red contour for grayscale images
+        ch_red = vmin + (vmax - vmin) * 0.7
+        ch_other = vmin
+        # assign channel values for contour
+        cell_img[cont, 0] = ch_red
+        cell_img[cont, 1] = ch_other
+        cell_img[cont, 2] = ch_other
+    elif contour_style == "lowest-level":
+        # use the lowest value from the colormap
+        # (used for e.g. phase images)
+        cell_img[cont] = cmap_levels[0]
+
+    return cell_img
+
+
+def image_zoom(cell_img, mask):
+    """Zoom in on the image"""
+    xv, yv = np.where(mask)
+    idminx = xv.min() - 5
+    idminy = yv.min() - 5
+    idmaxx = xv.max() + 5
+    idmaxy = yv.max() + 5
+    idminx = idminx if idminx >= 0 else 0
+    idminy = idminy if idminy >= 0 else 0
+    shx, shy = mask.shape
+    idmaxx = idmaxx if idmaxx < shx else shx
+    idmaxy = idmaxy if idmaxy < shy else shy
+    return cell_img[idminx:idmaxx, idminy:idmaxy]
+
+
+def prepare_event_image_image(
+        data,
+        zoom: bool = False,
+        draw_contour: bool = False,
+        auto_contrast: bool = False,
+        subtract_background: bool = False,
+        ) -> tuple[np.ndarray, float, float]:
+    """Prepare to draw a regular image event"""
+    cell_img = data["image"]
+
+    if zoom and "mask" in data:
+        cell_img = image_zoom(cell_img, data["mask"])
+
+    # apply background correction
+    if subtract_background and "image_bg" in data:
+
+        bgimg = data["image_bg"].astype(np.int16)
+        if zoom and "mask" in data:
+            bgimg = image_zoom(bgimg, data["mask"])
+
+        cell_img = cell_img.astype(np.int16)
+        cell_img = cell_img - bgimg + int(np.mean(bgimg))
+
+    # automatic contrast
+    if auto_contrast:
+        vmin, vmax = cell_img.min(), cell_img.max()
+    else:
+        vmin, vmax = (0, 255)
+
+    cell_img = convert_to_rgb(cell_img)
+
+    if draw_contour and "mask" in data:
+        mask = data["mask"]
+        if zoom:
+            mask = image_zoom(mask, mask)
+
+        cell_img = image_insert_contour(
+            cell_img,
+            mask,
+            cmap_levels=(vmin, vmax),
+            contour_style="red",
+        )
+
+    return cell_img, vmin, vmax
+
+
+def prepare_event_image_qpi_amp(
+        data,
+        zoom: bool = False,
+        draw_contour: bool = False,
+        auto_contrast: bool = False,
+        ) -> tuple[np.ndarray, float, float]:
+    """Prepare to draw a QPI amplitude event image"""
+    cell_img = data["qpi_amp"]
+
+    if zoom and "mask" in data:
+        cell_img = image_zoom(cell_img, data["mask"])
+
+    if auto_contrast:
+        vmin, vmax = cell_img.min(), cell_img.max()
+    else:
+        vmin, vmax = (0, 2)
+
+    cell_img = convert_to_rgb(cell_img)
+
+    if draw_contour and "mask" in data:
+        mask = data["mask"]
+        if zoom:
+            mask = image_zoom(mask, mask)
+
+        cell_img = image_insert_contour(
+            cell_img,
+            mask,
+            cmap_levels=(vmin, vmax),
+            contour_style="red",
+        )
+
+    return cell_img, vmin, vmax
+
+
+def prepare_event_image_qpi_pha(
+        data,
+        zoom: bool = False,
+        draw_contour: bool = False,
+        auto_contrast: bool = False,
+        ) -> tuple[np.ndarray, float, float, pg.ColorMap]:
+    """Prepare to draw a QPI phase event image"""
+    cell_img = np.copy(data["qpi_pha"])
+
+    if zoom and "mask" in data:
+        cell_img = image_zoom(cell_img, data["mask"])
+
+    if auto_contrast:
+        # phase values centered around zero
+        vmin_abs, vmax_abs = np.abs(cell_img.min()), np.abs(cell_img.max())
+        v_largest = max(vmax_abs, vmin_abs)
+        vmin, vmax = -v_largest, v_largest
+    else:
+        vmin, vmax = (-3.14, 3.14)
+
+    if draw_contour and "mask" in data:
+        # offset required for auto-contrast with contour
+        # two times the contrast range, divided by the cmap length
+        # this essentially adds a cmap point for our contour
+        offset = 2 * ((vmax - vmin) / len(cmap_pha.color))
+        vmin -= offset
+
+        mask = data["mask"]
+        if zoom:
+            mask = image_zoom(mask, mask)
+
+        cell_img = image_insert_contour(
+            cell_img,
+            mask,
+            cmap_levels=(vmin, vmax),
+            contour_style="lowest-level",
+        )
+        cmap = cmap_pha_with_black
+    else:
+        cmap = cmap_pha
+
+    return cell_img, vmin, vmax, cmap