diff --git a/bec_widgets/widgets/ivan_waveform/__init__.py b/bec_widgets/widgets/ivan_waveform/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/widgets/ivan_waveform/ivan_waveform.py b/bec_widgets/widgets/ivan_waveform/ivan_waveform.py new file mode 100644 index 00000000..89e32785 --- /dev/null +++ b/bec_widgets/widgets/ivan_waveform/ivan_waveform.py @@ -0,0 +1,154 @@ +import itertools +from threading import RLock + +import pyqtgraph as pg +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from qtpy.QtCore import Property as pyqtProperty, Slot as pyqtSlot + +from bec_widgets.utils.bec_dispatcher import BECDispatcher + +logger = bec_logger.logger + +pg.setConfigOptions(background="w", foreground="k", antialias=True) +COLORS = ["#fd7f6f", "#7eb0d5", "#b2e061", "#bd7ebe", "#ffb55a"] + + +class BECScanPlot(pg.GraphicsView): + def __init__(self, parent=None, background="default"): + super().__init__(parent, background) + BECDispatcher().connect_slot(self.on_scan_segment, MessageEndpoints.scan_segment()) + + self.view = pg.PlotItem() + self.setCentralItem(self.view) + + self._scanID = None + self._scanID_lock = RLock() + + self._x_channel = "" + self._y_channel_list = [] + + self.scan_curves = {} + self.dap_curves = {} + + self.x_channel = "samx" + self.y_channel_list = ["bpm3y", "bpm6y"] + + def reset_plots(self, _scan_segment, _metadata): + for plot_curve in {**self.scan_curves, **self.dap_curves}.values(): + plot_curve.setData(x=[], y=[]) + + @pyqtSlot(dict, dict) + def on_scan_segment(self, scan_segment, metadata): + # reset plots on scanID change + with self._scanID_lock: + scan_id = scan_segment["scan_id"] + if self._scanID != scan_id: + self._scanID = scan_id + self.reset_plots(scan_segment, metadata) + + if not self.x_channel: + return + + data = scan_segment["data"] + + if self.x_channel not in data: + logger.warning(f"Unknown channel `{self.x_channel}` for X data in {self.objectName()}") + return + + x_new = data[self.x_channel][self.x_channel]["value"] + for chan, plot_curve in self.scan_curves.items(): + if not chan: + continue + + if chan not in data: + logger.warning(f"Unknown channel `{chan}` for Y data in {self.objectName()}") + continue + + y_new = data[chan][chan]["value"] + x, y = plot_curve.getData() # TODO: is it a good approach? + if x is None: + x = [] + if y is None: + y = [] + + plot_curve.setData(x=[*x, x_new], y=[*y, y_new]) + + @pyqtSlot(dict, dict) + def redraw_dap(self, content, _metadata): + data = content["data"] + for chan, plot_curve in self.dap_curves.items(): + if not chan: + continue + + if chan not in data: + logger.warning(f"Unknown channel `{chan}` for DAP data in {self.objectName()}") + continue + + x_new = data[chan]["x"] + y_new = data[chan]["y"] + + plot_curve.setData(x=x_new, y=y_new) + + @pyqtProperty("QStringList") + def y_channel_list(self): + return self._y_channel_list + + @y_channel_list.setter + def y_channel_list(self, new_list): + bec_dispatcher = BECDispatcher() + # TODO: do we want to care about dap/not dap here? + chan_removed = [chan for chan in self._y_channel_list if chan not in new_list] + if chan_removed and chan_removed[0].startswith("dap."): + chan_removed = chan_removed[0].partition("dap.")[-1] + chan_removed_ep = MessageEndpoints.processed_data(chan_removed) + bec_dispatcher.disconnect_slot(self.redraw_dap, chan_removed_ep) + + self._y_channel_list = new_list + + # Prepare plot for a potentially different list of y channels + self.view.clear() + + self.view.addLegend() + colors = itertools.cycle(COLORS) + + for y_chan in new_list: + if y_chan.startswith("dap."): + y_chan = y_chan.partition("dap.")[-1] + curves = self.dap_curves + y_chan_ep = MessageEndpoints.processed_data(y_chan) + bec_dispatcher.connect_slot(self.redraw_dap, y_chan_ep) + else: + curves = self.scan_curves + + curves[y_chan] = self.view.plot( + x=[], y=[], pen=pg.mkPen(color=next(colors), width=2), name=y_chan + ) + + if len(new_list) == 1: + self.view.setLabel("left", new_list[0]) + + @pyqtProperty(str) + def x_channel(self): + return self._x_channel + + @x_channel.setter + def x_channel(self, new_val): + self._x_channel = new_val + self.view.setLabel("bottom", new_val) + + +if __name__ == "__main__": + import sys + + from qtpy.QtWidgets import QApplication + + app = QApplication(sys.argv) + + plot = BECScanPlot() + plot.x_channel = "samx" + plot.y_channel_list = ["bpm3y", "bpm6y"] + + plot.show() + + sys.exit(app.exec()) diff --git a/bec_widgets/widgets/ivan_waveform/ivan_waveform.pyproject b/bec_widgets/widgets/ivan_waveform/ivan_waveform.pyproject new file mode 100644 index 00000000..f8092aab --- /dev/null +++ b/bec_widgets/widgets/ivan_waveform/ivan_waveform.pyproject @@ -0,0 +1 @@ +{'files': ['ivan_waveform.py']} \ No newline at end of file diff --git a/bec_widgets/widgets/ivan_waveform/ivan_waveform_plugin.py b/bec_widgets/widgets/ivan_waveform/ivan_waveform_plugin.py new file mode 100644 index 00000000..64e90788 --- /dev/null +++ b/bec_widgets/widgets/ivan_waveform/ivan_waveform_plugin.py @@ -0,0 +1,56 @@ +from qtpy.QtGui import QIcon +from qtpy.QtDesigner import QDesignerCustomWidgetInterface + +from bec_widgets.widgets.ivan_waveform.ivan_waveform import BECScanPlot + + +class BECScanPlotPlugin(QDesignerCustomWidgetInterface): + def __init__(self, parent=None): + super().__init__(parent) + + self._initialized = False + + def initialize(self, formEditor): + if self._initialized: + return + + self._initialized = True + + def isInitialized(self): + return self._initialized + + def createWidget(self, parent): + return BECScanPlot(parent) + + def name(self): + return "BECScanPlot" + + def group(self): + return "BEC widgets" + + def icon(self): + return QIcon() + + def toolTip(self): + return "BEC plot for scans" + + def whatsThis(self): + return "BEC plot for scans" + + def isContainer(self): + return False + + def domXml(self): + return ( + '\n' + ' \n' + " BEC plot for scans\n" + " \n" + ' \n' + " BEC plot for scans in Python using PyQt.\n" + " \n" + "\n" + ) + + def includeFile(self): + return "scan_plot" diff --git a/bec_widgets/widgets/ivan_waveform/register_ivan_waveform.py b/bec_widgets/widgets/ivan_waveform/register_ivan_waveform.py new file mode 100644 index 00000000..f55ce376 --- /dev/null +++ b/bec_widgets/widgets/ivan_waveform/register_ivan_waveform.py @@ -0,0 +1,15 @@ +def main(): # pragma: no cover + from qtpy import PYSIDE6 + + if not PYSIDE6: + print("PYSIDE6 is not available in the environment. Cannot patch designer.") + return + from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection + + from bec_widgets.widgets.ivan_waveform.ivan_waveform_plugin import BECScanPlotPlugin + + QPyDesignerCustomWidgetCollection.addCustomWidget(BECScanPlotPlugin()) + + +if __name__ == "__main__": # pragma: no cover + main()