feat(csaxs): csaxs.aerotech.plot() shows the exposure centres Y vs X live

Opens (or reuses) the GUI window "aerotech" with a Waveform of the async
aerocapture centres, enc2_centre vs enc6_centre, updated per line during
aerotech_cont_grid; scan_number=... shows a finished scan.

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

csaxs_bec/bec_ipython_client/plugins/cSAXS/aerotech.py

@@ -0,0 +1,63 @@
+"""Aerotech SAXS-TT stage tools of the cSAXS session (csaxs.aerotech)."""
+
+from __future__ import annotations
+
+from bec_lib import bec_logger
+
+logger = bec_logger.logger
+
+
+class cSAXSAerotech:
+    """Tools for the Aerotech SAXS-TT stages and the aerotech_cont_grid scan."""
+
+    WINDOW = "aerotech"
+    PLOT = "aerotech_positions"
+    GUI_RPC_TIMEOUT = 20
+    CAPTURE_DEVICE = "aerocapture"
+    # exposure centres of the fast (X) and step (Y) axis, published per line
+    SIGNAL_X = "aerocapture_positions_enc6_centre"
+    SIGNAL_Y = "aerocapture_positions_enc2_centre"
+
+    def __init__(self, client):
+        self.client = client
+        self.gui = getattr(client, "gui", None)
+
+    def plot(self, scan_number: int | None = None):
+        """
+        Open a window that plots the measured exposure centres Y versus X of
+        aerotech_cont_grid, line by line while the scan runs.
+
+        Without scan_number the plot follows the running (or next) scan; with
+        scan_number it shows that scan from the history. Calling it again reuses the
+        window.
+
+        Examples
+        --------
+        csaxs.aerotech.plot()                  # live, follows the scans
+        csaxs.aerotech.plot(scan_number=8494)  # a finished scan
+        """
+        if self.gui is None:
+            raise RuntimeError("No BEC GUI available: start the BEC client with the GUI.")
+        if self.WINDOW in self.gui.windows:
+            window = self.gui.windows[self.WINDOW]
+        else:
+            window = self.gui.new(self.WINDOW, timeout=self.GUI_RPC_TIMEOUT)
+        waveform = window.new("Waveform", object_name=self.PLOT, timeout=self.GUI_RPC_TIMEOUT)
+        waveform.clear_all()
+        waveform.title = "aerotech_cont_grid: exposure centres"
+        waveform.x_label = "X centre (aerox)"
+        waveform.y_label = "Y centre (aeroy)"
+        waveform.lock_aspect_ratio = True
+        curve = waveform.plot(
+            device_x=self.CAPTURE_DEVICE,
+            signal_x=self.SIGNAL_X,
+            device_y=self.CAPTURE_DEVICE,
+            signal_y=self.SIGNAL_Y,
+            label="exposure centres",
+            scan_number=scan_number,
+        )
+        # markers for the exposures, a thin line shows the snake order
+        curve.set_symbol("o")
+        curve.set_symbol_size(4)
+        curve.set_pen_width(1)
+        return waveform

csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py

@@ -3,6 +3,7 @@ import inspect
 from bec_lib import bec_logger
 from typeguard import typechecked
 
+from csaxs_bec.bec_ipython_client.plugins.cSAXS.aerotech import cSAXSAerotech
 from csaxs_bec.bec_ipython_client.plugins.cSAXS.diagnostics import cSAXSDiagnostics
 from csaxs_bec.bec_ipython_client.plugins.cSAXS.filter_transmission import cSAXSFilterTransmission
 from csaxs_bec.bec_ipython_client.plugins.cSAXS.intensity_map_predict_gap import (
@@ -32,6 +33,7 @@ class cSAXS(
         self.OMNYTools = OMNYTools(self.client)
 
         self.diagnostics = cSAXSDiagnostics()
+        self.aerotech = cSAXSAerotech(client)
         super().__init__(client=client)
 
     # ------------------------------------------------------------------
@@ -104,6 +106,7 @@ class cSAXS(
             ("csaxs.diagnostics.bim", "BIM diagnostics — .show_all(), .gain(val)"),
             ("csaxs.diagnostics.beamstop", "Beamstop diode — .show_all(), .gain(val)"),
             ("csaxs.diagnostics.polarization", "Polarization diodes — .show_all(), .gain(val)"),
+            ("csaxs.aerotech.plot()", "Aerotech scan: exposure centres Y vs X, live"),
             ("csaxs.OMNYTools.*", "OMNY instrument tools"),
         ]:
             ns.add_row(access, desc)

docs/user/saxs/aerotech_stages/aerotech_stages.md

@@ -81,6 +81,14 @@ 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.
 
+To watch the measured positions while the scan runs, open the live plot of the
+exposure centres (Y versus X, one line after the other):
+
+```python
+csaxs.aerotech.plot()                  # follows the running and the next scans
+csaxs.aerotech.plot(scan_number=8494)  # a finished scan
+```
+
 ### Examples
 
 Fields of view centred on the current position (`relative=True`), X fast, Y step:

tests/tests_bec_ipython_client/test_cSAXS_aerotech.py

@@ -0,0 +1,50 @@
+from unittest import mock
+
+import pytest
+
+from csaxs_bec.bec_ipython_client.plugins.cSAXS.aerotech import cSAXSAerotech
+
+
+def _client(windows=None):
+    gui = mock.MagicMock()
+    gui.windows = windows if windows is not None else {}
+    return mock.MagicMock(gui=gui)
+
+
+def test_plot_opens_window_and_plots_centres():
+    client = _client()
+    waveform = client.gui.new.return_value.new.return_value
+
+    assert cSAXSAerotech(client).plot() is waveform
+
+    client.gui.new.assert_called_once_with("aerotech", timeout=20)
+    client.gui.new.return_value.new.assert_called_once_with(
+        "Waveform", object_name="aerotech_positions", timeout=20
+    )
+    waveform.clear_all.assert_called_once_with()
+    assert waveform.lock_aspect_ratio is True
+    waveform.plot.assert_called_once_with(
+        device_x="aerocapture",
+        signal_x="aerocapture_positions_enc6_centre",
+        device_y="aerocapture",
+        signal_y="aerocapture_positions_enc2_centre",
+        label="exposure centres",
+        scan_number=None,
+    )
+    waveform.plot.return_value.set_symbol.assert_called_once_with("o")
+
+
+def test_plot_reuses_open_window_and_shows_history():
+    window = mock.MagicMock()
+    client = _client(windows={"aerotech": window})
+
+    cSAXSAerotech(client).plot(scan_number=8494)
+
+    client.gui.new.assert_not_called()
+    assert window.new.return_value.plot.call_args.kwargs["scan_number"] == 8494
+
+
+def test_plot_without_gui_raises():
+    client = mock.MagicMock(spec=["device_manager"])
+    with pytest.raises(RuntimeError, match="No BEC GUI"):
+        cSAXSAerotech(client).plot()