fix: prel. support for config_plotter with widgets instead of items

Automatically generated by python-semantic-release

.git_hooks/pre-commit

@@ -0,0 +1,2 @@
+black --line-length=100 $(git diff --cached --name-only --diff-filter=ACM -- '***.py')
+git add $(git diff --cached --name-only --diff-filter=ACM -- '***.py')

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 <!--next-version-placeholder-->
 
+## v0.2.0 (2023-07-13)
+
+### Feature
+
+* Move ivan's qtwidgets to bec-widgets ([`34e5ed2`](https://gitlab.psi.ch/bec/bec-widgets/-/commit/34e5ed2cf7e6128a51c110db8870d9560f2b2831))
+
 ## v0.1.0 (2023-07-11)
 
 ### Feature

bec_widgets/cli.py

@@ -0,0 +1,98 @@
+import argparse
+import os
+from threading import RLock
+
+from PyQt5 import uic
+from PyQt5.QtCore import pyqtSignal
+from PyQt5.QtWidgets import QApplication, QMainWindow
+from scan_plot import BECScanPlot
+
+from bec_lib.core import BECMessage, MessageEndpoints, RedisConnector
+
+
+class BEC_UI(QMainWindow):
+    new_scan_data = pyqtSignal(dict)
+    new_dap_data = pyqtSignal(dict)  # signal per proc instance?
+    new_scan = pyqtSignal()
+
+    def __init__(self, uipath):
+        super().__init__()
+        self._scan_channels = set()
+        self._dap_channels = set()
+
+        self._scan_thread = None
+        self._dap_threads = []
+
+        ui = uic.loadUi(uipath, self)
+
+        _, fname = os.path.split(uipath)
+        self.setWindowTitle(fname)
+
+        for sp in ui.findChildren(BECScanPlot):
+            for chan in (sp.x_channel, *sp.y_channel_list):
+                if chan.startswith("dap."):
+                    chan = chan.partition("dap.")[-1]
+                    self._dap_channels.add(chan)
+                else:
+                    self._scan_channels.add(chan)
+
+            sp.initialize()  # TODO: move this elsewhere?
+
+            self.new_scan_data.connect(sp.redraw_scan)  # TODO: merge
+            self.new_dap_data.connect(sp.redraw_dap)
+            self.new_scan.connect(sp.clearData)
+
+        # Scan setup
+        self._scan_id = None
+        scan_lock = RLock()
+
+        def _scan_cb(msg):
+            msg = BECMessage.ScanMessage.loads(msg.value)
+            with scan_lock:
+                scan_id = msg[0].content["scanID"]
+                if self._scan_id != scan_id:
+                    self._scan_id = scan_id
+                    self.new_scan.emit()
+            self.new_scan_data.emit(msg[0].content["data"])
+
+        bec_connector = RedisConnector("localhost:6379")
+
+        if self._scan_channels:
+            scan_readback = MessageEndpoints.scan_segment()
+            self._scan_thread = bec_connector.consumer(
+                topics=scan_readback,
+                cb=_scan_cb,
+            )
+            self._scan_thread.start()
+
+        # DAP setup
+        def _proc_cb(msg):
+            msg = BECMessage.ProcessedDataMessage.loads(msg.value)
+            self.new_dap_data.emit(msg.content["data"])
+
+        if self._dap_channels:
+            for chan in self._dap_channels:
+                proc_ep = MessageEndpoints.processed_data(chan)
+                dap_thread = bec_connector.consumer(topics=proc_ep, cb=_proc_cb)
+                dap_thread.start()
+                self._dap_threads.append(dap_thread)
+
+        self.show()
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        prog="bec-pyqt", formatter_class=argparse.ArgumentDefaultsHelpFormatter
+    )
+
+    parser.add_argument("uipath", type=str, help="Path to a BEC ui file")
+
+    args, rem = parser.parse_known_args()
+
+    app = QApplication(rem)
+    BEC_UI(args.uipath)
+    app.exec_()
+
+
+if __name__ == "__main__":
+    main()

bec_widgets/config_plotter.py

@@ -2,11 +2,12 @@ from typing import List
 
 import numpy as np
 import pyqtgraph as pg
+from PyQt5.QtWidgets import QApplication, QGridLayout, QSizePolicy, QWidget
 from pyqtgraph import mkPen
-from pyqtgraph.Qt import QtCore, QtWidgets
+from pyqtgraph.Qt import QtCore
 
 
-class ConfigPlotter(pg.GraphicsWidget):
+class ConfigPlotter(QWidget):
     """
     ConfigPlotter is a widget that can be used to plot data from multiple channels
     in a grid layout. The layout is specified by a list of dicts, where each dict
@@ -28,7 +29,7 @@ class ConfigPlotter(pg.GraphicsWidget):
     """
 
     def __init__(self, configs: List[dict], parent=None):
-        super().__init__(parent)
+        super(ConfigPlotter, self).__init__()
         self.configs = configs
         self.plots = {}
         self._init_ui()
@@ -37,63 +38,73 @@ class ConfigPlotter(pg.GraphicsWidget):
     def _init_ui(self):
         pg.setConfigOption("background", "w")
         pg.setConfigOption("foreground", "k")
+        self.layout = QGridLayout()
+        self.setLayout(self.layout)
         self.pen = mkPen(color=(56, 76, 107), width=4, style=QtCore.Qt.SolidLine)
-
-        self.view = pg.GraphicsView()
-        self.view.setAntialiasing(True)
-        self.view.show()
-
-        self.layout = pg.GraphicsLayout()
-        self.view.setCentralWidget(self.layout)
+        self.show()
 
     def _init_plots(self):
         for config in self.configs:
             channels = config["config"]["channels"]
             for channel in channels:
-                item = pg.PlotItem()
-                self.layout.addItem(
-                    item,
-                    row=config["y"],
-                    col=config["x"],
-                    rowspan=config["rows"],
-                    colspan=config["cols"],
-                )
-
                 # call the corresponding init function, e.g. init_plotitem
                 init_func = getattr(self, f"init_{config['config']['item']}")
-                init_func(channel, config["config"], item)
+                init_func(channel, config)
 
                 # self.init_ImageItem(channel, config["config"], item)
 
-    def init_PlotItem(self, channel: str, config: dict, item: pg.GraphicsItem):
+    def init_PlotItem(self, channel: str, config: dict):
         """
         Initialize a PlotItem
 
         Args:
             channel(str): channel to plot
             config(dict): config dict for the channel
-            item(pg.GraphicsItem): PlotItem to plot the data
         """
         # pylint: disable=invalid-name
-        plot_data = item.plot(np.random.rand(100), pen=self.pen)
-        item.setLabel("left", channel)
-        self.plots[channel] = {"item": item, "plot_data": plot_data}
+        plot_widget = pg.PlotWidget()
+        plot_widget.setSizePolicy(QSizePolicy.Fixed, QSizePolicy.Fixed)
+        self.layout.addWidget(plot_widget, config["y"], config["x"], config["rows"], config["cols"])
+        plot_data = plot_widget.plot(np.random.rand(100), pen=self.pen)
+        # item.setLabel("left", channel)
+        # self.plots[channel] = {"item": item, "plot_data": plot_data}
 
-    def init_ImageItem(self, channel: str, config: dict, item: pg.GraphicsItem):
+    def init_ImageItem(self, channel: str, config: dict):
         """
         Initialize an ImageItem
 
         Args:
             channel(str): channel to plot
             config(dict): config dict for the channel
-            item(pg.GraphicsItem): ImageItem to plot the data
         """
         # pylint: disable=invalid-name
+        item = pg.PlotItem()
+        self.layout.addItem(
+            item,
+            row=config["y"],
+            col=config["x"],
+            rowspan=config["rows"],
+            colspan=config["cols"],
+        )
         img = pg.ImageItem()
         item.addItem(img)
         img.setImage(np.random.rand(100, 100))
         self.plots[channel] = {"item": item, "plot_data": img}
 
+    def init_ImageView(self, channel: str, config: dict):
+        """
+        Initialize an ImageView
+
+        Args:
+            channel(str): channel to plot
+            config(dict): config dict for the channel
+        """
+        # pylint: disable=invalid-name
+        img = pg.ImageView()
+        img.setImage(np.random.rand(100, 100))
+        self.layout.addWidget(img, config["y"], config["x"], config["rows"], config["cols"])
+        self.plots[channel] = {"item": img, "plot_data": img}
+
 
 if __name__ == "__main__":
     import sys
@@ -118,10 +129,10 @@ if __name__ == "__main__":
             "rows": 2,
             "y": 0,
             "x": 1,
-            "config": {"channels": ["c"], "label_xy": ["", "c"], "item": "ImageItem"},
+            "config": {"channels": ["c"], "label_xy": ["", "c"], "item": "ImageView"},
         },
     ]
 
-    app = QtWidgets.QApplication(sys.argv)
+    app = QApplication(sys.argv)
     win = ConfigPlotter(CONFIG)
     pg.exec()

bec_widgets/readme.md

@@ -0,0 +1,4 @@
+Add/modify the path in the following variable to make the plugin avaiable in Qt Designer:
+```
+$ export PYQTDESIGNERPATH=/<path to repo>/bec/bec_qtplugin:$PYQTDESIGNERPATH
+```

bec_widgets/scan_plot.py

@@ -0,0 +1,106 @@
+import itertools
+
+import pyqtgraph as pg
+from PyQt5.QtCore import pyqtProperty, pyqtSlot
+
+from bec_lib.core.logger import bec_logger
+
+logger = bec_logger.logger
+
+
+pg.setConfigOptions(background="w", foreground="k", antialias=True)
+COLORS = ["#fd7f6f", "#7eb0d5", "#b2e061", "#bd7ebe", "#ffb55a"]
+
+
+class BECScanPlot(pg.PlotWidget):
+    def __init__(self, parent=None, background="default"):
+        super().__init__(parent, background)
+
+        self._x_channel = ""
+        self._y_channel_list = []
+
+        self.scan_curves = {}
+        self.dap_curves = {}
+
+    def initialize(self):
+        plot_item = self.getPlotItem()
+        plot_item.addLegend()
+        colors = itertools.cycle(COLORS)
+
+        for y_chan in self.y_channel_list:
+            if y_chan.startswith("dap."):
+                y_chan = y_chan.partition("dap.")[-1]
+                curves = self.dap_curves
+            else:
+                curves = self.scan_curves
+
+            curves[y_chan] = plot_item.plot(
+                x=[], y=[], pen=pg.mkPen(color=next(colors), width=2), name=y_chan
+            )
+
+        plot_item.setLabel("bottom", self._x_channel)
+        if len(self.scan_curves) == 1:
+            plot_item.setLabel("left", next(iter(self.scan_curves)))
+
+    @pyqtSlot()
+    def clearData(self):
+        for plot_curve in {**self.scan_curves, **self.dap_curves}.values():
+            plot_curve.setData(x=[], y=[])
+
+    @pyqtSlot(dict)
+    def redraw_scan(self, data):
+        if not self.x_channel:
+            return
+
+        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)
+    def redraw_dap(self, 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):
+        self._y_channel_list = new_list
+
+    @pyqtProperty(str)
+    def x_channel(self):
+        return self._x_channel
+
+    @x_channel.setter
+    def x_channel(self, new_val):
+        self._x_channel = new_val

bec_widgets/scan_plot_plugin.py

@@ -0,0 +1,55 @@
+from PyQt5.QtDesigner import QPyDesignerCustomWidgetPlugin
+from PyQt5.QtGui import QIcon
+from scan_plot import BECScanPlot
+
+
+class BECScanPlotPlugin(QPyDesignerCustomWidgetPlugin):
+    def __init__(self, parent=None):
+        super(BECScanPlotPlugin, self).__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 (
+            '<widget class="BECScanPlot" name="BECScanPlot">\n'
+            ' <property name="toolTip" >\n'
+            "  <string>BEC plot for scans</string>\n"
+            " </property>\n"
+            ' <property name="whatsThis" >\n'
+            "  <string>BEC plot for scans in Python using PyQt.</string>\n"
+            " </property>\n"
+            "</widget>\n"
+        )
+
+    def includeFile(self):
+        return "scan_plot"

setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup
 
-__version__ = "0.1.0"
+__version__ = "0.2.0"
 
 if __name__ == "__main__":
     setup(