[S30-130][fix] Store e-beam current and voltage as metadata

src/odemis/gui/cont/stream.py

@@ -281,11 +281,14 @@ def _display_metadata(self):
         elif model.MD_DWELL_TIME in md:
             self.add_metadata(model.MD_DWELL_TIME, md[model.MD_DWELL_TIME], 's')
 
-        if model.MD_EBEAM_VOLTAGE in md:
-            self.add_metadata("Acceleration voltage", md[model.MD_EBEAM_VOLTAGE], 'V')
-
-        if model.MD_EBEAM_CURRENT in md:
-            self.add_metadata("Emission current", md[model.MD_EBEAM_CURRENT], 'A')
+        # On the SPARC, the CL streams (eg, spectrum) also have the beam voltage and current metadata,
+        # but it's not helpful to show it, as it's always the same as the associated SEM streams
+        if self.stream.acquisitionType.value in (model.MD_AT_EM, model.MD_AT_FIB):
+            if model.MD_BEAM_VOLTAGE in md:
+                self.add_metadata("Acceleration voltage", md[model.MD_BEAM_VOLTAGE], 'V')
+
+            if model.MD_BEAM_CURRENT in md:
+                self.add_metadata("Emission current", md[model.MD_BEAM_CURRENT], 'A')
 
     def pause(self):
         """ Pause (freeze) SettingEntry related control updates """

src/odemis/odemisd/mdupdater.py

@@ -590,21 +590,45 @@ def updateTriggerRate(rate, comp_affected=comp_affected):
 
         return True
 
-    def observeEbeam(self, ebeam, comp_affected):
-        """Add ebeam rotation to multibeam metadata to make sure that the thumbnails
-        are displayed correctly."""
+    def observeEbeam(self, ebeam: model.HwComponent, comp_affected: model.HwComponent) -> bool:
+        """
+        Update metadata of components affected by the e-beam when its parameters change.
 
-        if comp_affected.role != "multibeam":
-            return False
+        Propagates probeCurrent and accelVoltage to all affected components.
+        Additionally, for multibeam components, propagates the beam rotation.
 
-        def updateRotation(rot, comp_affected=comp_affected):
-            md = {model.MD_ROTATION: rot}
-            comp_affected.updateMetadata(md)
+        :param ebeam: the e-beam scanner component
+        :param comp_affected: a component affected by the e-beam
+        :return: True if at least one VA is observed, False otherwise
+        """
+        observed = False
 
-        ebeam.rotation.subscribe(updateRotation, init=True)
-        self._onTerminate.append((ebeam.rotation.unsubscribe, (updateRotation,)))
+        # Map of VA name -> metadata key for VAs that are directly copied
+        va_md_map = {
+            "probeCurrent": model.MD_BEAM_CURRENT,
+            "accelVoltage": model.MD_BEAM_VOLTAGE,
+        }
 
-        return True
+        for va_name, md_key in va_md_map.items():
+            if model.hasVA(ebeam, va_name):
+                def updateBeamParam(val, md_key=md_key, comp_affected=comp_affected):
+                    comp_affected.updateMetadata({md_key: val})
+
+                va = getattr(ebeam, va_name)
+                va.subscribe(updateBeamParam, init=True)
+                self._onTerminate.append((va.unsubscribe, (updateBeamParam,)))
+                observed = True
+
+        if comp_affected.role == "multibeam":
+            def updateRotation(rot, comp_affected=comp_affected):
+                md = {model.MD_ROTATION: rot}
+                comp_affected.updateMetadata(md)
+
+            ebeam.rotation.subscribe(updateRotation, init=True)
+            self._onTerminate.append((ebeam.rotation.unsubscribe, (updateRotation,)))
+            observed = True
+
+        return observed
 
     def terminate(self):
         self._mic.alive.unsubscribe(self._onAlive)

src/odemis/odemisd/test/mdupdater_test.py

@@ -25,7 +25,7 @@
 import numpy
 
 from odemis import model
-from odemis.driver import static
+from odemis.driver import simsem, static
 from odemis.model import Microscope
 from odemis.odemisd.mdupdater import MetadataUpdater
 from odemis.util import mock
@@ -88,5 +88,89 @@ def fake_get_component(name):
             mdup.terminate()
 
 
+class EbeamMDUpdaterTest(unittest.TestCase):
+    """
+    Tests for e-beam metadata propagation via MetadataUpdater, using mock components.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        # Create a minimal sparc2 microscope
+        cls.mic = Microscope("Fake SPARC2", "sparc2")
+
+        # Create a SimSEM with an e-beam scanner and an se-detector child
+        cls.sem = simsem.SimSEM(
+            "SEM",
+            "sem",
+            children={
+                "scanner": {"name": "e-beam scanner", "role": "e-beam"},
+                "detector0": {"name": "se-detector", "role": "se-detector"},
+            }
+        )
+        cls.ebeam = next(c for c in cls.sem.children.value if c.role == "e-beam")
+
+        # Create a fake CCD that will be affected by the e-beam
+        img = model.DataArray(numpy.empty((512, 768), dtype=numpy.uint16))
+        cls.ccd = mock.FakeCCD(img)
+
+        cls.ebeam.affects.value = [cls.ccd.name, "se-detector"]
+
+        # Mock model.getComponent() so MetadataUpdater can resolve component names
+        all_comps = list(cls.sem.children.value) + [cls.sem, cls.ccd]
+
+        def fake_get_component(name):
+            for c in all_comps:
+                if c.name == name:
+                    return c
+            raise LookupError(f"no component {name}")
+
+        cls._patch_get_component = unittest.mock.patch.object(model, "getComponent", fake_get_component)
+        cls._patch_get_component.start()
+
+        cls.mdup = MetadataUpdater("MDUpdater", cls.mic)
+        cls.mic.alive.value = set(cls.sem.children.value) | {cls.ccd}
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.mdup.terminate()
+        cls.sem.terminate()
+        cls._patch_get_component.stop()
+
+    def test_ebeam_beam_params_on_detector(self):
+        """
+        Verify that when probeCurrent or accelVoltage change on the e-beam,
+        MD_BEAM_CURRENT and MD_BEAM_VOLTAGE are updated on the affected detectors.
+        """
+        # Both VAs must be present on the simulated e-beam
+        self.assertTrue(model.hasVA(self.ebeam, "probeCurrent"),
+                        "e-beam has no probeCurrent VA")
+        self.assertTrue(model.hasVA(self.ebeam, "accelVoltage"),
+                        "e-beam has no accelVoltage VA")
+
+        # After setup, the metadata should already be populated (init=True on subscribe)
+        md = self.ccd.getMetadata()
+        self.assertIn(model.MD_BEAM_CURRENT, md,
+                      "MD_BEAM_CURRENT not set on CCD after startup")
+        self.assertIn(model.MD_BEAM_VOLTAGE, md,
+                      "MD_BEAM_VOLTAGE not set on CCD after startup")
+        self.assertAlmostEqual(md[model.MD_BEAM_CURRENT], self.ebeam.probeCurrent.value)
+        self.assertAlmostEqual(md[model.MD_BEAM_VOLTAGE], self.ebeam.accelVoltage.value)
+
+        # Change probeCurrent => MD_BEAM_CURRENT must be updated
+        new_current = self.ebeam.probeCurrent.choices - {self.ebeam.probeCurrent.value}
+        new_current = next(iter(new_current))  # pick any value different from current
+        self.ebeam.probeCurrent.value = new_current
+        md = self.ccd.getMetadata()
+        self.assertAlmostEqual(md[model.MD_BEAM_CURRENT], new_current)
+
+        # Change accelVoltage => MD_BEAM_VOLTAGE must be updated
+        orig_voltage = self.ebeam.accelVoltage.value
+        voltage_range = self.ebeam.accelVoltage.range
+        new_voltage = voltage_range[1] if orig_voltage == voltage_range[0] else voltage_range[0]
+        self.ebeam.accelVoltage.value = new_voltage
+        md = self.ccd.getMetadata()
+        self.assertAlmostEqual(md[model.MD_BEAM_VOLTAGE], new_voltage)
+
+
 if __name__ == "__main__":
     unittest.main()