feat(stats_signal): Test device to get sigma_x and y and eccentricity from stats module

2026-07-27 17:06:46 +02:00
parent acc9df84ff
commit f520456a39
1 changed files with 62 additions and 0 deletions
@@ -0,0 +1,62 @@
+from ophyd_devices import AsyncMultiSignal
+
+from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
+from ophyd import Component as Cpt
+from ophyd import EpicsSignalRO, Kind
+from collections import defaultdict
+from threading import RLock
+import time
+
+
+class StatsSignal(PSIDeviceBase, AsyncMultiSignal):
+
+    data = Cpt(
+        AsyncMultiSignal,
+        name="data",
+        signals=["sigma_x", "sigma_y", "eccentricity"],
+        ndim=1,
+        async_update={"type": "add", "max_shape": [None]},
+        max_size=1000,
+        kind=Kind.normal,
+        doc="Stats data signal containing sigma_x, sigma_y, and eccentricity values",
+    )
+
+    sigma_x_raw = Cpt(EpicsSignalRO, "Stats1:SigmaX_RBV", kind=Kind.omitted)
+    sigma_y_raw = Cpt(EpicsSignalRO, "Stats1:SigmaY_RBV", kind=Kind.omitted)
+    eccentricity_raw = Cpt(EpicsSignalRO, "Stats1:Eccentricity_RBV", kind=Kind.omitted)
+
+    def on_init(self):
+        self._rlock = RLock()
+        self._cached_average_data = defaultdict(float)
+
+    def on_connected(self):
+        self._cached_average_data.clear()
+        for sig in (self.sigma_x_raw, self.sigma_y_raw, self.eccentricity_raw):
+            sig.subscribe(self._update_average_data, run=False)
+
+    def _update_average_data(self, value, obj, **kwargs):
+        if value is None:
+            return
+        sig_name = obj.name
+
+        with self._rlock:
+            self._cached_average_data[sig_name] += value
+
+    def on_trigger(self):
+        with self._rlock:
+            self._cached_average_data.clear()
+        time.sleep(
+            self.scan_info.msg.scan_parameters.get("exp_time")
+        )  # Wait for the exposure time to finish
+
+        with self._rlock:
+            sigma_x = self._cached_average_data.get("sigma_x_raw", 0.0)
+            sigma_y = self._cached_average_data.get("sigma_y_raw", 0.0)
+            eccentricity = self._cached_average_data.get("eccentricity_raw", 0.0)
+
+        ret = {
+            "sigma_x": {"value": sigma_x},
+            "sigma_y": {"value": sigma_y},
+            "eccentricity": {"value": eccentricity},
+        }
+        self.data.put(ret)