Capture plan for any detector (bp.count); Phantom arms in prepare and waits for the event trigger; GUI tab

src/hextools/detectors/phantom.py

@@ -1,6 +1,7 @@
 """Ophyd async support for Phantom VEO camera at HEX."""
 
 import asyncio
+import contextlib
 from collections.abc import Sequence
 from typing import Annotated as A
 
@@ -413,31 +414,51 @@ class PhantomAcquireLogic(ADAcquireLogic):
     def __init__(self, driver: PhantomIO):
         super().__init__(driver, driver.acquire)
         self.driver = driver
+        self._readout: asyncio.Future[None] | None = None
 
     async def start_acquiring(self):
-        """Start the acquisition and the image data download.
+        """Arm the camera, and return once it is waiting for the event trigger.
 
-        Start the acquisition, wait for the event trigger,
-        wait for any post trigger frames, and start the download.
+        The camera records into its cine until the event trigger, then records the
+        post-trigger frames; the image download starts after that. Those waits run
+        in the background (see `_read_out_after_trigger`) and `wait_for_idle`
+        collects their result, because ophyd-async calls this from ``prepare`` for
+        externally triggered detectors: the trigger comes later, from the PandA once
+        a fly scan's motion starts at kickoff, or from an operator pressing it while
+        watching the live image. Waiting for it here would stop ``prepare`` ever
+        finishing.
 
         Raises
         ------
-        RuntimeError
-            If acquisition stops before we receive the event trigger.
         TimeoutError
-            If we timeout waiting for the cine to be marked as valid.
-        ValueError
-            If the number of post trigger frames recorded is not what was expected.
+            If the camera does not report waiting for a trigger.
         """
-        # Start the acquisition, and wait for waiting for trigger to be True
-        await set_and_wait_for_other_value(
+        # Not waiting for the Acquire write itself: an areaDetector Acquire put
+        # completes only when acquisition ends, after the trigger. wait_for_idle
+        # collects it.
+        self.acquire_status = await set_and_wait_for_other_value(
             self.driver.acquire,
             True,
             self.driver.waiting_for_trigger,
             True,
             timeout=DEFAULT_TIMEOUT,
+            wait_for_set_completion=False,
         )
+        self._readout = asyncio.ensure_future(self._read_out_after_trigger())
 
+    async def _read_out_after_trigger(self):
+        """Wait for the event trigger and the post-trigger frames; start the download.
+
+        Raises
+        ------
+        RuntimeError
+            If acquisition stops before we receive the event trigger, or more
+            frames are requested than the camera recorded.
+        TimeoutError
+            If we timeout waiting for the cine to be marked as valid.
+        ValueError
+            If the number of post trigger frames recorded is not what was expected.
+        """
         # Wait for trigger_received to go to 1. If trigger_received does not go
         # to 1 within the timeout, check if acquisition stopped, and if so raise
         # a timeout error indicating acquisition stopped while waiting for trigger.
@@ -503,14 +524,18 @@ async def start_acquiring(self):
         await self.driver.download.set(True)
 
     async def wait_for_idle(self):
-        """Wait for the camera to finish downloading and return to idle state.
+        """Wait for the trigger and readout, then for the download to finish.
 
         Raises
         ------
         TimeoutError
             If we timeout waiting for the download to complete.
         """
-        # First, make sure our arm process is complete.
+        # First, the readout started by start_acquiring: its errors surface here.
+        if self._readout is not None:
+            readout, self._readout = self._readout, None
+            await readout
+        # Then make sure our arm process is complete.
         if self.acquire_status:
             await self.acquire_status
 
@@ -553,6 +578,15 @@ async def wait_for_idle(self):
                     ) from err
                 last_download_count = current
 
+    async def ensure_stopped(self):
+        """Stop a readout still waiting for its trigger, then stop the camera."""
+        if self._readout is not None:
+            readout, self._readout = self._readout, None
+            readout.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await readout
+        await super().ensure_stopped()
+
 
 class PhantomDetector(AreaDetector[PhantomIO]):
     """Detector class for Phantom cameras."""

src/hextools/gui/__main__.py

@@ -74,7 +74,7 @@
 from hextools.gui.theme_switch import QtThemeSwitch
 from hextools.photon_delivery_system.dclm import change_beam_mode
 from hextools.tomography.alignment import tomo_alignment_scan
-from hextools.tomography.flyscans import tomo_1d_step_scan, tomo_2d_step_scan, tomo_flyscan
+from hextools.tomography.flyscans import capture, tomo_1d_step_scan, tomo_2d_step_scan, tomo_flyscan
 from hextools.tomography.radiography import take_radiograph
 from hextools.edxd import configure_test_pulses, edxd_2theta_tilt, edxd_calib_scan, edxd_count, edxd_custom_pos_list_grid, edxd_grid_scan, edxd_scan
 from hextools.photon_delivery_system import change_energy
@@ -269,6 +269,7 @@ def __init__(self, re_client: RunEngineClient | RunEngine, *args, **kwargs):
         plan_tabs.addTab(QtPlanWidget(re_client, tomo_2d_step_scan), "2D Step")
         plan_tabs.addTab(QtPlanWidget(re_client, take_radiograph), "Radiography")
         plan_tabs.addTab(QtPlanWidget(re_client, tomo_alignment_scan), "Alignment")
+        plan_tabs.addTab(QtPlanWidget(re_client, capture), "Capture")
 
         side = QVBoxLayout()
         side.addWidget(plan_tabs, stretch=1)

src/hextools/profiles/collection.py

@@ -64,6 +64,7 @@
 )
 
 from hextools.tomography import tomo_flyscan, tomo_alignment_scan, tomo_1d_step_scan, tomo_2d_step_scan, tomo_nd_step_scan
+from hextools.tomography import capture  # noqa: F401 - run by name from the GUI
 from hextools.edxd import configure_test_pulses, edxd_scan, edxd_count, edxd_grid_scan, edxd_2theta_tilt, edxd_calib_scan, edxd_custom_pos_list_grid
 
 

src/hextools/tomography/__init__.py

@@ -1,11 +1,12 @@
 """Tomography tools for HEX beamline."""
 
-from .flyscans import tomo_flyscan, tomo_1d_step_scan, tomo_2d_step_scan, tomo_nd_step_scan
+from .flyscans import capture, tomo_flyscan, tomo_1d_step_scan, tomo_2d_step_scan, tomo_nd_step_scan
 
 from .alignment import tomo_alignment_scan
 from .radiography import take_radiograph
 
 __all__ = [
+    "capture",
     "tomo_flyscan",
     "tomo_1d_step_scan",
     "tomo_2d_step_scan",

src/hextools/tomography/flyscans.py

@@ -7,7 +7,13 @@
 from bluesky.protocols import Collectable, Flyable, HasName, Movable, Readable
 from bluesky.utils import CustomPlanMetadata, MsgGenerator, plan
 from nslsii import detectors
-from ophyd_async.core import StandardMovable, DetectorTrigger, StandardFlyable, TriggerInfo
+from ophyd_async.core import (
+    DetectorTrigger,
+    StandardDetector,
+    StandardFlyable,
+    StandardMovable,
+    TriggerInfo,
+)
 from ophyd_async.epics.adkinetix import KinetixDetector, KinetixTriggerMode
 from ophyd_async.fastcs.panda import HDFPanda
 from ophyd_async.epics.motor import Motor as AsyncEpicsMotor
@@ -649,3 +655,55 @@ def tomo_2d_step_scan(
         md={"plan_name": "tomo_2d_step_scan"},
     ))
 
+
+def capture(
+    detectors: list[StandardDetector],
+    num_images: int,
+    exposure_time: float,
+    timeout: float = 600,
+    sample_name: str | None = None,
+):
+    """Take ``num_images`` frames on each detector as one reading (``bp.count``).
+
+    Works with any ophyd-async detector. A Phantom arms and waits for its event
+    trigger (the GUI's Trigger button, a hardware trigger or the camera's own
+    screen), then downloads the first ``num_images`` post-trigger frames; other
+    cameras take the frames straight away.
+
+    Parameters
+    ----------
+    detectors : list of StandardDetector
+        The detectors to capture with.
+    num_images : int
+        Frames to record in the reading, at least 1.
+    exposure_time : float
+        Exposure time per frame, in seconds.
+    timeout : float, optional
+        How long to wait for the frames, including any wait for a trigger, in
+        seconds. Default 600.
+    sample_name : str, optional
+        Recorded in the run's metadata.
+    """
+    info = TriggerInfo(
+        livetime=exposure_time,
+        collections_per_event=num_images,
+        exposure_timeout=timeout,
+    )
+    md = {
+        "plan_name": "capture",
+        "num_images": num_images,
+        "exposure_time": exposure_time,
+    }
+    if sample_name is not None:
+        md["sample_name"] = sample_name
+
+    # bp.count stages the detectors, which clears any earlier prepare, so prepare
+    # in per_shot, after staging and before the reading.
+    def _prepare_then_read(dets, take_reading=None):
+        for det in dets:
+            yield from bps.prepare(det, info, group="capture_prepare")
+        yield from bps.wait(group="capture_prepare")
+        yield from bps.one_shot(dets, take_reading=take_reading)
+
+    return (yield from bp.count(detectors, per_shot=_prepare_then_read, md=md))
+

tests/detectors/test_phantom.py

@@ -15,6 +15,7 @@
     TriggerInfo,
     callback_on_mock_put,
     init_devices,
+    set_mock_put_proceeds,
     set_mock_value,
 )
 from ophyd_async.epics.adcore import ADBaseDataType, ADWriterFactory, NDPluginFileIO
@@ -29,6 +30,12 @@
 )
 
 
+async def _read_out(logic: PhantomAcquireLogic) -> None:
+    """Wait for the background readout start_acquiring left running."""
+    assert logic._readout is not None
+    await logic._readout
+
+
 @pytest.fixture
 def phantom_io(RE: RunEngine) -> PhantomIO:
     with init_devices(mock=True):
@@ -159,10 +166,11 @@ async def _stop_acquisition():
 
     stop_task = asyncio.create_task(_stop_acquisition())
     try:
+        await phantom_arm_logic.start_acquiring()
         with pytest.raises(
             RuntimeError, match="Acquisition stopped while waiting for event trigger!"
         ):
-            await phantom_arm_logic.start_acquiring()
+            await phantom_arm_logic.wait_for_idle()
     finally:
         stop_task.cancel()
 
@@ -182,11 +190,12 @@ async def test_arm_logic_arm_timeout_saving_to_cine(
     # so the second loop times out
     set_mock_value(phantom_arm_logic.driver.post_trig_frames, 10)
 
+    await phantom_arm_logic.start_acquiring()
     with pytest.raises(
         TimeoutError,
         match="Received event trigger, but writing to cine was not completed!",
     ):
-        await phantom_arm_logic.start_acquiring()
+        await phantom_arm_logic.wait_for_idle()
 
 
 async def test_arm_logic_arm_post_trig_frames_incorrect(
@@ -206,11 +215,12 @@ async def test_arm_logic_arm_post_trig_frames_incorrect(
         phantom_arm_logic.driver.array_counter, 5
     )  # Different from post_trig_frames
 
+    await phantom_arm_logic.start_acquiring()
     with pytest.raises(
         ValueError,
         match="Expected number of post trig frames .* does not match actual number .*",
     ):
-        await phantom_arm_logic.start_acquiring()
+        await phantom_arm_logic.wait_for_idle()
 
 
 async def test_arm_logic_arm_success(
@@ -231,9 +241,47 @@ async def test_arm_logic_arm_success(
     )  # Matches post_trig_frames
 
     await phantom_arm_logic.start_acquiring()  # Should complete without exceptions
+    await _read_out(phantom_arm_logic)  # the trigger has already come
     assert await phantom_arm_logic.driver.download.get_value()
 
 
+async def test_arm_logic_start_returns_before_the_trigger(
+    phantom_arm_logic: PhantomAcquireLogic, monkeypatch
+):
+    # ophyd-async calls start_acquiring from prepare; the trigger comes later (the
+    # PandA after kickoff, or an operator). Waiting for it here hung prepare.
+    monkeypatch.setattr(hextools.detectors.phantom, "DEFAULT_TIMEOUT", 0.1)
+    set_mock_value(phantom_arm_logic.driver.waiting_for_trigger, True)
+    # Like the IOC: an Acquire put completes only when acquisition ends.
+    set_mock_put_proceeds(phantom_arm_logic.driver.acquire, False)
+
+    await asyncio.wait_for(phantom_arm_logic.start_acquiring(), timeout=1)
+
+    acquire = phantom_arm_logic.acquire_status
+    assert acquire is not None
+    assert not acquire.done  # the Acquire put is still in flight
+    assert phantom_arm_logic._readout is not None
+    assert not phantom_arm_logic._readout.done()  # still waiting for the trigger
+    set_mock_put_proceeds(phantom_arm_logic.driver.acquire, True)
+    await phantom_arm_logic.ensure_stopped()
+
+
+async def test_arm_logic_stop_cancels_a_readout_waiting_for_its_trigger(
+    phantom_arm_logic: PhantomAcquireLogic, monkeypatch
+):
+    monkeypatch.setattr(hextools.detectors.phantom, "DEFAULT_TIMEOUT", 0.1)
+    set_mock_value(phantom_arm_logic.driver.waiting_for_trigger, True)
+    await phantom_arm_logic.start_acquiring()
+    readout = phantom_arm_logic._readout
+    assert readout is not None
+
+    await phantom_arm_logic.ensure_stopped()
+
+    assert readout.cancelled()
+    assert phantom_arm_logic._readout is None
+    assert not await phantom_arm_logic.driver.download.get_value()
+
+
 async def test_arm_logic_wait_for_idle_timeout(
     phantom_arm_logic: PhantomAcquireLogic, monkeypatch
 ):

tests/tomography/test_capture.py

@@ -0,0 +1,31 @@
+from pathlib import Path
+
+import pytest
+from bluesky.run_engine import RunEngine
+from ophyd_async.core import StaticPathProvider, UUIDFilenameProvider, init_devices
+from ophyd_async.sim import PatternGenerator, SimBlobDetector
+
+from hextools.tomography.flyscans import capture
+
+
+@pytest.fixture
+def blob(RE: RunEngine, tmp_path: Path) -> SimBlobDetector:
+    path_provider = StaticPathProvider(UUIDFilenameProvider(), tmp_path)
+    with init_devices():
+        blob = SimBlobDetector(path_provider, PatternGenerator())
+    return blob
+
+
+# capture is for any detector, not only the Phantom (PR 92 review).
+def test_capture_takes_one_reading_of_num_images_frames(RE: RunEngine, blob):
+    docs: dict[str, list] = {}
+    RE(
+        capture([blob], num_images=3, exposure_time=0.01),
+        lambda name, doc: docs.setdefault(name, []).append(doc),
+    )
+
+    assert docs["start"][0]["plan_name"] == "capture"
+    assert docs["start"][0]["num_images"] == 3
+    assert len(docs["event"]) == 1
+    shapes = docs["descriptor"][0]["data_keys"]["blob"]["shape"]
+    assert shapes[0] == 3, f"expected 3 frames in the one reading, got {shapes}"