Refactor

Author: VesnaT

Orange/widgets/tests/base.py

@@ -1018,12 +1018,16 @@ def test_invalidated_embedding(self, timeout=DEFAULT_TIMEOUT):
 
     def test_saved_selection(self, timeout=DEFAULT_TIMEOUT):
         self.send_signal(self.widget.Inputs.data, self.data)
-        self.wait_until_stop_blocking()
+        if self.widget.isBlocking():
+            spy = QSignalSpy(self.widget.blockingStateChanged)
+            self.assertTrue(spy.wait(timeout))
         self.widget.graph.select_by_indices(list(range(0, len(self.data), 10)))
         settings = self.widget.settingsHandler.pack_data(self.widget)
         w = self.create_widget(self.widget.__class__, stored_settings=settings)
         self.send_signal(self.widget.Inputs.data, self.data, widget=w)
-        self.wait_until_stop_blocking(widget=w)
+        if w.isBlocking():
+            spy = QSignalSpy(w.blockingStateChanged)
+            self.assertTrue(spy.wait(timeout))
         self.assertEqual(np.sum(w.graph.selection), 15)
         np.testing.assert_equal(self.widget.graph.selection, w.graph.selection)
 

Orange/widgets/utils/concurrent.py

@@ -813,34 +813,33 @@ def __init__(self, *args):
             self.partial_result_ready, Qt.QueuedConnection)
 
     @property
-    def future(self):
-        # type: () -> Future
+    def future(self) -> Future:
         return self.__future
 
-    def set_status(self, text):
+    def set_status(self, text: str):
         self._p_status_changed.emit(text)
 
-    def set_progress_value(self, value):
+    def set_progress_value(self, value: float):
         if round(value, 1) > round(self.__progress, 1):
             # Only emit progress when it has changed sufficiently
             self._p_progress_changed.emit(value)
             self.__progress = value
 
-    def set_partial_results(self, value):
+    def set_partial_result(self, value: Any):
         self._p_partial_result_ready.emit(value)
 
-    def is_interruption_requested(self):
+    def is_interruption_requested(self) -> bool:
         return self.__interruption_requested
 
-    def start(self, executor, func=None):
-        # type: (concurrent.futures.Executor, Callable[[], Any]) -> Future
+    def start(self, executor: concurrent.futures.Executor,
+              func: Callable[[], Any] = None) -> Future:
         assert self.future is None
         assert not self.__interruption_requested
         self.__future = executor.submit(func)
         self.watcher.setFuture(self.future)
         return self.future
 
-    def cancel(self):
+    def cancel(self) -> bool:
         assert not self.__interruption_requested
         self.__interruption_requested = True
         if self.future is not None:
@@ -861,22 +860,55 @@ def __init__(self):
         self.__task = None  # type: Optional[TaskState]
 
     @property
-    def task(self):
+    def task(self) -> TaskState:
         return self.__task
 
-    def _prepare_task(self, state: TaskState) -> Callable[[], Any]:
+    def on_partial_result(self, result: Any) -> None:
+        """ Invoked from runner (by state) to send the partial results
+        The method should handle partial results, i.e. show them in the plot.
+
+        :param result: any data structure to hold temporary result
+        """
         raise NotImplementedError
 
-    def _set_partial_results(self, result: Any) -> None:
+    def on_done(self, result: Any) -> None:
+        """ Invoked when task is done.
+        The method should re-set the result (to double check it) and
+        perform operations with obtained results, eg. send data to the output.
+
+        :param result: any data structure to hold temporary result
+        """
         raise NotImplementedError
 
-    def _set_results(self, results: Any) -> None:
-        # NOTE: All of these have already been set by _set_partial_results,
-        # we double check that they are aliases
-        for key in results.__dict__:
-            value = getattr(results, key)
-            if value is not None:
-                setattr(self, key, value)
+    def start(self, task: Callable, *args, **kwargs):
+        """ Call from derived class to start the task.
+        :param task: runner - a method to run in a thread - should accept
+        `state` parameter
+        """
+        self.__cancel_task(wait=False)
+
+        if self.data is None:
+            self.__set_state_ready()
+            return
+
+        assert callable(task), "`task` must be callable!"
+        state = TaskState(self)
+        task = partial(task, *(args + (state,)), **kwargs)
+
+        self.__set_state_busy()
+        self.__start_task(task, state)
+
+    def cancel(self):
+        """ Call from derived class to stop the task. """
+        self.__cancel_task(wait=False)
+        self.__set_state_ready()
+
+    def shutdown(self):
+        """ Call from derived class when the widget is deleted
+         (in onDeleteWidget).
+        """
+        self.__cancel_task(wait=True)
+        self.__executor.shutdown(True)
 
     def __set_state_ready(self):
         self.progressBarFinished()
@@ -891,42 +923,28 @@ def __start_task(self, task: Callable[[], Any], state: TaskState):
         assert self.__task is None
         state.status_changed.connect(self.setStatusMessage)
         state.progress_changed.connect(self.progressBarSet)
-        state.partial_result_ready.connect(self._set_partial_results)
-        state.watcher.done.connect(self.on_done)
+        state.partial_result_ready.connect(self.on_partial_result)
+        state.watcher.done.connect(self.__on_task_done)
         state.start(self.__executor, task)
         state.setParent(self)
         self.__task = state
 
-    def __cancel_task(self, wait=True):
+    def __cancel_task(self, wait: bool = True):
         if self.__task is not None:
             state, self.__task = self.__task, None
             state.cancel()
-            state.partial_result_ready.disconnect(self._set_partial_results)
+            state.partial_result_ready.disconnect(self.on_partial_result)
             state.status_changed.disconnect(self.setStatusMessage)
             state.progress_changed.disconnect(self.progressBarSet)
-            state.watcher.done.disconnect(self.on_done)
+            state.watcher.done.disconnect(self.__on_task_done)
             if wait:
                 concurrent.futures.wait([state.future])
                 state.deleteLater()
             else:
                 w = FutureWatcher(state.future, parent=state)
                 w.done.connect(state.deleteLater)
 
-    def start(self):
-        """ Call to start the task. """
-        self.__cancel_task(wait=False)
-
-        if self.data is None:
-            self.__set_state_ready()
-            return
-
-        state = TaskState(self)
-        task = self._prepare_task(state)
-        self.__set_state_busy()
-        self.__start_task(task, state)
-
-    def on_done(self, future: Future):
-        """ Invoked when task is done. """
+    def __on_task_done(self, future: Future):
         assert future.done()
         assert self.__task is not None
         assert self.__task.future is future
@@ -935,14 +953,4 @@ def on_done(self, future: Future):
         task.deleteLater()
         self.__set_state_ready()
         result = future.result()
-        self._set_results(result)
-
-    def cancel(self):
-        """ Call to stop the task. """
-        self.__cancel_task(wait=False)
-        self.__set_state_ready()
-
-    def shutdown(self):
-        """ Call when widget is deleted (in onDeleteWidget). """
-        self.__cancel_task(wait=True)
-        self.__executor.shutdown(True)
+        self.on_done(result)

Orange/widgets/utils/tests/concurrent_example.py

@@ -1,7 +1,6 @@
 # pylint: disable=too-many-ancestors
 from typing import Optional
 from types import SimpleNamespace as namespace
-from functools import partial
 
 import numpy as np
 
@@ -13,42 +12,38 @@
 from Orange.widgets.visualize.utils.widget import OWDataProjectionWidget
 
 
-class Results(namespace):
+class Result(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 _ 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))
-                # Needs a copy because projection should not be modified
-                # inplace. If it is modified inplace, the widget and the thread
-                #  hold a reference to the same object. When the thread is
-                # interrupted it is still modifying the object, but the widget
-                # receives it (the modified object) with a delay.
-                embedding = np.vstack((x_data, y_data)).T.copy()
-            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(res)
-        return res
+def run(data: Table, embedding: Optional[np.ndarray], state: TaskState):
+    res = Result(embedding=embedding)
+
+    # simulate wasteful calculation (increase 'steps')
+    step, steps = 0, 10
+    state.set_status("Calculating...")
+    while step < steps:
+        for _ 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.random.rand(len(x_data))
+            # Needs a copy because projection should not be modified
+            # inplace. If it is modified inplace, the widget and the thread
+            #  hold a reference to the same object. When the thread is
+            # interrupted it is still modifying the object, but the widget
+            # receives it (the modified object) with a delay.
+            embedding = np.vstack((x_data, y_data)).T.copy()
+        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_result(res)
+    return res
 
 
 class OWConcurrentWidget(OWDataProjectionWidget, ConcurrentWidgetMixin):
@@ -72,7 +67,7 @@ def _add_controls(self):
         super()._add_controls()
 
     def __param_combo_changed(self):
-        super().start()
+        self._run()
 
     def _toggle_run(self):
         if self.data is None:
@@ -85,14 +80,14 @@ def _toggle_run(self):
             self.commit()
         # Resume task
         else:
-            self.start()
+            self._run()
 
-    # extend ConcurrentWidgetMixin
-    def _prepare_task(self, state: TaskState):
-        return partial(Runner.run, self.data,
-                       embedding=self.embedding, state=state)
+    def _run(self):
+        self.run_button.setText("Stop")
+        self.start(run, self.data, self.embedding)
 
-    def _set_partial_results(self, result: Results):
+    # ConcurrentWidgetMixin
+    def on_partial_result(self, result: Result):
         assert isinstance(result.embedding, np.ndarray)
         assert len(result.embedding) == len(self.data)
         first_result = self.embedding is None
@@ -103,25 +98,20 @@ def _set_partial_results(self, result: Results):
             self.graph.update_coordinates()
             self.graph.update_density()
 
-    def _set_results(self, result: Results):
+    def on_done(self, result: Result):
+        # NOTE: All of these have already been set by on_partial_result,
+        # we double check that they are aliased
         assert isinstance(result.embedding, np.ndarray)
         assert len(result.embedding) == len(self.data)
-        super()._set_results(result)
-
-    def start(self):
-        self.run_button.setText("Stop")
-        super().start()
-
-    def on_done(self, future):
-        super().on_done(future)
+        self.embedding = result.embedding
         self.run_button.setText("Start")
         self.commit()
 
-    # extend OWDataProjectionWidget
+    # OWDataProjectionWidget
     def set_data(self, data: Table):
         super().set_data(data)
         if self._invalidated:
-            self.start()
+            self._run()
 
     def get_embedding(self):
         if self.embedding is None: