OWConcurrentProjectionWidget: Add an example

Author: VesnaT

Orange/widgets/utils/tests/concurrent_example.py

@@ -0,0 +1,150 @@
+from typing import Any, Tuple, Optional
+from types import SimpleNamespace as namespace
+from functools import partial
+
+import numpy as np
+
+from Orange.data import Table
+from Orange.widgets import gui
+from Orange.widgets.settings import Setting
+from Orange.widgets.utils.concurrent import TaskState, ConcurrentWidgetMixin
+from Orange.widgets.utils.widgetpreview import WidgetPreview
+from Orange.widgets.visualize.utils.widget import OWDataProjectionWidget
+
+
+class Results(namespace):
+    embedding = None  # type: Optional[np.ndarray]
+
+
+class Runner:
+    @staticmethod
+    def run(data: Table, embedding: Optional[np.ndarray], state: TaskState):
+        res = Results(embedding=embedding)
+
+        # simulate wasteful calculation (increase 'steps')
+        step, steps = 0, 10
+        state.set_status("Calculating...")
+        while step < steps:
+            for i in range(steps):
+                x_data = np.array(np.mean(data.X, axis=1))
+                if x_data.ndim == 2:
+                    x_data = x_data.ravel()
+                y_data = np.ones(len(x_data))
+                y_data[::2] = step % 2
+                y_data = np.random.rand(len(x_data))
+                embedding = np.vstack((x_data, y_data)).T
+            step += 1
+            if step % (steps / 10) == 0:
+                state.set_progress_value(100 * step / steps)
+
+            if state.is_interruption_requested():
+                return res
+
+            res.embedding = embedding
+            state.set_partial_results(("embedding", res.embedding))
+        return res
+
+
+class OWConcurrentProjectionWidget(OWDataProjectionWidget,
+                                   ConcurrentWidgetMixin):
+    name = "Projection"
+    param = Setting(0)
+
+    def __init__(self):
+        OWDataProjectionWidget.__init__(self)
+        ConcurrentWidgetMixin.__init__(self)
+        self.embedding = None  # type: Optional[np.ndarray]
+
+    # GUI
+    def _add_controls(self):
+        box = gui.vBox(self.controlArea, True)
+        gui.comboBox(
+            box, self, "param", label="Parameter:",
+            items=["Param A", "Param B"], labelWidth=80,
+            callback=self.__param_combo_changed
+        )
+        self.run_button = gui.button(box, self, "Start", self._toggle_run)
+        super()._add_controls()
+
+    def __param_combo_changed(self):
+        super().start()
+
+    def _toggle_run(self):
+        if self.data is None:
+            return
+
+        # Pause task
+        if self.task is not None:
+            self.cancel()
+            self.run_button.setText("Resume")
+            self.commit()
+        # Resume task
+        else:
+            self.start()
+
+    # extend ConcurrentWidgetMixin
+    def _start_partial(self, state: TaskState):
+        return partial(Runner.run, self.data,
+                       embedding=self.embedding, state=state)
+
+    def _set_partial_results(self, result: Tuple[str, Any]):
+        which, res = result
+        if which == "embedding":
+            assert isinstance(res, np.ndarray) and len(res) == len(self.data)
+            first_result = self.embedding is None
+            self.embedding = res
+            if first_result:
+                self.setup_plot()
+            else:
+                self.graph.update_coordinates()
+                self.graph.update_density()
+        else:
+            assert False, which
+
+    def start(self):
+        self.run_button.setText("Stop")
+        super().start()
+
+    def finish(self, future):
+        super().finish(future)
+        self.run_button.setText("Start")
+        self.commit()
+
+    # extend OWDataProjectionWidget
+    def set_data(self, data: Table):
+        super().set_data(data)
+        if self._invalidated:
+            self.init_projection()
+            self.start()
+
+    def init_projection(self):
+        if self.data is None:
+            return
+        x_data = np.array(np.mean(self.data.X, axis=1))
+        if x_data.ndim == 2:
+            x_data = x_data.ravel()
+        y_data = np.ones(len(x_data))
+        self.embedding = np.vstack((x_data, y_data)).T
+
+    def get_embedding(self):
+        if self.embedding is None:
+            self.valid_data = None
+            return None
+
+        self.valid_data = np.all(np.isfinite(self.embedding), 1)
+        return self.embedding
+
+    def clear(self):
+        super().clear()
+        self.cancel_task_no_wait()
+        self.embedding = None
+
+    def onDeleteWidget(self):
+        self.on_delete_widget()
+        super().onDeleteWidget()
+
+
+if __name__ == "__main__":
+    table = Table("iris")
+    WidgetPreview(OWConcurrentProjectionWidget).run(
+        set_data=table, set_subset_data=table[::10])

Orange/widgets/utils/tests/test_concurrent_example.py

@@ -0,0 +1,43 @@
+# Test methods with long descriptive names can omit docstrings
+# pylint: disable=missing-docstring
+import unittest
+from Orange.widgets.tests.base import (
+    WidgetTest, WidgetOutputsTestMixin, ProjectionWidgetTestMixin
+)
+from Orange.widgets.utils.tests.concurrent_example import (
+    OWConcurrentProjectionWidget
+)
+
+
+class TestOWConcurrentProjectionWidget(WidgetTest, ProjectionWidgetTestMixin,
+                                       WidgetOutputsTestMixin):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        WidgetOutputsTestMixin.init(cls)
+
+        cls.signal_name = "Data"
+        cls.signal_data = cls.data
+        cls.same_input_output_domain = False
+
+    def setUp(self):
+        self.widget = self.create_widget(OWConcurrentProjectionWidget)
+
+    def test_button_no_data(self):
+        self.widget.run_button.click()
+        self.assertEqual(self.widget.run_button.text(), "Start")
+
+    def test_button_with_data(self):
+        self.send_signal(self.widget.Inputs.data, self.data)
+        self.assertEqual(self.widget.run_button.text(), "Stop")
+        self.wait_until_stop_blocking()
+        self.assertEqual(self.widget.run_button.text(), "Start")
+
+    def test_button_toggle(self):
+        self.send_signal(self.widget.Inputs.data, self.data)
+        self.widget.run_button.click()
+        self.assertEqual(self.widget.run_button.text(), "Resume")
+
+
+if __name__ == "__main__":
+    unittest.main()