Merge remote-tracking branch 'origin/main' into 772_beamline_specific…

…_zebra_constants

conftest.py

@@ -53,9 +53,9 @@ def patched_open(*args, **kwargs):
         requested_path = Path(args[0])
         if requested_path.is_absolute():
             for p in BANNED_PATHS:
-                assert not requested_path.is_relative_to(
-                    p
-                ), f"Attempt to open {requested_path} from inside a unit test"
+                assert not requested_path.is_relative_to(p), (
+                    f"Attempt to open {requested_path} from inside a unit test"
+                )
         return unpatched_open(*args, **kwargs)
 
     with patch("builtins.open", side_effect=patched_open):

src/dodal/beamlines/i03.py

@@ -24,6 +24,7 @@
 from dodal.devices.fast_grid_scan import PandAFastGridScan, ZebraFastGridScan
 from dodal.devices.flux import Flux
 from dodal.devices.focusing_mirror import FocusingMirrorWithStripes, MirrorVoltages
+from dodal.devices.i03.beamstop import Beamstop
 from dodal.devices.motors import XYZPositioner
 from dodal.devices.oav.oav_detector import OAV
 from dodal.devices.oav.oav_parameters import OAVConfig
@@ -104,6 +105,22 @@ def attenuator(
     )
 
 
+def beamstop(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Beamstop:
+    """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 device_instantiation(
+        Beamstop,
+        "beamstop",
+        "-MO-BS-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+        beamline_parameters=get_beamline_parameters(),
+    )
+
+
 def dcm(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) -> DCM:
     """Get the i03 DCM device, instantiate it if it hasn't already been.
     If this is called when already instantiated in i03, it will return the existing object.

src/dodal/common/crystal_metadata.py

@@ -55,7 +55,7 @@ def make_crystal_metadata_from_material(
     d_spacing = d_spacing_param or CrystalMetadata.calculate_default_d_spacing(
         material.value.lattice_parameter, reflection_plane
     )
-    assert all(
-        isinstance(i, int) and i > 0 for i in reflection_plane
-    ), "Reflection plane indices must be positive integers"
+    assert all(isinstance(i, int) and i > 0 for i in reflection_plane), (
+        "Reflection plane indices must be positive integers"
+    )
     return CrystalMetadata(usage, material.value.name, reflection_plane, d_spacing)

src/dodal/common/udc_directory_provider.py

@@ -46,7 +46,9 @@ async def update(self, *, directory: Path, suffix: str = "", **kwargs):
         self._filename_provider.suffix = suffix
 
     def __call__(self, device_name: str | None = None) -> PathInfo:
-        assert self._output_directory, "Directory unknown for PandA to write into, update() needs to be called at least once"
+        assert self._output_directory, (
+            "Directory unknown for PandA to write into, update() needs to be called at least once"
+        )
         return PathInfo(
             directory_path=self._output_directory,
             filename=self._filename_provider(device_name),

src/dodal/devices/attenuator/attenuator.py

@@ -106,7 +106,7 @@ def __init__(
         with self.add_children_as_readables():
             self.filters: DeviceVector[FilterMotor] = DeviceVector(
                 {
-                    index: FilterMotor(f"{prefix}MP{index+1}:", filter, name)
+                    index: FilterMotor(f"{prefix}MP{index + 1}:", filter, name)
                     for index, filter in enumerate(filter_selection)
                 }
             )

src/dodal/devices/bimorph_mirror.py

@@ -41,7 +41,7 @@ class BimorphMirrorStatus(StrictEnum):
     ERROR = "Error"
 
 
-class BimorphMirrorChannel(StandardReadable, EpicsDevice):
+class BimorphMirrorChannel(StandardReadable, Movable, EpicsDevice):
     """Collection of PVs comprising a single bimorph channel.
 
     Attributes:
@@ -56,6 +56,15 @@ class BimorphMirrorChannel(StandardReadable, EpicsDevice):
     status: A[SignalR[BimorphMirrorOnOff], PvSuffix("STATUS"), Format.CONFIG_SIGNAL]
     shift: A[SignalW[float], PvSuffix("SHIFT")]
 
+    @AsyncStatus.wrap
+    async def set(self, value: float):
+        """Sets channel's VOUT to given value.
+
+        Args:
+            value: float to set VOUT to
+        """
+        await self.output_voltage.set(value)
+
 
 class BimorphMirror(StandardReadable, Movable):
     """Class to represent CAENels Bimorph Mirrors.
@@ -91,7 +100,6 @@ def __init__(self, prefix: str, number_of_channels: int, name=""):
         self.commit_target_voltages = epics_signal_x(f"{prefix}ALLTRGT.PROC")
         self.status = epics_signal_r(BimorphMirrorStatus, f"{prefix}STATUS")
         self.err = epics_signal_r(str, f"{prefix}ERR")
-
         super().__init__(name=name)
 
     @AsyncStatus.wrap
@@ -106,7 +114,7 @@ async def set(self, value: Mapping[int, float], tolerance: float = 0.0001) -> No
 
         if any(key not in self.channels for key in value):
             raise ValueError(
-                f"Attempting to put to non-existent channels: {[key  for key in value if (key not in self.channels)]}"
+                f"Attempting to put to non-existent channels: {[key for key in value if (key not in self.channels)]}"
             )
 
         # Write target voltages:
@@ -129,7 +137,10 @@ async def set(self, value: Mapping[int, float], tolerance: float = 0.0001) -> No
                     timeout=DEFAULT_TIMEOUT,
                 )
                 for i, target in value.items()
-            ]
+            ],
+            wait_for_value(
+                self.status, BimorphMirrorStatus.IDLE, timeout=DEFAULT_TIMEOUT
+            ),
         )
 
 

src/dodal/devices/eiger_odin.py

@@ -91,10 +91,10 @@ def check_frames_dropped(self) -> tuple[bool, str]:
     def wait_for_no_errors(self, timeout) -> dict[SubscriptionStatus, str]:
         errors = {}
         for node_number, node_pv in enumerate(self.nodes):
-            errors[
-                await_value(node_pv.error_status, False, timeout)
-            ] = f"Filewriter {node_number} is in an error state with error message\
+            errors[await_value(node_pv.error_status, False, timeout)] = (
+                f"Filewriter {node_number} is in an error state with error message\
                      - {node_pv.error_message.get()}"
+            )
 
         return errors
 

src/dodal/devices/i03/beamstop.py

@@ -0,0 +1,85 @@
+from asyncio import gather
+from math import isclose
+
+from ophyd_async.core import StandardReadable, StrictEnum
+from ophyd_async.epics.motor import Motor
+
+from dodal.common.beamlines.beamline_parameters import GDABeamlineParameters
+from dodal.common.signal_utils import create_hardware_backed_soft_signal
+
+
+class BeamstopPositions(StrictEnum):
+    """
+    Beamstop positions.
+    GDA supports Standard/High/Low resolution positions, as well as parked and
+    robot load however all 3 resolution positions are the same. We also
+    do not use the robot load position in Hyperion.
+
+    Until we support moving the beamstop it is only necessary to check whether the
+    beamstop is in beam or not.
+
+    See Also:
+        https://github.com/DiamondLightSource/mx-bluesky/issues/484
+
+    Attributes:
+        DATA_COLLECTION: The beamstop is in beam ready for data collection
+        UNKNOWN: The beamstop is in some other position, check the device motor
+            positions to determine it.
+    """
+
+    DATA_COLLECTION = "Data Collection"
+    UNKNOWN = "Unknown"
+
+
+class Beamstop(StandardReadable):
+    """
+    Beamstop for I03.
+
+    Attributes:
+        x: beamstop x position in mm
+        y: beamstop y position in mm
+        z: beamstop z position in mm
+        selected_pos: Get the current position of the beamstop as an enum. Currently this
+            is read-only.
+    """
+
+    def __init__(
+        self,
+        prefix: str,
+        beamline_parameters: GDABeamlineParameters,
+        name: str = "",
+    ):
+        with self.add_children_as_readables():
+            self.x_mm = Motor(prefix + "X")
+            self.y_mm = Motor(prefix + "Y")
+            self.z_mm = Motor(prefix + "Z")
+            self.selected_pos = create_hardware_backed_soft_signal(
+                BeamstopPositions, self._get_selected_position
+            )
+
+        self._in_beam_xyz_mm = [
+            float(beamline_parameters[f"in_beam_{axis}_STANDARD"])
+            for axis in ("x", "y", "z")
+        ]
+        self._xyz_tolerance_mm = [
+            float(beamline_parameters[f"bs_{axis}_tolerance"])
+            for axis in ("x", "y", "z")
+        ]
+
+        super().__init__(name)
+
+    async def _get_selected_position(self) -> BeamstopPositions:
+        current_pos = await gather(
+            self.x_mm.user_readback.get_value(),
+            self.y_mm.user_readback.get_value(),
+            self.z_mm.user_readback.get_value(),
+        )
+        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
+            )
+        ):
+            return BeamstopPositions.DATA_COLLECTION
+        else:
+            return BeamstopPositions.UNKNOWN

src/dodal/devices/zebra/zebra.py

@@ -51,6 +51,11 @@ class I24Axes:
 
 
 class RotationDirection(StrictEnum):
+    """
+    Defines for a swept angle whether the scan width (sweep) is to be added or subtracted from
+    the initial angle to obtain the final angle.
+    """
+
     POSITIVE = "Positive"
     NEGATIVE = "Negative"
 
@@ -251,7 +256,7 @@ def __str__(self) -> str:
         for input, (source, invert) in enumerate(
             zip(self.sources, self.invert, strict=False)
         ):
-            input_strings.append(f"INP{input+1}={'!' if invert else ''}{source}")
+            input_strings.append(f"INP{input + 1}={'!' if invert else ''}{source}")
 
         return ", ".join(input_strings)
 

src/dodal/log.py

@@ -152,7 +152,7 @@ def set_up_graylog_handler(logger: Logger, host: str, port: int):
 def set_up_INFO_file_handler(logger, path: Path, filename: str):
     """Set up a file handler for the logger, at INFO level, which will keep 30 days
     of logs, rotating once per day. Creates the directory if necessary."""
-    print(f"Logging to INFO file handler {path/filename}")
+    print(f"Logging to INFO file handler {path / filename}")
     path.mkdir(parents=True, exist_ok=True)
     file_handler = TimedRotatingFileHandler(
         filename=path / filename, when="MIDNIGHT", backupCount=INFO_LOG_DAYS
@@ -169,7 +169,7 @@ def set_up_DEBUG_memory_handler(
     log file when it sees a message of severity ERROR. Creates the directory if
     necessary"""
     debug_path = path / "debug"
-    print(f"Logging to DEBUG handler {debug_path/filename}")
+    print(f"Logging to DEBUG handler {debug_path / filename}")
     debug_path.mkdir(parents=True, exist_ok=True)
     file_handler = TimedRotatingFileHandler(
         filename=debug_path / filename, when="H", backupCount=DEBUG_LOG_FILES_TO_KEEP

src/dodal/plans/wrapped.py

@@ -49,9 +49,9 @@ def count(
     Wraps bluesky.plans.count(det, num, delay, md=metadata) exposing only serializable
     parameters and metadata."""
     if isinstance(delay, Sequence):
-        assert (
-            len(delay) == num - 1
-        ), f"Number of delays given must be {num - 1}: was given {len(delay)}"
+        assert len(delay) == num - 1, (
+            f"Number of delays given must be {num - 1}: was given {len(delay)}"
+        )
     metadata = metadata or {}
     metadata["shape"] = (num,)
     yield from bp.count(tuple(detectors), num, delay=delay, md=metadata)

tests/common/beamlines/test_device_instantiation.py

@@ -29,9 +29,9 @@ def test_device_creation(RE, module_and_devices_for_beamline):
         for name, device in devices.items()
         if not follows_bluesky_protocols(device)
     ]
-    assert (
-        len(devices_not_following_bluesky_protocols) == 0
-    ), f"{devices_not_following_bluesky_protocols} do not follow bluesky protocols"
+    assert len(devices_not_following_bluesky_protocols) == 0, (
+        f"{devices_not_following_bluesky_protocols} do not follow bluesky protocols"
+    )
 
 
 @pytest.mark.parametrize(
@@ -56,6 +56,6 @@ def test_devices_are_identical(RE, module_and_devices_for_beamline):
     ]
     total_number_of_devices = len(devices_a)
     non_identical_number_of_devies = len(devices_a)
-    assert (
-        len(non_identical_names) == 0
-    ), f"{non_identical_number_of_devies}/{total_number_of_devices} devices were not identical: {non_identical_names}"
+    assert len(non_identical_names) == 0, (
+        f"{non_identical_number_of_devies}/{total_number_of_devices} devices were not identical: {non_identical_names}"
+    )

tests/devices/i03/test_beamstop.py

@@ -0,0 +1,66 @@
+from itertools import dropwhile
+from unittest.mock import Mock
+
+import pytest
+from bluesky import plan_stubs as bps
+from bluesky.preprocessors import run_decorator
+from bluesky.run_engine import RunEngine
+from ophyd_async.testing import set_mock_value
+
+from dodal.common.beamlines.beamline_parameters import GDABeamlineParameters
+from dodal.devices.i03.beamstop import Beamstop, BeamstopPositions
+
+
+@pytest.fixture
+def beamline_parameters() -> GDABeamlineParameters:
+    return GDABeamlineParameters.from_file(
+        "tests/test_data/test_beamline_parameters.txt"
+    )
+
+
+@pytest.mark.parametrize(
+    "x, y, z, expected_pos",
+    [
+        [0, 0, 0, BeamstopPositions.UNKNOWN],
+        [1.52, 44.78, 30.0, BeamstopPositions.DATA_COLLECTION],
+        [1.501, 44.776, 29.71, BeamstopPositions.DATA_COLLECTION],
+        [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(
+    beamline_parameters: GDABeamlineParameters,
+    RE: RunEngine,
+    x: float,
+    y: float,
+    z: float,
+    expected_pos: BeamstopPositions,
+):
+    beamstop = Beamstop("-MO-BS-01:", beamline_parameters, name="beamstop")
+    await beamstop.connect(mock=True)
+    set_mock_value(beamstop.x_mm.user_readback, x)
+    set_mock_value(beamstop.y_mm.user_readback, y)
+    set_mock_value(beamstop.z_mm.user_readback, z)
+
+    mock_callback = Mock()
+    RE.subscribe(mock_callback, "event")
+
+    @run_decorator()
+    def check_in_beam():
+        current_pos = yield from bps.rd(beamstop.selected_pos)
+        assert current_pos == expected_pos
+        yield from bps.create()
+        yield from bps.read(beamstop)
+        yield from bps.save()
+
+    RE(check_in_beam())
+
+    event_call = next(
+        dropwhile(lambda c: c.args[0] != "event", mock_callback.mock_calls)
+    )
+    data = event_call.args[1]["data"]
+    assert data["beamstop-x_mm"] == x
+    assert data["beamstop-y_mm"] == y
+    assert data["beamstop-z_mm"] == z
+    assert data["beamstop-selected_pos"] == expected_pos

tests/devices/unit_tests/detector/test_det_resolution.py

@@ -55,9 +55,9 @@ def test_resolution(
         detector_params.use_roi_mode = roi
         get_detector_max_size.return_value = 434.6
         actual_res = resolution(detector_params, wavelength_angstroms, det_distance_mm)
-    assert isclose(
-        expected_res, actual_res
-    ), f"expected={expected_res}, actual={actual_res}"
+    assert isclose(expected_res, actual_res), (
+        f"expected={expected_res}, actual={actual_res}"
+    )
 
 
 @pytest.mark.parametrize(
@@ -126,6 +126,6 @@ def test_resolution_with_roi_realistic(
 
     actual_res = resolution(detector_params, wavelength_angstroms, det_distance_mm)
 
-    assert isclose(
-        actual_res, expected_res, rtol=1e-3
-    ), f"expected={expected_res}, actual={actual_res}"
+    assert isclose(actual_res, expected_res, rtol=1e-3), (
+        f"expected={expected_res}, actual={actual_res}"
+    )