diff --git a/csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py b/csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py index 9c67462a..d0cc7ae1 100644 --- a/csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py +++ b/csaxs_bec/bec_ipython_client/plugins/cSAXS/cSAXS.py @@ -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)", diff --git a/csaxs_bec/bec_ipython_client/plugins/cSAXS/diagnostics.py b/csaxs_bec/bec_ipython_client/plugins/cSAXS/diagnostics.py index 0b1c7815..c0469d93 100644 --- a/csaxs_bec/bec_ipython_client/plugins/cSAXS/diagnostics.py +++ b/csaxs_bec/bec_ipython_client/plugins/cSAXS/diagnostics.py @@ -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() \ No newline at end of file + 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})") diff --git a/tests/tests_bec_ipython_client/test_csaxs_diagnostics_plots.py b/tests/tests_bec_ipython_client/test_csaxs_diagnostics_plots.py new file mode 100644 index 00000000..1edb4fe2 --- /dev/null +++ b/tests/tests_bec_ipython_client/test_csaxs_diagnostics_plots.py @@ -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