Rewrote plans to use single analyser detector with stronger tests

pyproject.toml

@@ -18,7 +18,8 @@ dependencies = [
     # it will be auto-pinned to the latest release version by the pre-release workflow
     #
     "bluesky",
-    "dls-dodal>=1.68.0",
+    #"my_bluesky @ file:///scratch/bluesky_development/my_bluesky",
+    "dls-dodal@git+https://github.com/DiamondLightSource/dodal.git@single_electron_analyser_detector",
     "ophyd-async[sim]",
     "scanspec",
 ]

src/sm_bluesky/electron_analyser/plan_stubs/analyser_per_step.py

@@ -1,25 +1,23 @@
-from collections.abc import Mapping, Sequence
-from typing import Any
+from collections.abc import Iterable, Mapping, Sequence
+from typing import Any, TypeVar
 
-from bluesky.plan_stubs import (
-    move_per_step,
-    trigger_and_read,
-)
+from bluesky.plan_stubs import move_per_step, mv, trigger_and_read
 from bluesky.protocols import Movable, Readable
 from bluesky.utils import (
     MsgGenerator,
     plan,
 )
-from dodal.devices.electron_analyser.base import (
-    ElectronAnalyserRegionDetector,
-    GenericElectronAnalyserRegionDetector,
-)
+from dodal.devices.electron_analyser.base import ElectronAnalyserDetector
 from dodal.log import LOGGER
 
+T = TypeVar("T")
 
-@plan
-def analyser_shot(detectors: Sequence[Readable], *args) -> MsgGenerator:
-    yield from analyser_nd_step(detectors, {}, {}, *args)
+
+def get_first_of_type(objects: Iterable[Any], target_type: type[T]) -> T:
+    for obj in objects:
+        if isinstance(obj, target_type):
+            return obj
+    raise ValueError(f"Cannot find object from {objects} with type {target_type}")
 
 
 @plan
@@ -46,14 +44,6 @@ def analyser_nd_step(
         mapping motors to their last-set positions
     """
 
-    analyser_detectors: list[GenericElectronAnalyserRegionDetector] = []
-    other_detectors = []
-    for det in detectors:
-        if isinstance(det, ElectronAnalyserRegionDetector):
-            analyser_detectors.append(det)
-        else:
-            other_detectors.append(det)
-
     # Step provides the map of motors to single position to move to. Move motors to
     # required positions.
     yield from move_per_step(step, pos_cache)
@@ -62,16 +52,20 @@ def analyser_nd_step(
     # them Readable so positions can be measured.
     motors: list[Readable] = [s for s in step.keys() if isinstance(s, Readable)]
 
-    # To get energy sources and open paired shutters, they need to be given in this
-    # plan. They could possibly come from step but we then have to extract them out.
-    # It would also mean forcefully adding in the devices at the wrapper level.
-    # It would easier if they are part of the detector and the plan just calls the
-    # common methods so it is more dynamic and configuration only for device.
-    for analyser_det in analyser_detectors:
-        dets = [analyser_det] + list(other_detectors) + list(motors)
-
-        LOGGER.info(f"Scanning region {analyser_det.region.name}.")
-        yield from trigger_and_read(
-            dets,
-            name=analyser_det.region.name,
-        )
+    readables = list(detectors) + motors
+
+    analyser = get_first_of_type(detectors, ElectronAnalyserDetector)
+
+    sequence = analyser.sequence_loader.sequence
+    if sequence is None:
+        raise ValueError(f"{analyser.sequence_loader.name}.sequence is None.")
+
+    for region in sequence.get_enabled_regions():
+        LOGGER.info(f"Scanning region {region.name}.")
+        yield from mv(analyser, region)
+        yield from trigger_and_read(readables, name=region.name)
+
+
+@plan
+def analyser_shot(detectors: Sequence[Readable], *args) -> MsgGenerator:
+    yield from analyser_nd_step(detectors, {}, {}, *args)

src/sm_bluesky/electron_analyser/plans/analyser_scans.py

@@ -1,73 +1,35 @@
 from collections.abc import Iterable, Sequence
 from typing import Any
 
-from bluesky.plans import PerShot, PerStep, count, grid_scan, scan
-from bluesky.protocols import (
-    Movable,
-    Readable,
-)
+from bluesky.plan_stubs import mv
+from bluesky.plans import count, grid_scan, scan
+from bluesky.protocols import Movable, Readable
 from bluesky.utils import (
     CustomPlanMetadata,
     MsgGenerator,
     ScalarOrIterableFloat,
     plan,
 )
-from dodal.devices.electron_analyser.base import (
-    ElectronAnalyserDetector,
-)
+from dodal.devices.electron_analyser.base import ElectronAnalyserDetector
 
 from sm_bluesky.electron_analyser.plan_stubs import (
     analyser_nd_step,
     analyser_shot,
 )
 
 
-def process_detectors_for_analyserscan(
-    detectors: Sequence[Readable],
-    sequence_file: str,
-) -> Sequence[Readable]:
-    """
-    Check for instance of ElectronAnalyserDetector in the detector list. Provide it with
-    sequence file to read and create list of ElectronAnalyserRegionDetector's. Replace
-    ElectronAnalyserDetector in list of detectors with the
-    list[ElectronAnalyserRegionDetector] and flatten.
-
-    Args:
-        detectors:
-            Devices to measure with for a scan.
-        sequence_file:
-            The file to read to create list[ElectronAnalyserRegionDetector].
-
-    Returns:
-        list of detectors, with any instances of ElectronAnalyserDetector replaced by
-        ElectronAnalyserRegionDetector by the number of regions in the sequence file.
-
-    """
-    analyser_detector = None
-    region_detectors = []
-    for det in detectors:
-        if isinstance(det, ElectronAnalyserDetector):
-            analyser_detector = det
-            region_detectors = det.create_region_detector_list(sequence_file)
-            break
-
-    expansions = (
-        region_detectors if e == analyser_detector else [e] for e in detectors
-    )
-    return [v for vals in expansions for v in vals]
-
-
 def analysercount(
-    detectors: Sequence[Readable],
+    analyser: ElectronAnalyserDetector,
     sequence_file: str,
+    detectors: Sequence[Readable],
     num: int = 1,
     delay: ScalarOrIterableFloat = 0.0,
     *,
-    per_shot: PerShot | None = None,
     md: CustomPlanMetadata | None = None,
 ) -> MsgGenerator:
+    yield from mv(analyser.sequence_loader, sequence_file)
     yield from count(
-        process_detectors_for_analyserscan(detectors, sequence_file),
+        list(detectors) + [analyser],
         num,
         delay,
         per_shot=analyser_shot,
@@ -77,15 +39,16 @@ def analysercount(
 
 @plan
 def analyserscan(
-    detectors: Sequence[Readable],
+    analyser: ElectronAnalyserDetector,
     sequence_file: str,
+    detectors: Sequence[Readable],
     *args: Movable | Any,
     num: int | None = None,
-    per_step: PerStep | None = None,
     md: CustomPlanMetadata | None = None,
 ) -> MsgGenerator:
+    yield from mv(analyser.sequence_loader, sequence_file)
     yield from scan(
-        process_detectors_for_analyserscan(detectors, sequence_file),
+        list(detectors) + [analyser],
         *args,
         num,
         per_step=analyser_nd_step,
@@ -95,14 +58,16 @@ def analyserscan(
 
 @plan
 def grid_analyserscan(
-    detectors: Sequence[Readable],
+    analyser: ElectronAnalyserDetector,
     sequence_file: str,
+    detectors: Sequence[Readable],
     *args,
     snake_axes: Iterable | bool | None = None,
     md: CustomPlanMetadata | None = None,
 ) -> MsgGenerator:
+    yield from mv(analyser.sequence_loader, sequence_file)
     yield from grid_scan(
-        process_detectors_for_analyserscan(detectors, sequence_file),
+        list(detectors) + [analyser],
         *args,
         snake_axes=snake_axes,
         per_step=analyser_nd_step,

tests/electron_analyser/conftest.py

@@ -6,6 +6,7 @@
 )
 from dodal.devices.electron_analyser.specs import SpecsDetector
 from dodal.devices.electron_analyser.vgscienta import VGScientaDetector
+from dodal.devices.selectable_source import SourceSelector
 from dodal.testing.electron_analyser import create_detector
 from ophyd_async.core import init_devices
 from ophyd_async.sim import SimMotor
@@ -14,29 +15,41 @@
     TEST_SPECS_SEQUENCE,
     TEST_VGSCIENTA_SEQUENCE,
 )
+from tests.electron_analyser.util import analyser_setup_for_scan
+
+
+@pytest.fixture
+async def dcm_energy() -> SimMotor:
+    with init_devices():
+        dcm_energy = SimMotor()
+    await dcm_energy.set(2200)
+    return dcm_energy
 
 
 @pytest.fixture
 async def pgm_energy() -> SimMotor:
     with init_devices():
         pgm_energy = SimMotor()
+    await pgm_energy.set(500)
     return pgm_energy
 
 
 @pytest.fixture
-async def dcm_energy() -> SimMotor:
-    with init_devices():
-        dcm_energy = SimMotor()
-    return dcm_energy
+async def source_selector() -> SourceSelector:
+    with init_devices(mock=True):
+        source_selector = SourceSelector()
+    return source_selector
 
 
 @pytest.fixture
 async def dual_energy_source(
-    dcm_energy: SimMotor, pgm_energy: SimMotor
+    dcm_energy: SimMotor, pgm_energy: SimMotor, source_selector: SourceSelector
 ) -> DualEnergySource:
     with init_devices():
         dual_energy_source = DualEnergySource(
-            dcm_energy.user_readback, pgm_energy.user_readback
+            dcm_energy.user_readback,
+            pgm_energy.user_readback,
+            source_selector.selected_source,
         )
     return dual_energy_source
 
@@ -49,14 +62,17 @@ async def dual_energy_source(
 )
 async def sim_analyser(
     request: pytest.FixtureRequest,
+    source_selector: SourceSelector,
     dual_energy_source: DualEnergySource,
 ) -> ElectronAnalyserDetector:
     with init_devices(mock=True):
         sim_analyser = create_detector(
             request.param,
             prefix="TEST:",
             energy_source=dual_energy_source,
+            source_selector=source_selector,
         )
+    analyser_setup_for_scan(sim_analyser)
     return sim_analyser