Distances: Offload work to a separate thread

Author: VesnaT

Orange/widgets/unsupervised/owdistances.py

@@ -6,6 +6,7 @@
 import Orange.misc
 from Orange import distance
 from Orange.widgets import gui, settings
+from Orange.widgets.utils.concurrent import TaskState, ConcurrentWidgetMixin
 from Orange.widgets.utils.sql import check_sql_input
 from Orange.widgets.utils.widgetpreview import WidgetPreview
 from Orange.widgets.widget import OWWidget, Msg, Input, Output
@@ -26,7 +27,22 @@
 ]
 
 
-class OWDistances(OWWidget):
+class DistanceRunner:
+    @staticmethod
+    def run(data: Orange.data.Table, metric: distance, normalized_dist: bool,
+            axis: int, state: TaskState) -> Orange.misc.DistMatrix:
+        if data is None:
+            return None
+
+        state.set_status("Calculating...")
+        if metric.supports_normalization and normalized_dist:
+            return metric(data, axis=1 - axis, impute=True,
+                          normalize=True)
+        else:
+            return metric(data, axis=1 - axis, impute=True)
+
+
+class OWDistances(OWWidget, ConcurrentWidgetMixin):
     name = "Distances"
     description = "Compute a matrix of pairwise distances."
     icon = "icons/Distance.svg"
@@ -65,13 +81,15 @@ class Warning(OWWidget.Warning):
         imputing_data = Msg("Missing values were imputed")
 
     def __init__(self):
-        super().__init__()
+        OWWidget.__init__(self)
+        ConcurrentWidgetMixin.__init__(self)
 
         self.data = None
 
-        gui.radioButtons(self.controlArea, self, "axis", ["Rows", "Columns"],
-                         box="Distances between", callback=self._invalidate
-                        )
+        gui.radioButtons(
+            self.controlArea, self, "axis", ["Rows", "Columns"],
+            box="Distances between", callback=self._invalidate
+        )
         box = gui.widgetBox(self.controlArea, "Distance Metric")
         self.metrics_combo = gui.comboBox(
             box, self, "metric_idx",
@@ -93,6 +111,7 @@ def __init__(self):
     @Inputs.data
     @check_sql_input
     def set_data(self, data):
+        self.cancel()
         self.data = data
         self.refresh_metrics()
         self.unconditional_commit()
@@ -106,8 +125,7 @@ def refresh_metrics(self):
     def commit(self):
         # pylint: disable=invalid-sequence-index
         metric = METRICS[self.metric_idx][1]
-        dist = self.compute_distances(metric, self.data)
-        self.Outputs.distances.send(dist)
+        self.compute_distances(metric, self.data)
 
     def compute_distances(self, metric, data):
         def _check_sparse():
@@ -152,22 +170,34 @@ def _fix_missing():
             return True
 
         self.clear_messages()
-        if data is None:
-            return None
-        for check in (_check_sparse,
-                      _fix_discrete, _fix_missing, _fix_nonbinary):
-            if not check():
-                return None
-        try:
-            if metric.supports_normalization and self.normalized_dist:
-                return metric(data, axis=1 - self.axis, impute=True,
-                              normalize=True)
-            else:
-                return metric(data, axis=1 - self.axis, impute=True)
-        except ValueError as e:
+        if data is not None:
+            for check in (_check_sparse, _fix_discrete,
+                          _fix_missing, _fix_nonbinary):
+                if not check():
+                    data = None
+                    break
+
+        self.start(DistanceRunner.run, data, metric,
+                   self.normalized_dist, self.axis)
+
+    def on_partial_result(self, _):
+        pass
+
+    def on_done(self, dist: Orange.misc.DistMatrix):
+        assert isinstance(dist, Orange.misc.DistMatrix) or dist is None
+        self.Outputs.distances.send(dist)
+
+    def on_exception(self, e):
+        if isinstance(e, ValueError):
             self.Error.distances_value_error(e)
-        except MemoryError:
+        elif isinstance(e, MemoryError):
             self.Error.distances_memory_error()
+        else:
+            raise e
+
+    def onDeleteWidget(self):
+        self.shutdown()
+        super().onDeleteWidget()
 
     def _invalidate(self):
         self.commit()

Orange/widgets/unsupervised/tests/test_owdistances.py

@@ -1,15 +1,46 @@
 # Test methods with long descriptive names can omit docstrings
 # pylint: disable=missing-docstring
+import unittest
 from unittest.mock import Mock
 
 import numpy as np
 
 from Orange.data import Table, Domain
 from Orange import distance
-from Orange.widgets.unsupervised.owdistances import OWDistances, METRICS
+from Orange.widgets.unsupervised.owdistances import OWDistances, METRICS, \
+    DistanceRunner
 from Orange.widgets.tests.base import WidgetTest
 
 
+class TestDistanceRunner(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.iris = Table("iris")[::5]
+
+    def test_run(self):
+        for _, metric in METRICS:
+            # between rows, normalized
+            dist1 = DistanceRunner.run(self.iris, metric, True, 0, Mock())
+            dist2 = metric(self.iris, axis=1, impute=True, normalize=True)
+            np.testing.assert_array_equal(dist1, dist2)
+
+            # between rows, not normalized
+            dist1 = DistanceRunner.run(self.iris, metric, False, 0, Mock())
+            dist2 = metric(self.iris, axis=1, impute=True, normalize=False)
+            np.testing.assert_array_equal(dist1, dist2)
+
+            # between columns, normalized
+            dist1 = DistanceRunner.run(self.iris, metric, True, 1, Mock())
+            dist2 = metric(self.iris, axis=0, impute=True, normalize=True)
+            np.testing.assert_array_equal(dist1, dist2)
+
+            # between columns, not normalized
+            dist1 = DistanceRunner.run(self.iris, metric, False, 1, Mock())
+            dist2 = metric(self.iris, axis=0, impute=True, normalize=False)
+            np.testing.assert_array_equal(dist1, dist2)
+
+
 class TestOWDistances(WidgetTest):
     @classmethod
     def setUpClass(cls):
@@ -27,6 +58,7 @@ def test_distance_combo(self):
         for i, (_, metric) in enumerate(METRICS):
             self.widget.metrics_combo.activated.emit(i)
             self.widget.metrics_combo.setCurrentIndex(i)
+            self.wait_until_stop_blocking()
             self.send_signal(self.widget.Inputs.data, self.iris)
             if metric.supports_normalization:
                 expected = metric(self.iris, normalize=self.widget.normalized_dist)
@@ -42,8 +74,10 @@ def test_error_message(self):
         data is removed from input"""
         self.widget.metric_idx = 2
         self.send_signal(self.widget.Inputs.data, self.iris)
+        self.wait_until_stop_blocking()
         self.assertFalse(self.widget.Error.no_continuous_features.is_shown())
         self.send_signal(self.widget.Inputs.data, self.titanic)
+        self.wait_until_stop_blocking()
         self.assertTrue(self.widget.Error.no_continuous_features.is_shown())
         self.send_signal(self.widget.Inputs.data, None)
         self.assertFalse(self.widget.Error.no_continuous_features.is_shown())
@@ -53,32 +87,39 @@ def test_jaccard_messages(self):
             if name == "Jaccard":
                 break
         self.send_signal(self.widget.Inputs.data, self.iris)
+        self.wait_until_stop_blocking()
         self.assertTrue(self.widget.Error.no_binary_features.is_shown())
         self.assertFalse(self.widget.Warning.ignoring_nonbinary.is_shown())
 
         self.send_signal(self.widget.Inputs.data, None)
+        self.wait_until_stop_blocking()
         self.assertFalse(self.widget.Error.no_binary_features.is_shown())
         self.assertFalse(self.widget.Warning.ignoring_nonbinary.is_shown())
 
         self.send_signal(self.widget.Inputs.data, self.titanic)
+        self.wait_until_stop_blocking()
         self.assertFalse(self.widget.Error.no_binary_features.is_shown())
         self.assertTrue(self.widget.Warning.ignoring_nonbinary.is_shown())
 
         self.send_signal(self.widget.Inputs.data, None)
+        self.wait_until_stop_blocking()
         self.assertFalse(self.widget.Error.no_binary_features.is_shown())
         self.assertFalse(self.widget.Warning.ignoring_nonbinary.is_shown())
 
         self.send_signal(self.widget.Inputs.data, self.titanic)
+        self.wait_until_stop_blocking()
         self.assertFalse(self.widget.Error.no_binary_features.is_shown())
         self.assertTrue(self.widget.Warning.ignoring_nonbinary.is_shown())
 
         dom = self.titanic.domain
         dom = Domain(dom.attributes[1:], dom.class_var)
         self.send_signal(self.widget.Inputs.data, self.titanic.transform(dom))
+        self.wait_until_stop_blocking()
         self.assertFalse(self.widget.Error.no_binary_features.is_shown())
         self.assertFalse(self.widget.Warning.ignoring_nonbinary.is_shown())
 
         self.send_signal(self.widget.Inputs.data, Table("heart_disease"))
+        self.wait_until_stop_blocking()
         self.assertFalse(self.widget.Error.no_binary_features.is_shown())
         self.assertFalse(self.widget.Warning.ignoring_discrete.is_shown())
 
@@ -93,10 +134,12 @@ def test_too_big_array(self):
 
         mock = Mock(side_effect=ValueError)
         self.widget.compute_distances(mock, self.iris)
+        self.wait_until_stop_blocking()
         self.assertTrue(self.widget.Error.distances_value_error.is_shown())
 
         mock = Mock(side_effect=MemoryError)
         self.widget.compute_distances(mock, self.iris)
+        self.wait_until_stop_blocking()
         self.assertEqual(len(self.widget.Error.active), 1)
         self.assertTrue(self.widget.Error.distances_memory_error.is_shown())