Add devices required for Hyperion beamstop check

src/dodal/beamlines/i03.py

@@ -33,6 +33,7 @@
 from dodal.devices.i03 import Beamstop
 from dodal.devices.i03.dcm import DCM
 from dodal.devices.i03.undulator_dcm import UndulatorDCM
+from dodal.devices.ipin import IPin
 from dodal.devices.motors import XYZStage
 from dodal.devices.oav.oav_detector import OAVBeamCentreFile
 from dodal.devices.oav.oav_parameters import OAVConfigBeamCentre
@@ -456,3 +457,11 @@ def collimation_table() -> CollimationTable:
     If this is called when already instantiated in i03, it will return the existing object.
     """
     return CollimationTable(prefix=f"{PREFIX.beamline_prefix}-MO-TABLE-01")
+
+
+@device_factory()
+def ipin() -> IPin:
+    """Get the i03 ipin device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in i04, it will return the existing object.
+    """
+    return IPin(f"{PREFIX.beamline_prefix}-EA-PIN-01:")

src/dodal/devices/ipin.py

@@ -1,5 +1,22 @@
-from ophyd_async.core import StandardReadable, StandardReadableFormat
-from ophyd_async.epics.core import epics_signal_r
+from ophyd_async.core import StandardReadable, StandardReadableFormat, SubsetEnum
+from ophyd_async.epics.core import epics_signal_r, epics_signal_rw
+
+
+class IPinGain(SubsetEnum):
+    GAIN_10E3_LOW_NOISE = "10^3 low noise"
+    GAIN_10E4_LOW_NOISE = "10^4 low noise"
+    GAIN_10E5_LOW_NOISE = "10^5 low noise"
+    GAIN_10E6_LOW_NOISE = "10^6 low noise"
+    GAIN_10E7_LOW_NOISE = "10^7 low noise"
+    GAIN_10E8_LOW_NOISE = "10^8 low noise"
+    GAIN_10E9_LOW_NOISE = "10^9 low noise"
+    GAIN_10E5_HIGH_SPEED = "10^5 high speed"
+    GAIN_10E6_HIGH_SPEED = "10^6 high speed"
+    GAIN_10E7_HIGH_SPEED = "10^7 high speed"
+    GAIN_10E8_HIGH_SPEED = "10^8 high speed"
+    GAIN_10E9_HIGH_SPEED = "10^9 high speed"
+    GAIN_10E10_HIGH_SPEED = "10^10 high spd"
+    GAIN_10E11_HIGH_SPEED = "10^11 high spd"
 
 
 class IPin(StandardReadable):
@@ -10,4 +27,5 @@ def __init__(self, prefix: str, name: str = "") -> None:
             format=StandardReadableFormat.HINTED_SIGNAL
         ):
             self.pin_readback = epics_signal_r(float, prefix + "I")
+            self.gain = epics_signal_rw(IPinGain, prefix + "GAIN")
         super().__init__(name)

src/dodal/devices/mx_phase1/beamstop.py

@@ -10,6 +10,8 @@
 
 from dodal.common.beamlines.beamline_parameters import GDABeamlineParameters
 
+_BEAMSTOP_OUT_DELTA_Y_MM = -2
+
 
 class BeamstopPositions(StrictEnum):
     """
@@ -28,6 +30,7 @@ class BeamstopPositions(StrictEnum):
     """
 
     DATA_COLLECTION = "Data Collection"
+    OUT_OF_BEAM = "Out"
     UNKNOWN = "Unknown"
 
 
@@ -63,6 +66,10 @@ def __init__(
             float(beamline_parameters[f"in_beam_{axis}_STANDARD"])
             for axis in ("x", "y", "z")
         ]
+
+        self._out_of_beam_xyz_mm = list(self._in_beam_xyz_mm)
+        self._out_of_beam_xyz_mm[1] += _BEAMSTOP_OUT_DELTA_Y_MM
+
         self._xyz_tolerance_mm = [
             float(beamline_parameters[f"bs_{axis}_tolerance"])
             for axis in ("x", "y", "z")
@@ -72,24 +79,36 @@ def __init__(
 
     def _get_selected_position(self, x: float, y: float, z: float) -> BeamstopPositions:
         current_pos = [x, y, z]
-        if all(
-            isclose(axis_pos, axis_in_beam, abs_tol=axis_tolerance)
-            for axis_pos, axis_in_beam, axis_tolerance in zip(
-                current_pos, self._in_beam_xyz_mm, self._xyz_tolerance_mm, strict=False
-            )
-        ):
+        if self._is_near_position(current_pos, self._in_beam_xyz_mm):
             return BeamstopPositions.DATA_COLLECTION
+        elif self._is_near_position(current_pos, self._out_of_beam_xyz_mm):
+            return BeamstopPositions.OUT_OF_BEAM
         else:
             return BeamstopPositions.UNKNOWN
 
+    def _is_near_position(
+        self, current_pos: list[float], target_pos: list[float]
+    ) -> bool:
+        return all(
+            isclose(axis_pos, axis_in_beam, abs_tol=axis_tolerance)
+            for axis_pos, axis_in_beam, axis_tolerance in zip(
+                current_pos, target_pos, self._xyz_tolerance_mm, strict=False
+            )
+        )
+
     async def _set_selected_position(self, position: BeamstopPositions) -> None:
         match position:
             case BeamstopPositions.DATA_COLLECTION:
-                # Move z first as it could be under the table
-                await self.z_mm.set(self._in_beam_xyz_mm[2])
-                await asyncio.gather(
-                    self.x_mm.set(self._in_beam_xyz_mm[0]),
-                    self.y_mm.set(self._in_beam_xyz_mm[1]),
-                )
+                await self._safe_move_above_table(self._in_beam_xyz_mm)
+            case BeamstopPositions.OUT_OF_BEAM:
+                await self._safe_move_above_table(self._out_of_beam_xyz_mm)
             case _:
                 raise ValueError(f"Cannot set beamstop to position {position}")
+
+    async def _safe_move_above_table(self, pos: list[float]):
+        # Move z first as it could be under the table
+        await self.z_mm.set(pos[2])
+        await asyncio.gather(
+            self.x_mm.set(pos[0]),
+            self.y_mm.set(pos[1]),
+        )

tests/devices/mx_phase1/test_beamstop.py

@@ -6,11 +6,11 @@
 from bluesky import plan_stubs as bps
 from bluesky.preprocessors import run_decorator
 from bluesky.run_engine import RunEngine
-from ophyd_async.testing import get_mock_put, set_mock_value
+from ophyd_async.testing import get_mock, get_mock_put, set_mock_value
 
 from dodal.common.beamlines.beamline_parameters import GDABeamlineParameters
 from dodal.devices.i03 import Beamstop, BeamstopPositions
-from dodal.testing import patch_motor
+from dodal.testing import patch_all_motors, patch_motor
 from tests.common.beamlines.test_beamline_parameters import TEST_BEAMLINE_PARAMETERS_TXT
 
 
@@ -25,12 +25,13 @@ def beamline_parameters() -> GDABeamlineParameters:
         [0, 0, 0, BeamstopPositions.UNKNOWN],
         [1.52, 44.78, 30.0, BeamstopPositions.DATA_COLLECTION],
         [1.501, 44.776, 29.71, BeamstopPositions.DATA_COLLECTION],
+        [1.52, 42.78, 29.71, BeamstopPositions.OUT_OF_BEAM],
         [1.499, 44.776, 29.71, BeamstopPositions.UNKNOWN],
         [1.501, 44.774, 29.71, BeamstopPositions.UNKNOWN],
         [1.501, 44.776, 29.69, BeamstopPositions.UNKNOWN],
     ],
 )
-async def test_beamstop_pos_select(
+async def test_beamstop_pos_read_selected_pos(
     beamline_parameters: GDABeamlineParameters,
     run_engine: RunEngine,
     x: float,
@@ -67,8 +68,18 @@ def check_in_beam():
     assert data["beamstop-selected_pos"] == expected_pos
 
 
-async def test_set_beamstop_position_to_data_collection_moves_beamstop_into_beam(
-    beamline_parameters: GDABeamlineParameters, run_engine: RunEngine
+@pytest.mark.parametrize(
+    "demanded_pos, expected_coords",
+    [
+        [BeamstopPositions.DATA_COLLECTION, (1.52, 44.78, 30.0)],
+        [BeamstopPositions.OUT_OF_BEAM, (1.52, 42.78, 30.0)],
+    ],
+)
+async def test_set_beamstop_position_to_data_collection_moves_beamstop(
+    demanded_pos: BeamstopPositions,
+    expected_coords: tuple[float, float, float],
+    beamline_parameters: GDABeamlineParameters,
+    run_engine: RunEngine,
 ):
     beamstop = Beamstop("-MO-BS-01:", beamline_parameters, name="beamstop")
     await beamstop.connect(mock=True)
@@ -86,13 +97,11 @@ async def test_set_beamstop_position_to_data_collection_moves_beamstop_into_beam
     parent_mock.attach_mock(get_mock_put(y_mock), "beamstop_y")
     parent_mock.attach_mock(get_mock_put(z_mock), "beamstop_z")
 
-    run_engine(
-        bps.abs_set(beamstop.selected_pos, BeamstopPositions.DATA_COLLECTION, wait=True)
-    )
+    run_engine(bps.abs_set(beamstop.selected_pos, demanded_pos, wait=True))
 
-    assert get_mock_put(x_mock).call_args_list == [call(1.52, wait=True)]
-    assert get_mock_put(y_mock).call_args_list == [call(44.78, wait=True)]
-    assert get_mock_put(z_mock).call_args_list == [call(30.0, wait=True)]
+    assert get_mock_put(x_mock).call_args_list == [call(expected_coords[0], wait=True)]
+    assert get_mock_put(y_mock).call_args_list == [call(expected_coords[1], wait=True)]
+    assert get_mock_put(z_mock).call_args_list == [call(expected_coords[2], wait=True)]
 
     assert parent_mock.method_calls[0] == call.beamstop_z(30.0, wait=True)
 
@@ -107,3 +116,25 @@ async def test_set_beamstop_position_to_unknown_raises_error(
             bps.abs_set(beamstop.selected_pos, BeamstopPositions.UNKNOWN, wait=True)
         )
         assert isinstance(e.value.args[0].exception(), ValueError)
+
+
+async def test_beamstop_select_pos_moves_z_axis_first(
+    run_engine: RunEngine, beamline_parameters: GDABeamlineParameters
+):
+    beamstop = Beamstop("-MO-BS-01:", beamline_parameters, name="beamstop")
+    await beamstop.connect(mock=True)
+    patch_all_motors(beamstop)
+
+    run_engine(
+        bps.abs_set(beamstop.selected_pos, BeamstopPositions.DATA_COLLECTION, wait=True)
+    )
+
+    parent = get_mock(beamstop)
+    parent.assert_has_calls(
+        [
+            call.selected_pos.put(BeamstopPositions.DATA_COLLECTION, wait=True),
+            call.z_mm.user_setpoint.put(30.0, wait=True),
+            call.x_mm.user_setpoint.put(1.52, wait=True),
+            call.y_mm.user_setpoint.put(44.78, wait=True),
+        ]
+    )