Make ray an optional depencency (#45)

* Make `ray` an optional depencency

Due to the large size of ray and all its dependencies, make it optional for deployments where it's not needed.

hiclass will fallback to joblib.Parallel if ray is not installed and `n_jobs` is not 1.

Author: flo-ri-an

Pipfile

@@ -7,7 +7,6 @@ name = "pypi"
 networkx = "*"
 numpy = "*"
 scikit-learn = "*"
-ray = "*"
 
 [dev-packages]
 pytest = "*"
@@ -17,3 +16,6 @@ pytest-cov = "*"
 twine = "*"
 sphinx = "4.1.1"
 sphinx-rtd-theme = "0.5.2"
+
+[extras]
+ray = "*"

Pipfile.lock

@@ -4,7 +4,14 @@
 
 import networkx as nx
 import numpy as np
-import ray
+
+try:
+    import ray
+except ImportError:
+    _has_ray = False
+    from joblib import Parallel, delayed, effective_n_jobs
+else:
+    _has_ray = True
 from sklearn.base import BaseEstimator
 from sklearn.linear_model import LogisticRegression
 
@@ -291,13 +298,21 @@ def _remove_separator(self, y):
 
     def _fit_node_classifier(self, nodes, local_mode):
         if self.n_jobs > 1:
-            ray.init(
-                num_cpus=self.n_jobs, local_mode=local_mode, ignore_reinit_error=True
-            )
-            lcppn = ray.put(self)
-            _parallel_fit = ray.remote(self._fit_classifier)
-            results = [_parallel_fit.remote(lcppn, node) for node in nodes]
-            classifiers = ray.get(results)
+            if _has_ray:
+                ray.init(
+                    num_cpus=self.n_jobs,
+                    local_mode=local_mode,
+                    ignore_reinit_error=True,
+                )
+                lcppn = ray.put(self)
+                _parallel_fit = ray.remote(self._fit_classifier)
+                results = [_parallel_fit.remote(lcppn, node) for node in nodes]
+                classifiers = ray.get(results)
+            else:
+                classifiers = Parallel(n_jobs=self.n_jobs)(
+                    delayed(self._fit_classifier)(self, node) for node in nodes
+                )
+
         else:
             classifiers = [self._fit_classifier(self, node) for node in nodes]
         for classifier, node in zip(classifiers, nodes):

hiclass/HierarchicalClassifier.py

@@ -6,7 +6,14 @@
 from copy import deepcopy
 
 import numpy as np
-import ray
+
+try:
+    import ray
+except ImportError:
+    _has_ray = False
+    from joblib import Parallel, delayed, effective_n_jobs
+else:
+    _has_ray = True
 from sklearn.base import BaseEstimator
 from sklearn.utils.validation import check_array, check_is_fitted
 
@@ -194,16 +201,24 @@ def _initialize_local_classifiers(self):
     def _fit_digraph(self, local_mode: bool = False):
         self.logger_.info("Fitting local classifiers")
         if self.n_jobs > 1:
-            ray.init(
-                num_cpus=self.n_jobs, local_mode=local_mode, ignore_reinit_error=True
-            )
-            lcpl = ray.put(self)
-            _parallel_fit = ray.remote(self._fit_classifier)
-            results = [
-                _parallel_fit.remote(lcpl, level, self.separator_)
-                for level in range(len(self.local_classifiers_))
-            ]
-            classifiers = ray.get(results)
+            if _has_ray:
+                ray.init(
+                    num_cpus=self.n_jobs,
+                    local_mode=local_mode,
+                    ignore_reinit_error=True,
+                )
+                lcpl = ray.put(self)
+                _parallel_fit = ray.remote(self._fit_classifier)
+                results = [
+                    _parallel_fit.remote(lcpl, level, self.separator_)
+                    for level in range(len(self.local_classifiers_))
+                ]
+                classifiers = ray.get(results)
+            else:
+                classifiers = Parallel(n_jobs=self.n_jobs)(
+                    delayed(self._fit_classifier)(self, level, self.separator_)
+                    for level in range(len(self.local_classifiers_))
+                )
         else:
             classifiers = [
                 self._fit_classifier(self, level, self.separator_)

hiclass/LocalClassifierPerLevel.py

@@ -26,11 +26,11 @@
 KEYWORDS = ["hierarchical classification"]
 DACS_SOFTWARE = "https://gitlab.com/dacs-hpi"
 # What packages are required for this module to be executed?
-REQUIRED = ["networkx", "numpy", "scikit-learn", "ray", "protobuf<4.0"]
+REQUIRED = ["networkx", "numpy", "scikit-learn", "protobuf<4.0"]
 
 # What packages are optional?
 # 'fancy feature': ['django'],}
-EXTRAS = {}
+EXTRAS = {"ray": ["ray>=1.11.0"]}
 
 # The rest you shouldn't have to touch too much :)
 # ------------------------------------------------

setup.py

@@ -59,6 +59,28 @@ def test_fit_digraph(digraph_logistic_regression):
     assert 1
 
 
+def test_fit_digraph_joblib_multiprocessing(digraph_logistic_regression):
+    from joblib import Parallel, delayed
+
+    LocalClassifierPerLevel._has_ray = False
+
+    classifiers = [
+        LogisticRegression(),
+        LogisticRegression(),
+    ]
+    digraph_logistic_regression.n_jobs = 2
+    digraph_logistic_regression.local_classifiers_ = classifiers
+    from joblib import Parallel, delayed, effective_n_jobs
+
+    digraph_logistic_regression._fit_digraph(local_mode=True)
+    for classifier in digraph_logistic_regression.local_classifiers_:
+        try:
+            check_is_fitted(classifier)
+        except NotFittedError as e:
+            pytest.fail(repr(e))
+    assert 1
+
+
 def test_fit_1_class():
     lcpl = LocalClassifierPerLevel(local_classifier=LogisticRegression(), n_jobs=2)
     y = np.array([["1", "2"]])

tests/test_LocalClassifierPerLevel.py

@@ -111,6 +111,30 @@ def test_fit_digraph(digraph_logistic_regression):
     assert 1
 
 
+def test_fit_digraph_joblib_multiprocessing(digraph_logistic_regression):
+    from joblib import Parallel, delayed
+
+    LocalClassifierPerNode._has_ray = False
+
+    classifiers = {
+        "b": {"classifier": LogisticRegression()},
+        "c": {"classifier": LogisticRegression()},
+    }
+    digraph_logistic_regression.n_jobs = 2
+    nx.set_node_attributes(digraph_logistic_regression.hierarchy_, classifiers)
+    digraph_logistic_regression._fit_digraph(local_mode=True)
+    with pytest.raises(KeyError):
+        check_is_fitted(digraph_logistic_regression.hierarchy_.nodes["a"]["classifier"])
+    for node in ["b", "c"]:
+        try:
+            check_is_fitted(
+                digraph_logistic_regression.hierarchy_.nodes[node]["classifier"]
+            )
+        except NotFittedError as e:
+            pytest.fail(repr(e))
+    assert 1
+
+
 def test_fit_1_class():
     lcpn = LocalClassifierPerNode(local_classifier=LogisticRegression(), n_jobs=2)
     y = np.array([["1", "2"]])

tests/test_LocalClassifierPerNode.py

@@ -1,5 +1,6 @@
 import logging
 import tempfile
+import builtins
 
 import networkx as nx
 import numpy as np
@@ -66,6 +67,29 @@ def test_fit_digraph(digraph_logistic_regression):
     assert 1
 
 
+def test_fit_digraph_joblib_multiprocessing(digraph_logistic_regression):
+    from joblib import Parallel, delayed
+
+    LocalClassifierPerParentNode._has_ray = False
+
+    classifiers = {
+        "a": {"classifier": LogisticRegression()},
+    }
+    digraph_logistic_regression.n_jobs = 2
+    nx.set_node_attributes(digraph_logistic_regression.hierarchy_, classifiers)
+    digraph_logistic_regression._fit_digraph(local_mode=True)
+    try:
+        check_is_fitted(digraph_logistic_regression.hierarchy_.nodes["a"]["classifier"])
+    except NotFittedError as e:
+        pytest.fail(repr(e))
+    for node in ["b", "c"]:
+        with pytest.raises(KeyError):
+            check_is_fitted(
+                digraph_logistic_regression.hierarchy_.nodes[node]["classifier"]
+            )
+    assert 1
+
+
 def test_fit_1_class():
     lcppn = LocalClassifierPerParentNode(
         local_classifier=LogisticRegression(), n_jobs=2