Refactor camera wizards and fix probe positioning functionality

laserstudio/config_schema/probe.schema.json

@@ -14,10 +14,10 @@
       "type": "array",
       "minItems": 2,
       "maxItems": 2,
-      "description": "The relative position from the center of the camera, in micrometers.",
+      "description": "The relative position from the center of the camera without distortion, in pixels.",
       "items": {
-        "type": "integer",
-        "suffix": "um"
+        "type": "number",
+        "suffix": "px"
       }
     }
   }

laserstudio/instruments/camera.py

@@ -1,7 +1,8 @@
 import os
 import logging
-from typing import Optional, Literal, cast, Any
+from typing import Literal, cast, Any
 import numpy
+from numpy.typing import NDArray
 import cv2
 from PyQt6.QtCore import QTimer, pyqtSignal, Qt
 from PyQt6.QtGui import QImage, QTransform
@@ -42,19 +43,15 @@ def __init__(self, config: dict[str, Any]):
             list[float], config.get("pixel_size_in_um", [1.0, 1.0])
         )
 
-        # Objective
-        self.objective = cast(float, config.get("objective", 1.0))
-        self.select_objective(self.objective)
-
         # Correction matrix
-        self.correction_matrix: Optional[QTransform] = None
+        self.correction_matrix: QTransform | None = None
 
         # Shutter
         shutter = config.get("shutter")
-        self.shutter: Optional[ShutterInstrument] = None
-        if type(shutter) is dict and shutter.get("enable", True):
+        self.shutter: ShutterInstrument | None = None
+        if isinstance(shutter, dict) and shutter.get("enable", True):
             try:
-                if (device_type := shutter.get("type")) == "TIC":
+                if (device_type := cast(str, shutter.get("type"))) == "TIC":
                     self.shutter = TicShutterInstrument(shutter)
                 else:
                     logging.getLogger("laserstudio").error(
@@ -65,16 +62,16 @@ def __init__(self, config: dict[str, Any]):
                     f"Shutter is enabled but device could not be created: {str(e)}... Skipping."
                 )
 
-        # Whtie and black levels adjustment
+        # White and black levels adjustment
         self.black_level = 0.0
         self.white_level = 1.0
 
         # Image averaging
-        self._last_frame_accumulator: Optional[numpy.ndarray] = None
+        self._last_frame_accumulator: NDArray[Any] | None = None
         # The number of images to average
         self._image_averaging = 1
         # The number of images that have been averaged
-        self.number_of_averaged_images = 0
+        self.number_of_averaged_images: int = 0
 
         self._last_neg = None
         self._last_pos = numpy.zeros((self.width, self.height), dtype=numpy.uint8)
@@ -83,18 +80,21 @@ def __init__(self, config: dict[str, Any]):
         # When the number of images to average is hit, and a new frame is retrieved,
         # the oldest one is removed from the accumulator and the new one is added.
         self.windowed_averaging = True
-        self._last_frames: list[numpy.ndarray] = []
+        self._last_frames: list[NDArray[Any]] = []
 
         # Reference image feature
-        self.reference_image_accumulators: dict[str, numpy.ndarray] = {}
+        self.reference_image_accumulators: dict[str, NDArray[Any]] = {}
         self.current_reference_image = "Reference 0"
         self.show_negative_values = True
 
         # The value of a white pixel
         self.white_value = 2**8 - 1
 
+        # Objective
+        self.objective = cast(float, config.get("objective", 1.0))
+
     @property
-    def reference_image_accumulator(self) -> Optional[numpy.ndarray]:
+    def reference_image_accumulator(self) -> NDArray[Any] | None:
         """
         Returns the current reference image.
 
@@ -103,7 +103,7 @@ def reference_image_accumulator(self) -> Optional[numpy.ndarray]:
         return self.reference_image_accumulators.get(self.current_reference_image)
 
     @reference_image_accumulator.setter
-    def reference_image_accumulator(self, value: Optional[numpy.ndarray]):
+    def reference_image_accumulator(self, value: NDArray[Any] | None):
         if (
             value is None
             and self.current_reference_image in self.reference_image_accumulators
@@ -114,7 +114,7 @@ def reference_image_accumulator(self, value: Optional[numpy.ndarray]):
         # Do nothing...
 
     @property
-    def last_frame_accumulator(self) -> Optional[numpy.ndarray]:
+    def last_frame_accumulator(self) -> NDArray[Any] | None:
         """
         Returns the last frame accumulator. See accumulate_frame, and image_averaging for more details.
 
@@ -129,11 +129,33 @@ def last_frame_accumulator(self) -> Optional[numpy.ndarray]:
     def select_objective(self, factor: float):
         """Select an objective with a magnifying factor.
 
-        :param factor: The magnifying factor of the objective (5x, 10x, 20x, 50x...)
+        :param factor: The magnifying factor of the objective (1x, 5x, 10x, 20x, 50x...)
         """
         self.objective = factor
-        self.width_um = self.width * self.pixel_size_in_um[0] / factor
-        self.height_um = self.height * self.pixel_size_in_um[1] / factor
+        logging.getLogger("laserstudio").debug(
+            f"Camera's objective changed to {factor}x"
+        )
+        logging.getLogger("laserstudio").debug(
+            f"Camera's width: {self.width}px, height: {self.height}px"
+        )
+        logging.getLogger("laserstudio").debug(
+            f"Image's dimension {self.width_um}\xa0µm; {self.height_um}\xa0µm (considering the objective)"
+        )
+        self.parameter_changed.emit("objective", factor)
+
+    @property
+    def width_um(self) -> float:
+        """
+        Returns the width in micrometers, considering the objective.
+        """
+        return self.width * self.pixel_size_in_um[0] / self.objective
+
+    @property
+    def height_um(self) -> float:
+        """
+        Returns the height in micrometers, considering the objective.
+        """
+        return self.height * self.pixel_size_in_um[1] / self.objective
 
     def get_last_qimage(self) -> QImage:
         """
@@ -163,7 +185,7 @@ def get_last_pil_image(self) -> Image.Image:
             im = Image.frombytes(mode=mode, data=data, size=size)
         return im
 
-    def capture_image(self) -> Optional[numpy.ndarray]:
+    def capture_image(self) -> NDArray[Any] | None:
         """
         To be overridden by the subclasses or CameraInstrument
 
@@ -173,7 +195,7 @@ def capture_image(self) -> Optional[numpy.ndarray]:
 
     def get_last_image(
         self,
-    ) -> tuple[int, int, Literal["L", "I;16", "RGB"], Optional[bytes]]:
+    ) -> tuple[int, int, Literal["L", "I;16", "RGB"], bytes | None]:
         """
         Capture an image and construct a Gray, 16bit Gray or RGB byte array.
 
@@ -232,7 +254,7 @@ def clear_averaged_images(self):
         self._last_frame_accumulator = None
         self.number_of_averaged_images = 0
 
-    def accumulate_frame(self, new_frame: numpy.ndarray):
+    def accumulate_frame(self, new_frame: NDArray[Any]):
         """
         Accumulates the given frame and removes the oldest one
           if windowed averaging is active.
@@ -277,7 +299,7 @@ def average_count(self) -> int:
         """
         return self.number_of_averaged_images
 
-    def apply_levels(self, image: numpy.ndarray) -> numpy.ndarray:
+    def apply_levels(self, image: NDArray[Any]) -> NDArray[Any]:
         """
         Apply the black and white levels to the image before displaying it.
 
@@ -295,7 +317,7 @@ def apply_levels(self, image: numpy.ndarray) -> numpy.ndarray:
         )
         return image.clip(min=0).astype(type_)
 
-    def compute_histogram(self, frame: numpy.ndarray, width: int = -1):
+    def compute_histogram(self, frame: NDArray[Any], width: int = -1):
         """
         Computes the histogram of the given frame.
 
@@ -313,7 +335,7 @@ def compute_histogram(self, frame: numpy.ndarray, width: int = -1):
             range=(0, numpy.iinfo(frame.dtype).max),
         )
 
-    def histogram_to_string(self, hist: numpy.ndarray, nlines=2):
+    def histogram_to_string(self, hist: NDArray[Any], nlines: int = 2) -> list[str]:
         """
         Returns the histogram as a string representation.
 
@@ -323,7 +345,7 @@ def histogram_to_string(self, hist: numpy.ndarray, nlines=2):
         """
         bar = " ▁▂▃▄▅▆▇█"
         hist = nlines * (hist / max(hist)) * (len(bar) - 1)
-        hists = []
+        hists: list[str] = []
         for i in range(nlines):
             offset = i * len(bar)
             val = [int(i) - offset for i in hist]
@@ -349,7 +371,7 @@ def levels_to_string(self, width: int = -1) -> tuple[str, str]:
 
     def show_histogram_terminal(
         self,
-        frame: Optional[numpy.ndarray] = None,
+        frame: NDArray[Any] | None = None,
         nlines: int = 5,
         nbins: int = 0,
     ):
@@ -397,7 +419,7 @@ def take_reference_image(self, do_take: bool):
 
     def substract_reference_image(
         self,
-    ) -> tuple[numpy.ndarray, Optional[numpy.ndarray]]:
+    ) -> tuple[NDArray[Any], NDArray[Any] | None]:
         """
         Substract the reference_image_accumulator from the current accumulator
 
@@ -424,7 +446,7 @@ def substract_reference_image(
         return self._last_pos, self._last_neg
 
     @property
-    def last_frame(self) -> numpy.ndarray:
+    def last_frame(self) -> NDArray[Any]:
         """
         Return the frame that should be analysed or displayed.
 
@@ -435,8 +457,8 @@ def last_frame(self) -> numpy.ndarray:
         return self.construct_display_image(pos, neg)
 
     def construct_display_image(
-        self, pos: numpy.ndarray, neg: Optional[numpy.ndarray] = None
-    ) -> numpy.ndarray:
+        self, pos: NDArray[Any], neg: NDArray[Any] | None = None
+    ) -> NDArray[Any]:
         """
         Construct the display image from the positive and negative images.
 
@@ -528,7 +550,6 @@ def settings(self, data: dict[str, Any]):
             )
         if "objective" in data:
             self.select_objective(data["objective"])
-            self.parameter_changed.emit("objective", data["objective"])
 
     @property
     def laplacian_std_dev(self) -> float:
@@ -537,8 +558,7 @@ def laplacian_std_dev(self) -> float:
 
         :return: The standard deviation of the Laplacian operator on the last image.
         """
-        if (last_frame := self.last_frame) is None:
-            return 0.0
+        last_frame = self.last_frame
         # KSIZE (3): Aperture size used to compute the
         #   second-derivative filters. See getDerivKernels for details.
         #   The size must be positive and odd.

laserstudio/instruments/camera_raptor.py

@@ -182,8 +182,8 @@ def __init__(self, config: dict):
         except SerialException as e:
             raise ConnectionFailure() from e
 
+        # Because of the ratio being wrong (v4l to cv2 conversion)
         self.width = self.width // 2
-        self.width_um = self.width_um // 2
 
         self.last_frame_number = 0
 

laserstudio/instruments/camera_usb.py

@@ -23,16 +23,6 @@ def __init__(self, config: dict):
             config.get("height", self.__video_capture.get(cv2.CAP_PROP_FRAME_HEIGHT))
         )
 
-        self.width_um = self.width * self.pixel_size_in_um[0]
-        self.height_um = self.height * self.pixel_size_in_um[1]
-
-        logging.getLogger("laserstudio").info(
-            f"Camera's resolution {self.width}px; {self.height}px"
-        )
-        logging.getLogger("laserstudio").info(
-            f"Image's dimension {self.width_um}\xa0µm; {self.height_um}\xa0µm (without considering any magnifier)"
-        )
-
     def __del__(self):
         self.__video_capture.release()
 

laserstudio/instruments/pdm.py

@@ -52,9 +52,9 @@ def __init__(self, config: dict[str, Any]):
             )
             try:
                 link = Link(dev)
-                logging.getLogger("laserstudio").info("OK")
+                logging.getLogger("laserstudio").info("Connection to PDM OK")
             except ConnectionFailure:
-                logging.getLogger("laserstudio").info("Failed")
+                logging.getLogger("laserstudio").info("Connection to PDM failed")
                 raise
             PDMInstrument.__PDM_LINKS[dev] = link
         self.pdm = pdm = PDM(config["num"], link)

laserstudio/instruments/probe.py

@@ -1,6 +1,7 @@
 from PyQt6.QtCore import pyqtSignal
 from typing import Any
 from .instrument import Instrument
+import logging
 
 
 class ProbeInstrument(Instrument):
@@ -11,15 +12,19 @@ def __init__(self, config: dict[str, Any]):
         self._offset_pos: tuple[float, float] | None = None
         if "offset_pos" in config:
             self._offset_pos = tuple(config["offset_pos"])
+        spot_size = config.get("spot_size_um", config.get("spot_size"))
+        self._spot_size_um = float(spot_size) if spot_size is not None else 10.0
 
     # Signal emited when fixed pos parameter changed
     offset_pos_changed = pyqtSignal()
+    spot_size_changed = pyqtSignal()
 
     @property
     def settings(self) -> dict[str, Any]:
         data = super().settings
         if self.offset_pos is not None:
             data["offset_pos"] = list(self.offset_pos)
+        data["spot_size_um"] = self.spot_size_um
         return data
 
     @settings.setter
@@ -29,6 +34,10 @@ def settings(self, data: dict[str, Any]):
         Instrument.settings.fset(self, data)
         offset_pos = data.get("offset_pos", None)
         self.offset_pos = tuple(offset_pos) if offset_pos is not None else None
+        if "spot_size_um" in data:
+            self.spot_size_um = data["spot_size_um"]
+        elif "spot_size" in data:
+            self.spot_size_um = data["spot_size"]
 
     @property
     def offset_pos(self) -> tuple[float, float] | None:
@@ -37,4 +46,17 @@ def offset_pos(self) -> tuple[float, float] | None:
     @offset_pos.setter
     def offset_pos(self, offset_pos: tuple[float, float] | None):
         self._offset_pos = offset_pos
+        logging.getLogger("laserstudio").debug(f"Offset pos changed to {offset_pos}")
         self.offset_pos_changed.emit()
+
+    @property
+    def spot_size_um(self) -> float:
+        return self._spot_size_um
+
+    @spot_size_um.setter
+    def spot_size_um(self, value: float):
+        if value <= 0:
+            raise ValueError("Spot size must be positive")
+        self._spot_size_um = value
+        logging.getLogger("laserstudio").debug(f"Spot size changed to {value}")
+        self.spot_size_changed.emit()