feat(csaxs): csaxs.diagnostics.show_plots() opens beam diagnostics waveform window
CI for csaxs_bec / test (push) Successful in 4m17s
CI for csaxs_bec / test (pull_request) Successful in 4m15s

Opens a BEC GUI window with waveform plots of the XBPM3 blade signals and
the XBIM signal by default. Further plots by name (xbpm3_position, xbpm2,
beamstop, polarization), or only the named ones with defaults=False.
Devices not in the config are skipped; calling it again rebuilds the plots
in the open window.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017NrH3poJsYBCTVq2Y9fuHZ
This commit is contained in:
x12saandClaude Opus 5.5 committed 2026-10-28 11:24:47 +01:00
1 parent 00479fd7d3
commit 602eac82ba
3 files changed
+206 -3

No files matched your search

@@ -31,7 +31,7 @@ class cSAXS(
self.device_manager = client.device_manager
self.OMNYTools = OMNYTools(self.client)
self.diagnostics = cSAXSDiagnostics()
self.diagnostics = cSAXSDiagnostics(client)
super().__init__(client=client)
# EPICS does not keep the slit offsets reliably; restore them from config
@@ -100,6 +100,10 @@ class cSAXS(
ns.add_column("Description")
for access, desc in [
("csaxs.diagnostics.show_all()", "All diagnostic device readbacks"),
(
"csaxs.diagnostics.show_plots(*extra)",
"Plot window: XBPM3 blades + XBIM; extra e.g. 'xbpm3_position', 'beamstop'",
),
(
"csaxs.diagnostics.xbpm2 / .xbpm3",
"BPM diagnostics — .show_all(), .gain(val)",
@@ -1,5 +1,6 @@
from __future__ import annotations
import builtins
import time
# ------------------------------------------------------------
# Access to dev (as before)
@@ -167,7 +168,32 @@ class BPMDiag:
class cSAXSDiagnostics:
"""User-visible diagnostics namespace."""
def __init__(self):
PLOT_WINDOW_NAME = "csaxs_diagnostics"
GUI_RPC_TIMEOUT = 20
_BLADES = ["left_top", "right_top", "left_bot", "right_bot"]
# name -> (plot title, [(device, signal suffix), ...]); one waveform plot
# each, curves of devices missing from the config are skipped
PLOTS = {
"xbpm3": ("XBPM3 blades (ES hutch, XBox 2)", [("xbpm3_slowrb", b) for b in _BLADES]),
"xbim": ("XBIM (ES hutch, XBox 3)", [("xbim_slowrb", "signal")]),
"xbpm3_position": (
"XBPM3 position",
[("xbpm3_slowrb", "pos_x"), ("xbpm3_slowrb", "pos_y")],
),
"xbpm2": ("XBPM2 blades (OP hutch, XBox 1)", [("xbpm2_slowrb", b) for b in _BLADES]),
"beamstop": ("Beamstop diode (flight tube)", [("beamstop_intensity", "signal")]),
"polarization": (
"Polarization diodes (OP hutch, XBox 1)",
[
("diode_horizontal_xbox1_slowrb", "signal"),
("diode_vertical_xbox1_slowrb", "signal"),
],
),
}
DEFAULT_PLOTS = ("xbpm3", "xbim")
def __init__(self, client=None):
self._client = client
# BPMs
self.xbpm2 = BPMDiag(
name="xbpm2",
@@ -217,4 +243,50 @@ class cSAXSDiagnostics:
self.xbpm3.show_all()
self.xbim.show_all()
self.beamstop.show_all()
self.polarization.show_all()
self.polarization.show_all()
def show_plots(self, *extra: str, defaults: bool = True):
"""Open a BEC GUI window with waveform plots of the beam diagnostics,
e.g. for an ID gap scan. By default the XBPM3 blade signals and the
XBIM signal; further plots by name, e.g.
show_plots("xbpm3_position", "beamstop"), or only the named ones with
defaults=False. Calling it again rebuilds the plots in the open window."""
unknown = [p for p in extra if p not in self.PLOTS]
if unknown:
print(f"Unknown plot(s) {unknown}. Available: {', '.join(self.PLOTS)}")
return
gui = getattr(self._client, "gui", None)
if gui is None:
print("No BEC GUI is available in this session.")
return
names = list(self.DEFAULT_PLOTS) if defaults else []
names += [p for p in extra if p not in names]
plots = []
for plot_name in names:
title, curves = self.PLOTS[plot_name]
curves = [(d, s) for d, s in curves if _safe_dev(d) is not None]
if curves:
plots.append((plot_name, title, curves))
else:
print(f"Skipping plot '{plot_name}': its devices are not in the current config.")
if not plots:
return
name = self.PLOT_WINDOW_NAME
if name in gui.windows:
window = getattr(gui, name)
window.delete_all(timeout=self.GUI_RPC_TIMEOUT)
else:
gui.new(name, timeout=self.GUI_RPC_TIMEOUT)
time.sleep(1)
window = getattr(gui, name)
for plot_name, title, curves in plots:
waveform = window.new(
"Waveform", object_name=f"{plot_name}_plot", timeout=self.GUI_RPC_TIMEOUT
)
waveform.title = title
for device, signal in curves:
waveform.plot(device_y=device, signal_y=f"{device}_{signal}")
others = [p for p in self.PLOTS if p not in names]
if others:
options = ", ".join(repr(p) for p in others)
print(f"More plots: csaxs.diagnostics.show_plots({options})")
@@ -0,0 +1,127 @@
"""csaxs.diagnostics.show_plots(): one waveform per available diagnostic
device, in its own BEC GUI window."""
import types
import csaxs_bec.bec_ipython_client.plugins.cSAXS.diagnostics as diag_module
from csaxs_bec.bec_ipython_client.plugins.cSAXS.diagnostics import cSAXSDiagnostics
class _Waveform:
def __init__(self):
self.title = None
self.curves = []
def plot(self, device_y, signal_y):
self.curves.append((device_y, signal_y))
class _Window:
def __init__(self):
self.widgets = {}
self.deleted = 0
def new(self, widget, object_name, timeout):
assert widget == "Waveform"
self.widgets[object_name] = _Waveform()
return self.widgets[object_name]
def delete_all(self, timeout):
self.deleted += 1
self.widgets = {}
class _Gui:
def __init__(self):
self.windows = {}
def new(self, name, timeout):
self.windows[name] = _Window()
setattr(self, name, self.windows[name])
def _diagnostics(monkeypatch, devices):
fake_dev = types.SimpleNamespace(**{d: object() for d in devices})
monkeypatch.setattr(diag_module, "dev", fake_dev, raising=False)
monkeypatch.setattr(diag_module.time, "sleep", lambda _s: None)
gui = _Gui()
return cSAXSDiagnostics(types.SimpleNamespace(gui=gui)), gui
_ALL = [
"xbpm2_slowrb",
"xbpm3_slowrb",
"xbim_slowrb",
"beamstop_intensity",
"diode_horizontal_xbox1_slowrb",
"diode_vertical_xbox1_slowrb",
]
def test_show_plots_default_is_xbpm3_blades_and_xbim(monkeypatch):
diagnostics, gui = _diagnostics(monkeypatch, _ALL)
diagnostics.show_plots()
window = gui.windows["csaxs_diagnostics"]
assert list(window.widgets) == ["xbpm3_plot", "xbim_plot"]
assert window.widgets["xbpm3_plot"].curves == [
("xbpm3_slowrb", "xbpm3_slowrb_left_top"),
("xbpm3_slowrb", "xbpm3_slowrb_right_top"),
("xbpm3_slowrb", "xbpm3_slowrb_left_bot"),
("xbpm3_slowrb", "xbpm3_slowrb_right_bot"),
]
assert window.widgets["xbim_plot"].curves == [("xbim_slowrb", "xbim_slowrb_signal")]
assert window.widgets["xbim_plot"].title.startswith("XBIM")
def test_show_plots_extra_and_only(monkeypatch):
diagnostics, gui = _diagnostics(monkeypatch, _ALL)
diagnostics.show_plots("xbpm3_position", "polarization")
window = gui.windows["csaxs_diagnostics"]
assert list(window.widgets) == [
"xbpm3_plot",
"xbim_plot",
"xbpm3_position_plot",
"polarization_plot",
]
assert window.widgets["polarization_plot"].curves == [
("diode_horizontal_xbox1_slowrb", "diode_horizontal_xbox1_slowrb_signal"),
("diode_vertical_xbox1_slowrb", "diode_vertical_xbox1_slowrb_signal"),
]
diagnostics.show_plots("beamstop", defaults=False)
assert list(window.widgets) == ["beamstop_plot"]
def test_show_plots_skips_devices_not_in_config(monkeypatch, capsys):
diagnostics, gui = _diagnostics(monkeypatch, ["xbpm3_slowrb"])
diagnostics.show_plots("xbpm2")
assert list(gui.windows["csaxs_diagnostics"].widgets) == ["xbpm3_plot"]
out = capsys.readouterr().out
assert "Skipping plot 'xbim'" in out and "Skipping plot 'xbpm2'" in out
def test_show_plots_unknown_name(monkeypatch, capsys):
diagnostics, gui = _diagnostics(monkeypatch, _ALL)
diagnostics.show_plots("nonsense")
assert gui.windows == {}
assert "Unknown plot(s)" in capsys.readouterr().out
def test_show_plots_again_rebuilds_in_same_window(monkeypatch):
diagnostics, gui = _diagnostics(monkeypatch, _ALL)
diagnostics.show_plots()
diagnostics.show_plots()
window = gui.windows["csaxs_diagnostics"]
assert window.deleted == 1
assert list(window.widgets) == ["xbpm3_plot", "xbim_plot"]
def test_show_plots_without_gui(monkeypatch, capsys):
monkeypatch.setattr(diag_module, "dev", types.SimpleNamespace(), raising=False)
cSAXSDiagnostics(types.SimpleNamespace(gui=None)).show_plots()
assert "No BEC GUI" in capsys.readouterr().out