Create beamsize devices

src/dodal/beamlines/i03.py

@@ -31,6 +31,7 @@
 from dodal.devices.focusing_mirror import FocusingMirrorWithStripes, MirrorVoltages
 from dodal.devices.hutch_shutter import HutchShutter
 from dodal.devices.i03 import Beamstop
+from dodal.devices.i03.beamsize import Beamsize
 from dodal.devices.i03.dcm import DCM
 from dodal.devices.i03.undulator_dcm import UndulatorDCM
 from dodal.devices.motors import XYZStage
@@ -456,3 +457,13 @@ 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 beamsize() -> Beamsize:
+    """Get the i03 beamstop device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in i03, it will return the existing object.
+    """
+    return Beamsize(
+        aperture_scatterguard=aperture_scatterguard(),
+    )

src/dodal/beamlines/i04.py

@@ -20,6 +20,7 @@
 from dodal.devices.fast_grid_scan import ZebraFastGridScanThreeD
 from dodal.devices.flux import Flux
 from dodal.devices.i03.dcm import DCM
+from dodal.devices.i04.beamsize import Beamsize
 from dodal.devices.i04.constants import RedisConstants
 from dodal.devices.i04.murko_results import MurkoResultsDevice
 from dodal.devices.i04.transfocator import Transfocator
@@ -377,3 +378,14 @@ def scintillator() -> Scintillator:
         Reference(aperture_scatterguard()),
         get_beamline_parameters(),
     )
+
+
+@device_factory()
+def beamsize() -> Beamsize:
+    """Get the i04 beamstop 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 Beamsize(
+        transfocator=transfocator(),
+        aperture_scatterguard=aperture_scatterguard(),
+    )

src/dodal/devices/beamsize/beamsize.py

@@ -0,0 +1,9 @@
+from ophyd_async.core import SignalR, StandardReadable
+
+
+class BeamsizeBase(StandardReadable):
+    x_um: SignalR[float]
+    y_um: SignalR[float]
+
+    def __init__(self, name="beamsize"):
+        super().__init__(name=name)

src/dodal/devices/i03/beamsize.py

@@ -0,0 +1,34 @@
+from ophyd_async.core import Reference, derived_signal_r
+
+from dodal.devices.aperturescatterguard import ApertureScatterguard
+from dodal.devices.beamsize.beamsize import BeamsizeBase
+from dodal.devices.i03.constants import BeamsizeConstants
+
+
+class Beamsize(BeamsizeBase):
+    def __init__(self, aperture_scatterguard: ApertureScatterguard, name="beamsize"):
+        super().__init__(name=name)
+        self._aperture_scatterguard_ref = Reference(aperture_scatterguard)
+
+        self.x_um = derived_signal_r(
+            self._get_beamsize_x,
+            aperture_radius=self._aperture_scatterguard_ref().radius,
+            derived_units="µm",
+        )
+        self.y_um = derived_signal_r(
+            self._get_beamsize_y,
+            aperture_radius=self._aperture_scatterguard_ref().radius,
+            derived_units="µm",
+        )
+
+    def _get_beamsize_x(
+        self,
+        aperture_radius: float,
+    ) -> float:
+        return min(aperture_radius, BeamsizeConstants.BEAM_WIDTH_UM)
+
+    def _get_beamsize_y(
+        self,
+        aperture_radius: float,
+    ) -> float:
+        return min(aperture_radius, BeamsizeConstants.BEAM_HEIGHT_UM)

src/dodal/devices/i03/constants.py

@@ -0,0 +1,7 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class BeamsizeConstants:
+    BEAM_WIDTH_UM = 80.0
+    BEAM_HEIGHT_UM = 20.0

src/dodal/devices/i04/beamsize.py

@@ -0,0 +1,44 @@
+from ophyd_async.core import Reference, derived_signal_r
+
+from dodal.devices.aperturescatterguard import ApertureScatterguard
+from dodal.devices.beamsize.beamsize import BeamsizeBase
+from dodal.devices.i04.transfocator import Transfocator
+
+
+class Beamsize(BeamsizeBase):
+    def __init__(
+        self,
+        transfocator: Transfocator,
+        aperture_scatterguard: ApertureScatterguard,
+        name="beamsize",
+    ):
+        super().__init__(name=name)
+        self._transfocator_ref = Reference(transfocator)
+        self._aperture_scatterguard_ref = Reference(aperture_scatterguard)
+
+        self.x_um = derived_signal_r(
+            self._get_beamsize_x,
+            transfocator_size_x=self._transfocator_ref().current_horizontal_size_rbv,
+            aperture_radius=self._aperture_scatterguard_ref().radius,
+            derived_units="µm",
+        )
+        self.y_um = derived_signal_r(
+            self._get_beamsize_y,
+            transfocator_size_y=self._transfocator_ref().current_vertical_size_rbv,
+            aperture_radius=self._aperture_scatterguard_ref().radius,
+            derived_units="µm",
+        )
+
+    def _get_beamsize_x(
+        self,
+        transfocator_size_x: float,
+        aperture_radius: float,
+    ) -> float:
+        return min(transfocator_size_x, aperture_radius)
+
+    def _get_beamsize_y(
+        self,
+        transfocator_size_y: float,
+        aperture_radius: float,
+    ) -> float:
+        return min(transfocator_size_y, aperture_radius)

src/dodal/devices/i04/transfocator.py

@@ -30,6 +30,7 @@ def __init__(self, prefix: str, name: str = ""):
 
         with self.add_children_as_readables():
             self.number_filters_sp = epics_signal_rw(int, prefix + "NUM_FILTERS")
+            self.current_horizontal_size_rbv = epics_signal_r(float, prefix + "HOR")
             self.current_vertical_size_rbv = epics_signal_r(float, prefix + "VER")
 
         self.TIMEOUT = 120

tests/devices/conftest.py

@@ -0,0 +1,96 @@
+import asyncio
+from collections.abc import AsyncGenerator
+
+import pytest
+from ophyd_async.core import init_devices
+
+from dodal.common.beamlines.beamline_parameters import GDABeamlineParameters
+from dodal.devices.aperturescatterguard import (
+    AperturePosition,
+    ApertureScatterguard,
+    ApertureValue,
+    load_positions_from_beamline_parameters,
+)
+from dodal.testing import patch_all_motors
+
+
+@pytest.fixture
+def aperture_positions() -> dict[ApertureValue, AperturePosition]:
+    return load_positions_from_beamline_parameters(
+        GDABeamlineParameters(
+            params={
+                "miniap_x_LARGE_APERTURE": 2.389,
+                "miniap_y_LARGE_APERTURE": 40.986,
+                "miniap_z_LARGE_APERTURE": 15.8,
+                "sg_x_LARGE_APERTURE": 5.25,
+                "sg_y_LARGE_APERTURE": 4.43,
+                "miniap_x_MEDIUM_APERTURE": 2.384,
+                "miniap_y_MEDIUM_APERTURE": 44.967,
+                "miniap_z_MEDIUM_APERTURE": 15.8,
+                "sg_x_MEDIUM_APERTURE": 5.285,
+                "sg_y_MEDIUM_APERTURE": 0.46,
+                "miniap_x_SMALL_APERTURE": 2.430,
+                "miniap_y_SMALL_APERTURE": 48.974,
+                "miniap_z_SMALL_APERTURE": 15.8,
+                "sg_x_SMALL_APERTURE": 5.3375,
+                "sg_y_SMALL_APERTURE": -3.55,
+                "miniap_x_ROBOT_LOAD": 2.386,
+                "miniap_y_ROBOT_LOAD": 31.40,
+                "miniap_z_ROBOT_LOAD": 15.8,
+                "sg_x_ROBOT_LOAD": 5.25,
+                "sg_y_ROBOT_LOAD": 4.43,
+                "miniap_x_MANUAL_LOAD": -4.91,
+                "miniap_y_MANUAL_LOAD": -48.70,
+                "miniap_z_MANUAL_LOAD": -10.0,
+                "sg_x_MANUAL_LOAD": -4.7,
+                "sg_y_MANUAL_LOAD": 1.8,
+            }
+        )
+    )
+
+
+@pytest.fixture
+def aperture_tolerances():
+    return AperturePosition.tolerances_from_gda_params(
+        GDABeamlineParameters(
+            {
+                "miniap_x_tolerance": 0.004,
+                "miniap_y_tolerance": 0.1,
+                "miniap_z_tolerance": 0.1,
+                "sg_x_tolerance": 0.1,
+                "sg_y_tolerance": 0.1,
+            }
+        )
+    )
+
+
+@pytest.fixture
+async def ap_sg(
+    aperture_positions: dict[ApertureValue, AperturePosition],
+    aperture_tolerances: AperturePosition,
+) -> AsyncGenerator[ApertureScatterguard]:
+    async with init_devices(mock=True):
+        ap_sg = ApertureScatterguard(
+            aperture_prefix="-MO-MAPT-01:",
+            scatterguard_prefix="-MO-SCAT-01:",
+            name="test_ap_sg",
+            loaded_positions=aperture_positions,
+            tolerances=aperture_tolerances,
+        )
+
+    with patch_all_motors(ap_sg):
+        yield ap_sg
+
+
+async def set_to_position(
+    aperture_scatterguard: ApertureScatterguard, position: AperturePosition
+):
+    aperture_x, aperture_y, aperture_z, scatterguard_x, scatterguard_y = position.values
+
+    await asyncio.gather(
+        aperture_scatterguard.aperture.x.set(aperture_x),
+        aperture_scatterguard.aperture.y.set(aperture_y),
+        aperture_scatterguard.aperture.z.set(aperture_z),
+        aperture_scatterguard.scatterguard.x.set(scatterguard_x),
+        aperture_scatterguard.scatterguard.y.set(scatterguard_y),
+    )

tests/devices/i03/test_beamsize.py

@@ -0,0 +1,41 @@
+import pytest
+from ophyd_async.testing import set_mock_value
+
+from dodal.devices.aperturescatterguard import (
+    AperturePosition,
+    ApertureScatterguard,
+    ApertureValue,
+)
+from dodal.devices.i03.beamsize import Beamsize
+from tests.devices.conftest import set_to_position
+
+
+@pytest.mark.parametrize(
+    "selected_aperture, aperture_signal_values, expected_beamsize",
+    [
+        (ApertureValue.SMALL, (1, 0, 0), (20.0, 20.0)),
+        (ApertureValue.MEDIUM, (0, 1, 0), (50.0, 20.0)),
+        (ApertureValue.LARGE, (0, 0, 1), (80.0, 20.0)),
+        (ApertureValue.OUT_OF_BEAM, (0, 0, 0), (0.0, 0.0)),
+        (ApertureValue.PARKED, (0, 0, 0), (0.0, 0.0)),
+    ],
+)
+async def test_beamsize_gives_min_of_aperture_and_beam_width_and_height(
+    selected_aperture: ApertureValue,
+    aperture_signal_values: tuple[int, int, int],
+    expected_beamsize: tuple[float, float],
+    ap_sg: ApertureScatterguard,
+    aperture_positions: dict[ApertureValue, AperturePosition],
+):
+    await set_to_position(ap_sg, aperture_positions[selected_aperture])
+
+    for i, signal in enumerate(
+        (ap_sg.aperture.small, ap_sg.aperture.medium, ap_sg.aperture.large)
+    ):
+        set_mock_value(signal, aperture_signal_values[i])
+
+    beamsize = Beamsize(aperture_scatterguard=ap_sg)
+    beamsize_x = await beamsize.x_um.get_value()
+    beamsize_y = await beamsize.y_um.get_value()
+    assert beamsize_x == expected_beamsize[0]
+    assert beamsize_y == expected_beamsize[1]

tests/devices/i04/conftest.py

@@ -0,0 +1,11 @@
+import pytest
+from ophyd_async.core import init_devices
+
+from dodal.devices.i04.transfocator import Transfocator
+
+
+@pytest.fixture
+async def fake_transfocator() -> Transfocator:
+    async with init_devices(mock=True):
+        transfocator = Transfocator(prefix="", name="transfocator")
+    return transfocator

tests/devices/i04/test_beamsize.py

@@ -0,0 +1,57 @@
+import pytest
+from ophyd_async.testing import set_mock_value
+
+from dodal.devices.aperturescatterguard import (
+    AperturePosition,
+    ApertureScatterguard,
+    ApertureValue,
+)
+from dodal.devices.i04.beamsize import Beamsize
+from dodal.devices.i04.transfocator import Transfocator
+from tests.devices.conftest import set_to_position
+
+
+@pytest.mark.parametrize(
+    "selected_aperture, aperture_signal_values, transfocator_sizes, expected_beamsize",
+    [
+        (ApertureValue.SMALL, (1, 0, 0), (10.0, 10.0), (10.0, 10.0)),
+        (ApertureValue.SMALL, (1, 0, 0), (30.0, 10.0), (20.0, 10.0)),
+        (ApertureValue.SMALL, (1, 0, 0), (10.0, 30.0), (10.0, 20.0)),
+        (ApertureValue.SMALL, (1, 0, 0), (55.0, 30.0), (20.0, 20.0)),
+        (ApertureValue.MEDIUM, (0, 1, 0), (10.0, 10.0), (10.0, 10.0)),
+        (ApertureValue.MEDIUM, (0, 1, 0), (30.0, 70.0), (30.0, 50.0)),
+        (ApertureValue.MEDIUM, (0, 1, 0), (70.0, 30.0), (50.0, 30.0)),
+        (ApertureValue.MEDIUM, (0, 1, 0), (90.0, 90.0), (50.0, 50.0)),
+        (ApertureValue.LARGE, (0, 0, 1), (90.0, 55.0), (90.0, 55.0)),
+        (ApertureValue.LARGE, (0, 0, 1), (110.0, 55.0), (100.0, 55.0)),
+        (ApertureValue.LARGE, (0, 0, 1), (90.0, 110.0), (90.0, 100.0)),
+        (ApertureValue.LARGE, (0, 0, 1), (120.0, 120.0), (100.0, 100.0)),
+        (ApertureValue.OUT_OF_BEAM, (0, 0, 0), (120.0, 10.0), (0.0, 0.0)),
+        (ApertureValue.OUT_OF_BEAM, (0, 0, 0), (10.0, 120.0), (0.0, 0.0)),
+        (ApertureValue.PARKED, (0, 0, 0), (10.0, 10.0), (0.0, 0.0)),
+        (ApertureValue.PARKED, (0, 0, 0), (120.0, 120.0), (0.0, 0.0)),
+    ],
+)
+async def test_beamsize_gives_min_of_aperture_and_transfocator_width_and_height(
+    selected_aperture: ApertureValue,
+    aperture_signal_values: tuple[int, int, int],
+    transfocator_sizes: tuple[float, float],
+    expected_beamsize: tuple[float, float],
+    fake_transfocator: Transfocator,
+    ap_sg: ApertureScatterguard,
+    aperture_positions: dict[ApertureValue, AperturePosition],
+):
+    await set_to_position(ap_sg, aperture_positions[selected_aperture])
+
+    for i, signal in enumerate(
+        (ap_sg.aperture.small, ap_sg.aperture.medium, ap_sg.aperture.large)
+    ):
+        set_mock_value(signal, aperture_signal_values[i])
+    set_mock_value(fake_transfocator.current_horizontal_size_rbv, transfocator_sizes[0])
+    set_mock_value(fake_transfocator.current_vertical_size_rbv, transfocator_sizes[1])
+
+    beamsize = Beamsize(transfocator=fake_transfocator, aperture_scatterguard=ap_sg)
+
+    beamsize_x = await beamsize.x_um.get_value()
+    beamsize_y = await beamsize.y_um.get_value()
+    assert (beamsize_x, beamsize_y) == expected_beamsize