Share waveform data and plot navigation and retain processing results

python/tests/test_generator_data.py

@@ -0,0 +1,37 @@
+"""Waveform data helpers run independently of SETGUI and Qt."""
+import io
+import json
+import unittest
+from set_devices.generator_data import (
+    parse_signal_csv, signal_csv_channels, preset_points, generate,
+    load_recipe, resample_signal, nearest_f407_sample_rate,
+)
+from set_devices.waveform import recipe
+from set_devices.plot_processing import Axis, Series, Snapshot, prepare, process, write_csv
+
+
+class GeneratorDataTests(unittest.TestCase):
+    def test_processing_csv_roundtrip_into_generator(self):
+        request = prepare(Snapshot((Series('a', 'A', [(1000, 0), (1010, 2), (1020, 0)]),),
+                                   Axis('Время', 'мс', 'unix_ms')), 'a', 'linear', 21)
+        output = io.StringIO()
+        write_csv(process(request), output)
+        self.assertEqual(1, len(signal_csv_channels(output.getvalue())))
+        points, rate = parse_signal_csv(output.getvalue())
+        self.assertEqual((0, 0), points[0])
+        self.assertEqual((20, 0), points[-1])
+        self.assertIsNone(rate)
+
+    def test_samples_recipe_and_resampling(self):
+        points = preset_points(0, 3.3)
+        samples = resample_signal(points, 10000)
+        self.assertEqual(200, len(samples))
+        wave = generate(samples, 10000, method='samples')
+        restored, settings = load_recipe(json.dumps(recipe(wave.points, 10000, 3.3, 'samples', 2)))
+        self.assertEqual(wave.points, restored)
+        self.assertEqual('samples', settings['method'])
+        self.assertEqual(10000, nearest_f407_sample_rate(10001))
+
+
+if __name__ == '__main__':
+    unittest.main()

python/tests/test_plot_processing_qt.py

@@ -36,14 +36,81 @@ class ProcessingQtTests(unittest.TestCase):
         self.assertIsNone(self.panel._job)
         self.assertIsNotNone(self.panel.curve, self.panel.status.text())
 
-    def test_all_methods_are_available_and_parameter_changes_clear_result(self):
-        for method in ("polynomial", "linear", "pchip", "spline"):
+    def test_all_methods_accumulate_and_parameter_changes_keep_results(self):
+        for index, method in enumerate(("polynomial", "linear", "pchip", "spline")):
             self.panel.method.setCurrentIndex(self.panel.method.findData(method))
             self.calculate()
             self.assertEqual(method, self.panel.curve.request.method)
             self.panel.count.setValue(self.panel.count.value() + 1)
-            self.assertIsNone(self.panel.curve)
-            self.assertFalse(self.panel.export_button.isEnabled())
+            self.assertEqual(index + 1, len(self.panel.curves))
+            self.assertTrue(self.panel.export_button.isEnabled())
+        specs, data = self.panel.overlay()
+        self.assertEqual(4, len(specs))
+        self.assertEqual(4, len(data))
+        self.assertEqual(4, len({spec[3] for spec in specs}))
+        self.panel.clear()
+        self.assertEqual((), self.panel.curves)
+        self.assertIsNone(self.panel.curve)
+
+    def test_repeated_request_does_not_duplicate_curve(self):
+        self.calculate()
+        self.calculate()
+        self.assertEqual(1, len(self.panel.curves))
+
+    def test_repeated_fft_workers_keep_gui_owned_signal_emitters(self):
+        self.panel.set_snapshot(replace(self.snapshot, axis=Axis("Frequency", "Hz"), source="FFT"))
+        for _ in range(5):
+            for method in ("linear", "pchip", "spline", "polynomial"):
+                self.panel.method.setCurrentIndex(self.panel.method.findData(method))
+                self.panel.calculate()
+                self.assertIs(self.panel, self.panel._job.signals.parent())
+                deadline = time.monotonic() + 5
+                while self.panel._job is not None and time.monotonic() < deadline:
+                    self.app.processEvents()
+                    time.sleep(.002)
+                self.assertIsNone(self.panel._job)
+                self.assertEqual(method, self.panel.curve.request.method)
+        self.assertEqual(4, len(self.panel.curves))
+
+    def test_switching_channel_preserves_results_and_invalidates_only_changed_source(self):
+        snapshot = replace(self.snapshot, series=self.snapshot.series +
+                           (Series("b", "B", [(0, 2), (1, 3), (2, 2)]),))
+        self.panel.set_snapshot(snapshot)
+        self.calculate()
+        first = self.panel.curve
+        self.panel.channel.setCurrentIndex(1)
+        self.panel.set_snapshot(snapshot)
+        self.assertEqual((first,), self.panel.curves)
+        self.calculate()
+        second = self.panel.curve
+        self.assertEqual((first, second), self.panel.curves)
+        second_spec = self.panel.overlay()[0][1]
+        self.panel.set_snapshot(replace(snapshot, series=(
+            Series("a", "A", [(0, 5), (1, 6)]), snapshot.series[1])))
+        self.assertEqual((second,), self.panel.curves)
+        self.assertEqual(second_spec, self.panel.overlay()[0][0])
+
+    def test_navigation_during_second_calculation_preserves_both_results(self):
+        self.panel.preserve_on_view_change = True
+        self.calculate()
+        first = self.panel.curve
+        self.panel.method.setCurrentIndex(self.panel.method.findData("linear"))
+        with patch("set_devices.qt_ports.plot_processing.QThreadPool"):
+            self.panel.calculate()
+        job = self.panel._job
+        self.panel.set_snapshot(replace(self.snapshot, x_range=(0, 1)))
+        self.panel._finished(job.signature, process(job.signature), "")
+        self.assertEqual(2, len(self.panel.curves))
+        self.assertIs(first, self.panel.curves[0])
+
+    def test_failed_calculation_keeps_previous_curve(self):
+        self.calculate()
+        previous = self.panel.curve
+        self.panel.count.setValue(123)
+        with patch("set_devices.qt_ports.plot_processing.QThreadPool"):
+            self.panel.calculate()
+        self.panel._finished(self.panel._job.signature, None, "test error")
+        self.assertEqual((previous,), self.panel.curves)
 
     def test_source_units_and_blocking_invalidate_result(self):
         for changed in (replace(self.snapshot, axis=Axis("Frequency", "Hz")),