Add snapshots plan

pyproject.toml

@@ -25,7 +25,7 @@ dependencies = [
     "scanspec",
     "deepdiff",
     "blueapi >= 1.0.0",
-    "ophyd-async[ca,pva]==v0.17a4",
+    "ophyd-async[ca,pva] @ git+https://github.com/bluesky/ophyd-async.git@add_cb_trigger",
     "bluesky >= 1.13.1",
     "daq-config-server>=1.3.1",
     "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@main",

src/crystallography_bluesky/i15_1/plans/__init__.py

@@ -4,12 +4,14 @@
     robot_load,
     robot_unload,
 )
+from .snapshots import take_snapshot
 from .static_collection import static_collection_plan
 
 __all__ = [
     "robot_load",
     "move_hexapod_to_home_position",
     "prepare_beamline_for_robot_load",
     "robot_unload",
+    "take_snapshot",
     "static_collection_plan",
 ]

src/crystallography_bluesky/i15_1/plans/snapshots.py

@@ -0,0 +1,13 @@
+import bluesky.plan_stubs as bps
+import bluesky.preprocessors as bpp
+from bluesky.utils import MsgGenerator
+from ophyd_async.core import TriggerInfo
+from ophyd_async.epics.adcore import ContAcqDetector
+
+
+@bpp.run_decorator()
+def take_snapshot(camera: ContAcqDetector) -> MsgGenerator:
+    yield from bps.stage(camera)
+    yield from bps.prepare(camera, TriggerInfo(number_of_events=1), wait=True)
+    yield from bps.trigger_and_read([camera])
+    yield from bps.unstage(camera)

tests/unit_tests/i15_1/conftest.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 
 import pytest
+from bluesky import RunEngine
 from dodal.devices.beamlines.i15_1.robot import Robot
 from dodal.devices.tetramm import TetrammDetector
 from dodal.devices.zebra.zebra import Zebra, ZebraMapping
@@ -9,6 +10,11 @@
 from ophyd_async.fastcs.eiger import EigerDetector
 
 
+@pytest.fixture
+def run_engine():
+    return RunEngine()
+
+
 @pytest.fixture
 def path_provider() -> StaticPathProvider:
     return StaticPathProvider(StaticFilenameProvider(""), Path(""))

tests/unit_tests/i15_1/test_snapshots.py

@@ -0,0 +1,59 @@
+from unittest.mock import AsyncMock
+
+import pytest
+from bluesky.simulators import RunEngineSimulator, assert_message_and_return_remaining
+from dodal.common.visit import DataCollectionIdentifier, StaticVisitPathProvider
+from ophyd_async.core import init_devices
+from ophyd_async.epics.adcore import ADImageMode, ContAcqDetector
+
+from crystallography_bluesky.i15_1.plans.snapshots import take_snapshot
+
+
+@pytest.fixture
+async def camera(tmp_path) -> ContAcqDetector:
+    path_provider = StaticVisitPathProvider("i15_1", tmp_path)
+    path_provider._filename_provider.collectionId = DataCollectionIdentifier(
+        collectionNumber=0
+    )
+    async with init_devices(mock=True):
+        camera = ContAcqDetector("", path_provider, name="cam_1")
+
+    await camera.driver.image_mode.set(ADImageMode.CONTINUOUS)
+    await camera.driver.acquire.set(True)
+
+    camera.writer.file_path_exists.get_value = AsyncMock(return_value=True)
+
+    return camera
+
+
+def test_take_snapshot_plan_makes_expected_calls(camera: ContAcqDetector):
+    run_engine = RunEngineSimulator()
+
+    msgs = run_engine.simulate_plan(take_snapshot(camera))
+
+    msgs = assert_message_and_return_remaining(
+        msgs, predicate=lambda msg: msg.command == "open_run"
+    )
+    msgs = assert_message_and_return_remaining(
+        msgs,
+        predicate=lambda msg: msg.command == "stage" and msg.obj.name == "cam_1",
+    )
+    msgs = assert_message_and_return_remaining(
+        msgs,
+        predicate=lambda msg: msg.command == "prepare" and msg.obj.name == "cam_1",
+    )
+    msgs = assert_message_and_return_remaining(
+        msgs,
+        predicate=lambda msg: msg.command == "trigger" and msg.obj.name == "cam_1",
+    )
+    msgs = assert_message_and_return_remaining(
+        msgs,
+        predicate=lambda msg: msg.command == "read" and msg.obj.name == "cam_1",
+    )
+    msgs = assert_message_and_return_remaining(
+        msgs,
+        predicate=lambda msg: msg.command == "unstage" and msg.obj.name == "cam_1",
+    )
+    msgs = assert_message_and_return_remaining(
+        msgs, predicate=lambda msg: msg.command == "close_run"
+    )