feat(aerotech): progress bar from the captured lines, statistics shown without MCS

The IPython client prints client-info messages only from a scan's progress
display. With trigger_source="panda" or None the scan registered none, so the
end-of-scan statistics were sent but never shown (scan 8511). The position
capture now publishes a ProgressSignal (lines read / num_points) and the scan
uses it for the progress bar when there is no MCS card.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M8zoDzDfsCTY2o5bT3J7Wv

csaxs_bec/devices/aerotech/a3200.py

@@ -27,7 +27,7 @@ from ophyd import Component as Cpt
 from ophyd import Device, PositionerBase, Signal
 from ophyd.status import wait as status_wait
 from ophyd.utils import LimitError, ReadOnlyError
-from ophyd_devices import AsyncMultiSignal, PSIDeviceBase
+from ophyd_devices import AsyncMultiSignal, ProgressSignal, PSIDeviceBase
 from ophyd_devices.utils.controller import Controller, retry_once, threadlocked
 from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
 from prettytable import PrettyTable
@@ -841,6 +841,7 @@ class A3200PositionCapture(PSIDeviceBase):
         kind="normal",
         doc="Positions latched at each trigger, in user units, one sub-signal per channel.",
     )
+    progress = Cpt(ProgressSignal, doc="Lines captured, for the scan progress bar")
 
     def __init__(
         self,
@@ -872,6 +873,8 @@ class A3200PositionCapture(PSIDeviceBase):
         self._signs = None  # channel -> sign of the BEC motor, resolved in calibrate()
         self._n_triggers = 0
         self._correlation_group = "monitored"
+        self._lines_total = 0
+        self._lines_read = 0
         super().__init__(
             prefix, name=name, scan_info=scan_info, device_manager=device_manager, **kwargs
         )
@@ -893,6 +896,11 @@ class A3200PositionCapture(PSIDeviceBase):
         frames = getattr(getattr(self, "scan_info", None), "msg", None)
         frames_per_trigger = getattr(frames, "frames_per_trigger", 1) if frames else 1
         self._correlation_group = "monitored" if frames_per_trigger == 1 else "burst_group"
+        # one readout per line; the scan's num_points is the number of lines
+        self._lines_total = int(getattr(frames, "num_points", 0) or 0) if frames else 0
+        self._lines_read = 0
+        if self._lines_total:
+            self.progress.put(value=0, max_value=self._lines_total, done=False)
 
     def calibrate(self) -> dict[str, float]:
         """
@@ -983,4 +991,11 @@ class A3200PositionCapture(PSIDeviceBase):
             },
             correlation_group=self._correlation_group,
         )
+        self._lines_read += 1
+        if self._lines_total:
+            self.progress.put(
+                value=self._lines_read,
+                max_value=self._lines_total,
+                done=self._lines_read >= self._lines_total,
+            )
         return {key: value.tolist() for key, value in result.items()}

csaxs_bec/scans/scans_v4/aerotech_cont_grid.py

@@ -345,8 +345,12 @@ class AerotechContGrid(ScanBase):
             readout_time=self._line["readout_time"],
         )
 
+        # progress bar in the client, which also prints the statistics at the end: the
+        # MCS frames at the beamline, otherwise the captured lines
         if self.mcs is not None:
             self.actions.add_scan_report_instruction_device_progress(self.mcs)
+        else:
+            self.actions.add_scan_report_instruction_device_progress(self.capture)
         self._baseline_readout_status = self.actions.read_baseline_devices(wait=False)
         self._premove_motor_status = self.actions.set(
             self.motors, [self._line["line_start"], self.stepper_start], wait=False

docs/user/saxs/aerotech_stages/aerotech_stages.md

@@ -80,7 +80,9 @@ the number of lines `ceil(|stepper_stop - stepper_start| / stepper_step_size) +
 The scan velocity can be up to 50 mm/s (`max_scan_velocity` of `aerox`). After the
 scan, also after an aborted one, X returns to its normal move velocity.
 
-With `trigger_source="ddg"` the client shows a progress bar of the frames.
+The client shows a progress bar (frames of the MCS card at the beamline, captured
+lines otherwise) and, at the end of the scan, a short summary of the positioning
+and the line times.
 
 To watch the measured positions while the scan runs, open the live plot of the
 exposure centres (Y versus X, forward and backward lines in two colours, updated

tests/tests_devices/test_a3200.py

@@ -497,3 +497,19 @@ def test_capture_without_motor_uses_controller_coordinates():
     _, _, capture = _capture_setup()  # DMMock without motors
     capture.calibrate()
     assert capture._signs == {6: 1, 2: 1}
+
+
+def test_capture_progress_counts_lines():
+    from types import SimpleNamespace
+
+    x, _, capture = _capture_setup()
+    capture.scan_info = SimpleNamespace(msg=SimpleNamespace(frames_per_trigger=5, num_points=2))
+    capture.on_stage()
+    assert capture.progress.get().value == 0 and capture.progress.get().max_value == 2
+    capture.calibrate()
+    for target in (-1.0, 1.0):
+        capture.arm(5)
+        x.move(target, wait=True)
+        capture.read_line()
+    progress = capture.progress.get()
+    assert progress.value == 2 and progress.done

tests/tests_scans/test_aerotech_cont_grid_scan.py

@@ -593,3 +593,24 @@ def test_ddg_without_shutter_delay_argument_leaves_ddg1_alone(v4_scan_assembler,
     scan.fast_axis = mock.MagicMock()
     scan.on_exception(RuntimeError("aborted"))
     scan.ddg1.set_shutter_to_open_delay.assert_not_called()
+
+
+def _progress_devices(scan):
+    return [
+        instr["device_progress"][0]
+        for instr in scan.scan_info.scan_report_instructions
+        if "device_progress" in instr
+    ]
+
+
+def test_progress_bar_from_the_mcs_at_the_beamline(v4_scan_assembler, device_manager):
+    scan = _assemble(v4_scan_assembler, device_manager)
+    scan.prepare_scan()
+    assert _progress_devices(scan) == ["mcs"]
+
+
+def test_progress_bar_from_the_capture_without_mcs(v4_scan_assembler, device_manager):
+    # a progress display is what prints the statistics (client messages) at the end
+    scan = _assemble(v4_scan_assembler, device_manager, trigger_source="panda")
+    scan.prepare_scan()
+    assert _progress_devices(scan) == ["aerocapture"]