diff --git a/mx_bec/devices/beam_profile.py b/mx_bec/devices/beam_profile.py index ead1ae8..2157376 100644 --- a/mx_bec/devices/beam_profile.py +++ b/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) diff --git a/mx_bec/devices/beam_steering.py b/mx_bec/devices/beam_steering.py index 5eb1aa7..f154a29 100644 --- a/mx_bec/devices/beam_steering.py +++ b/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'"""