Add load metadata plan for fastcs eiger

pyproject.toml

@@ -24,9 +24,9 @@ dependencies = [
     "requests",
     "graypy",
     "pydantic>=2.0",
-    "opencv-python-headless",                                          # For pin-tip detection.
-    "aioca",                                                           # Required for CA support with ophyd-async.
-    "p4p",                                                             # Required for PVA support with ophyd-async.
+    "opencv-python-headless", # For pin-tip detection.
+    "aioca",                  # Required for CA support with ophyd-async.
+    "p4p",                    # Required for PVA support with ophyd-async.
     "numpy",
     "aiofiles",
     "aiohttp",

src/dodal/beamlines/i03.py

@@ -1,3 +1,4 @@
+from ophyd_async.fastcs.eiger import EigerDetector as FastEiger
 from ophyd_async.fastcs.panda import HDFPanda
 
 from dodal.common.beamlines.beamline_parameters import get_beamline_parameters
@@ -172,6 +173,20 @@ def eiger(mock: bool = False) -> EigerDetector:
     )
 
 
+@device_factory()
+def fastcs_eiger() -> FastEiger:
+    """Get the i03 FastCS Eiger 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 FastEiger(
+        prefix=PREFIX.beamline_prefix,
+        path_provider=get_path_provider(),
+        drv_suffix="-EA-EIGER-02:",
+        hdf_suffix="-EA-EIGER-01:OD:",
+    )
+
+
 @device_factory()
 def zebra_fast_grid_scan() -> ZebraFastGridScan:
     """Get the i03 zebra_fast_grid_scan device, instantiate it if it hasn't already been.

src/dodal/plans/configure_arm_trigger_and_disarm_detector.py

@@ -0,0 +1,167 @@
+import time
+
+import bluesky.plan_stubs as bps
+from bluesky import preprocessors as bpp
+from bluesky.run_engine import RunEngine
+from ophyd_async.core import DetectorTrigger
+from ophyd_async.fastcs.eiger import EigerDetector, EigerTriggerInfo
+
+from dodal.beamlines.i03 import fastcs_eiger
+from dodal.devices.detector import DetectorParams
+from dodal.log import LOGGER, do_default_logging_setup
+
+
+@bpp.run_decorator()
+def configure_arm_trigger_and_disarm_detector(
+    eiger: EigerDetector,
+    detector_params: DetectorParams,
+    trigger_info: EigerTriggerInfo,
+):
+    assert detector_params.expected_energy_ev
+    start = time.time()
+    yield from bps.unstage(eiger, wait=True)
+    LOGGER.info(f"Stopping Eiger-Odin: {time.time() - start}s")
+    start = time.time()
+    yield from set_cam_pvs(eiger, detector_params, wait=True)
+    LOGGER.info(f"Setting CAM PVs: {time.time() - start}s")
+    start = time.time()
+    yield from change_roi_mode(eiger, detector_params, wait=True)
+    LOGGER.info(f"Changing ROI Mode: {time.time() - start}s")
+    start = time.time()
+    yield from bps.abs_set(eiger.odin.num_frames_chunks, 1)
+    LOGGER.info(f"Setting # of Frame Chunks: {time.time() - start}s")
+    start = time.time()
+    yield from set_mx_settings_pvs(eiger, detector_params, wait=True)
+    LOGGER.info(f"Setting MX PVs: {time.time() - start}s")
+    start = time.time()
+    yield from bps.prepare(eiger, trigger_info, wait=True)
+    LOGGER.info(f"Preparing Eiger: {time.time() - start}s")
+    start = time.time()
+    yield from bps.kickoff(eiger, wait=True)
+    LOGGER.info(f"Kickoff Eiger: {time.time() - start}s")
+    start = time.time()
+    yield from bps.trigger(eiger.drv.detector.trigger)  # type: ignore
+    LOGGER.info(f"Triggering Eiger: {time.time() - start}s")
+    start = time.time()
+    yield from bps.complete(eiger, wait=True)
+    LOGGER.info(f"Completing Capture: {time.time() - start}s")
+    start = time.time()
+    yield from bps.unstage(eiger, wait=True)
+    LOGGER.info(f"Disarming Eiger: {time.time() - start}s")
+
+
+def set_cam_pvs(
+    eiger: EigerDetector,
+    detector_params: DetectorParams,
+    wait: bool,
+    group="cam_pvs",
+):
+    yield from bps.abs_set(
+        eiger.drv.detector.count_time, detector_params.exposure_time_s, group=group
+    )
+    yield from bps.abs_set(
+        eiger.drv.detector.frame_time, detector_params.exposure_time_s, group=group
+    )
+    yield from bps.abs_set(eiger.drv.detector.nexpi, 1, group=group)
+
+    if wait:
+        yield from bps.wait(group)
+
+
+def change_roi_mode(
+    eiger: EigerDetector,
+    detector_params: DetectorParams,
+    wait: bool,
+    group="roi_mode",
+):
+    detector_dimensions = (
+        detector_params.detector_size_constants.roi_size_pixels
+        if detector_params.use_roi_mode
+        else detector_params.detector_size_constants.det_size_pixels
+    )
+
+    yield from bps.abs_set(
+        eiger.drv.detector.roi_mode,
+        1 if detector_params.use_roi_mode else 0,
+        group=group,
+    )
+    yield from bps.abs_set(
+        eiger.odin.image_height,
+        detector_dimensions.height,
+        group=group,
+    )
+    yield from bps.abs_set(
+        eiger.odin.image_width,
+        detector_dimensions.width,
+        group=group,
+    )
+    yield from bps.abs_set(
+        eiger.odin.num_row_chunks,
+        detector_dimensions.height,
+        group=group,
+    )
+    yield from bps.abs_set(
+        eiger.odin.num_col_chunks,
+        detector_dimensions.width,
+        group=group,
+    )
+
+    if wait:
+        yield from bps.wait(group)
+
+
+def set_mx_settings_pvs(
+    eiger: EigerDetector,
+    detector_params: DetectorParams,
+    wait: bool,
+    group="mx_settings",
+):
+    beam_x_pixels, beam_y_pixels = detector_params.get_beam_position_pixels(
+        detector_params.detector_distance
+    )
+
+    yield from bps.abs_set(eiger.drv.detector.beam_center_x, beam_x_pixels, group)
+    yield from bps.abs_set(eiger.drv.detector.beam_center_y, beam_y_pixels, group)
+    yield from bps.abs_set(
+        eiger.drv.detector.detector_distance, detector_params.detector_distance, group
+    )
+
+    yield from bps.abs_set(
+        eiger.drv.detector.omega_start, detector_params.omega_start, group
+    )
+    yield from bps.abs_set(
+        eiger.drv.detector.omega_increment, detector_params.omega_increment, group
+    )
+
+    if wait:
+        yield from bps.wait(group)
+
+
+if __name__ == "__main__":
+    RE = RunEngine()
+    do_default_logging_setup()
+    eiger = fastcs_eiger(connect_immediately=True)
+    RE(
+        configure_arm_trigger_and_disarm_detector(
+            eiger=eiger,
+            detector_params=DetectorParams(
+                expected_energy_ev=12800,
+                exposure_time_s=0.01,
+                directory="/dls/i03/data/2025/cm40607-2/test_new_eiger/",
+                prefix="",
+                detector_distance=255,
+                omega_start=0,
+                omega_increment=0.1,
+                num_images_per_trigger=1,
+                num_triggers=1,
+                use_roi_mode=False,
+                det_dist_to_beam_converter_path="/dls_sw/i03/software/daq_configuration/lookup/DetDistToBeamXYConverter.txt",
+            ),
+            trigger_info=EigerTriggerInfo(
+                number_of_events=1,
+                energy_ev=12800,
+                trigger=DetectorTrigger.INTERNAL,
+                deadtime=0.0001,
+            ),
+        )
+    )

tests/conftest.py

@@ -1,12 +1,15 @@
 import importlib
 import os
+from pathlib import Path
 from types import ModuleType
 from unittest.mock import patch
 
 import pytest
 
 from conftest import mock_attributes_table
 from dodal.common.beamlines import beamline_parameters, beamline_utils
+from dodal.devices.detector import DetectorParams
+from dodal.devices.detector.det_dim_constants import EIGER2_X_16M_SIZE
 from dodal.utils import (
     DeviceInitializationController,
     collect_factories,
@@ -39,3 +42,22 @@ def mock_beamline_module_filepaths(bl_name: str, bl_module: ModuleType):
         beamline_parameters.BEAMLINE_PARAMETER_PATHS[bl_name] = (
             "tests/test_data/i04_beamlineParameters"
         )
+
+
+@pytest.fixture
+def eiger_params(tmp_path: Path) -> DetectorParams:
+    return DetectorParams(
+        expected_energy_ev=100.0,
+        exposure_time_s=1.0,
+        directory=str(tmp_path),
+        prefix="test",
+        run_number=0,
+        detector_distance=1.0,
+        omega_start=0.0,
+        omega_increment=1.0,
+        num_images_per_trigger=1,
+        num_triggers=2000,
+        use_roi_mode=False,
+        det_dist_to_beam_converter_path="tests/devices/unit_tests/test_lookup_table.txt",
+        detector_size_constants=EIGER2_X_16M_SIZE.det_type_string,  # type: ignore
+    )

tests/devices/unit_tests/test_eiger.py

@@ -1,6 +1,5 @@
 # type: ignore # Eiger will soon be ophyd-async https://github.com/DiamondLightSource/dodal/issues/700
 import threading
-from pathlib import Path
 from unittest.mock import ANY, MagicMock, Mock, call, create_autospec, patch
 
 import pytest
@@ -16,49 +15,19 @@
 from dodal.devices.util.epics_util import run_functions_without_blocking
 from dodal.log import LOGGER
 
-TEST_DETECTOR_SIZE_CONSTANTS = EIGER2_X_16M_SIZE
-
-TEST_EXPECTED_ENERGY = 100.0
-TEST_EXPOSURE_TIME = 1.0
 TEST_PREFIX = "test"
 TEST_RUN_NUMBER = 0
-TEST_DETECTOR_DISTANCE = 1.0
-TEST_OMEGA_START = 0.0
-TEST_OMEGA_INCREMENT = 1.0
-TEST_NUM_IMAGES_PER_TRIGGER = 1
-TEST_NUM_TRIGGERS = 2000
-TEST_USE_ROI_MODE = False
-TEST_DET_DIST_TO_BEAM_CONVERTER_PATH = "tests/devices/unit_tests/test_lookup_table.txt"
 
 
 class StatusException(Exception):
     pass
 
 
 @pytest.fixture
-def params(tmp_path: Path) -> DetectorParams:
-    return DetectorParams(
-        expected_energy_ev=TEST_EXPECTED_ENERGY,
-        exposure_time_s=TEST_EXPOSURE_TIME,
-        directory=str(tmp_path),
-        prefix=TEST_PREFIX,
-        run_number=TEST_RUN_NUMBER,
-        detector_distance=TEST_DETECTOR_DISTANCE,
-        omega_start=TEST_OMEGA_START,
-        omega_increment=TEST_OMEGA_INCREMENT,
-        num_images_per_trigger=TEST_NUM_IMAGES_PER_TRIGGER,
-        num_triggers=TEST_NUM_TRIGGERS,
-        use_roi_mode=TEST_USE_ROI_MODE,
-        det_dist_to_beam_converter_path=TEST_DET_DIST_TO_BEAM_CONVERTER_PATH,
-        detector_size_constants=TEST_DETECTOR_SIZE_CONSTANTS.det_type_string,
-    )
-
-
-@pytest.fixture
-def fake_eiger(request, params: DetectorParams):
+def fake_eiger(request, eiger_params: DetectorParams):
     FakeEigerDetector: EigerDetector = make_fake_device(EigerDetector)
     fake_eiger: EigerDetector = FakeEigerDetector.with_params(
-        params=params, name=f"test fake Eiger: {request.node.name}"
+        params=eiger_params, name=f"test fake Eiger: {request.node.name}"
     )
     return fake_eiger
 
@@ -255,8 +224,8 @@ def test_disable_roi_mode_sets_correct_roi_mode(fake_eiger):
 @pytest.mark.parametrize(
     "roi_mode, expected_detector_dimensions",
     [
-        (True, TEST_DETECTOR_SIZE_CONSTANTS.roi_size_pixels),
-        (False, TEST_DETECTOR_SIZE_CONSTANTS.det_size_pixels),
+        (True, EIGER2_X_16M_SIZE.roi_size_pixels),
+        (False, EIGER2_X_16M_SIZE.det_size_pixels),
     ],
 )
 def test_change_roi_mode_sets_correct_detector_size_constants(
@@ -733,10 +702,10 @@ def test_when_eiger_is_stopped_then_dev_shm_disabled(fake_eiger: EigerDetector):
     assert fake_eiger.odin.fan.dev_shm_enable.get() == 0
 
 
-def test_for_other_beamlines_i03_used_as_default(params: DetectorParams):
+def test_for_other_beamlines_i03_used_as_default(eiger_params: DetectorParams):
     FakeEigerDetector: EigerDetector = make_fake_device(EigerDetector)
     fake_eiger: EigerDetector = FakeEigerDetector.with_params(
-        params=params, beamline="ixx"
+        params=eiger_params, beamline="ixx"
     )
     assert fake_eiger.beamline == "ixx"
     assert fake_eiger.timeouts == AVAILABLE_TIMEOUTS["i03"]

tests/plans/test_configure_arm_trigger_and_disarm_detector.py

@@ -0,0 +1,66 @@
+from collections.abc import AsyncGenerator
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from bluesky.run_engine import RunEngine
+from ophyd_async.core import DetectorTrigger
+from ophyd_async.fastcs.eiger import EigerDetector as FastEiger
+from ophyd_async.fastcs.eiger import EigerTriggerInfo
+from ophyd_async.testing import callback_on_mock_put, set_mock_value
+
+from dodal.plans.configure_arm_trigger_and_disarm_detector import (
+    configure_arm_trigger_and_disarm_detector,
+)
+
+
+@pytest.fixture
+async def fake_eiger():
+    fake_eiger = FastEiger("", MagicMock())
+    await fake_eiger.connect(mock=True)
+    fake_eiger.drv.detector.arm.trigger = AsyncMock()
+    fake_eiger.drv.detector.disarm.trigger = AsyncMock()
+    fake_eiger._writer.observe_indices_written = fake_observe_indices_written
+    return fake_eiger
+
+
+async def fake_observe_indices_written(timeout: float) -> AsyncGenerator[int, None]:
+    yield 1
+
+
+async def test_configure_arm_trigger_and_disarm_detector(
+    fake_eiger, eiger_params, RE: RunEngine
+):
+    trigger_info = EigerTriggerInfo(
+        # Manual trigger, so setting number of triggers to 1.
+        number_of_events=1,
+        energy_ev=eiger_params.expected_energy_ev,
+        trigger=DetectorTrigger.INTERNAL,
+        deadtime=0.0001,
+    )
+
+    def set_meta_active(*args, **kwargs) -> None:
+        set_mock_value(fake_eiger.odin.meta_active, "Active")
+
+    def set_capture_rbv_meta_writing_and_detector_state(*args, **kwargs) -> None:
+        # Mimics capturing and immediete completion status on Eiger.
+        set_mock_value(fake_eiger.odin.capture_rbv, "Capturing")
+        set_mock_value(fake_eiger.odin.meta_writing, "Writing")
+        set_mock_value(fake_eiger.drv.detector.state, "idle")
+
+    callback_on_mock_put(fake_eiger.odin.num_to_capture, set_meta_active)
+    callback_on_mock_put(
+        fake_eiger.odin.capture, set_capture_rbv_meta_writing_and_detector_state
+    )
+
+    RE(
+        configure_arm_trigger_and_disarm_detector(
+            fake_eiger, eiger_params, trigger_info
+        )
+    )
+    fake_eiger.drv.detector.arm.trigger.assert_called_once()
+    # Disarm occurs at the start and end of the plan.
+    assert len(fake_eiger.drv.detector.disarm.trigger.call_args_list) == 2
+    assert (
+        await fake_eiger.drv.detector.photon_energy.get_value()
+        == eiger_params.expected_energy_ev
+    )