diff --git a/pyproject.toml b/pyproject.toml index 02f08728..a6a72025 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/src/sm_bluesky/electron_analyser/plan_stubs/analyser_per_step.py b/src/sm_bluesky/electron_analyser/plan_stubs/analyser_per_step.py index f9592042..4494c944 100644 --- a/src/sm_bluesky/electron_analyser/plan_stubs/analyser_per_step.py +++ b/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) diff --git a/src/sm_bluesky/electron_analyser/plans/analyser_scans.py b/src/sm_bluesky/electron_analyser/plans/analyser_scans.py index c9ba2bef..761c5f53 100644 --- a/src/sm_bluesky/electron_analyser/plans/analyser_scans.py +++ b/src/sm_bluesky/electron_analyser/plans/analyser_scans.py @@ -1,20 +1,16 @@ 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, @@ -22,52 +18,18 @@ ) -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, diff --git a/tests/electron_analyser/conftest.py b/tests/electron_analyser/conftest.py index 3b0a3e2f..e465b77f 100644 --- a/tests/electron_analyser/conftest.py +++ b/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,6 +62,7 @@ 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): @@ -56,7 +70,9 @@ async def sim_analyser( request.param, prefix="TEST:", energy_source=dual_energy_source, + source_selector=source_selector, ) + analyser_setup_for_scan(sim_analyser) return sim_analyser diff --git a/tests/electron_analyser/plan_stubs/test_analyser_per_step.py b/tests/electron_analyser/plan_stubs/test_analyser_per_step.py index a59bb185..c90bad2b 100644 --- a/tests/electron_analyser/plan_stubs/test_analyser_per_step.py +++ b/tests/electron_analyser/plan_stubs/test_analyser_per_step.py @@ -9,25 +9,27 @@ from bluesky import RunEngine from bluesky import plan_stubs as bps from bluesky.protocols import Movable, Readable, Triggerable -from dodal.devices.electron_analyser.base import ( - ElectronAnalyserDetector, - ElectronAnalyserRegionDetector, - GenericElectronAnalyserDetector, - GenericElectronAnalyserRegionDetector, -) +from dodal.devices.electron_analyser.base import GenericElectronAnalyserDetector from ophyd_async.core import AsyncStatus from ophyd_async.sim import SimMotor from sm_bluesky.electron_analyser.plan_stubs import analyser_per_step as aps -from tests.electron_analyser.util import analyser_setup_for_scan @pytest.fixture -def region_detectors( - sim_analyser: ElectronAnalyserDetector, sequence_file: str -) -> Sequence[ElectronAnalyserRegionDetector]: - analyser_setup_for_scan(sim_analyser) - return sim_analyser.create_region_detector_list(sequence_file) +async def analyser_with_sequence( + sim_analyser: GenericElectronAnalyserDetector, sequence_file: str +) -> GenericElectronAnalyserDetector: + await sim_analyser.sequence_loader.set(sequence_file) + assert sim_analyser.sequence_loader.sequence is not None + return sim_analyser + + +@pytest.fixture +def number_of_regions(analyser_with_sequence: GenericElectronAnalyserDetector) -> float: + sequence = analyser_with_sequence.sequence_loader.sequence + assert sequence is not None + return len(sequence.get_enabled_regions()) @pytest.fixture(params=[0, 1, 2]) @@ -39,9 +41,10 @@ def other_detectors( @pytest.fixture def all_detectors( - region_detectors: Sequence[Readable], other_detectors: Sequence[Readable] + analyser_with_sequence: GenericElectronAnalyserDetector, + other_detectors: Sequence[Readable], ) -> Sequence[Readable]: - return list(region_detectors) + list(other_detectors) + return [analyser_with_sequence] + list(other_detectors) @pytest.fixture @@ -83,32 +86,33 @@ def test_analyser_nd_step_func_has_expected_driver_set_calls( analyser_nd_step: Callable, all_detectors: Sequence[Readable], sim_analyser: GenericElectronAnalyserDetector, - region_detectors: Sequence[GenericElectronAnalyserRegionDetector], step: dict[Movable, Any], pos_cache: dict[Movable, Any], ) -> None: # Mock driver.set to track expected calls controller = sim_analyser._controller controller.setup_with_region = AsyncMock(side_effect=fake_status) - expected_driver_set_calls = [call(r_det.region) for r_det in region_detectors] + sequence = sim_analyser.sequence_loader.sequence + assert sequence is not None + expected_driver_set_calls = [ + call(region) for region in sequence.get_enabled_regions() + ] run_engine(analyser_nd_step(all_detectors, step, pos_cache)) - # Our driver instance is shared between each region detector instance. - # Check that each driver.set was called once with the correct region + # Check that controller method was called with the number of regions. assert controller.setup_with_region.call_args_list == expected_driver_set_calls async def test_analyser_nd_step_func_calls_detectors_trigger_and_read_correctly( run_engine: RunEngine, analyser_nd_step: Callable, + sim_analyser: GenericElectronAnalyserDetector, all_detectors: Sequence[Readable], - other_detectors: Sequence[Readable], - region_detectors: Sequence[GenericElectronAnalyserRegionDetector], step: dict[Movable, Any], pos_cache: dict[Movable, Any], ) -> None: - for det in other_detectors: + for det in all_detectors: if isinstance(det, Triggerable): det.trigger = MagicMock(side_effect=fake_status) @@ -118,22 +122,17 @@ async def test_analyser_nd_step_func_calls_detectors_trigger_and_read_correctly( else: det.read = MagicMock(return_value=det.read()) - for r_det in region_detectors: - r_det.trigger = MagicMock(side_effect=fake_status) - r_det.read = MagicMock(return_value=r_det.read()) - run_engine(analyser_nd_step(all_detectors, step, pos_cache)) - for r_det in region_detectors: - r_det.trigger.assert_called_once() # type: ignore - r_det.read.assert_called_once() # type: ignore + sequence = sim_analyser.sequence_loader.sequence + assert sequence is not None + n_regions = len(sequence.get_enabled_regions()) - # Check that the other detectors are triggered and read by the number of region - # detectors. - for det in other_detectors: + # Check that alldetectors are triggered and read by the number of regions. + for det in all_detectors: if isinstance(det, Triggerable): - assert det.trigger.call_count == len(region_detectors) # type: ignore - assert det.read.call_count == len(region_detectors) # type: ignore + assert det.trigger.call_count == n_regions # type: ignore + assert det.read.call_count == n_regions # type: ignore async def test_analyser_nd_step_func_moves_motors_before_detector_trigger( diff --git a/tests/electron_analyser/plans/test_analyser_scans.py b/tests/electron_analyser/plans/test_analyser_scans.py index 1044b28e..018d0d53 100644 --- a/tests/electron_analyser/plans/test_analyser_scans.py +++ b/tests/electron_analyser/plans/test_analyser_scans.py @@ -1,13 +1,13 @@ -from collections.abc import Sequence +import math +from collections.abc import Mapping, Sequence import pytest from bluesky import RunEngine -from bluesky.protocols import Readable +from bluesky.protocols import Readable, Reading from dodal.devices.electron_analyser.base import ( - ElectronAnalyserDetector, - ElectronAnalyserRegionDetector, + AbstractEnergySource, + DualEnergySource, GenericElectronAnalyserDetector, - GenericElectronAnalyserRegionDetector, ) from ophyd_async.sim import SimMotor @@ -15,65 +15,113 @@ analysercount, analyserscan, grid_analyserscan, - process_detectors_for_analyserscan, ) -from tests.electron_analyser.util import analyser_setup_for_scan +from tests.electron_analyser.util import ( + assert_mapped_data_equals_expected, + expected_analyser_config, +) -@pytest.fixture(params=[0, 1, 2]) -def extra_detectors( - request: pytest.FixtureRequest, -) -> list[Readable]: - return [SimMotor("det" + str(i + 1)) for i in range(request.param)] +def add_energy_source_monitor(energy_source: AbstractEnergySource) -> list[float]: + energy_values = [] + def energy_monitor(reading: dict[str, Reading[float]], *args, **kwargs) -> None: + value = reading[energy_source.energy.name]["value"] + energy_values.append(value) -@pytest.fixture -def all_detectors( - sim_analyser: ElectronAnalyserDetector, extra_detectors: list[Readable] -) -> Sequence[Readable]: - return [sim_analyser] + extra_detectors + energy_source.energy.subscribe_reading(energy_monitor) + return energy_values -async def test_process_detectors_for_analyserscan_func_correctly_replaces_detectors( - sequence_file: str, - sim_analyser: GenericElectronAnalyserDetector, - extra_detectors: Sequence[Readable], - all_detectors: Sequence[Readable], +def assert_analyserscan_config( + run_engine_documents: Mapping[str, list[dict[str, Reading]]], + analyser: GenericElectronAnalyserDetector, + energy_values: list[float], ) -> None: - sequence = sim_analyser.load_sequence(sequence_file) + """Check that the configuration for the analyser device is correct.""" + drv = analyser._controller.driver + + sequence = analyser.sequence_loader.sequence + assert sequence is not None - analyserscan_detectors: Sequence[Readable] = process_detectors_for_analyserscan( - all_detectors, sequence_file + configuration_region_names = [] + + for i, descriptor in enumerate(run_engine_documents["descriptor"]): + analyser_config = descriptor["configuration"][analyser.name]["data"] + + region_name = analyser_config[drv.region_name.name] + configuration_region_names.append(region_name) + + region = sequence.get_region_by_name(region_name) + assert region is not None + + epics_region = region.prepare_for_epics(energy_values[i]) + + assert_mapped_data_equals_expected( + analyser_config, expected_analyser_config(drv, epics_region) + ) + + assert configuration_region_names == sequence.get_enabled_region_names(), ( + "The saved region names are not same as the sequence region names!" ) - # Check analyser detector is removed from detector list - assert sim_analyser not in analyserscan_detectors - # Check all extra detectors are still present in detector list - for extra_det in extra_detectors: - assert extra_det in analyserscan_detectors - region_detectors: list[GenericElectronAnalyserRegionDetector] = [ - ad - for ad in analyserscan_detectors - if isinstance(ad, ElectronAnalyserRegionDetector) - ] - # Check length of region_detectors list is length of sequence enabled regions - assert len(region_detectors) == len(sequence.get_enabled_region_names()) +def assert_other_devices_config( + run_engine_documents: Mapping[str, list[dict[str, Reading]]], + extra_detectors: Sequence[Readable], + args: list[SimMotor | int], +) -> None: + motors = [a for a in args if isinstance(a, SimMotor)] + for descriptor in run_engine_documents["descriptor"]: + for m in motors: + assert descriptor["configuration"][m.name]["data"] + for d in extra_detectors: + assert descriptor["configuration"][d.name]["data"] + + +def assert_analyser_event_data( + run_engine_documents: Mapping[str, list[dict[str, Reading]]], + analyser: GenericElectronAnalyserDetector, + motor_iterations: int, +) -> None: + sequence = analyser.sequence_loader.sequence + assert sequence is not None + number_of_regions = sequence.get_enabled_regions() + assert ( + len(run_engine_documents["event"]) == len(number_of_regions) * motor_iterations + ) + drv = analyser._controller.driver + + for event in run_engine_documents["event"]: + assert drv.spectrum.name in event["data"] + assert drv.image.name in event["data"] + assert drv.total_intensity.name in event["data"] + - # ToDo - We cannot compare that the region detectors are the same without override - # equals method. For now, just compare that region name is the same. - for region_det in region_detectors: - assert region_det.region.name in sequence.get_enabled_region_names() +@pytest.fixture(params=[0, 1, 2]) +def extra_detectors( + request: pytest.FixtureRequest, +) -> list[Readable]: + return [SimMotor("det" + str(i + 1)) for i in range(request.param)] async def test_analysercount( run_engine: RunEngine, - sim_analyser: ElectronAnalyserDetector, + run_engine_documents: Mapping[str, list[dict[str, Reading]]], + sim_analyser: GenericElectronAnalyserDetector, sequence_file: str, - all_detectors: Sequence[Readable], + extra_detectors: Sequence[Readable], + dual_energy_source: DualEnergySource, ) -> None: - analyser_setup_for_scan(sim_analyser) - run_engine(analysercount(all_detectors, sequence_file)) + energy_monitor_values = add_energy_source_monitor(dual_energy_source) + run_engine(analysercount(sim_analyser, sequence_file, extra_detectors)) + assert_analyserscan_config( + run_engine_documents, + sim_analyser, + energy_monitor_values, + ) + assert_other_devices_config(run_engine_documents, extra_detectors, []) + assert_analyser_event_data(run_engine_documents, sim_analyser, 1) @pytest.mark.parametrize( @@ -85,28 +133,55 @@ async def test_analysercount( ) async def test_analyserscan( run_engine: RunEngine, - sim_analyser: ElectronAnalyserDetector, + run_engine_documents: Mapping[str, list[dict[str, Reading]]], + sim_analyser: GenericElectronAnalyserDetector, sequence_file: str, - all_detectors: Sequence[Readable], + extra_detectors: Sequence[Readable], args: list[SimMotor | int], + dual_energy_source: DualEnergySource, ) -> None: - analyser_setup_for_scan(sim_analyser) - run_engine(analyserscan(all_detectors, sequence_file, *args, num=10)) + energy_monitor_values = add_energy_source_monitor(dual_energy_source) + motor_iterations = 3 + run_engine( + analyserscan( + sim_analyser, sequence_file, extra_detectors, *args, num=motor_iterations + ) + ) + assert_analyserscan_config( + run_engine_documents, + sim_analyser, + energy_monitor_values, + ) + assert_other_devices_config(run_engine_documents, extra_detectors, args) + assert_analyser_event_data(run_engine_documents, sim_analyser, motor_iterations) @pytest.mark.parametrize( "args", [ - [SimMotor("motor1"), 1, 10, 1], - [SimMotor("motor1"), 1, 10, 1, SimMotor("motor2"), 1, 5, 1], + [SimMotor("motor1"), 1, 3, 3], + [SimMotor("motor1"), 1, 3, 3, SimMotor("motor2"), 1, 2, 2], ], ) async def test_grid_analyserscan( run_engine: RunEngine, - sim_analyser: ElectronAnalyserDetector, + run_engine_documents: Mapping[str, list[dict[str, Reading]]], + sim_analyser: GenericElectronAnalyserDetector, sequence_file: str, - all_detectors: Sequence[Readable], + extra_detectors: Sequence[Readable], args: list[SimMotor | int], + dual_energy_source: DualEnergySource, ) -> None: - analyser_setup_for_scan(sim_analyser) - run_engine(grid_analyserscan(all_detectors, sequence_file, *args)) + energy_monitor_values = add_energy_source_monitor(dual_energy_source) + run_engine(grid_analyserscan(sim_analyser, sequence_file, extra_detectors, *args)) + assert_analyserscan_config( + run_engine_documents, + sim_analyser, + energy_monitor_values, + ) + assert_other_devices_config(run_engine_documents, extra_detectors, args) + + # For args, start at index 3, get every 4th value + dimensions: list[int] = [v for v in args[3::4] if isinstance(v, int)] + motor_iterations = math.prod(dimensions) + assert_analyser_event_data(run_engine_documents, sim_analyser, motor_iterations) diff --git a/tests/electron_analyser/util.py b/tests/electron_analyser/util.py index cf9dd586..1a5c8117 100644 --- a/tests/electron_analyser/util.py +++ b/tests/electron_analyser/util.py @@ -1,7 +1,21 @@ +from typing import Any +from unittest.mock import ANY + from dodal.devices.electron_analyser.base import ( ElectronAnalyserDetector, + GenericAnalyserDriverIO, + GenericRegion, +) +from dodal.devices.electron_analyser.specs import AcquisitionMode as SpecsAcqusitionMode +from dodal.devices.electron_analyser.specs import ( + SpecsAnalyserDriverIO, + SpecsDetector, + SpecsRegion, +) +from dodal.devices.electron_analyser.vgscienta import ( + VGScientaAnalyserDriverIO, + VGScientaRegion, ) -from dodal.devices.electron_analyser.specs import SpecsDetector from ophyd_async.core import set_mock_value @@ -14,3 +28,92 @@ def analyser_setup_for_scan(sim_analyser: ElectronAnalyserDetector): set_mock_value(sim_analyser.driver.slices, dummy_val) set_mock_value(sim_analyser.driver.low_energy, dummy_val) set_mock_value(sim_analyser.driver.high_energy, dummy_val) + + +def expected_analyser_config( + drv: GenericAnalyserDriverIO, + epics_region: GenericRegion, +) -> dict[str, Any]: + if isinstance(drv, VGScientaAnalyserDriverIO) and isinstance( + epics_region, VGScientaRegion + ): + return expected_vgscienta_analyser_config(drv, epics_region) + elif isinstance(drv, SpecsAnalyserDriverIO) and isinstance( + epics_region, SpecsRegion + ): + return expected_specs_analyser_config(drv, epics_region) + else: + raise TypeError( + f"Not a valid type for driver {type(drv)} and region {type(epics_region)} " + ) + + +def expected_vgscienta_analyser_config( + drv: VGScientaAnalyserDriverIO, + epics_region: VGScientaRegion, +) -> dict[str, Any]: + return { + drv.region_name.name: epics_region.name, + drv.low_energy.name: epics_region.low_energy, + drv.centre_energy.name: epics_region.centre_energy, + drv.high_energy.name: epics_region.high_energy, + drv.slices.name: epics_region.slices, + drv.lens_mode.name: epics_region.lens_mode, + drv.pass_energy.name: epics_region.pass_energy, + drv.iterations.name: epics_region.iterations, + drv.acquire_time.name: epics_region.acquire_time, + drv.acquisition_mode.name: epics_region.acquisition_mode, + drv.energy_step.name: epics_region.energy_step, + drv.detector_mode.name: epics_region.detector_mode, + drv.region_min_x.name: epics_region.min_x, + drv.region_size_x.name: epics_region.size_x, + drv.region_min_y.name: epics_region.min_y, + drv.region_size_y.name: epics_region.size_y, + drv.energy_mode.name: epics_region.energy_mode, + } + + +def expected_specs_analyser_config( + drv: SpecsAnalyserDriverIO, + epics_region: SpecsRegion, +) -> dict[str, Any]: + if epics_region.acquisition_mode == SpecsAcqusitionMode.FIXED_TRANSMISSION: + energy_step = epics_region.energy_step + else: + energy_step = ANY + + if epics_region.acquisition_mode == SpecsAcqusitionMode.FIXED_ENERGY: + centre_energy = epics_region.centre_energy + else: + centre_energy = ANY + + return { + drv.region_name.name: epics_region.name, + drv.low_energy.name: epics_region.low_energy, + drv.high_energy.name: epics_region.high_energy, + drv.slices.name: epics_region.slices, + drv.acquire_time.name: epics_region.acquire_time, + drv.lens_mode.name: epics_region.lens_mode, + drv.pass_energy.name: epics_region.pass_energy, + drv.iterations.name: epics_region.iterations, + drv.acquisition_mode.name: epics_region.acquisition_mode, + drv.snapshot_values.name: epics_region.values, + drv.psu_mode.name: epics_region.psu_mode, + drv.energy_mode.name: epics_region.energy_mode, + drv.slices.name: epics_region.slices, + drv.energy_mode.name: epics_region.energy_mode, + drv.psu_mode.name: epics_region.psu_mode, + drv.snapshot_values.name: epics_region.values, + drv.energy_step.name: energy_step, + drv.centre_energy.name: centre_energy, + } + + +def assert_mapped_data_equals_expected( + data: dict[str, Any], expected: dict[str, Any], skip_expected_is_none: bool = True +) -> None: + for key, exp in expected.items(): + assert key in data + if skip_expected_is_none and exp is None: + continue + assert data[key] == exp diff --git a/uv.lock b/uv.lock index f7760b39..7e3f5081 100644 --- a/uv.lock +++ b/uv.lock @@ -1198,8 +1198,8 @@ wheels = [ [[package]] name = "dls-dodal" -version = "1.68.0" -source = { registry = "https://pypi.org/simple" } +version = "1.68.1.dev23+g44cade940" +source = { git = "https://github.com/DiamondLightSource/dodal.git?rev=single_electron_analyser_detector#44cade940879f04154268f256a63b3c9e3f10c84" } dependencies = [ { name = "aiofiles" }, { name = "aiohttp" }, @@ -1221,10 +1221,6 @@ dependencies = [ { name = "scanspec" }, { name = "zocalo" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/8f/a2/5e9b52b7e9c33a49d7ed498de811a10cf50f9d3e78414c9cea234ed8c98a/dls_dodal-1.68.0.tar.gz", hash = "sha256:e24b1c555c0e02e90fab327ed1024480555eecc0ca74f90389d6ac8519f97794", size = 1345787, upload-time = "2025-12-18T10:49:23.995Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/47/69/d16626e7cb65d6baa8022abb13acbdf1655cc03886f77fb00215646f35b0/dls_dodal-1.68.0-py3-none-any.whl", hash = "sha256:7a20c9b0140ea5363f906d67e762d1f24f382b4738682022a962b390446dabd5", size = 337957, upload-time = "2025-12-18T10:49:22.527Z" }, -] [[package]] name = "dnspython" @@ -5355,7 +5351,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "bluesky" }, - { name = "dls-dodal", specifier = ">=1.68.0" }, + { name = "dls-dodal", git = "https://github.com/DiamondLightSource/dodal.git?rev=single_electron_analyser_detector" }, { name = "ophyd-async", extras = ["sim"] }, { name = "scanspec" }, ]