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,19 @@ 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]:
         raise NotImplementedError
 
-    def _set_partial_results(self, result: Any) -> None:
+    def _on_partial_result(self, result: Any) -> None:
         raise NotImplementedError
 
-    def _set_results(self, results: Any) -> None:
-        # NOTE: All of these have already been set by _set_partial_results,
+    def _on_done(self, result: Any) -> None:
+        # NOTE: All of these have already been set by _on_partial_result,
         # 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)
+        raise NotImplementedError
 
     def __set_state_ready(self):
         self.progressBarFinished()
@@ -891,27 +887,39 @@ 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 __on_task_done(self, future: Future):
+        """ Invoked when task is done. """
+        assert future.done()
+        assert self.__task is not None
+        assert self.__task.future is future
+        assert self.__task.watcher.future() is future
+        self.__task, task = None, self.__task
+        task.deleteLater()
+        self.__set_state_ready()
+        result = future.result()
+        self._on_done(result)
+
     def start(self):
         """ Call to start the task. """
         self.__cancel_task(wait=False)
@@ -925,18 +933,6 @@ def start(self):
         self.__set_state_busy()
         self.__start_task(task, state)
 
-    def on_done(self, future: Future):
-        """ Invoked when task is done. """
-        assert future.done()
-        assert self.__task is not None
-        assert self.__task.future is future
-        assert self.__task.watcher.future() is future
-        self.__task, task = None, self.__task
-        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)

Orange/widgets/utils/tests/concurrent_example.py

@@ -13,42 +13,40 @@
 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.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_result(res)
+    return res
 
 
 class OWConcurrentWidget(OWDataProjectionWidget, ConcurrentWidgetMixin):
@@ -89,10 +87,9 @@ def _toggle_run(self):
 
     # extend ConcurrentWidgetMixin
     def _prepare_task(self, state: TaskState):
-        return partial(Runner.run, self.data,
-                       embedding=self.embedding, state=state)
+        return partial(run, self.data, embedding=self.embedding, state=state)
 
-    def _set_partial_results(self, result: Results):
+    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,20 +100,17 @@ 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):
         assert isinstance(result.embedding, np.ndarray)
         assert len(result.embedding) == len(self.data)
-        super()._set_results(result)
+        self.embedding = result.embedding
+        self.run_button.setText("Start")
+        self.commit()
 
     def start(self):
         self.run_button.setText("Stop")
         super().start()
 
-    def on_done(self, future):
-        super().on_done(future)
-        self.run_button.setText("Start")
-        self.commit()
-
     # extend OWDataProjectionWidget
     def set_data(self, data: Table):
         super().set_data(data)

Orange/widgets/visualize/owfreeviz.py

@@ -21,41 +21,40 @@
 from Orange.widgets.visualize.utils.widget import OWAnchorProjectionWidget
 
 
-class Results(namespace):
+class Result(namespace):
     projector = None  # type: FreeViz
     projection = None  # type: FreeVizModel
 
 
-class Runner:
-    MAX_ITERATIONS = 1000
+MAX_ITERATIONS = 1000
 
-    @staticmethod
-    def run(data: Table, projector: FreeViz,
-            projection: FreeVizModel, state: TaskState):
-        res = Results(projector=projector,
-                      projection=projection)
-        step, steps = 0, Runner.MAX_ITERATIONS
-        initial = res.projector.components_.T
-        state.set_status("Calculating...")
-        while True:
-            # 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.
-            res.projection = res.projector(data).copy()
-            anchors = res.projector.components_.T
-            res.projector.initial = anchors
-
-            state.set_partial_results(res)
-            if np.allclose(initial, anchors, rtol=1e-5, atol=1e-4):
-                return res
-            initial = anchors
-
-            step += 1
-            state.set_progress_value(100 * step / steps)
-            if state.is_interruption_requested():
-                return res
+
+def run_freeviz(data: Table, projector: FreeViz,
+                projection: FreeVizModel, state: TaskState):
+    res = Result(projector=projector,
+                 projection=projection)
+    step, steps = 0, MAX_ITERATIONS
+    initial = res.projector.components_.T
+    state.set_status("Calculating...")
+    while True:
+        # 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.
+        res.projection = res.projector(data).copy()
+        anchors = res.projector.components_.T
+        res.projector.initial = anchors
+
+        state.set_partial_result(res)
+        if np.allclose(initial, anchors, rtol=1e-5, atol=1e-4):
+            return res
+        initial = anchors
+
+        step += 1
+        state.set_progress_value(100 * step / steps)
+        if state.is_interruption_requested():
+            return res
 
 
 class OWFreeVizGraph(OWGraphWithAnchors):
@@ -127,7 +126,6 @@ def items():
 
 
 class OWFreeViz(OWAnchorProjectionWidget, ConcurrentWidgetMixin):
-    MAX_ITERATIONS = 1000
     MAX_INSTANCES = 10000
 
     name = "FreeViz"
@@ -205,35 +203,33 @@ def _toggle_run(self):
 
     # extend ConcurrentWidgetMixin
     def _prepare_task(self, state: TaskState):
-        return partial(Runner.run, self.effective_data,
+        return partial(run_freeviz, self.effective_data,
                        projector=self.projector,
                        projection=self.projection,
                        state=state)
 
-    def _set_partial_results(self, result: Results):
+    def _on_partial_result(self, result: Result):
         assert isinstance(result.projector, FreeViz)
-        self.projector = result.projector
         assert isinstance(result.projection, FreeVizModel)
+        self.projector = result.projector
         self.projection = result.projection
         self.graph.update_coordinates()
         self.graph.update_density()
 
-    def _set_results(self, result: Results):
-        assert isinstance(result.projector, FreeViz)
-        assert isinstance(result.projection, FreeVizModel)
-        super()._set_results(result)
+    def _on_done(self, results: Result):
+        assert isinstance(results.projector, FreeViz)
+        assert isinstance(results.projection, FreeVizModel)
+        self.projector = results.projector
+        self.projection = results.projection
+        self.graph.set_sample_size(None)
+        self.run_button.setText("Start")
+        self.commit()
 
     def start(self):
         self.graph.set_sample_size(self.SAMPLE_SIZE)
         self.run_button.setText("Stop")
         super().start()
 
-    def on_done(self, future):
-        super().on_done(future)
-        self.graph.set_sample_size(None)
-        self.run_button.setText("Start")
-        self.commit()
-
     def check_data(self):
         def error(err):
             err()