Create a new API for custom plot1d

bliss/flint/client/plots.py

@@ -14,6 +14,7 @@ from typing import Optional
 
 import numpy
 import gevent
+import contextlib
 
 from . import proxy
 from bliss.common import event
@@ -327,6 +328,9 @@ class Plot1D(_DataPlot):
         )
 
     def add_curve(self, x, y, **kwargs):
+        """
+        Create a curve in this plot.
+        """
         if x is None:
             x = numpy.arange(len(y))
         if y is None:
@@ -355,6 +359,58 @@ class Plot1D(_DataPlot):
         flint = self._flint
         flint.run_method(self._plot_id, "setYAxisLogarithmic", [value == "log"], {})
 
+    def clear_items(self):
+        """Remove all the items described in this plot
+
+        If no transaction was open, it will update the plot and refresh the plot
+        view.
+        """
+        self.submit("clearItems")
+
+    def add_curve_item(self, xname: str, yname: str, legend: str = None, **kwargs):
+        """Define a specific curve item
+
+        If no transaction was open, it will update the plot and refresh the plot
+        view.
+        """
+        self.submit("addCurveItem", xname, yname, legend=legend, **kwargs)
+
+    def remove_item(self, legend: str):
+        """Remove a specific item.
+
+        If no transaction was open, it will update the plot and refresh the plot
+        view.
+        """
+        self.submit("removeItem", legend)
+
+    def set_data(self, **kwargs):
+        """Set data named from keys with associated values.
+
+        If no transaction was open, it will update the plot and refresh the plot
+        view.
+        """
+        self.submit("setData", **kwargs)
+
+    def append_data(self, **kwargs):
+        """Append data named from keys with associated values.
+
+        If no transaction was open, it will update the plot and refresh the plot
+        view.
+        """
+        self.submit("appendData", **kwargs)
+
+    @contextlib.contextmanager
+    def transaction(self, resetzoom=True):
+        """Context manager to handle a set of changes and a single refresh of
+        the plot. This is needed cause the action are done on the plot
+        asynchronously"""
+        self.submit("setAutoUpdatePlot", False)
+        try:
+            yield
+        finally:
+            self.submit("setAutoUpdatePlot", True)
+            self.submit("updatePlot", resetzoom=resetzoom)
+
 
 class ScatterView(_DataPlot):
 

bliss/flint/custom_plots/silx_plots.py

@@ -7,11 +7,15 @@
 
 from __future__ import annotations
 
+import typing
 import numpy
+import logging
 
 from silx.gui import qt
 from silx.gui import plot as silx_plot
 
+_logger = logging.getLogger(__name__)
+
 
 class _DataWidget(qt.QWidget):
     def __init__(self, parent=None):
@@ -97,11 +101,102 @@ class Plot1D(_DataWidget):
     # Name of the method to add data to the plot
     METHOD = "addCurve"
 
+    class CurveItem(typing.NamedTuple):
+        xdata: str
+        ydata: str
+        style: typing.Dict[str, object]
+
+    def __init__(self, parent=None):
+        _DataWidget.__init__(self, parent=parent)
+        self.__items = {}
+        self.__autoUpdatePlot = True
+        self.__raiseOnException = False
+
+    def setRaiseOnException(self, raises):
+        """To simplify remote debug"""
+        self.__raiseOnException = raises
+
     def _createSilxWidget(self, parent):
         widget = silx_plot.Plot1D(parent=parent)
         widget.setDataMargins(0.05, 0.05, 0.05, 0.05)
         return widget
 
+    def setAutoUpdatePlot(self, update="bool"):
+        """Set to true to enable or disable update of plot for each changes of
+        the data or items"""
+        self.__autoUpdatePlot = update
+
+    def clearItems(self):
+        """Remove the item definitions"""
+        self.__items.clear()
+        self.__updatePlotIfNeeded()
+
+    def removeItem(self, legend: str):
+        """Remove a specific item by name"""
+        del self.__items[legend]
+        self.__updatePlotIfNeeded()
+
+    def addCurveItem(self, xdata: str, ydata: str, legend: str = None, **kwargs):
+        """Define an item which have to be displayed with the specified data
+        name
+        """
+        if legend is None:
+            legend = ydata + " -> " + xdata
+        self.__items[legend] = self.CurveItem(xdata, ydata, kwargs)
+        self.__updatePlotIfNeeded()
+
+    def setData(self, **kwargs):
+        dataDict = self.dataDict()
+        for k, v in kwargs.items():
+            dataDict[k] = v
+        self.__updatePlotIfNeeded()
+
+    def appendData(self, **kwargs):
+        dataDict = self.dataDict()
+        for k, v in kwargs.items():
+            d = dataDict.get(k, None)
+            if d is None:
+                d = v
+            else:
+                d = numpy.concatenate((d, v))
+            dataDict[k] = d
+        self.__updatePlotIfNeeded()
+
+    def clear(self):
+        super(Plot1D, self).clear()
+        self.__updatePlotIfNeeded()
+
+    def updatePlot(self, resetzoom: bool = True):
+        try:
+            self.__updatePlot()
+        except Exception:
+            _logger.error("Error while updating the plot", exc_info=True)
+            if self.__raiseOnException:
+                raise
+        if resetzoom:
+            self.resetZoom()
+
+    def __updatePlotIfNeeded(self):
+        if self.__autoUpdatePlot:
+            self.updatePlot(resetzoom=True)
+
+    def __updatePlot(self):
+        plot = self.silxPlot()
+        plot.clear()
+        dataDict = self.dataDict()
+        for legend, item in self.__items.items():
+            xData = dataDict.get(item.xdata)
+            yData = dataDict.get(item.ydata)
+            if xData is None or yData is None:
+                continue
+            if len(yData) != len(xData):
+                size = min(len(yData), len(xData))
+                xData = xData[0:size]
+                yData = yData[0:size]
+            if len(yData) == 0:
+                continue
+            plot.addCurve(xData, yData, legend=legend, **item.style, resetzoom=False)
+
 
 class Plot2D(_DataWidget):
     """Generic plot to display 2D data"""

tests/flint_client/test_custom_plots.py

@@ -59,7 +59,7 @@ def test_curveplot__create_empty(flint_session):
 def test_curveplot__bliss_1_8(flint_session):
     """Check custom plot curve API from BLISS <= 1.8"""
     flint = plot.get_flint()
-    p = flint.get_plot(plot_class="curve", name="foo-cp")
+    p = flint.get_plot(plot_class="curve", name="bliss-1.8")
 
     data1 = numpy.array([4, 5, 6])
     data2 = numpy.array([2, 5, 2])
@@ -85,6 +85,56 @@ def test_curveplot__bliss_1_8(flint_session):
     assert data == []
 
 
+def test_curveplot__bliss_1_9(flint_session):
+    """Check custom plot curve API from BLISS >= 1.9"""
+    flint = plot.get_flint()
+    p = flint.get_plot(plot_class="curve", name="bliss-1.9")
+    p.submit("setRaiseOnException", True)
+
+    y1 = numpy.array([4, 5, 6])
+    y2 = numpy.array([2, 5, 2])
+    x = numpy.array([1, 2, 3])
+
+    # Setup the items
+    p.add_curve_item("x", "y1", legend="item1", color="red", yaxis="left")
+    p.add_curve_item("x", "y2", legend="item2", color="blue", yaxis="right")
+    vrange = p.get_data_range()
+    assert vrange == [None, None, None]
+
+    # Update the data
+    p.set_data(x=x, y1=y1, y2=y2)
+    vrange = p.get_data_range()
+    assert vrange == [[1, 3], [4, 6], [2, 5]]
+
+    # Clear the data
+    p.clear_data()
+    vrange = p.get_data_range()
+    assert vrange == [None, None, None]
+
+    # Append the data
+    for i in range(len(x)):
+        p.append_data(x=x[i : i + 1], y1=y1[i : i + 1], y2=y2[i : i + 1])
+    vrange = p.get_data_range()
+    assert vrange == [[1, 3], [4, 6], [2, 5]]
+
+    # Or change the item layout
+    p.remove_item("item2")
+    vrange = p.get_data_range()
+    assert vrange == [[1, 3], [4, 6], None]
+
+    # Check transaction
+    p.clear_data()
+    with p.transaction(resetzoom=False):
+        p.set_data(x=x, y=y1)
+        vrange = p.get_data_range()
+        assert vrange == [None, None, None]
+        p.add_curve_item("x", "y", legend="item")
+        vrange = p.get_data_range()
+        assert vrange == [None, None, None]
+    vrange = p.get_data_range()
+    assert vrange == [[1, 3], [4, 6], None]
+
+
 def test_curveplot__reuse_plot(flint_session):
     flint = plot.get_flint()
     p = flint.get_plot(plot_class="curve", unique_name="foo-reuse")