Add snapshots plan

pyproject.toml

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

src/crystallography_bluesky/i15_1/plans/__init__.py

@@ -4,6 +4,7 @@
     robot_load,
     robot_unload,
 )
+from .snapshots import take_snapshot
 from .take_i0_data import take_i0_data
 
 __all__ = [
@@ -12,4 +13,5 @@
     "prepare_beamline_for_robot_load",
     "robot_unload",
     "take_i0_data",
+    "take_snapshot",
 ]

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

@@ -0,0 +1,7 @@
+import pytest
+from bluesky import RunEngine
+
+
+@pytest.fixture
+def run_engine():
+    return RunEngine()

tests/unit_tests/i15_1/test_snapshots.py

@@ -0,0 +1,28 @@
+import pytest
+from bluesky import RunEngine
+from dodal.beamlines.i15_1 import cam_1
+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:
+    assert tmp_path.exists()
+    path_provider = StaticVisitPathProvider("i15_1", tmp_path)
+    path_provider._filename_provider.collectionId = DataCollectionIdentifier(
+        collectionNumber=0
+    )
+    async with init_devices(mock=True):
+        camera = cam_1(path_provider)
+
+    await camera.driver.image_mode.set(ADImageMode.CONTINUOUS)
+    await camera.driver.acquire.set(True)
+
+    return camera
+
+
+def test_take_snapshot_plan(run_engine: RunEngine, camera: ContAcqDetector):
+    run_engine(take_snapshot(camera))