Replace use of deprecated dodal plan writing utils (#99)

Co-authored-by: Stanislaw Malinowski <[email protected]>

pyproject.toml

@@ -12,10 +12,10 @@ classifiers = [
 ]
 description = "Plans and behaviours specific to the i22 beamline at DiamondLightSource."
 dependencies = [
-    "dls_dodal>=1.32",
-    "ophyd_async>=0.6,<0.7",
+    "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@main",
+    "ophyd_async",
     "numpy",
-    "pydantic>=2.0",
+    "pydantic",
 ] # Add project dependencies here, e.g. ["click", "numpy"]
 dynamic = ["version"]
 license.file = "LICENSE"

src/i22_bluesky/panda/fly_scanning.py

@@ -1,9 +1,9 @@
 import asyncio
 from dataclasses import dataclass
 
-from dls_bluesky_core.core import in_micros
+from dodal.common.maths import in_micros
 from ophyd_async.core import DetectorTrigger, TriggerInfo, TriggerLogic, wait_for_value
-from ophyd_async.fastcs.panda import SeqBlock, SeqTable
+from ophyd_async.fastcs.panda import SeqBlock, SeqTable, TimeUnits
 
 
 @dataclass
@@ -51,7 +51,7 @@ async def prepare(self, value: RepeatedTrigger):
             + SeqTable.row(time2=in_micros(self.shutter_time)),
         )
         await asyncio.gather(
-            self.seq.prescale_units.set("us"),
+            self.seq.prescale_units.set(TimeUnits.us),
             self.seq.enable.set("ZERO"),
         )
         await asyncio.gather(

src/i22_bluesky/plans/linkam.py

@@ -1,7 +1,7 @@
 from typing import Annotated, Any
 
 import bluesky.preprocessors as bpp
-from dodal.common import MsgGenerator, inject
+from bluesky.utils import MsgGenerator
 from dodal.common.maths import step_to_num
 from dodal.devices.linkam3 import Linkam3
 from dodal.plans.data_session_metadata import attach_data_session_metadata_decorator
@@ -14,20 +14,19 @@
     LinkamTrajectory,
     capture_linkam_segment,
 )
-from i22_bluesky.util.baseline import (
-    DEFAULT_DETECTORS,
-    DEFAULT_LINKAM,
-    DEFAULT_PANDA,
+from i22_bluesky.util.default_devices import (
+    DETECTORS,
+    LINKAM,
+    PANDA,
+    STAMPED_DETECTOR,
 )
 from i22_bluesky.util.settings import (
     get_device_save_dir,
     stamp_temp_pv,
 )
 
-DEFAULT_STAMPED_DETECTOR: StandardDetector = inject("saxs")
 
-
-def save_linkam(panda: HDFPanda = DEFAULT_PANDA) -> MsgGenerator:
+def save_linkam(panda: HDFPanda = PANDA) -> MsgGenerator:
     yield from save_device(
         panda,
         get_device_save_dir(linkam_plan.__name__),
@@ -39,20 +38,20 @@ def save_linkam(panda: HDFPanda = DEFAULT_PANDA) -> MsgGenerator:
 @validate_call(config={"arbitrary_types_allowed": True})
 def linkam_plan(
     trajectory: Annotated[LinkamTrajectory, "Trajectory for the scan to follow."],
-    linkam: Annotated[Linkam3, "Temperature controller."] = DEFAULT_LINKAM,
+    linkam: Annotated[Linkam3, "Temperature controller."] = LINKAM,
     panda: Annotated[
         HDFPanda,
         "Panda with sequence table configured and connected to \
         FastShutter (outa) and each of detectors (outb).",
-    ] = DEFAULT_PANDA,
+    ] = PANDA,
     stamped_detector: Annotated[
         StandardDetector,
         "AreaDetector to configure to stamp the Linkam temperature. \
             Will be automatically added to detectors if not included.",
-    ] = DEFAULT_STAMPED_DETECTOR,
+    ] = STAMPED_DETECTOR,
     detectors: Annotated[
         set[StandardDetector], "Detectors to capture at each temperature value"
-    ] = DEFAULT_DETECTORS,
+    ] = DETECTORS,
     shutter_time: Annotated[
         float, "Time allowed for opening shutter before triggering detectors."
     ] = 0.04,
@@ -62,28 +61,13 @@ def linkam_plan(
     """
     Follow a trajectory in temperature, collecting a number of frames either at equally
     spaced positions or while continually scanning. e.g. for 2 segments, the first
-    stepped and the 2nd flown:
-    trajectory start   v             v final segment stop
-                       \\           /
-       stepped segment__\\__       /
-                           \\     /  flown segment
-           1st segment stop \\__ /
+    stepped and the 2nd flown:\n
+    trajectory start   v             v final segment stop\n
+                       \\           /\n
+       stepped segment__\\__       /\n
+                           \\     /  flown segment\n
+           1st segment stop \\__ /\n
         exposures:    xx  xx  xx   1/N seconds
-    Args:
-        start_temp: Initial temperature to reach before starting experiment
-        trajectory: Trajectory to follow: each segment begins at the end of the previous
-        num_frames: Default number of frames at each captured point
-        exposure: Default exposure for each frame
-        linkam: Linkam temperature stage
-        panda: PandA for controlling flyable motion
-        stamped_detector: Detector to stamp temperature PV to H5 file
-        detectors: Other StandardDetectors to capture
-
-    Returns:
-        MsgGenerator: Plan
-
-    Yields:
-        Iterator[MsgGenerator]: Bluesky messages
     """
     flyer = StandardFlyer(StaticSeqTableTriggerLogic(panda.seq[1]))
     detectors = detectors | {stamped_detector}

src/i22_bluesky/plans/stopflow.py

@@ -23,7 +23,7 @@
 import bluesky.plans as bp
 import bluesky.preprocessors as bpp
 from bluesky.protocols import Readable
-from dodal.common import MsgGenerator
+from bluesky.utils import MsgGenerator
 from dodal.devices.tetramm import TetrammDetector
 from dodal.plans.data_session_metadata import attach_data_session_metadata_decorator
 from ophyd_async.core import (
@@ -44,11 +44,11 @@
 )
 from ophyd_async.plan_stubs import fly_and_collect
 
-from i22_bluesky.util.baseline import (
-    DEFAULT_BASELINE_MEASUREMENTS,
-    DEFAULT_DETECTORS,
-    DEFAULT_PANDA,
+from i22_bluesky.util.default_devices import (
+    BASELINE_DEVICES,
+    DETECTORS,
     FAST_DETECTORS,
+    PANDA,
 )
 from i22_bluesky.util.settings import get_device_save_dir
 
@@ -60,7 +60,7 @@
 @attach_data_session_metadata_decorator()
 def check_detectors_for_stopflow(
     num_frames: int = 1,
-    devices: set[Readable] = DEFAULT_DETECTORS | DEFAULT_BASELINE_MEASUREMENTS,
+    devices: set[Readable] = DETECTORS | BASELINE_DEVICES,
 ) -> MsgGenerator:
     """
     Take a reading from all devices that are used in the
@@ -78,9 +78,9 @@ def check_detectors_for_stopflow(
 
 
 def check_stopflow_assembly(
-    panda: HDFPanda = DEFAULT_PANDA,
-    detectors: set[StandardDetector] = DEFAULT_DETECTORS,
-    baseline: set[Readable] = DEFAULT_BASELINE_MEASUREMENTS,
+    panda: HDFPanda = PANDA,
+    detectors: set[StandardDetector] = DETECTORS,
+    baseline: set[Readable] = BASELINE_DEVICES,
 ) -> MsgGenerator:
     """
     Simplified version of the stopflow plan that should catch most
@@ -100,9 +100,9 @@ def check_stopflow_assembly(
 
 
 def check_stopflow_experiment(
-    panda: HDFPanda = DEFAULT_PANDA,
-    detectors: set[StandardDetector] = DEFAULT_DETECTORS,
-    baseline: set[Readable] = DEFAULT_BASELINE_MEASUREMENTS,
+    panda: HDFPanda = PANDA,
+    detectors: set[StandardDetector] = DETECTORS,
+    baseline: set[Readable] = BASELINE_DEVICES,
 ) -> MsgGenerator:
     """
     Full test of stopflow experiment functionality with sensible values
@@ -123,9 +123,9 @@ def stress_test_stopflow(
     exposure: float = 1.0 / 250.0,
     post_stop_frames: int = 2000,
     pre_stop_frames: int = 8000,
-    panda: HDFPanda = DEFAULT_PANDA,
+    panda: HDFPanda = PANDA,
     detectors: set[StandardDetector] = FAST_DETECTORS,
-    baseline: set[Readable] = DEFAULT_BASELINE_MEASUREMENTS,
+    baseline: set[Readable] = BASELINE_DEVICES,
 ) -> MsgGenerator:
     yield from stopflow(
         exposure=exposure,
@@ -138,7 +138,7 @@ def stress_test_stopflow(
     )
 
 
-def save_stopflow(panda: HDFPanda = DEFAULT_PANDA) -> MsgGenerator:
+def save_stopflow(panda: HDFPanda = PANDA) -> MsgGenerator:
     yield from save_device(
         panda,
         get_device_save_dir(stopflow.__name__),
@@ -151,9 +151,9 @@ def stopflow(
     post_stop_frames: int,
     pre_stop_frames: int = 0,
     shutter_time: float = 4e-3,
-    panda: HDFPanda = DEFAULT_PANDA,
-    detectors: set[StandardDetector] = DEFAULT_DETECTORS,
-    baseline: set[Readable] = DEFAULT_BASELINE_MEASUREMENTS,
+    panda: HDFPanda = PANDA,
+    detectors: set[StandardDetector] = DETECTORS,
+    baseline: set[Readable] = BASELINE_DEVICES,
     metadata: dict[str, Any] | None = None,
 ) -> MsgGenerator:
     """

src/i22_bluesky/stubs/linkam.py

@@ -2,7 +2,7 @@
 
 import bluesky.plan_stubs as bps
 import numpy as np
-from dodal.common import MsgGenerator
+from bluesky.utils import MsgGenerator
 from dodal.common.coordination import group_uuid
 from dodal.devices.linkam3 import Linkam3
 from ophyd_async.core import StandardDetector, StandardFlyer

src/i22_bluesky/util/default_devices.py

@@ -0,0 +1,53 @@
+from bluesky.protocols import Readable
+from dodal.beamlines.i22 import (
+    dcm,
+    fswitch,
+    hfm,
+    i0,
+    it,
+    linkam,
+    oav,
+    panda1,
+    saxs,
+    slits_1,
+    slits_2,
+    slits_3,
+    slits_4,
+    slits_5,
+    slits_6,
+    synchrotron,
+    undulator,
+    vfm,
+    waxs,
+)
+from dodal.devices.linkam3 import Linkam3
+from ophyd_async.core import StandardDetector
+from ophyd_async.fastcs.panda import HDFPanda
+
+FAST_DETECTORS: set[StandardDetector] = {
+    saxs(),
+    waxs(),
+    i0(),
+    it(),
+}
+
+DETECTORS: set[StandardDetector] = FAST_DETECTORS | {oav()}
+
+BASELINE_DEVICES: set[Readable] = {
+    fswitch(),
+    slits_1(),
+    slits_2(),
+    slits_3(),
+    slits_4(),
+    slits_5(),
+    slits_6(),
+    hfm(),
+    vfm(),
+    undulator(),
+    dcm(),
+    synchrotron(),
+}
+
+PANDA: HDFPanda = panda1()
+LINKAM: Linkam3 = linkam()
+STAMPED_DETECTOR: StandardDetector = saxs()

tests/plans/test_stopflow.py

@@ -1,4 +1,5 @@
 import asyncio
+from collections.abc import Iterable
 from unittest.mock import Mock, patch
 
 import numpy as np
@@ -296,11 +297,20 @@ def test_stopflow_seq_table(
     generated_seq_table: SeqTable,
     expected_seq_table: SeqTable,
 ):
-    for element in generated_seq_table.__dict__:
-        assert all(
-            getattr(expected_seq_table, element)
-            == getattr(generated_seq_table, element)
-        )
+    for attr_name in generated_seq_table.__dict__:
+        generated_value = getattr(generated_seq_table, attr_name)
+        expected_value = getattr(expected_seq_table, attr_name)
+
+        # Check if the attribute is iterable and not a string or bytes
+        if isinstance(generated_value, Iterable) and not isinstance(
+            generated_value, str | bytes
+        ):
+            assert all(
+                g == e for g, e in zip(generated_value, expected_value, strict=False)
+            ), f"Mismatch in {attr_name}"
+        else:
+            # Direct comparison for non-iterable attributes like booleans
+            assert generated_value == expected_value, f"Mismatch in {attr_name}"
 
 
 @pytest.mark.xfail(reason="Strange import behavior, to be investigated")