Add unittest for VectorView

extra_foam/special_suite/tests/test_special_analysis_base.py

@@ -292,6 +292,22 @@ def testRoiCtrl(self):
         pass
 
     def testSqueezeCameraImage(self):
+        a1d = np.ones((4, ))
+        a2d = np.ones((2, 1))
+        a3d = np.ones((3, 3, 1))
+
+        func = functools.partial(self._win._worker_st.squeezeToVector, 1234)
+
+        assert func(None) is None
+        assert func(a3d) is None
+
+        ret_1d = func(a1d)
+        np.testing.assert_array_equal(a1d, ret_1d)
+
+        ret_2d = func(a2d)
+        np.testing.assert_array_equal(a2d.squeeze(axis=-1), ret_2d)
+
+    def testSqueezeToVector(self):
         a1d = np.ones((4, ))
         a2d = np.ones((2, 2))
         a3d = np.ones((3, 3, 1))

extra_foam/special_suite/tests/test_vectorview.py

@@ -0,0 +1,108 @@
+from collections import Counter
+
+import pytest
+import numpy as np
+
+from extra_foam.pipeline.tests import _TestDataMixin
+
+from extra_foam.special_suite import logger, mkQApp
+from extra_foam.pipeline.exceptions import ProcessingError
+from extra_foam.special_suite.vector_view import (
+    VectorViewProcessor, VectorViewWindow, VectorPlot, VectorCorrelationPlot,
+    InTrainVectorCorrelationPlot
+)
+
+from . import _SpecialSuiteWindowTestBase, _SpecialSuiteProcessorTestBase
+
+
+app = mkQApp()
+
+logger.setLevel('INFO')
+
+
+class TestCamViewWindow(_SpecialSuiteWindowTestBase):
+    @classmethod
+    def setUpClass(cls):
+        cls._win = VectorViewWindow('SCS')
+
+    @classmethod
+    def tearDownClass(cls):
+        # explicitly close the MainGUI to avoid error in GuiLogger
+        cls._win.close()
+
+    @staticmethod
+    def data4visualization():
+        """Override."""
+        return {
+            "vector1": np.arange(10),
+            "vector2": np.arange(10) + 5,
+            "vector1_full": np.arange(100),
+            "vector2_full": np.arange(100) + 5,
+        }
+
+    def testWindow(self):
+        win = self._win
+
+        self.assertEqual(3, len(win._plot_widgets_st))
+        counter = Counter()
+        for key in win._plot_widgets_st:
+            counter[key.__class__] += 1
+
+        self.assertEqual(1, counter[VectorPlot])
+        self.assertEqual(1, counter[InTrainVectorCorrelationPlot])
+        self.assertEqual(1, counter[VectorCorrelationPlot])
+
+        self._check_update_plots()
+
+    def testCtrl(self):
+        win = self._win
+        ctrl_widget = win._ctrl_widget_st
+        proc = win._worker_st
+
+        # test default values
+        self.assertEqual('XGM intensity', proc._vector1)
+        self.assertEqual('', proc._vector2)
+
+        # test set new values
+        widget = ctrl_widget.vector1_cb
+        widget.setCurrentText("ROI FOM")
+        self.assertEqual("ROI FOM", proc._vector1)
+
+        widget = ctrl_widget.vector2_cb
+        widget.setCurrentText("Digitizer pulse integral")
+        self.assertEqual("Digitizer pulse integral", proc._vector2)
+
+
+class TestVectorViewProcessor(_TestDataMixin, _SpecialSuiteProcessorTestBase):
+    @pytest.fixture(autouse=True)
+    def setUp(self):
+        self._proc = VectorViewProcessor(object(), object())
+        self._proc._vector1 = "XGM intensity"
+
+    def testProcessing(self):
+        data, processed = self.simple_data(1001, (4, 2, 2))
+        proc = self._proc
+
+        with pytest.raises(ProcessingError, match="XGM intensity is not available"):
+            proc.process(data)
+
+        processed.pulse.xgm.intensity = np.random.rand(4)
+        ret = proc.process(data)
+        self._check_processed_data_structure(ret)
+
+        self._proc._vector2 = "ROI FOM"
+        processed.pulse.roi.fom = np.random.rand(5)
+        with pytest.raises(ProcessingError, match="Vectors have different lengths"):
+            proc.process(data)
+        processed.pulse.roi.fom = np.random.rand(4)
+        proc.process(data)
+
+        self._proc._vector2 = "Digitizer pulse integral"
+        processed.pulse.digitizer.ch_normalizer = 'B'
+        processed.pulse.digitizer['B'].pulse_integral = np.random.rand(4)
+        proc.process(data)
+
+    def _check_processed_data_structure(self, ret):
+        """Override."""
+        data_gt = TestCamViewWindow.data4visualization().keys()
+        assert set(ret.keys()) == set(data_gt)

extra_foam/special_suite/vector_view.py

@@ -41,8 +41,6 @@ def __init__(self, *args, **kwargs):
         self._vector1 = ''
         self._vector2 = ''
 
-        self._digitizer_ch = ''
-
         self._vector1_full = SimpleSequence(max_len=6000)
         self._vector2_full = SimpleSequence(max_len=6000)
 
@@ -52,17 +50,12 @@ def onVector1Change(self, value: str):
     def onVector2Change(self, value: str):
         self._vector2 = value
 
-    def onDigitizerChannelChange(self, value: str):
-        self._digitizer_ch = value
-
     @profiler("Vector view processor")
     def process(self, data):
         """Override."""
-        meta = data["meta"]
-
-        tid = self.getTrainId(meta)
+        processed = data["processed"]
 
-        vec1, vec2 = self._fetch_data(data["processed"])
+        vec1, vec2 = self._fetch_data(processed)
 
         if vec1 is not None and vec2 is not None:
             if len(vec1) != len(vec2):
@@ -74,7 +67,7 @@ def process(self, data):
             self._vector1_full.extend(vec1)
             self._vector2_full.extend(vec2)
 
-        self.log.info(f"Train {tid} processed")
+        self.log.info(f"Train {processed.tid} processed")
 
         return {
             "vector1": vec1,
@@ -90,17 +83,20 @@ def _fetch_data(self, processed):
             if name == 'ROI FOM':
                 vec = processed.pulse.roi.fom
                 if vec is None:
-                    raise ProcessingError("ROI FOM is not available!")
+                    raise ProcessingError(
+                        "Pulse-resolved ROI FOM is not available!")
 
             elif name == 'XGM intensity':
                 vec = processed.pulse.xgm.intensity
                 if vec is None:
                     raise ProcessingError("XGM intensity is not available!")
 
             elif name == 'Digitizer pulse integral':
-                vec = processed.pulse.digitizer[self._digitizer_ch].pulse_integral
+                digit = processed.pulse.digitizer
+                vec = digit[digit.ch_normalizer].pulse_integral
                 if vec is None:
-                    raise ProcessingError("Digitizer APD is not available!")
+                    raise ProcessingError(
+                        "Digitizer pulse integral is not available!")
 
             ret.append(vec)
         return ret
@@ -119,20 +115,17 @@ def __init__(self, *args, **kwargs):
 
         self.vector1_cb = QComboBox()
         for item in _PREDEFINED_VECTORS:
-            self.vector1_cb.addItem(item)
+            if item:
+                # vector1 cannot be empty
+                self.vector1_cb.addItem(item)
 
         self.vector2_cb = QComboBox()
         for item in _PREDEFINED_VECTORS:
             self.vector2_cb.addItem(item)
 
-        self.channel_cb = QComboBox()
-        for item in _DIGITIZER_CHANNELS:
-            self.channel_cb.addItem(item)
-
         self._non_reconfigurable_widgets = [
             self.vector1_cb,
             self.vector2_cb,
-            self.channel_cb,
         ]
 
         self.initUI()
@@ -144,7 +137,6 @@ def initUI(self):
 
         layout.addRow("Vector 1: ", self.vector1_cb)
         layout.addRow("Vector 2: ", self.vector2_cb)
-        layout.addRow("Digitizer channel: ", self.channel_cb)
 
     def initConnections(self):
         """Override."""
@@ -276,8 +268,3 @@ def initConnections(self):
             self._worker_st.onVector2Change)
         self._ctrl_widget_st.vector2_cb.currentTextChanged.emit(
             self._ctrl_widget_st.vector2_cb.currentText())
-
-        self._ctrl_widget_st.channel_cb.currentTextChanged.connect(
-            self._worker_st.onDigitizerChannelChange)
-        self._ctrl_widget_st.channel_cb.currentTextChanged.emit(
-            self._ctrl_widget_st.channel_cb.currentText())