feat: add histo beam profile option

mx_bec/devices/beam_profile.py

@@ -1,15 +1,41 @@
 """A device for determining beam shape and location from sample camera image analysis."""
 
+from abc import ABC, abstractmethod
 from typing import Literal
 
+import numpy as np
 from ophyd import Component as Cpt
 from ophyd import EpicsSignal, EpicsSignalRO, Kind
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
 
 
-class BeamProfile(PSIDeviceBase):
+class BeamProfile(PSIDeviceBase, ABC):
+    """Common interface for different methods of getting beam location and size"""
+
+    # Readback
+    x_pos_px = Cpt[EpicsSignalRO]
+    y_pos_px = Cpt[EpicsSignalRO]
+    x_sig_px = Cpt[EpicsSignalRO]
+    y_sig_px = Cpt[EpicsSignalRO]
+
+    @abstractmethod
+    def prepare_plugin(self):
+        """Set parameters, etc., which only need to be set up once before using
+        the profile results (e.g. threshold value...)"""
+
+    @abstractmethod
+    def enable_computation(self):
+        """Ensure plugins or whatever else is needed is configured to provide
+        analysis results"""
+
+    @abstractmethod
+    def disable_computation(self):
+        """Save resources by turning computations off again"""
+
+
+class GaussianBeamProfile(BeamProfile):
     """Use image analysis of the scintillator to determine the beam centre and width on the sample
-    camera image and convert it to physical units, by fitting a Gaussian to the image profile.
+    camera image, by fitting a Gaussian to the image profile.
     The analysis is provided by AD plugins in EPICS, we merely configure it and read the results."""
 
     # Config
@@ -23,6 +49,8 @@ class BeamProfile(PSIDeviceBase):
     x_sig_px = Cpt(EpicsSignalRO, name="x_sig_px", suffix="X:Sigma_RBV", kind=Kind.normal)
     y_sig_px = Cpt(EpicsSignalRO, name="x_sig_px", suffix="Y:Sigma_RBV", kind=Kind.normal)
 
+    def prepare_plugin(self): ...
+
     def enable_computation(self):
         """Ensure all the configuration parameters are set."""
         # TODO: add making sure the correct plugins and callbacks are wired together.
@@ -38,3 +66,52 @@ class BeamProfile(PSIDeviceBase):
         st2 = self.x_compute.set(enabled)
         st3 = self.y_compute.set(enabled)
         return st1 and st2 and st3
+
+
+class CentroidBeamProfile(BeamProfile):
+    """Use image analysis of the scintillator to determine the beam centre and width on the sample
+    camera image, by finding the pixels over a threshold in a histogram of each direction.
+    The analysis is provided by AD plugins in EPICS, we merely configure it and read the results."""
+
+    # Config
+    histo_enabled = Cpt(
+        EpicsSignal, name="histo_enabled", suffix="ComputeHistogram", kind=Kind.config
+    )
+    centroid_enabled = Cpt(
+        EpicsSignal, name="centroid_enabled", suffix="ComputeCentroid", kind=Kind.config
+    )
+    centroid_thresh = Cpt(
+        EpicsSignal, name="centroid_thresh", suffix="CentroidThreshold", kind=Kind.config
+    )
+
+    # Data for setup
+    histo_arr = Cpt(EpicsSignal, name="histo_arr", suffix="Histogram_RBV", kind=Kind.omitted)
+
+    def prepare_plugin(self):
+        self._update_threshold()
+
+    def enable_computation(self):
+        """Ensure all the configuration parameters are set."""
+        # TODO: add making sure the correct plugins and callbacks are wired together.
+        # TODO: make sure we are at max zoom
+        self._set_computation(1).wait()
+
+    def disable_computation(self):
+        """Ensure all the image analysis computations are disabled."""
+        self._set_computation(0).wait()
+
+    def _update_threshold(self):
+        self.histo_enabled.set(1).wait()
+        histogram: np.typing.NDArray[np.int64] = self.histo_arr.get()  # type: ignore
+        cs = np.cumsum(histogram)
+        cs_thresh = np.sum(histogram) * 0.97
+        exceeds_thresh_indices = np.where(cs > cs_thresh)
+        self.centroid_thresh.set(exceeds_thresh_indices[0][0]).wait()
+
+    def _set_computation(self, enabled: Literal[0, 1]):
+        return self.centroid_enabled.set(enabled)
+
+    x_pos_px = Cpt(EpicsSignalRO, name="x_pos_px", suffix="CentroidX_RBV", kind=Kind.normal)
+    y_pos_px = Cpt(EpicsSignalRO, name="y_pos_px", suffix="CentroidY_RBV", kind=Kind.normal)
+    x_sig_px = Cpt(EpicsSignalRO, name="x_sig_px", suffix="SigmaX_RBV", kind=Kind.normal)
+    y_sig_px = Cpt(EpicsSignalRO, name="x_sig_px", suffix="SigmaY_RBV", kind=Kind.normal)

mx_bec/devices/beam_steering.py

@@ -9,6 +9,7 @@ from ophyd import Component as Cpt
 from ophyd import Kind, Signal
 from ophyd_devices import EpicsMotorEC
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
+from ophyd_devices.utils.psi_device_base_utils import Status
 
 from mx_bec.devices.beam_profile import BeamProfile
 
@@ -43,7 +44,10 @@ class BeamSteerer(PSIDeviceBase):
 
         hfm_motor = cast(EpicsMotorEC | None, self.device_manager.devices.get("hfm_yr"))
         vfm_motor = cast(EpicsMotorEC | None, self.device_manager.devices.get("vfm_yw"))
-        bp = cast(BeamProfile | None, self.device_manager.devices.get("beam_profile"))
+        profile_device = self.device_manager.devices.get("beam_steering").user_parameter.get(
+            "profile_provider"
+        )
+        bp = cast(BeamProfile | None, self.device_manager.devices.get(profile_device))
         if hfm_motor is None or vfm_motor is None or bp is None:
             raise RuntimeError(
                 "This device needs the focussing mirrors and the beam profile, "
@@ -89,7 +93,7 @@ class BeamSteerer(PSIDeviceBase):
             return x_status
         if y_status is not None:
             return y_status
-        return None
+        return Status(done=True)
 
     def trigger(self):
         """External interface for 'step_towards_centre'"""