From 3ae911bb83eb78c6e9463cc431d48166ba7b3d20 Mon Sep 17 00:00:00 2001 From: x06da Date: Thu, 13 Aug 2026 14:48:09 +0200 Subject: [PATCH] feat: beam steering devices --- mx_bec/devices/beam_profile.py | 40 ++++++++++++++ mx_bec/devices/beam_steering.py | 96 +++++++++++++++++++++++++++++++++ 2 files changed, 136 insertions(+) create mode 100644 mx_bec/devices/beam_profile.py create mode 100644 mx_bec/devices/beam_steering.py diff --git a/mx_bec/devices/beam_profile.py b/mx_bec/devices/beam_profile.py new file mode 100644 index 0000000..ead1ae8 --- /dev/null +++ b/mx_bec/devices/beam_profile.py @@ -0,0 +1,40 @@ +"""A device for determining beam shape and location from sample camera image analysis.""" + +from typing import Literal + +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): + """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. + The analysis is provided by AD plugins in EPICS, we merely configure it and read the results.""" + + # Config + cbs_enabled = Cpt(EpicsSignal, name="cbs_enabled", suffix="EnableCallbacks", kind=Kind.config) + x_compute = Cpt(EpicsSignal, name="x_compute", suffix="X:Compute", kind=Kind.config) + y_compute = Cpt(EpicsSignal, name="y_compute", suffix="X:Compute", kind=Kind.config) + + # Readback + x_pos_px = Cpt(EpicsSignalRO, name="x_pos_px", suffix="X:Mu_RBV", kind=Kind.normal) + y_pos_px = Cpt(EpicsSignalRO, name="y_pos_px", suffix="Y:Mu_RBV", kind=Kind.normal) + 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 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 _set_computation(self, enabled: Literal[0, 1]): + st1 = self.cbs_enabled.set(enabled) + st2 = self.x_compute.set(enabled) + st3 = self.y_compute.set(enabled) + return st1 and st2 and st3 diff --git a/mx_bec/devices/beam_steering.py b/mx_bec/devices/beam_steering.py new file mode 100644 index 0000000..5eb1aa7 --- /dev/null +++ b/mx_bec/devices/beam_steering.py @@ -0,0 +1,96 @@ +"""A device for stepping the beam towards the centre""" + +from __future__ import annotations + +from typing import Literal, cast + +from bec_lib.logger import bec_logger +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 mx_bec.devices.beam_profile import BeamProfile + +logger = bec_logger.logger + + +class BeamSteerer(PSIDeviceBase): + """Based on the beam profile, step towards the centre by moving the focussing mirrors. + Assumes everything is done at max zoom.""" + + # Config + hfm_step_per_x_px = Cpt(Signal, name="hfm_step_per_x_px", kind=Kind.config) + vfm_step_per_y_px = Cpt(Signal, name="hfm_step_per_y_px", kind=Kind.config) + sample_loc_x_px = Cpt(Signal, name="sample_loc_x_px", kind=Kind.config) + sample_loc_y_px = Cpt(Signal, name="sample_loc_y_px", kind=Kind.config) + px_tolerance = Cpt(Signal, name="px_tolerance", kind=Kind.config) + step_fraction = Cpt(Signal, name="step_fraction", kind=Kind.config) + + def _px_from_centre(self, bp: BeamProfile, axis: Literal["x", "y"]): + sample_loc = self.sample_loc_x_px if axis == "x" else self.sample_loc_y_px + curr_loc = bp.x_pos_px if axis == "x" else bp.y_pos_px + return float(sample_loc.get() - curr_loc.get()) # type: ignore + + def step_towards_centre(self): + """Take a step towards the beam centre by adjusting the HFM and VFM. Uses BeamProfile to + determine the x and y location on the samcam in px, the step_per_[x,y]_px parameters to + determine how far to move the mirrors, and multiplies this by the step fraction.""" + if self.device_manager is None: + raise RuntimeError( + "This can only be called in a connected BEC session with access to other devices." + ) + + 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")) + 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, " + "supply them in 'needs' in the BEC device config." + ) + logger.info( + f"Beam steering stepping towards target {self.sample_loc_x_px.get()}, {self.sample_loc_y_px.get()} (x, y px on samcam image)..." + ) + return self._step_towards_centre(hfm_motor, vfm_motor, bp) + + def _step_towards_centre( + self, hfm_motor: EpicsMotorEC, vfm_motor: EpicsMotorEC, beam_profile: BeamProfile + ): + x_dist = self._px_from_centre(beam_profile, "x") + y_dist = self._px_from_centre(beam_profile, "y") + tol = float(self.px_tolerance.get()) # type: ignore + damp_factor = float(self.step_fraction.get()) # type: ignore + hfm_step_per_x_px = float(self.hfm_step_per_x_px.get()) # type: ignore + vfm_step_per_y_px = float(self.vfm_step_per_y_px.get()) # type: ignore + logger.info(f"Beam steering - samcam distance: {x_dist=} px, {y_dist=} px") + + if abs(x_dist) > tol: + x_full_step = x_dist * hfm_step_per_x_px + x_dampened_step = x_full_step * damp_factor + new_x_pos = hfm_motor.user_setpoint.get() + x_dampened_step # type: ignore + logger.info(f"Moving HFM_YR {x_dampened_step} units to {new_x_pos}") + x_status = hfm_motor.move(new_x_pos) + else: + x_status = None + + if abs(y_dist) > tol: + y_full_step = y_dist * vfm_step_per_y_px + y_dampened_step = y_full_step * damp_factor + new_y_pos = vfm_motor.user_setpoint.get() + y_dampened_step # type: ignore + logger.info(f"Moving VFM_YW {y_dampened_step} units to {new_y_pos}") + y_status = vfm_motor.move(new_y_pos) + else: + y_status = None + + if x_status is not None and y_status is not None: + return x_status and y_status + if x_status is not None: + return x_status + if y_status is not None: + return y_status + return None + + def trigger(self): + """External interface for 'step_towards_centre'""" + return self.step_towards_centre()