[fix] fixed _checkcancelled and merged unittests goffset + goffset_alignment

Author: yxkdejong

src/odemis/acq/align/goffset.py

@@ -226,7 +226,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

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()