add parallel flowSOM notebook

Author: berombau

docs/api.md

@@ -63,7 +63,9 @@ For more background information, see the paper for this software package {cite:p
     :toctree: generated
 
     models.FlowSOMEstimator
+    models.BatchFlowSOMEstimator
     models.SOMEstimator
+    models.BatchSOMEstimator
     models.ConsensusCluster
     models.BaseClusterEstimator
     models.BaseFlowSOMEstimator

docs/index.md

@@ -7,6 +7,7 @@
 :maxdepth: 1
 
 notebooks/example
+notebooks/parallel
 api.md
 changelog.md
 contributing.md

docs/notebooks/parallel.ipynb

@@ -2,5 +2,6 @@
 from .base_cluster_estimator import BaseClusterEstimator  # isort:skip
 from .som_estimator import SOMEstimator  # isort:skip
 from .base_flowsom_estimator import BaseFlowSOMEstimator  # isort:skip
-from .consensus_cluster import ConsensusCluster
-from .flowsom_estimator import FlowSOMEstimator
+from .consensus_cluster import ConsensusCluster  # isort:skip
+from .flowsom_estimator import FlowSOMEstimator  # isort:skip
+from .batch_flowsom_estimator import BatchFlowSOMEstimator  # isort:skip

src/flowsom/models/__init__.py

@@ -1,2 +1,2 @@
 from ._som import SOM_Batch, map_data_to_codes  # isort:skip
-from .som_estimator import SOMEstimator_batch_init  # isort:skip
+from .som_estimator import BatchSOMEstimator  # isort:skip

src/flowsom/models/batch/__init__.py

@@ -43,7 +43,7 @@ def SOM_Batch(
     radii: tuple,
     ncodes: int,
     rlen: int,
-    nr_batches: int = 10,
+    num_batches: int = 10,
     distf: Callable[[np.ndarray, np.ndarray], float] = eucl_without_sqrt,
     seed=None,
 ):
@@ -57,7 +57,7 @@ def SOM_Batch(
         radii (tuple): The radii.
         ncodes (int): The number of codes.
         rlen (int): The number of iterations.
-        nr_batches (int): The number of batches.
+        num_batches (int): The number of batches.
         distf (function): The distance function.
         seed (int): The seed for the random number generator.
 
@@ -82,7 +82,7 @@ def SOM_Batch(
     treshold_step = (radii[0] - radii[1]) / niter
 
     # Keep the temporary codes, using the given codes as the initial codes, for every batch
-    tmp_codes_all = np.empty((nr_batches, ncodes, px), dtype=np.float64)
+    tmp_codes_all = np.empty((num_batches, ncodes, px), dtype=np.float64)
 
     # Copy the codes as a float64, because the codes are updated in the algorithm
     copy_codes = codes.copy().astype(np.float64)
@@ -117,14 +117,14 @@ def SOM_Batch(
         init_threshold -= treshold_step
 
     # Choose random data points, for the different batches, and the rlen iterations
-    data_points_random = np.random.choice(n, nr_batches * rlen * n, replace=True)
+    data_points_random = np.random.choice(n, num_batches * rlen * n, replace=True)
 
     # Decrease the number of iterations, because the first iterations are already done
     rlen = int(rlen / 2)
 
     for iteration in range(rlen):
         # Execute the batches in parallel
-        for batch_nr in prange(nr_batches):
+        for batch_nr in prange(num_batches):
             # Keep the temporary codes, using the given codes as the initial codes
             tmp_codes = copy_codes.copy()
 

src/flowsom/models/batch/_som.py

@@ -8,7 +8,7 @@
 from . import SOM_Batch, map_data_to_codes
 
 
-class SOMEstimator_batch_init(BaseClusterEstimator):
+class BatchSOMEstimator(BaseClusterEstimator):
     """Estimate a Self-Organizing Map (SOM) clustering model."""
 
     def __init__(
@@ -23,6 +23,7 @@ def __init__(
         map=True,
         codes=None,
         importance=None,
+        num_batches=10,
         seed=None,
     ):
         super().__init__()
@@ -36,6 +37,7 @@ def __init__(
         self.map = map
         self.codes = codes
         self.importance = importance
+        self.num_batches = num_batches
         self.seed = seed
 
     # Core of the algorithm, where the SOM is executed
@@ -109,15 +111,15 @@ def fit(
             alpha = [tuple(alpha[i : i + 2]) for i in range(mst)]
 
         # Define the number of batches
-        nr_batches = 10
+        num_batches = self.num_batches
 
         # Split the data for the different batches, where batch with number 0 contains datapoint 0, batch_size, 2*batch_size, ...
         data = []
-        for i in range(nr_batches):
-            data.append(X[i::nr_batches, :])
+        for i in range(num_batches):
+            data.append(X[i::num_batches, :])
 
         # Make sure all the batches have the same amount of data, if not add the last data point to the last batch
-        for i in range(nr_batches):
+        for i in range(num_batches):
             if data[i].shape[0] < data[0].shape[0]:
                 data[i] = np.vstack([data[i], X[-1, :]])
 
@@ -132,7 +134,7 @@ def fit(
                 ncodes=n_codes,
                 rlen=self.rlen,
                 seed=self.seed,
-                nr_batches=nr_batches,
+                num_batches=num_batches,
             )
             if mst != 1:
                 nhbrdist: list[list[int]] = _dist_mst(codes)
@@ -159,7 +161,7 @@ def fit_predict(self, X, y=None):
         return self.labels_
 
 
-def _dist_mst(codes):
+def _dist_mst(codes) -> list[list[int]]:
     adjacency = cdist(
         codes,
         codes,

src/flowsom/models/batch/som_estimator.py

@@ -1,13 +1,13 @@
-from . import BaseFlowSOMEstimator, ConsensusCluster # isort:skip
-from .batch import SOMEstimator_batch_init # isort:skip
+from . import BaseFlowSOMEstimator, ConsensusCluster  # isort:skip
+from .batch import BatchSOMEstimator  # isort:skip
 
 
 class BatchFlowSOMEstimator(BaseFlowSOMEstimator):
     """A class that implements the FlowSOM model."""
 
     def __init__(
         self,
-        cluster_model=SOMEstimator_batch_init,
+        cluster_model=BatchSOMEstimator,
         metacluster_model=ConsensusCluster,
         **kwargs,
     ):

src/flowsom/models/batch_flowsom_estimator.py

@@ -1,5 +1,5 @@
-from . import PyFlowSOM_SOMEstimator
 from . import BaseFlowSOMEstimator, ConsensusCluster
+from .pyFlowSOM_som_estimator import PyFlowSOM_SOMEstimator
 
 
 class PyFlowSOMEstimator(BaseFlowSOMEstimator):
@@ -11,7 +11,6 @@ def __init__(
         metacluster_model=ConsensusCluster,
         **kwargs,
     ):
-        """Initialize the FlowSOMEstimator object."""
         super().__init__(
             cluster_model=cluster_model,
             metacluster_model=metacluster_model,

src/flowsom/models/pyflowsom_estimator.py

@@ -1,14 +1,21 @@
+import pytest
 from sklearn.metrics import v_measure_score
 
-from flowsom.models import PyFlowSOMEstimator
+# optional import if pyFlowSOM is installed, otherwise use regular FlowSOM for type checking
+try:
+    from flowsom.models.pyflowsom_estimator import PyFlowSOMEstimator
+except ImportError:
+    from flowsom.models import FlowSOMEstimator as PyFlowSOMEstimator
 
 
+@pytest.importorskip("flowsom.models.pyflowsom_estimator")
 def test_clustering(X):
     fsom = PyFlowSOMEstimator(n_clusters=10)
     y_pred = fsom.fit_predict(X)
     assert y_pred.shape == (100,)
 
 
+@pytest.importorskip("flowsom.models.pyflowsom_estimator")
 def test_clustering_v_measure(X_and_y):
     som = PyFlowSOMEstimator(n_clusters=10)
     X, y_true = X_and_y
@@ -17,6 +24,7 @@ def test_clustering_v_measure(X_and_y):
     assert score > 0.7
 
 
+@pytest.importorskip("flowsom.models.pyflowsom_estimator")
 def test_reproducibility_no_seed(X):
     fsom_1 = PyFlowSOMEstimator(n_clusters=10)
     fsom_2 = PyFlowSOMEstimator(n_clusters=10)
@@ -26,6 +34,7 @@ def test_reproducibility_no_seed(X):
     assert not all(y_pred_1 == y_pred_2)
 
 
+@pytest.importorskip("flowsom.models.pyflowsom_estimator")
 def test_reproducibility_seed(X):
     fsom_1 = PyFlowSOMEstimator(n_clusters=10, seed=0)
     fsom_2 = PyFlowSOMEstimator(n_clusters=10, seed=0)