From 003abc11bc3c67d97c6b5b982cea3bcc5c6ce575 Mon Sep 17 00:00:00 2001 From: x06da Date: Wed, 19 Aug 2026 11:10:37 +0200 Subject: [PATCH] feat: add options for beam profile provider --- mx_bec/devices/beam_profile.py | 66 +++++++++++++++++++++++++++++++-- mx_bec/devices/beam_steering.py | 7 +++- 2 files changed, 67 insertions(+), 6 deletions(-) diff --git a/mx_bec/devices/beam_profile.py b/mx_bec/devices/beam_profile.py index 05dabcc..2157376 100644 --- a/mx_bec/devices/beam_profile.py +++ b/mx_bec/devices/beam_profile.py @@ -3,6 +3,7 @@ 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 @@ -18,14 +19,23 @@ class BeamProfile(PSIDeviceBase, ABC): y_sig_px = Cpt[EpicsSignalRO] @abstractmethod - def enable_computation(self): ... + 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 disable_computation(self): ... + 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 @@ -39,6 +49,8 @@ class GaussianBeamProfile(BeamProfile): 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. @@ -56,4 +68,50 @@ class GaussianBeamProfile(BeamProfile): return st1 and st2 and st3 -class ThresholdBeamProfile(BeamProfile): ... +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 68cb600..f154a29 100644 --- a/mx_bec/devices/beam_steering.py +++ b/mx_bec/devices/beam_steering.py @@ -9,7 +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 MoveStatus, Status +from ophyd_devices.utils.psi_device_base_utils import Status from mx_bec.devices.beam_profile import BeamProfile @@ -44,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, "