diff --git a/src/hextools/gui/__main__.py b/src/hextools/gui/__main__.py index 2718200..cea32ff 100644 --- a/src/hextools/gui/__main__.py +++ b/src/hextools/gui/__main__.py @@ -114,6 +114,7 @@ ) from hextools.tomography.radiography import take_radiograph from hextools.utils import is_running_in_ci, reset_scan_id +from hextools.xrd import xrd_calibration, xrd_count, xrd_dark, xrd_scan from ._event_loop import get_our_app_name, gui_qt from ._threading import wait_for_workers_to_quit @@ -351,16 +352,28 @@ def __init__(self, re_client: RunEngineClient | RunEngine, *args, **kwargs): class QtXRDView(QWidget): - """X-ray diffraction view: the live Perkin Elmer area detector viewer.""" + """X-ray diffraction view: the live Perkin Elmer viewer and XRD plans.""" def __init__(self, re_client: RunEngineClient | RunEngine, *args, **kwargs): super().__init__(*args, **kwargs) self._re_client = re_client - vbox = QVBoxLayout() + top = QHBoxLayout() detectors = QtTabbedDetectorsWidget(re_client) detectors.add_detector("perkin_elmer", "XF:27ID1-ES{PE-Det:1}") - vbox.addWidget(detectors, stretch=1) + top.addWidget(detectors, stretch=3) + + plan_tabs = QTabWidget() + plan_tabs.addTab(QtPlanWidget(re_client, xrd_count), "Count") + plan_tabs.addTab(QtPlanWidget(re_client, xrd_scan), "Scan") + plan_tabs.addTab(QtPlanWidget(re_client, xrd_dark), "Dark") + plan_tabs.addTab(QtPlanWidget(re_client, xrd_calibration), "Calibration") + side = QVBoxLayout() + side.addWidget(plan_tabs, stretch=1) + top.addLayout(side, stretch=1) + + vbox = QVBoxLayout() + vbox.addLayout(top, stretch=1) self.setLayout(vbox) diff --git a/src/hextools/profiles/collection.py b/src/hextools/profiles/collection.py index 52bd741..ab630b7 100644 --- a/src/hextools/profiles/collection.py +++ b/src/hextools/profiles/collection.py @@ -55,6 +55,7 @@ print_proposal_info, print_version_info, ) +from hextools.xrd import * # Adjust the default timeout for ophyd async to 60s DEFAULT_TIMEOUT = 60.0 diff --git a/src/hextools/xrd/__init__.py b/src/hextools/xrd/__init__.py index 86b23d7..ae6558a 100644 --- a/src/hextools/xrd/__init__.py +++ b/src/hextools/xrd/__init__.py @@ -1 +1,12 @@ """X-Ray Diffraction (XRD) tools for HEX beamline.""" + +from .calibration import xrd_calibration +from .plans import ensure_cont_acq_exposure, xrd_count, xrd_dark, xrd_scan + +__all__ = [ + "xrd_calibration", + "xrd_count", + "xrd_dark", + "xrd_scan", + "ensure_cont_acq_exposure", +] diff --git a/src/hextools/xrd/calibration.py b/src/hextools/xrd/calibration.py index 8b495a0..3b0bcc1 100644 --- a/src/hextools/xrd/calibration.py +++ b/src/hextools/xrd/calibration.py @@ -5,11 +5,12 @@ from bluesky import Msg from bluesky import plan_stubs as bps from bluesky import plans as bp -from ophyd_async.epics.adcore import AreaDetector +from ophyd_async.epics.adcore import AreaDetector, ContAcqDetector from ophyd_async.epics.adcore._io import ADBaseIOT from ophyd_async.epics.motor import Motor as AsyncEpicsMotor from hextools.photon_delivery_system import Shutter +from hextools.xrd.plans import ensure_cont_acq_exposure def xrd_calibration( @@ -26,17 +27,18 @@ def xrd_calibration( if start_position is None: start_position: float = yield from bps.rd(motor.user_readback) - yield from bps.abs_set( - detector.driver.acquire_time, - exposure_time, - detector.driver.acquire_period, - 0.0, - detector.driver.num_images, - 1, - wait=True, - ) - try: + if isinstance(detector, ContAcqDetector): + yield from ensure_cont_acq_exposure(detector, exposure_time) + else: + yield from bps.mv( + detector.driver.acquire_time, + exposure_time, + detector.driver.acquire_period, + 0.0, + detector.driver.num_images, + 1, + ) yield from bps.mv(photon_shutter, True) _md = { @@ -52,5 +54,7 @@ def xrd_calibration( md=_md, ) finally: - yield from bps.mv(photon_shutter, False) - yield from bps.mv(motor, start_position) + try: + yield from bps.mv(photon_shutter, False) + finally: + yield from bps.mv(motor, start_position) diff --git a/src/hextools/xrd/plans.py b/src/hextools/xrd/plans.py new file mode 100644 index 0000000..18f1d6d --- /dev/null +++ b/src/hextools/xrd/plans.py @@ -0,0 +1,318 @@ +"""XRD acquisition plans for the HEX beamline. + +Thin wrappers over :func:`~hextools.tomography.radiography.take_radiograph` and +``bluesky.plans.scan`` that add XRD metadata, shutter handling and the Perkin +Elmer's continuous-acquisition exposure setup. +""" + +from collections.abc import Generator, Mapping, Sequence +from functools import partial +from typing import Any + +from bluesky import Msg +from bluesky import plan_stubs as bps +from bluesky import plans as bp +from bluesky import preprocessors as bpp +from bluesky.protocols import Movable, Readable +from ophyd_async.core import DetectorTrigger, TriggerInfo, set_and_wait_for_value +from ophyd_async.epics.adcore import ADImageMode, AreaDetector, ContAcqDetector + +from hextools.photon_delivery_system import Shutter +from hextools.photon_delivery_system.shutter import ( + ensure_shutter_closed, + ensure_shutter_open, +) +from hextools.tomography.radiography import take_radiograph +from hextools.utils import Steppable, ensure_available + + +def _resolve_detector(detector: ContAcqDetector | None) -> ContAcqDetector: + if detector is not None: + return detector + return ensure_available(ContAcqDetector, perkin_elmer=None) + + +def _with_plan_name(plan: Generator[Msg, Any, Any], plan_name: str): + """Run ``plan`` with ``plan_name`` in each of its start documents.""" + + def _rename(msg: Msg) -> Msg: + if msg.command == "open_run": + return msg._replace(kwargs={**msg.kwargs, "plan_name": plan_name}) + return msg + + return (yield from bpp.msg_mutator(plan, _rename)) + + +def ensure_cont_acq_exposure(detector: AreaDetector, exposure_time: float): + """Have a continuous-acquisition detector acquiring at ``exposure_time``. + + A ``ContAcqDetector`` only checks the exposure time it is prepared with, so + the driver is stopped, set and restarted here, as the pyepics scripts did. + Other detectors set their exposure when prepared, so this does nothing. + """ + if not isinstance(detector, ContAcqDetector): + return + acquiring = yield from bps.rd(detector.driver.acquire) + acquire_time = yield from bps.rd(detector.driver.acquire_time) + image_mode = yield from bps.rd(detector.driver.image_mode) + if ( + acquiring + and acquire_time == exposure_time + and image_mode == ADImageMode.CONTINUOUS + ): + return + if acquiring: + yield from bps.abs_set(detector.driver.acquire, False, wait=True) + yield from bps.mv( + detector.driver.acquire_time, + exposure_time, + detector.driver.image_mode, + ADImageMode.CONTINUOUS, + ) + # Acquire's put completes only when acquisition stops: wait on its readback. + (started,) = yield from bps.wait_for( + [ + partial( + set_and_wait_for_value, + detector.driver.acquire, + True, + wait_for_set_completion=False, + ) + ] + ) + started.result() + + +def xrd_count( + exposure_time: float, + num_images: int = 1, + images_to_average: int = 1, + num_iterations: int = 1, + time_gap: float = 0.0, + sample_name: str | None = None, + description: str | None = None, + use_shutter: bool = True, + detector: ContAcqDetector | None = None, + fe_shutter: Shutter | None = None, + photon_shutter: Shutter | None = None, +): + """Take XRD images without moving anything. + + Parameters + ---------- + exposure_time : float + exposure time per frame, in seconds + num_images : int + number of images saved per run + images_to_average : int + number of frames averaged into each saved image + num_iterations : int + number of runs to take, each its own file + time_gap : float + idle time between runs, in seconds + sample_name : str, optional + name of the sample + description : str, optional + description of the acquisition + use_shutter : bool + whether to open the photon shutter for the acquisition + detector : ContAcqDetector, optional + the XRD detector; the ``perkin_elmer`` in the session if not given + fe_shutter : Shutter, optional + the front-end shutter to check before opening the photon shutter + photon_shutter : Shutter, optional + the photon shutter to open and close around the acquisition + """ + detector = _resolve_detector(detector) + photon_shutter = ensure_available(Shutter, photon_shutter=photon_shutter) + + def _body(): + yield from ensure_cont_acq_exposure(detector, exposure_time) + return ( + yield from _with_plan_name( + take_radiograph( + [detector], + exposure_time, + num_images=num_images, + num_iterations=num_iterations, + external_trigger=False, + time_gap=time_gap, + images_to_average=images_to_average, + sample_name=sample_name, + description=description, + use_shutter=use_shutter, + fe_shutter=fe_shutter, + photon_shutter=photon_shutter, + ), + "xrd_count", + ) + ) + + def _cleanup(): + # take_radiograph closes it too; this covers a setup failure before it + yield from ensure_shutter_closed(photon_shutter, allow_actuation=True) + + return (yield from bpp.finalize_wrapper(_body(), _cleanup())) + + +def xrd_dark( + exposure_time: float, + num_images: int = 1, + images_to_average: int = 1, + sample_name: str | None = None, + description: str | None = None, + detector: ContAcqDetector | None = None, + fe_shutter: Shutter | None = None, + photon_shutter: Shutter | None = None, +): + """Take XRD dark images with the photon shutter closed. + + Parameters + ---------- + exposure_time : float + exposure time per frame, in seconds; use the one the data will use + num_images : int + number of dark images saved + images_to_average : int + number of frames averaged into each saved image + sample_name : str, optional + name of the sample + description : str, optional + description of the acquisition + detector : ContAcqDetector, optional + the XRD detector; the ``perkin_elmer`` in the session if not given + fe_shutter : Shutter, optional + the front-end shutter (not opened for darks) + photon_shutter : Shutter, optional + the photon shutter, closed before the darks are taken + """ + detector = _resolve_detector(detector) + photon_shutter = ensure_available(Shutter, photon_shutter=photon_shutter) + yield from ensure_shutter_closed(photon_shutter, allow_actuation=True) + yield from ensure_cont_acq_exposure(detector, exposure_time) + return ( + yield from _with_plan_name( + take_radiograph( + [detector], + exposure_time, + num_images=num_images, + external_trigger=False, + images_to_average=images_to_average, + sample_name=sample_name, + description=description, + use_shutter=False, + fe_shutter=fe_shutter, + photon_shutter=photon_shutter, + ), + "xrd_dark", + ) + ) + + +def xrd_scan( + motor: Steppable, + start: float, + stop: float, + num_points: int, + exposure_time: float, + images_to_average: int = 1, + num_iterations: int = 1, + time_between_iterations: float = 0.0, + sample_name: str | None = None, + description: str | None = None, + reset_position: bool = True, + use_shutter: bool = True, + detector: ContAcqDetector | None = None, + fe_shutter: Shutter | None = None, + photon_shutter: Shutter | None = None, +): + """Step one motor and take an XRD image at each point. + + Parameters + ---------- + motor : Steppable + the motor to step + start : float + first motor position + stop : float + last motor position + num_points : int + number of points, including start and stop + exposure_time : float + exposure time per frame, in seconds + images_to_average : int + number of frames averaged into the image saved at each point + num_iterations : int + number of times to repeat the scan, each its own run + time_between_iterations : float + idle time between repeats, in seconds + sample_name : str, optional + name of the sample + description : str, optional + description of the scan + reset_position : bool + whether to move the motor back to ``start`` afterwards + use_shutter : bool + whether to open the photon shutter for the scan + detector : ContAcqDetector, optional + the XRD detector; the ``perkin_elmer`` in the session if not given + fe_shutter : Shutter, optional + the front-end shutter to check before opening the photon shutter + photon_shutter : Shutter, optional + the photon shutter to open and close around the scan + """ + detector = _resolve_detector(detector) + fe_shutter = ensure_available(Shutter, fe_shutter=fe_shutter) + photon_shutter = ensure_available(Shutter, photon_shutter=photon_shutter) + trigger_info = TriggerInfo( + trigger=DetectorTrigger.INTERNAL, + livetime=exposure_time, + exposures_per_collection=images_to_average, + ) + + def _per_step( + detectors: Sequence[Readable], + step: Mapping[Movable, Any], + pos_cache: dict[Movable, Any], + take_reading: bps.TakeReading | None = None, + ): + # stage() at the start of each run clears what prepare() set + for det in detectors: + yield from bps.prepare(det, trigger_info, wait=True) + return ( + yield from bps.one_nd_step( + detectors, + step, + pos_cache, + take_reading=take_reading or bps.trigger_and_read, + ) + ) + + _md: dict[str, Any] = {"plan_name": "xrd_scan"} + if sample_name is not None: + _md["sample_name"] = sample_name + if description is not None: + _md["description"] = description + + def _body(): + if use_shutter: + yield from ensure_shutter_open(fe_shutter) + yield from ensure_shutter_open(photon_shutter, allow_actuation=True) + yield from ensure_cont_acq_exposure(detector, exposure_time) + yield from bp.scan( + [detector], motor, start, stop, num_points, per_step=_per_step, md=_md + ) + + def _cleanup(): + if use_shutter: + yield from ensure_shutter_closed(photon_shutter, allow_actuation=True) + if reset_position: + yield from bps.mv(motor, start) + + return ( + yield from bps.repeat( + lambda: bpp.finalize_wrapper(_body(), _cleanup()), + num=num_iterations, + delay=time_between_iterations, + ) + ) diff --git a/tests/gui/test_xrd_view.py b/tests/gui/test_xrd_view.py new file mode 100644 index 0000000..39d2801 --- /dev/null +++ b/tests/gui/test_xrd_view.py @@ -0,0 +1,29 @@ +import os + +import pytest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") +pytest.importorskip("qtpy") +pytest.importorskip("pyqtgraph") + +from bluesky import RunEngine # noqa: E402 +from IPython.core.interactiveshell import InteractiveShell # noqa: E402 +from qtpy.QtWidgets import QApplication, QTabWidget # noqa: E402 + +from hextools.gui.__main__ import QtXRDView # noqa: E402 + + +def test_xrd_tab_offers_the_xrd_plans(RE: RunEngine): + _app = QApplication.instance() or QApplication([]) + # the plan widgets validate their arguments in the IPython session + InteractiveShell.instance() + view = QtXRDView(RE) + plan_tabs = [ + tabs + for tabs in view.findChildren(QTabWidget) + if tabs.count() and tabs.tabText(0) == "Count" + ] + assert len(plan_tabs) == 1 + labels = [plan_tabs[0].tabText(i) for i in range(plan_tabs[0].count())] + assert labels == ["Count", "Scan", "Dark", "Calibration"] + view.deleteLater() diff --git a/tests/xrd/test_plans.py b/tests/xrd/test_plans.py new file mode 100644 index 0000000..909ebb9 --- /dev/null +++ b/tests/xrd/test_plans.py @@ -0,0 +1,353 @@ +"""XRD plans against a mocked Perkin Elmer (a continuous-acquisition detector).""" + +import asyncio +from typing import Any + +import pytest +from bluesky import RunEngine +from ophyd_async.core import ( + StaticPathProvider, + callback_on_mock_execute, + callback_on_mock_put, + get_mock_put, + init_devices, + set_mock_put_proceeds, + set_mock_value, +) +from ophyd_async.epics.adcore import ( + ADImageMode, + ADWriterFactory, + ContAcqDetector, + NDCircularBuffIO, + NDFileHDF5IO, + NDProcessIO, +) +from ophyd_async.epics.motor import Motor as AsyncEpicsMotor +from ophyd_async.sim import SimMotor + +from hextools.photon_delivery_system import Shutter, ShutterStatus +from hextools.xrd import ( + ensure_cont_acq_exposure, + xrd_calibration, + xrd_count, + xrd_dark, + xrd_scan, +) + + +@pytest.fixture +def perkin_elmer(static_path_provider: StaticPathProvider) -> ContAcqDetector: + """A Perkin Elmer acquiring continuously at 0.1 s, as the profile builds it.""" + with init_devices(mock=True): + pe = ContAcqDetector( + "PE:", + ADWriterFactory.hdf(static_path_provider, hinted=False), + name="perkin_elmer", + proc_suffix="Proc1:", + ) + set_mock_value(pe.driver.image_mode, ADImageMode.CONTINUOUS) + set_mock_value(pe.driver.acquire, True) + set_mock_value(pe.driver.acquire_time, 0.1) + set_mock_value(pe.driver.array_size_x, 2048) + set_mock_value(pe.driver.array_size_y, 2048) + hdf = pe.get_plugin("hdf", NDFileHDF5IO) + set_mock_value(hdf.file_path_exists, True) + cb = pe.get_plugin("cb", NDCircularBuffIO) + proc = pe.get_plugin("proc", NDProcessIO) + + async def _buffer_frames(value): + # The CB plugin passes post_count frames; Proc1 averages num_filter of them. + if not value: + return + frames = await cb.post_count.get_value() + frames //= max(1, await proc.num_filter.get_value()) + set_mock_value(hdf.num_captured, await hdf.num_captured.get_value() + frames) + + async def _done(): + await asyncio.sleep(0.01) + set_mock_value(cb.trigger_, False) + + asyncio.get_running_loop().create_task(_done()) + + callback_on_mock_put(cb.trigger_, _buffer_frames) + # each run opens a new file, which starts the count again + callback_on_mock_put(hdf.capture, lambda _: set_mock_value(hdf.num_captured, 0)) + return pe + + +@pytest.fixture +def open_fe(two_shutters: tuple[Shutter, Shutter]) -> tuple[Shutter, Shutter]: + fe_shutter, photon_shutter = two_shutters + set_mock_value(fe_shutter.status, ShutterStatus.OPEN) + set_mock_value(photon_shutter.status, ShutterStatus.CLOSED) + return fe_shutter, photon_shutter + + +def _collect(RE: RunEngine, plan) -> dict[str, list[Any]]: + docs: dict[str, list[Any]] = {} + RE(plan, lambda name, doc: docs.setdefault(name, []).append(doc)) + return docs + + +def test_ensure_cont_acq_exposure_leaves_a_matching_detector_alone( + RE: RunEngine, perkin_elmer: ContAcqDetector +): + RE(ensure_cont_acq_exposure(perkin_elmer, 0.1)) + get_mock_put(perkin_elmer.driver.acquire).assert_not_called() + get_mock_put(perkin_elmer.driver.acquire_time).assert_not_called() + + +async def test_ensure_cont_acq_exposure_restarts_at_a_new_exposure( + RE: RunEngine, perkin_elmer: ContAcqDetector +): + RE(ensure_cont_acq_exposure(perkin_elmer, 0.5)) + acquire_puts = [ + c.args[0] for c in get_mock_put(perkin_elmer.driver.acquire).call_args_list + ] + assert acquire_puts == [False, True] + assert await perkin_elmer.driver.acquire_time.get_value() == 0.5 + assert await perkin_elmer.driver.image_mode.get_value() == ADImageMode.CONTINUOUS + + +async def test_xrd_count_takes_averaged_images_in_each_run( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + fe_shutter, photon_shutter = open_fe + docs = _collect( + RE, + xrd_count( + 0.2, + num_images=3, + images_to_average=5, + num_iterations=2, + sample_name="LaB6", + detector=perkin_elmer, + fe_shutter=fe_shutter, + photon_shutter=photon_shutter, + ), + ) + assert [d["plan_name"] for d in docs["start"]] == ["xrd_count", "xrd_count"] + assert all(d["sample_name"] == "LaB6" for d in docs["start"]) + assert len(docs["event"]) == 2 + # one event per run, holding num_images averaged frames + assert [d["indices"] for d in docs["stream_datum"]] == [{"start": 0, "stop": 1}] * 2 + assert docs["descriptor"][0]["data_keys"]["perkin_elmer"]["shape"][0] == 3 + assert await perkin_elmer.driver.acquire_time.get_value() == 0.2 + proc = perkin_elmer.get_plugin("proc", NDProcessIO) + assert await proc.num_filter.get_value() == 5 + assert await photon_shutter.status.get_value() == ShutterStatus.CLOSED + + +async def test_xrd_dark_never_opens_the_photon_shutter( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + fe_shutter, photon_shutter = open_fe + set_mock_value(photon_shutter.status, ShutterStatus.OPEN) + opened: list[object] = [] + callback_on_mock_execute(photon_shutter.open_cmd, lambda *a: opened.append(a)) + docs = _collect( + RE, + xrd_dark( + 0.1, + num_images=4, + detector=perkin_elmer, + fe_shutter=fe_shutter, + photon_shutter=photon_shutter, + ), + ) + assert [d["plan_name"] for d in docs["start"]] == ["xrd_dark"] + assert opened == [] + assert await photon_shutter.status.get_value() == ShutterStatus.CLOSED + + +async def test_xrd_scan_takes_one_averaged_image_per_point( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + fe_shutter, photon_shutter = open_fe + with init_devices(): + det_z = SimMotor(name="det_z", instant=True) + docs = _collect( + RE, + xrd_scan( + det_z, + 0.0, + 4.0, + 5, + 0.1, + images_to_average=3, + num_iterations=2, + detector=perkin_elmer, + fe_shutter=fe_shutter, + photon_shutter=photon_shutter, + ), + ) + assert [d["plan_name"] for d in docs["start"]] == ["xrd_scan", "xrd_scan"] + assert len(docs["event"]) == 10 + positions = [e["data"]["det_z"] for e in docs["event"]] + assert positions == pytest.approx([0.0, 1.0, 2.0, 3.0, 4.0] * 2) + proc = perkin_elmer.get_plugin("proc", NDProcessIO) + assert await proc.num_filter.get_value() == 3 + assert await det_z.user_readback.get_value() == 0.0 + assert await photon_shutter.status.get_value() == ShutterStatus.CLOSED + + +async def test_xrd_scan_closes_the_shutter_and_returns_the_motor_on_failure( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + fe_shutter, photon_shutter = open_fe + with init_devices(): + det_z = SimMotor(name="det_z", instant=True) + hdf = perkin_elmer.get_plugin("hdf", NDFileHDF5IO) + cb = perkin_elmer.get_plugin("cb", NDCircularBuffIO) + triggers: list[bool] = [] + + async def _fault_on_second_point(value): + if not value: + return + triggers.append(value) + if len(triggers) == 2: + raise RuntimeError("detector fault") + set_mock_value(hdf.num_captured, await hdf.num_captured.get_value() + 1) + asyncio.get_running_loop().call_later(0.01, set_mock_value, cb.trigger_, False) + + callback_on_mock_put(cb.trigger_, _fault_on_second_point) + with pytest.raises(Exception, match="detector fault"): + RE( + xrd_scan( + det_z, + 1.0, + 3.0, + 3, + 0.1, + detector=perkin_elmer, + fe_shutter=fe_shutter, + photon_shutter=photon_shutter, + ) + ) + assert await det_z.user_readback.get_value() == 1.0 + assert await photon_shutter.status.get_value() == ShutterStatus.CLOSED + + +async def test_xrd_scan_after_a_restart_waits_for_the_acquire_readback_not_its_put( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + """Acquire is a busy record: its put completes only when acquisition stops.""" + fe_shutter, photon_shutter = open_fe + acquire = perkin_elmer.driver.acquire + timed_out: list[bool] = [] + with init_devices(): + det_z = SimMotor(name="det_z", instant=True) + + def _busy_acquire(value): + if not value: + set_mock_put_proceeds(acquire, True) + return None + set_mock_put_proceeds(acquire, False) + loop = asyncio.get_running_loop() + loop.call_later(0.2, set_mock_value, acquire, True) + # bound the test if the plan waits on the put rather than the readback + loop.call_later(15.0, _release_after_timeout) + return False + + def _release_after_timeout(): + timed_out.append(True) + set_mock_put_proceeds(acquire, True) + + callback_on_mock_put(acquire, _busy_acquire) + docs = _collect( + RE, + xrd_scan( + det_z, + 0.0, + 1.0, + 2, + 0.3, + detector=perkin_elmer, + fe_shutter=fe_shutter, + photon_shutter=photon_shutter, + ), + ) + RE.loop.call_soon_threadsafe(set_mock_put_proceeds, acquire, True) + assert len(docs["event"]) == 2 + assert timed_out == [] + assert await perkin_elmer.driver.acquire_time.get_value() == 0.3 + + +async def test_xrd_calibration_scans_the_perkin_elmer( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + _, photon_shutter = open_fe + with init_devices(mock=True): + det_z = AsyncEpicsMotor("DETZ:") + set_mock_value(det_z.velocity, 10000) + set_mock_value(det_z.acceleration_time, 0.0001) + docs = _collect( + RE, + xrd_calibration( + perkin_elmer, 0.1, det_z, photon_shutter, num_steps=2, gap=10.0 + ), + ) + assert [d["plan_name"] for d in docs["start"]] == ["xrd_calibration"] + assert len(docs["event"]) == 2 + assert await det_z.user_readback.get_value() == 0.0 + assert await photon_shutter.status.get_value() == ShutterStatus.CLOSED + + +def _fail_exposure_setup(detector: ContAcqDetector) -> None: + def _fail(value, **_): + raise RuntimeError("exposure setup failed") + + callback_on_mock_put(detector.driver.acquire_time, _fail) + + +async def test_xrd_count_closes_an_open_shutter_when_exposure_setup_fails( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + fe_shutter, photon_shutter = open_fe + set_mock_value(photon_shutter.status, ShutterStatus.OPEN) + _fail_exposure_setup(perkin_elmer) + with pytest.raises(Exception, match="exposure setup failed"): + RE( + xrd_count( + 0.3, + detector=perkin_elmer, + fe_shutter=fe_shutter, + photon_shutter=photon_shutter, + ) + ) + assert await photon_shutter.status.get_value() == ShutterStatus.CLOSED + + +async def test_xrd_calibration_closes_an_open_shutter_when_exposure_setup_fails( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + _, photon_shutter = open_fe + set_mock_value(photon_shutter.status, ShutterStatus.OPEN) + with init_devices(mock=True): + det_z = AsyncEpicsMotor("DETZ:") + _fail_exposure_setup(perkin_elmer) + with pytest.raises(Exception, match="exposure setup failed"): + RE(xrd_calibration(perkin_elmer, 0.3, det_z, photon_shutter, num_steps=2)) + assert await photon_shutter.status.get_value() == ShutterStatus.CLOSED + + +async def test_xrd_calibration_returns_the_motor_when_the_shutter_fails_to_close( + RE: RunEngine, perkin_elmer: ContAcqDetector, open_fe: tuple[Shutter, Shutter] +): + _, photon_shutter = open_fe + with init_devices(mock=True): + det_z = AsyncEpicsMotor("DETZ:") + set_mock_value(det_z.velocity, 10000) + set_mock_value(det_z.acceleration_time, 0.0001) + + def _stuck_open(*_): + raise RuntimeError("shutter stuck") + + callback_on_mock_execute(photon_shutter.close_cmd, _stuck_open) + with pytest.raises(Exception, match="shutter stuck"): + RE( + xrd_calibration( + perkin_elmer, 0.1, det_z, photon_shutter, num_steps=2, gap=10.0 + ) + ) + assert await det_z.user_readback.get_value() == 0.0