[fix] fixed _checkcancelled and merged unittests goffset + goffset_alignment

Author: yxkdejong

src/odemis/acq/align/goffset.py

@@ -91,7 +91,7 @@ def find_peak_position(data: numpy.ndarray, window_radius: int = 15) -> float:
 
 def estimate_goffset_scale(spgr: model.Actuator, detector: model.Detector, delta=5.0, retries = 1) -> Tuple[float, float, float]:
     """
-    Estimate the scale factor between a change in the grating offset ('goffset') 
+    Estimate the scale factor between a change in the grating offset ('goffset')
     and the resulting shift of the spectral peak on the detector.
 
     The function moves the actuator by a small step (delta) and measures the peak position before and after the move.
@@ -106,7 +106,6 @@ def estimate_goffset_scale(spgr: model.Actuator, detector: model.Detector, delta
     :param delta: The relative goffset step size to apply when measuring the scale (default: 5.0).
                   The actual step may be negated to avoid exceeding hardware limits.
     :param retries: number of retries allowed if the estimated scale is unreliable (default: 1).
-                  
     :return: Tuple (scale, p0, p1)
          scale: estimated pixels per unit of goffset
          p0: peak position at the initial goffset
@@ -226,7 +225,7 @@ def _do_sparc_auto_grating_offset(future: model.ProgressiveFuture,
         max_step = 0.1 * (maxv - minv)  # max 10% of range
 
         for i in range(max_it):
-            _checkCancelled()
+            _checkCancelled(future)
 
             if i == 0:
                 peak_px = p0
@@ -253,7 +252,8 @@ def _do_sparc_auto_grating_offset(future: model.ProgressiveFuture,
             logging.debug(
                 "DEBUG | Iter: %d | Peak: %.1f | Error: %.1f | Move: %.4f | Total Change: %.4f",
                 i, peak_px, error_px, delta_goffset, total_goffset_displacement
-            )            spgr.moveRelSync({"goffset": delta_goffset})
+            )
+            spgr.moveRelSync({"goffset": delta_goffset})
 
             future.set_progress(
                 end=time.time() + (max_it - i - 1) * 0.5)  # update estimated end time
@@ -464,4 +464,4 @@ def _cancel_auto_align_grating_detector_offsets(future: model.ProgressiveFuture)
         future._subfuture.cancel()
         logging.debug("Auto-alignment cancellation requested")
 
-    return True
+    return True

src/odemis/acq/align/goffset_alignment.py

@@ -197,4 +197,4 @@ def _cancel_auto_align_grating_detector_offsets(future: model.ProgressiveFuture)
         future._subfuture.cancel()
         logging.debug("Auto-alignment cancellation requested")
 
-    return True
+    return True

src/odemis/acq/align/test/goffset_alignment_test.py

@@ -114,12 +114,11 @@ def test_multi_detector_iteration(self):
         # move to spectral camera
         self.selector.moveAbsSync({"rx": 1.5707963267948966})
         data = spccd.data.get(asap=False)
-
         # check data is not flat
         if data.max() == data.min():
             print("WARNING: sp-ccd is returning a flat image!")
         else:
             print(f"sp-ccd signal range: {data.min()} to {data.max()}")
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()

src/odemis/acq/align/test/goffset_test.py

@@ -8,17 +8,16 @@
 import unittest
 import logging
 import numpy as np
+import odemis
 
 from concurrent.futures._base import CancelledError
-
 from odemis import model
 from odemis.util import testing, timeout
 
 from odemis.acq.align.goffset import (find_peak_position,
     estimate_goffset_scale,
-    sparc_auto_grating_offset)
-
-import odemis
+    sparc_auto_grating_offset,
+    auto_align_grating_detector_offsets)
 
 logging.getLogger().setLevel(logging.DEBUG)
 
@@ -59,7 +58,7 @@ def test_find_peak_position_synthetic(self):
         """
         x = np.arange(200)
         true_center = 83.4
-        spectrum = np.exp(-0.5*((x-true_center)/3.0)**2)
+        spectrum = np.exp(-0.5 * ((x - true_center) / 3.0)**2)
 
         peak = find_peak_position(spectrum)
         self.assertAlmostEqual(peak, true_center, places=1)
@@ -70,7 +69,7 @@ def test_find_peak_position_2d(self):
         """
         x = np.arange(200)
         true_center = 120.0
-        line = np.exp(-0.5*((x-true_center)/4.0)**2)
+        line = np.exp(-0.5 * ((x - true_center) / 4.0)**2)
         image = np.tile(line, (50, 1))
 
         peak = find_peak_position(image)
@@ -115,6 +114,94 @@ def test_cancel(self):
         except:
             pass
         self.assertTrue(f.done())
+        
+        
+
+class TestAutoAlignGratingDetectorOffsets(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.spgr = model.getComponent(role="spectrograph")
+        cls.ccd = model.getComponent(role="ccd")
+        #cls.spccd = model.getComponent(role="sp-ccd")
+        cls.selector = model.getComponent(role="spec-det-selector")
+
+    def setUp(self):
+        # Speed up acquisition
+        self.ccd.exposureTime.value = self.ccd.exposureTime.range[0]
+
+    @timeout(1000)
+    def test_single_detector_iteration(self):
+        f = auto_align_grating_detector_offsets(spectrograph=self.spgr, detectors=[self.ccd], selector=self.selector)
+        res = f.result(timeout=900)
+
+        n_gratings = len(self.spgr.axes["grating"].choices)
+        n_detectors = 1
+        expected = n_detectors + (n_gratings - 1)
+
+        self.assertEqual(len(res), expected)
+
+        first_grating = list(self.spgr.axes["grating"].choices.keys())[0]
+        dets_first = [d for (g, d) in res.keys() if g == first_grating]
+
+        self.assertEqual(len(dets_first), n_detectors)
+
+    def test_multi_detector_iteration(self):
+        spccd = model.getComponent(role="sp-ccd")
+        spccd.exposureTime.value = spccd.exposureTime.range[0]
+
+        detectors = [self.ccd, spccd]
+
+        # run alignment
+        f = auto_align_grating_detector_offsets(spectrograph=self.spgr, detectors=detectors, selector=self.selector)
+        res = f.result(timeout=900)
+
+        # calculate expected results
+        n_gratings = len(self.spgr.axes["grating"].choices)
+        n_detectors = len(detectors)
+        expected_count = n_detectors + (n_gratings - 1)
+
+        self.assertEqual(len(res), expected_count, f"Expected {expected_count} results, got {len(res)}")
+
+        # verify that every detector was used for the first grating
+        gratings_list = list(self.spgr.axes["grating"].choices.keys())
+        first_grating = gratings_list[0]
+
+        dets_for_first_grating = [d for (g, d) in res.keys() if g == first_grating]
+        self.assertEqual(len(dets_for_first_grating), n_detectors)
+        self.assertIn(self.ccd.name, dets_for_first_grating)
+        self.assertIn(spccd.name, dets_for_first_grating)
+
+        # verify that only first detector is used for remaining gratings
+        for g in gratings_list[1:]:
+            # Ensure only the first detector (index 0) is present for these gratings
+            dets_for_this_grating = [d for (grating, d) in res.keys() if grating == g]
+            self.assertEqual(len(dets_for_this_grating), 1)
+            self.assertEqual(dets_for_this_grating[0], detectors[0].name)
+
+        # move to spectral camera
+        self.selector.moveAbsSync({"rx": 1.5707963267948966})
+        data = spccd.data.get(asap=False)
+
+        # check data is not flat
+        if data.max() == data.min():
+            print("WARNING: sp-ccd is returning a flat image!")
+        else:
+            print(f"sp-ccd signal range: {data.min()} to {data.max()}")
+
+    @timeout(100)
+    def test_cancel(self):
+        f = auto_align_grating_detector_offsets(spectrograph=self.spgr, detectors=[self.ccd],)
+
+        time.sleep(1)
+
+        cancelled = f.cancel()
+        self.assertTrue(cancelled)
+        self.assertTrue(f.cancelled())
+
+        with self.assertRaises(CancelledError):
+            f.result(timeout=900)
+
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()