Profile conversion 5c: xrd_temperature_series

src/pdftools/profiles/collection.py

@@ -35,7 +35,13 @@
     print_version_info,
     remove_md_keys,
 )
-from pdftools.xrd import xrd_count, xrd_dark, xrd_jog, xrd_scan  # noqa: F401
+from pdftools.xrd import (  # noqa: F401  for the session
+    xrd_count,
+    xrd_dark,
+    xrd_jog,
+    xrd_scan,
+    xrd_temperature_series,
+)
 
 # Mechanism for passing arguments to profile.
 _parser = argparse.ArgumentParser("Collection profile for pdftools.")

src/pdftools/xrd.py

@@ -19,10 +19,13 @@
 from ophyd_async.epics.motor import Motor
 from ophyd_async.sim import SimMotor
 
+from pdftools.eurotherm3k import Eurotherm3kLoop
 from pdftools.shutter import FastShutter
 from pdftools.utils import ensure_available
 
 Steppable = Motor | SimMotor
+# Moving one waits until its reading has settled at the setpoint.
+TemperatureController = Eurotherm3kLoop
 
 
 def ensure_cont_acq_exposure(detector: StandardDetector, exposure_time: float):
@@ -420,3 +423,100 @@ def _cleanup():
 
     velocity = yield from bps.rd(motor.velocity)
     return (yield from bpp.finalize_wrapper(_body(), _cleanup()))
+
+
+def xrd_temperature_series(
+    detectors: Sequence[StandardDetector],
+    controller: TemperatureController,
+    temperatures: Sequence[float],
+    exposure_time: float,
+    num_images: int = 1,
+    images_to_average: int = 1,
+    num_iterations: int = 1,
+    time_gap: float = 0.0,
+    ramp_rate: float | None = None,
+    sample_name: str | None = None,
+    description: str | None = None,
+    use_shutter: bool = True,
+    shutter: FastShutter | None = None,
+    md: dict | None = None,
+):
+    """Take XRD images at each of a list of temperatures.
+
+    For each temperature the controller is moved, which waits until its reading
+    has settled there (set ``tolerance`` and ``settle_time`` on the controller,
+    e.g. ``eurotherm.loop1.tolerance = 0.5``), then the images are taken as in
+    ``xrd_count``. The shutter is closed while the temperature changes. Replaces
+    pdf-profile-collection ``pbar_set_temp`` followed by a count.
+
+    Parameters
+    ----------
+    detectors : list of detectors
+        the detectors to read, e.g. ``[pe1c]``
+    controller : temperature controller
+        the loop to set, e.g. ``eurotherm.loop1``
+    temperatures : list of float
+        the setpoints, in the order to visit them
+    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 combined into each saved image
+    num_iterations : int
+        number of runs at each temperature, each its own file
+    time_gap : float
+        idle time between runs at one temperature, in seconds
+    ramp_rate : float, optional
+        ramp rate to set once before the first temperature, in the controller's
+        units; left as it is if not given
+    sample_name : str, optional
+        name of the sample
+    description : str, optional
+        description of the acquisition
+    use_shutter : bool
+        whether to open the fast shutter for each count and close it after
+    shutter : FastShutter, optional
+        the fast shutter; the session's ``fs`` if not given
+    md : dict, optional
+        extra metadata for each start document
+    """
+    if use_shutter:
+        shutter = ensure_available(FastShutter, fs=shutter)
+
+    def _body():
+        for det in detectors:
+            yield from ensure_cont_acq_exposure(det, exposure_time)
+        if ramp_rate is not None:
+            yield from bps.mv(controller.ramp_rate, ramp_rate)
+        for setpoint in temperatures:
+            if use_shutter:
+                yield from bps.mv(shutter, "Close")
+            yield from bps.mv(controller, setpoint)
+            if use_shutter:
+                yield from bps.mv(shutter, "Open")
+            for i in range(num_iterations):
+                if i > 0:
+                    yield from bps.sleep(time_gap)
+                readback = yield from bps.rd(controller.readback)
+                run_md = _md("xrd_temperature_series", sample_name, description, md)
+                run_md["temperature"] = {
+                    "controller": controller.name,
+                    "setpoint": setpoint,
+                    "readback": readback,
+                }
+                yield from _count_runs(
+                    detectors,
+                    exposure_time,
+                    num_images,
+                    images_to_average,
+                    1,
+                    0.0,
+                    run_md,
+                )
+
+    def _cleanup():
+        if use_shutter:
+            yield from bps.mv(shutter, "Close")
+
+    return (yield from bpp.finalize_wrapper(_body(), _cleanup()))

tests/test_xrd_temperature.py

@@ -0,0 +1,167 @@
+"""The temperature-series plan against a mocked Eurotherm, Perkin Elmer and shutter."""
+
+from typing import Any
+
+import pytest
+from bluesky.run_engine import RunEngine
+from ophyd_async.core import (
+    callback_on_mock_put,
+    get_mock_put,
+    init_devices,
+    set_mock_value,
+)
+from ophyd_async.epics.adcore import ContAcqDetector
+
+from pdftools.eurotherm3k import Eurotherm3k
+from pdftools.shutter import FastShutter, OnOff
+from pdftools.xrd import xrd_temperature_series
+
+
+@pytest.fixture
+def fs(RE: RunEngine) -> FastShutter:
+    with init_devices(mock=True):
+        fs = FastShutter("FS:", name="fs", settle_time=0)
+    set_mock_value(fs.status, OnOff.ON)  # closed
+    callback_on_mock_put(fs.cmd, lambda value, **_: set_mock_value(fs.status, value))
+    return fs
+
+
+@pytest.fixture
+def eurotherm(RE: RunEngine) -> Eurotherm3k:
+    with init_devices(mock=True):
+        eurotherm = Eurotherm3k("ET:", name="eurotherm")
+    set_mock_value(eurotherm.loop1.readback, 25.0)
+    return eurotherm
+
+
+def _run(RE: RunEngine, plan) -> tuple[dict[str, list[Any]], list]:
+    docs: dict[str, list[Any]] = {}
+    msgs: list = []
+    RE.msg_hook = msgs.append  # pyright: ignore[reportAttributeAccessIssue]
+    try:
+        RE(plan, lambda name, doc: docs.setdefault(name, []).append(doc))
+    finally:
+        RE.msg_hook = None
+    return docs, msgs
+
+
+async def test_xrd_temperature_series_counts_at_each_settled_temperature(
+    RE: RunEngine, pe: ContAcqDetector, fs: FastShutter, eurotherm: Eurotherm3k
+):
+    loop = eurotherm.loop1
+    callback_on_mock_put(
+        loop.setpoint, lambda value, **_: set_mock_value(loop.readback, value + 0.2)
+    )
+    docs, msgs = _run(
+        RE,
+        xrd_temperature_series(
+            [pe],
+            loop,
+            [100.0, 200.0, 300.0],
+            0.1,
+            num_iterations=2,
+            sample_name="Ni",
+            shutter=fs,
+        ),
+    )
+    assert [c.args[0] for c in get_mock_put(loop.setpoint).call_args_list] == [
+        100.0,
+        200.0,
+        300.0,
+    ]
+    starts = docs["start"]
+    assert len(starts) == 6  # two runs at each temperature
+    assert {s["plan_name"] for s in starts} == {"xrd_temperature_series"}
+    assert {s["sample_name"] for s in starts} == {"Ni"}
+    assert [s["temperature"] for s in starts[::2]] == [
+        {"controller": "eurotherm-loop1", "setpoint": t, "readback": t + 0.2}
+        for t in (100.0, 200.0, 300.0)
+    ]
+    # closed while the temperature moves, open only to count
+    moves = [
+        (m.obj.name, m.args[0])
+        for m in msgs
+        if m.command == "set" and m.obj in (fs, loop)
+    ]
+    assert moves == [
+        step
+        for t in (100.0, 200.0, 300.0)
+        for step in (("fs", "Close"), ("eurotherm-loop1", t), ("fs", "Open"))
+    ] + [("fs", "Close")]
+    assert await fs.status.get_value() == OnOff.ON
+    assert get_mock_put(loop.ramp_rate).call_count == 0
+
+
+async def test_xrd_temperature_series_sets_the_ramp_rate_once(
+    RE: RunEngine, pe: ContAcqDetector, fs: FastShutter, eurotherm: Eurotherm3k
+):
+    loop = eurotherm.loop1
+    callback_on_mock_put(
+        loop.setpoint, lambda value, **_: set_mock_value(loop.readback, value)
+    )
+    _run(
+        RE,
+        xrd_temperature_series(
+            [pe], loop, [50.0, 60.0], 0.1, ramp_rate=5.0, shutter=fs
+        ),
+    )
+    assert [c.args[0] for c in get_mock_put(loop.ramp_rate).call_args_list] == [5.0]
+
+
+async def test_xrd_temperature_series_closes_the_shutter_if_never_settled(
+    RE: RunEngine, pe: ContAcqDetector, fs: FastShutter, eurotherm: Eurotherm3k
+):
+    loop = eurotherm.loop1
+    # mock=True gives a StandardMovable InstantMovableMock: replace it so the
+    # readback stays at 25 instead of jumping to the setpoint
+    callback_on_mock_put(loop.setpoint, lambda *_, **__: None)
+    loop.move_timeout = 0.2
+    docs: dict[str, list[Any]] = {}
+    with pytest.raises(Exception):  # noqa: B017  # bluesky wraps the timeout
+        RE(
+            xrd_temperature_series([pe], loop, [300.0], 0.1, shutter=fs),
+            lambda name, doc: docs.setdefault(name, []).append(doc),
+        )
+    assert "start" not in docs  # nothing counted off temperature
+    assert await fs.status.get_value() == OnOff.ON
+
+
+async def test_xrd_temperature_series_closes_an_open_shutter_before_moving(
+    RE: RunEngine, pe: ContAcqDetector, fs: FastShutter, eurotherm: Eurotherm3k
+):
+    loop = eurotherm.loop1
+    set_mock_value(fs.status, OnOff.OFF)  # left open by an earlier plan
+    shutter_at_move: list[OnOff] = []
+
+    async def _reach(value, **_):
+        shutter_at_move.append(await fs.status.get_value())
+        set_mock_value(loop.readback, value)
+
+    callback_on_mock_put(loop.setpoint, _reach)
+    _run(RE, xrd_temperature_series([pe], loop, [100.0], 0.1, shutter=fs))
+    assert shutter_at_move == [OnOff.ON]
+
+
+async def test_xrd_temperature_series_reads_the_temperature_before_each_run(
+    RE: RunEngine, pe: ContAcqDetector, fs: FastShutter, eurotherm: Eurotherm3k
+):
+    loop = eurotherm.loop1
+    callback_on_mock_put(
+        loop.setpoint, lambda value, **_: set_mock_value(loop.readback, value)
+    )
+    drift = iter([100.4, 100.7])
+
+    def _drift(name, doc):
+        if name == "stop":  # the temperature wanders between runs
+            set_mock_value(loop.readback, next(drift, 100.9))
+
+    RE.subscribe(_drift)
+    docs, _ = _run(
+        RE,
+        xrd_temperature_series([pe], loop, [100.0], 0.1, num_iterations=3, shutter=fs),
+    )
+    assert [s["temperature"]["readback"] for s in docs["start"]] == [
+        100.0,
+        100.4,
+        100.7,
+    ]