ENH/DEV: Formula Viewer for Archive Viewer

docs/source/widgets/archiver_timeplot.rst

@@ -67,3 +67,11 @@ ArchivePlotCurveItem
 .. autoclass:: pydm.widgets.archiver_time_plot.ArchivePlotCurveItem
    :members:
    :show-inheritance:
+
+#######################
+FormulaCurveItem
+#######################
+
+.. autoclass:: pydm.widgets.archiver_time_plot.FormulaCurveItem
+   :members:
+   :show-inheritance:

examples/archiver_time_plot/formula_curve_example.py

@@ -0,0 +1,63 @@
+from pydm import Display
+
+from qtpy import QtCore
+from qtpy.QtWidgets import QHBoxLayout, QApplication, QCheckBox, QLineEdit, QPushButton
+from pydm.widgets import PyDMArchiverTimePlot
+
+
+class archiver_time_plot_example(Display):
+    def __init__(self, parent=None, args=None, macros=None):
+        super(archiver_time_plot_example, self).__init__(parent=parent, args=args, macros=None)
+        self.app = QApplication.instance()
+        self.setup_ui()
+
+    def minimumSizeHint(self):
+        return QtCore.QSize(100, 100)
+
+    def ui_filepath(self):
+        return None
+
+    def setup_ui(self):
+        self.main_layout = QHBoxLayout()
+        self.setLayout(self.main_layout)
+        self.plot = PyDMArchiverTimePlot(background=[255, 255, 255, 255])
+        self.chkbx_live = QCheckBox()
+        self.chkbx_live.setChecked(True)
+        self.formula_box = QLineEdit(self)
+        self.formula_box.setText("f://{A}")
+        self.ok_button = QPushButton("OK", self)
+        self.ok_button.clicked.connect(self.set_formula)
+        self.main_layout.addWidget(self.formula_box)
+        self.main_layout.addWidget(self.ok_button)
+        self.main_layout.addWidget(self.chkbx_live)
+        self.main_layout.addWidget(self.plot)
+        self.curve = self.plot.addYChannel(
+            y_channel="ca://MTEST:Float",
+            name="name",
+            color="red",
+            yAxisName="Axis",
+            useArchiveData=True,
+            liveData=True,
+        )
+
+        pvdict = dict()
+        pvdict["A"] = self.curve
+        self.formula_curve = self.plot.addFormulaChannel(
+            formula=r"f://2*{A}",
+            pvs=pvdict,
+            yAxisName="Axis",
+            color="green",
+            useArchiveData=True,
+            liveData=True,
+        )
+        self.chkbx_live.stateChanged.connect(lambda x: self.set_live(self.curve, self.formula_curve, x))
+
+    @staticmethod
+    def set_live(curve, formula_curve, live):
+        curve.liveData = live
+        formula_curve.liveData = live
+
+    def set_formula(self):
+        print("assuming formula is valid, attempting to use formula")
+        self.formula_curve.formula = self.formula_box.text()
+        self.formula_curve.redrawCurve()

pydm/tests/widgets/test_archiver_timeplot.py

@@ -3,7 +3,7 @@
 from qtpy.QtCore import Slot
 
 from ..conftest import ConnectionSignals
-from ...widgets.archiver_time_plot import ArchivePlotCurveItem, PyDMArchiverTimePlot
+from ...widgets.archiver_time_plot import ArchivePlotCurveItem, PyDMArchiverTimePlot, FormulaCurveItem
 
 
 @pytest.mark.parametrize("address", ["ca://LINAC:PV1", "pva://LINAC:PV1", "LINAC:PV1"])
@@ -150,3 +150,53 @@ def test_request_data_from_archiver(qtbot):
     # Because the oldest live timestamp in the data buffer was 300, the ending timestamp for the request should
     # be one less than that.
     assert inspect_data_request.max_x == 299
+
+
+def test_formula_curve_item():
+    # Create two ArchivePlotCurveItems which we will make a few formulas out of
+    curve_item1 = ArchivePlotCurveItem()
+    curve_item1.archive_data_buffer = np.array([[100, 105, 110, 115, 120, 125], [2, 3, 4, 5, 6, 7]], dtype=float)
+    curve_item1.archive_points_accumulated = 6
+
+    curve_item2 = ArchivePlotCurveItem()
+    curve_item2.archive_data_buffer = np.array([[101, 106, 111, 116, 121, 126], [1, 2, 3, 4, 5, 6]], dtype=float)
+    curve_item2.archive_points_accumulated = 6
+
+    # Dictionary of PVS
+    curves1 = dict()
+    curves1["A"] = curve_item1
+    curves2 = dict()
+    curves2["A"] = curve_item1
+    curves2["B"] = curve_item2
+
+    formula1 = r"f://5*{A}"
+    formula2 = r"f://log({A})"
+    formula3 = r"f://{A}+{B}"
+    formula4 = r"f://{A}*{B}"
+
+    # Create the curves with the correct inputs
+    formula_curve_1 = FormulaCurveItem(formula=formula1, pvs=curves1)
+    formula_curve_2 = FormulaCurveItem(formula=formula2, pvs=curves1)
+    formula_curve_3 = FormulaCurveItem(formula=formula3, pvs=curves2)
+    formula_curve_4 = FormulaCurveItem(formula=formula4, pvs=curves2)
+
+    expected1 = np.array([[100, 105, 110, 115, 120, 125], [10, 15, 20, 25, 30, 35]], dtype=float)
+    expected2 = np.array([[100, 105, 110, 115, 120, 125], np.log([2, 3, 4, 5, 6, 7])], dtype=float)
+    expected3 = np.array(
+        [[101, 105, 106, 110, 111, 115, 116, 120, 121, 125], [3, 4, 5, 6, 7, 8, 9, 10, 11, 12]], dtype=float
+    )
+    expected4 = np.array(
+        [[101, 105, 106, 110, 111, 115, 116, 120, 121, 125], [2, 3, 6, 8, 12, 15, 20, 24, 30, 35]], dtype=float
+    )
+
+    # Evaluate them all
+    formula_curve_1.evaluate()
+    formula_curve_2.evaluate()
+    formula_curve_3.evaluate()
+    formula_curve_4.evaluate()
+
+    # They should match our precalculated outcomes
+    assert np.array_equal(formula_curve_1.archive_data_buffer, expected1)
+    assert np.array_equal(formula_curve_2.archive_data_buffer, expected2)
+    assert np.array_equal(formula_curve_3.archive_data_buffer, expected3)
+    assert np.array_equal(formula_curve_4.archive_data_buffer, expected4)