Merge pull request #243 from DiamondLightSource/hyperion_986_set_ener…

…gy_apply_lookup_table_value_for_pitch_roll

Create devices needed for aligning the DCM and mirrors when changing energy

.pre-commit-config.yaml

@@ -25,7 +25,7 @@ repos:
         types: [python]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.4.1
+    rev: v1.7.1
     hooks:
       - id: mypy
         files: 'src/.*\.py$'

pyproject.toml

@@ -80,8 +80,8 @@ markers = [
     "s03: marks tests as requiring the s03 simulator running (deselect with '-m \"not s03\"')",
 ]
 addopts = """
-    --tb=native -vv --doctest-modules --doctest-glob="*.rst"
     --cov=dodal --cov-report term --cov-report xml:cov.xml
+    --tb=native -vv --doctest-modules --doctest-glob="*.rst"
     """
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs src tests"

src/dodal/beamlines/i03.py

@@ -11,8 +11,10 @@
 from dodal.devices.eiger import EigerDetector
 from dodal.devices.fast_grid_scan import FastGridScan
 from dodal.devices.flux import Flux
+from dodal.devices.focusing_mirror import FocusingMirror, VFMMirrorVoltages
 from dodal.devices.oav.oav_detector import OAV, OAVConfigParams
 from dodal.devices.oav.pin_image_recognition import PinTipDetection
+from dodal.devices.qbpm1 import QBPM1
 from dodal.devices.s4_slit_gaps import S4SlitGaps
 from dodal.devices.sample_shutter import SampleShutter
 from dodal.devices.smargon import Smargon
@@ -29,6 +31,7 @@
     "/dls_sw/i03/software/gda/configurations/i03-config/xml/jCameraManZoomLevels.xml"
 )
 DISPLAY_CONFIG = "/dls_sw/i03/software/gda_versions/var/display.configuration"
+DAQ_CONFIGURATION_PATH = "/dls_sw/i03/software/daq_configuration"
 
 BL = get_beamline_name("s03")
 set_log_beamline(BL)
@@ -41,14 +44,69 @@ def dcm(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) ->
     If this is called when already instantiated in i03, it will return the existing object.
     """
     return device_instantiation(
-        device_factory=DCM,
-        name="dcm",
+        DCM,
+        "dcm",
+        "",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+        daq_configuration_path=DAQ_CONFIGURATION_PATH,
+    )
+
+
+@skip_device(lambda: BL == "s03")
+def qbpm1(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) -> QBPM1:
+    return device_instantiation(
+        device_factory=QBPM1,
+        name="qbpm1",
         prefix="",
         wait=wait_for_connection,
         fake=fake_with_ophyd_sim,
     )
 
 
+@skip_device(lambda: BL == "s03")
+def vfm(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> FocusingMirror:
+    mirror = device_instantiation(
+        device_factory=FocusingMirror,
+        name="vfm",
+        prefix="-OP-VFM-01:",
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+        bragg_to_lat_lut_path=DAQ_CONFIGURATION_PATH
+        + "/lookup/BeamLineEnergy_DCM_VFM_x_converter.txt",
+    )
+    return mirror
+
+
+@skip_device(lambda: BL == "s03")
+def vfm_mirror_voltages(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> VFMMirrorVoltages:
+    return device_instantiation(
+        device_factory=VFMMirrorVoltages,
+        name="vfm_mirror_voltages",
+        prefix="-MO-PSU-01:",
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+        daq_configuration_path=DAQ_CONFIGURATION_PATH,
+    )
+
+
+@skip_device(lambda: BL == "s03")
+def hfm(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> FocusingMirror:
+    return device_instantiation(
+        device_factory=FocusingMirror,
+        name="hfm",
+        prefix="-OP-HFM-01:",
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+    )
+
+
 def aperture_scatterguard(
     wait_for_connection: bool = True,
     fake_with_ophyd_sim: bool = False,

src/dodal/beamlines/i04.py

@@ -30,6 +30,7 @@
     "/dls_sw/i04/software/gda/configurations/i04-config/xml/jCameraManZoomLevels.xml"
 )
 DISPLAY_CONFIG = "/dls_sw/i04/software/gda_versions/var/display.configuration"
+DAQ_CONFIGURATION_PATH = "/dls_sw/i04/software/daq_configuration"
 
 BL = get_beamline_name("s04")
 set_log_beamline(BL)
@@ -192,6 +193,7 @@ def dcm(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) ->
         "",
         wait_for_connection,
         fake_with_ophyd_sim,
+        daq_configuration_path=DAQ_CONFIGURATION_PATH,
     )
 
 

src/dodal/devices/DCM.py

@@ -3,6 +3,15 @@
 
 
 class DCM(Device):
+    def __init__(self, *args, daq_configuration_path: str, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.dcm_pitch_converter_lookup_table_path = (
+            daq_configuration_path + "/lookup/BeamLineEnergy_DCM_Pitch_converter.txt"
+        )
+        self.dcm_roll_converter_lookup_table_path = (
+            daq_configuration_path + "/lookup/BeamLineEnergy_DCM_Roll_converter.txt"
+        )
+
     """
     A double crystal monochromator (DCM), used to select the energy of the beam.
 
@@ -12,12 +21,12 @@ class DCM(Device):
     offset ensures that the beam exits the DCM at the same point, regardless of energy.
     """
 
-    bragg = Cpt(EpicsMotor, "-MO-DCM-01:BRAGG")
-    roll = Cpt(EpicsMotor, "-MO-DCM-01:ROLL")
-    offset = Cpt(EpicsMotor, "-MO-DCM-01:OFFSET")
-    perp = Cpt(EpicsMotor, "-MO-DCM-01:PERP")
+    bragg_in_degrees = Cpt(EpicsMotor, "-MO-DCM-01:BRAGG")
+    roll_in_mrad = Cpt(EpicsMotor, "-MO-DCM-01:ROLL")
+    offset_in_mm = Cpt(EpicsMotor, "-MO-DCM-01:OFFSET")
+    perp_in_mm = Cpt(EpicsMotor, "-MO-DCM-01:PERP")
     energy_in_kev = Cpt(EpicsMotor, "-MO-DCM-01:ENERGY", kind=Kind.hinted)
-    pitch = Cpt(EpicsMotor, "-MO-DCM-01:PITCH")
+    pitch_in_mrad = Cpt(EpicsMotor, "-MO-DCM-01:PITCH")
     wavelength = Cpt(EpicsMotor, "-MO-DCM-01:WAVELENGTH")
 
     # temperatures
@@ -28,3 +37,9 @@ class DCM(Device):
     backplate_temp = Cpt(EpicsSignalRO, "-MO-DCM-01:TEMP5")
     perp_temp = Cpt(EpicsSignalRO, "-MO-DCM-01:TEMP6")
     perp_sub_assembly_temp = Cpt(EpicsSignalRO, "-MO-DCM-01:TEMP7")
+
+
+def fixed_offset_from_beamline_params(gda_beamline_parameters):
+    """I03 configures the DCM Perp as a side effect of applying this fixed value to the DCM Offset after an energy change"""
+    # Nb this parameter is misleadingly named to confuse you
+    return gda_beamline_parameters["DCM_Perp_Offset_FIXED"]

src/dodal/devices/eiger.py

@@ -8,7 +8,7 @@
 from dodal.devices.detector import DetectorParams, TriggerMode
 from dodal.devices.eiger_odin import EigerOdin
 from dodal.devices.status import await_value
-from dodal.devices.utils import run_functions_without_blocking
+from dodal.devices.util.epics_util import run_functions_without_blocking
 from dodal.log import LOGGER
 
 FREE_RUN_MAX_IMAGES = 1000000

src/dodal/devices/focusing_mirror.py

@@ -0,0 +1,133 @@
+from enum import Enum
+from typing import Any
+
+from ophyd import Component, Device, EpicsMotor, EpicsSignal
+from ophyd.status import Status, StatusBase
+
+from dodal.log import LOGGER
+
+VOLTAGE_POLLING_DELAY_S = 0.5
+
+# The default timeout is 60 seconds as voltage slew rate is typically ~2V/s
+DEFAULT_SETTLE_TIME_S = 60
+
+DEMAND_ACCEPTED_OK = 1
+
+
+class MirrorStripe(Enum):
+    RHODIUM = "Rhodium"
+    BARE = "Bare"
+    PLATINUM = "Platinum"
+
+
+class MirrorVoltageDevice(Device):
+    """Abstract the bimorph mirror voltage PVs into a single device that can be set asynchronously and returns when
+    the demanded voltage setpoint is accepted, without blocking the caller as this process can take significant time.
+    """
+
+    _actual_v: EpicsSignal = Component(EpicsSignal, "R")
+    _setpoint_v: EpicsSignal = Component(EpicsSignal, "D")
+    _demand_accepted: EpicsSignal = Component(EpicsSignal, "DSEV")
+
+    def set(self, value, *args, **kwargs) -> StatusBase:
+        """Combine the following operations into a single set:
+        1. apply the value to the setpoint PV
+        2. Return to the caller with a Status future
+        3. Wait until demand is accepted
+        4. When either demand is accepted or DEFAULT_SETTLE_TIME expires, signal the result on the Status
+        """
+
+        setpoint_v = self._setpoint_v
+        demand_accepted = self._demand_accepted
+
+        if setpoint_v.get() == value:
+            LOGGER.debug(f"{setpoint_v.name} already at {value} - skipping set")
+            return Status(success=True, done=True)
+
+        if demand_accepted.get() != DEMAND_ACCEPTED_OK:
+            raise AssertionError(
+                f"Attempted to set {setpoint_v.name} when demand is not accepted."
+            )
+
+        LOGGER.debug(f"setting {setpoint_v.name} to {value}")
+        demand_accepted_status = Status(self, DEFAULT_SETTLE_TIME_S)
+
+        subscription: dict[str, Any] = {"handle": None}
+
+        def demand_check_callback(old_value, value, **kwargs):
+            LOGGER.debug(f"Got event old={old_value} new={value}")
+            if old_value != DEMAND_ACCEPTED_OK and value == DEMAND_ACCEPTED_OK:
+                LOGGER.debug(f"Demand accepted for {setpoint_v.name}")
+                subs_handle = subscription.pop("handle", None)
+                if subs_handle is None:
+                    raise AssertionError("Demand accepted before set attempted")
+                demand_accepted.unsubscribe(subs_handle)
+
+                demand_accepted_status.set_finished()
+            # else timeout handled by parent demand_accepted_status
+
+        subscription["handle"] = demand_accepted.subscribe(demand_check_callback)
+        setpoint_status = setpoint_v.set(value)
+        status = setpoint_status & demand_accepted_status
+        return status
+
+
+class VFMMirrorVoltages(Device):
+    def __init__(self, *args, daq_configuration_path: str, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.voltage_lookup_table_path = (
+            daq_configuration_path + "/json/mirrorFocus.json"
+        )
+
+    _channel14_voltage_device = Component(MirrorVoltageDevice, "BM:V14")
+    _channel15_voltage_device = Component(MirrorVoltageDevice, "BM:V15")
+    _channel16_voltage_device = Component(MirrorVoltageDevice, "BM:V16")
+    _channel17_voltage_device = Component(MirrorVoltageDevice, "BM:V17")
+    _channel18_voltage_device = Component(MirrorVoltageDevice, "BM:V18")
+    _channel19_voltage_device = Component(MirrorVoltageDevice, "BM:V19")
+    _channel20_voltage_device = Component(MirrorVoltageDevice, "BM:V20")
+    _channel21_voltage_device = Component(MirrorVoltageDevice, "BM:V21")
+
+    @property
+    def voltage_channels(self) -> list[MirrorVoltageDevice]:
+        return [
+            self._channel14_voltage_device,
+            self._channel15_voltage_device,
+            self._channel16_voltage_device,
+            self._channel17_voltage_device,
+            self._channel18_voltage_device,
+            self._channel19_voltage_device,
+            self._channel20_voltage_device,
+            self._channel21_voltage_device,
+        ]
+
+
+class FocusingMirror(Device):
+    """Focusing Mirror"""
+
+    def __init__(self, bragg_to_lat_lut_path, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.bragg_to_lat_lookup_table_path = bragg_to_lat_lut_path
+
+    yaw_mrad: EpicsMotor = Component(EpicsMotor, "YAW")
+    pitch_mrad: EpicsMotor = Component(EpicsMotor, "PITCH")
+    fine_pitch_mm: EpicsMotor = Component(EpicsMotor, "FPMTR")
+    roll_mrad: EpicsMotor = Component(EpicsMotor, "ROLL")
+    vert_mm: EpicsMotor = Component(EpicsMotor, "VERT")
+    lat_mm: EpicsMotor = Component(EpicsMotor, "LAT")
+    jack1_mm: EpicsMotor = Component(EpicsMotor, "Y1")
+    jack2_mm: EpicsMotor = Component(EpicsMotor, "Y2")
+    jack3_mm: EpicsMotor = Component(EpicsMotor, "Y3")
+    translation1_mm: EpicsMotor = Component(EpicsMotor, "X1")
+    translation2_mm: EpicsMotor = Component(EpicsMotor, "X2")
+
+    stripe: EpicsSignal = Component(EpicsSignal, "STRP:DVAL", string=True)
+    # apply the current set stripe setting
+    apply_stripe: EpicsSignal = Component(EpicsSignal, "CHANGE.PROC")
+
+    def energy_to_stripe(self, energy_kev):
+        # In future, this should be configurable per-mirror
+        if energy_kev < 7:
+            return MirrorStripe.BARE
+        else:
+            return MirrorStripe.RHODIUM

src/dodal/devices/qbpm1.py

@@ -3,4 +3,6 @@
 
 
 class QBPM1(Device):
+    """Quadrant Beam Position Monitor"""
+
     intensity = Cpt(EpicsSignalRO, "-DI-QBPM-01:INTEN", kind=Kind.normal)

src/dodal/devices/smargon.py

@@ -7,7 +7,7 @@
 
 from dodal.devices.motors import MotorLimitHelper, XYZLimitBundle
 from dodal.devices.status import await_approx_value
-from dodal.devices.utils import SetWhenEnabled
+from dodal.devices.util.epics_util import SetWhenEnabled
 
 
 class StubPosition(Enum):

src/dodal/devices/util/adjuster_plans.py

@@ -0,0 +1,24 @@
+"""
+All the methods in this module return a bluesky plan generator that adjusts a value
+according to some criteria either via feedback, preset positions, lookup tables etc.
+"""
+from typing import Callable, Generator
+
+from bluesky import plan_stubs as bps
+from bluesky.run_engine import Msg
+from ophyd.epics_motor import EpicsMotor
+
+from dodal.log import LOGGER
+
+
+def lookup_table_adjuster(
+    lookup_table: Callable[[float], float], output_device: EpicsMotor, input
+):
+    """Returns a callable that adjusts a value according to a lookup table"""
+
+    def adjust(group=None) -> Generator[Msg, None, None]:
+        setpoint = lookup_table(input)
+        LOGGER.info(f"lookup_table_adjuster setting {output_device.name} to {setpoint}")
+        yield from bps.abs_set(output_device, setpoint, group=group)
+
+    return adjust

src/dodal/devices/util/lookup_tables.py

@@ -0,0 +1,42 @@
+"""
+All the public methods in this module return a lookup table of some kind that
+converts the source value s to a target value t for different values of s.
+"""
+from collections.abc import Sequence
+from typing import Callable
+
+import numpy as np
+from numpy import interp, loadtxt
+
+from dodal.log import LOGGER
+
+
+def linear_interpolation_lut(filename: str) -> Callable[[float], float]:
+    """Returns a callable that converts values by linear interpolation of lookup table values"""
+    LOGGER.info(f"Using lookup table {filename}")
+    s_and_t_vals = zip(*loadtxt(filename, comments=["#", "Units"]))
+
+    s_values: Sequence
+    t_values: Sequence
+    s_values, t_values = s_and_t_vals
+
+    # numpy interp expects x-values to be increasing
+    if not np.all(np.diff(s_values) > 0):
+        LOGGER.info(
+            f"Configuration file {filename} values are not ascending, trying reverse order..."
+        )
+        s_values = list(reversed(s_values))
+        t_values = list(reversed(t_values))
+        if not np.all(np.diff(s_values) > 0):
+            raise AssertionError(
+                f"Configuration file {filename} lookup table does not monotonically increase or decrease."
+            )
+
+    def s_to_t2(s: float) -> float:
+        if s < s_values[0] or s > s_values[len(s_values) - 1]:
+            raise ValueError(
+                f"Lookup table does not support extrapolation from file {filename}, s={s}"
+            )
+        return float(interp(s, s_values, t_values))
+
+    return s_to_t2