Repository: scikit-activeml/scikit-activeml (language: python)

Issue:
FR: Implement the pool-based query strategy `ProbCover`
### Describe the workflow you want to enable

`ProbCover` (Probability Coverage) [1] is a pool-based query strategy suited for low-budget active learning scenarios. 

[1] [Yehuda, Ofer, Avihu Dekel, Guy Hacohen, and Daphna Weinshall. "Active learning through a covering lens." Advances in Neural Information Processing Systems 35 (2022): 22354-22367.](https://proceedings.neurips.cc/paper_files/paper/2022/file/8c64bc3f7796d31caa7c3e6b969bf7da-Paper-Conference.pdf)

### Describe your proposed solution

The authors detail their algorithm in the paper [1] and provide a Pytorch implementation at [GitHub](https://github.com/avihu111/TypiClust).

### Describe alternatives you've considered, if relevant

_No response_

### Additional context

_No response_


Discussion / hints:

Fix patch:
diff --git a/docs/examples/pool/_prob_cover.json b/docs/examples/pool/_prob_cover.json
new file mode 100644
index 00000000..1f5a8aea
--- /dev/null
+++ b/docs/examples/pool/_prob_cover.json
@@ -0,0 +1,18 @@
+[
+{
+    "class" : "ProbCover",
+    "package" : "pool",
+    "method" : "Probability Coverage (ProbCover)",
+    "category" : "Others",
+    "template" : "examples/template_pool.py",
+    "tags" : ["pool", "classification", "single-annotator"],
+    "title" : "Probability Coverage (ProbCover)",
+    "refs" : ["yehuda2022active"],
+    "sequence" : ["title", "text_0", "plot", "refs"],
+    "text_0" : "",
+    "import_misc" : "from skactiveml.pool import ProbCover",
+    "init_qs" : "ProbCover(n_classes=2, random_state=0, cluster_algo_dict={\"random_state\": 0})",
+    "query_params" : "X=X, y=y"
+}
+]
+
diff --git a/docs/refs.bib b/docs/refs.bib
index bfb58e3b..ab725bc7 100644
--- a/docs/refs.bib
+++ b/docs/refs.bib
@@ -298,4 +298,10 @@ @inproceedings{hacohen2022active
   author={Hacohen, Guy and Dekel, Avihu and Weinshall, Daphna},
   booktitle={International Conference on Machine Learning},
   year={2022},
+}
+@inproceedings{yehuda2022active,
+  title={{Active Learning Through a Covering Lens}},
+  author={Yehuda, Ofer and Dekel, Avihu and Hacohen, Guy and Weinshall, Daphna},
+  booktitle={Advances in Neural Information Processing Systems},
+  year={2022}
 }
\ No newline at end of file
diff --git a/skactiveml/pool/__init__.py b/skactiveml/pool/__init__.py
index 173472cd..3df42c6e 100644
--- a/skactiveml/pool/__init__.py
+++ b/skactiveml/pool/__init__.py
@@ -35,6 +35,7 @@
 from ._core_set import CoreSet, k_greedy_center
 from ._typi_clust import TypiClust
 from ._badge import Badge
+from ._prob_cover import ProbCover
 from ._wrapper import SubSamplingWrapper, ParallelUtilityEstimationWrapper
 
 __all__ = [
@@ -68,6 +69,7 @@
     "k_greedy_center",
     "TypiClust",
     "Badge",
+    "ProbCover",
     "GreedyBALD",
     "RegressionTreeBasedAL",
     "SubSamplingWrapper",
diff --git a/skactiveml/pool/_prob_cover.py b/skactiveml/pool/_prob_cover.py
new file mode 100644
index 00000000..bf1ee5c0
--- /dev/null
+++ b/skactiveml/pool/_prob_cover.py
@@ -0,0 +1,236 @@
+"""
+Module implementing `ProbCover`, which is a deep active learning strategy
+suited for low budgets.
+"""
+
+import numpy as np
+import warnings
+
+from sklearn.metrics import pairwise_distances
+from sklearn.cluster import KMeans
+from sklearn.utils.validation import column_or_1d
+
+from ..base import SingleAnnotatorPoolQueryStrategy
+from ..utils import (
+    MISSING_LABEL,
+    rand_argmax,
+    check_scalar,
+)
+
+
+class ProbCover(SingleAnnotatorPoolQueryStrategy):
+    """Probability Coverage
+
+    This class implements the Probability Coverage (ProbCover) query strategy
+    [1], which aims at maximizing the probability coverage in a meaningful
+    sample embedding space.
+
+    Parameters
+    ----------
+    n_classes : None or int, default=None
+        This parameter is used to determine the delta value. If
+        `n_classes=None`, the number of classes is extracted from the
+        given labels. If this extracted number of classes is below 2,
+        `n_classes=2` is used as a fallback.
+    deltas : None or array-like of shape (n_deltas,), default=None
+        List of deltas (ball radii) to be tested for finding the maximum
+        value satisfying a sample coverage >= `alpha`. If no value in
+        `deltas` satisfies this constraint, a warning is raised where
+        the minimum `delta` value is used. If `deltas=None`, the values
+        `np.arange(0.1, 2.1, 0.1)` are used.
+    alpha : float in (0, 1), alpha=0.95
+        Minimum coverage as a constraint for the `delta` selection.
+    cluster_algo : ClusterMixin.__class__, default=sklearn.cluster.KMeans
+        The cluster algorithm to be used for determining the best delta value.
+    cluster_algo_dict : dict, default=None
+        The parameters passed to the clustering algorithm `cluster_algo`,
+        excluding the parameter for the number of clusters.
+    n_cluster_param_name : string, default="n_clusters"
+        The name of the parameter for the number of clusters.
+    distance_func : callable, default=sklearn.metrics.pairwise_distances
+        Takes as input `X` to compute the distances between each pair of
+        samples. This function can also only return the precomputed distances
+        of each pair in `X` for speedup.
+    missing_label : scalar or string or np.nan or None, default=np.nan
+        Value to represent a missing label.
+    random_state : None or int or np.random.RandomState, default=None
+        The random state to use.
+
+    References
+    ----------
+    .. [1] Yehuda, Ofer, Avihu Dekel, Guy Hacohen, and Daphna Weinshall.
+       "Active Learning Through a Covering Lens." NeurIPS, 2022.
+    """
+
+    def __init__(
+        self,
+        n_classes=None,
+        deltas=None,
+        alpha=0.95,
+        cluster_algo=KMeans,
+        cluster_algo_dict=None,
+        n_cluster_param_name="n_clusters",
+        distance_func=pairwise_distances,
+        missing_label=MISSING_LABEL,
+        random_state=None,
+    ):
+        super().__init__(
+            missing_label=missing_label, random_state=random_state
+        )
+        self.deltas = deltas
+        self.alpha = alpha
+        self.n_classes = n_classes
+        self.cluster_algo = cluster_algo
+        self.cluster_algo_dict = cluster_algo_dict
+        self.n_cluster_param_name = n_cluster_param_name
+        self.distance_func = distance_func
+
+    def query(
+        self,
+        X,
+        y,
+        candidates=None,
+        batch_size=1,
+        return_utilities=False,
+        update=False,
+    ):
+        """Query the next samples to be labeled
+
+        Parameters
+        ----------
+        X : array-like of shape (n_samples, n_features)
+            Training data set, usually complete, i.e. including the labeled and
+            unlabeled samples.
+        y : array-like of shape (n_samples,)
+            Labels of the training data set (possibly including unlabeled ones
+            indicated by self.missing_label).
+        candidates : None or array-like of shape (n_candidates) with \
+                dtype=int, default=None
+            If `candidates` is None, the unlabeled samples from `(X, y)`
+            are considered as candidates.
+            If `candidates` is of shape `(n_candidates,)` and of type int,
+            `candidates` is considered as a list of the indices of the samples
+            in `(X, y)`.
+        batch_size : int, default=1
+            The number of samples to be selected in one AL cycle.
+        return_utilities : bool, default=False
+            If True, also return the utilities based on the query strategy.
+        update : bool, default=False
+            This boolean flag determines whether the computed `delta_max_`
+            and the `distances_` shall be updated in the `query`. For the first
+            call of `query`, this parameter has no impact because both
+            quantities are computed for the first time.
+
+        Returns
+        ----------
+        query_indices : numpy.ndarray of shape (batch_size,)
+            The `query_indices` indicate for which candidate sample a label is
+            to be queried, e.g., `query_indices[0]` indicates the first
+            selected sample.
+            If `candidates` in `None` or of shape `(n_candidates,)`, the
+            indexing refers to samples in `X`.
+        utilities : numpy.ndarray of shape (batch_size, n_samples)
+            The utilities of samples for selecting each sample of the batch.
+            Here, utilities mean the out-degree of the candidate samples.
+            If `candidates` is `None` or of shape `(n_candidates,)`, the
+            indexing refers to the samples in `X`.
+        """
+        # Check parameters.
+        X, y, candidates, batch_size, return_utilities = self._validate_data(
+            X, y, candidates, batch_size, return_utilities, reset=True
+        )
+        _, mapping = self._transform_candidates(
+            candidates, X, y, enforce_mapping=True
+        )
+        is_candidate = np.full(len(X), fill_value=False)
+        is_candidate[mapping] = True
+        n_classes = self.n_classes
+        if n_classes is None:
+            n_classes = max(len(np.unique(y[~is_candidate])), 2)
+        check_scalar(
+            n_classes,
+            "n_classes",
+            min_val=2,
+            min_inclusive=True,
+            target_type=int,
+        )
+        if self.deltas is None:
+            deltas = np.arange(0.2, 2.2, 0.2)
+        else:
+            deltas = column_or_1d(self.deltas, dtype=float)
+            deltas = np.sort(deltas)
+            if (deltas < 0).any():
+                raise ValueError("`deltas` must contain non-negative floats.")
+        check_scalar(
+            self.alpha,
+            "alpha",
+            min_val=0,
+            max_val=1,
+            min_inclusive=False,
+            max_inclusive=False,
+            target_type=float,
+        )
+        if not (
+            isinstance(self.cluster_algo_dict, dict)
+            or self.cluster_algo_dict is None
+        ):
+            raise TypeError(
+                "Pass a dictionary with corresponding parameter names and "
+                "values according to the `init` function of `cluster_algo`."
+            )
+        cluster_algo_dict = (
+            {}
+            if self.cluster_algo_dict is None
+            else self.cluster_algo_dict.copy()
+        )
+        check_scalar(update, name="update", target_type=bool)
+
+        if update or not hasattr(self, "delta_max_"):
+            # Compute distances between each pair of observed samples.
+            self.distances_ = self.distance_func(X)
+
+            # Compute the maximum `delta` value satisfying a purity >= `alpha`.
+            self.delta_max_ = deltas[0]
+            max_purity = -1
+            if len(deltas) > 1:
+                cluster_algo_dict[self.n_cluster_param_name] = n_classes
+                cluster_obj = self.cluster_algo(**cluster_algo_dict)
+                y_cluster = cluster_obj.fit_predict(X)
+                is_impure = y_cluster[:, None] != y_cluster
+                for delta in deltas:
+                    edges = self.distances_ <= delta
+                    purity = 1 - (edges * is_impure).any(axis=1).mean()
+                    max_purity = max(max_purity, purity)
+                    if purity < self.alpha:
+                        break
+                    self.delta_max_ = delta
+
+            # Check whether condition defined by `alpha` was satisfied.
+            if max_purity < self.alpha:
+                warnings.warn(
+                    f"The maximum purity was {max_purity} being smaller "
+                    f"than the required value `alpha={self.alpha}`. You must"
+                    f"provide smaller values in `deltas` to avoid "
+                    f"this warning."
+                )
+
+        # Compute edges of the graph with the samples as vertices.
+        edges = self.distances_ <= self.delta_max_
+
+        # Perform sample-wise selection of the batch.
+        query_indices = np.full(batch_size, fill_value=-1, dtype=int)
+        utilities = np.full((batch_size, len(X)), fill_value=np.nan)
+        for b in range(batch_size):
+            # Step (ii) in [1]: Remove incoming edges for covered samples.
+            is_covered = edges[~is_candidate].any(axis=0)
+            edges[:, is_covered] = False
+            # Step (i) in [1]: Query the sample with the highest out-degree.
+            utilities[b][is_candidate] = edges[is_candidate].sum(axis=1)
+            idx = rand_argmax(utilities[b], random_state=self.random_state_)[0]
+            is_candidate[idx] = False
+            query_indices[b] = idx
+
+        if return_utilities:
+            return query_indices, utilities
+        else:
+            return query_indices


Test patch:
diff --git a/skactiveml/pool/tests/test_prob_cover.py b/skactiveml/pool/tests/test_prob_cover.py
new file mode 100644
index 00000000..5372e125
--- /dev/null
+++ b/skactiveml/pool/tests/test_prob_cover.py
@@ -0,0 +1,207 @@
+import unittest
+
+import numpy as np
+from sklearn.cluster import KMeans, SpectralClustering
+from sklearn.datasets import make_blobs
+from sklearn.metrics import pairwise_distances, euclidean_distances
+from skactiveml.pool import ProbCover
+from skactiveml.exceptions import MappingError
+from skactiveml.utils import MISSING_LABEL
+from skactiveml.tests.template_query_strategy import (
+    TemplateSingleAnnotatorPoolQueryStrategy,
+)
+
+
+class TestProbCover(
+    TemplateSingleAnnotatorPoolQueryStrategy, unittest.TestCase
+):
+    def setUp(self):
+        query_default_params_clf = {
+            "X": np.linspace(0, 1, 20).reshape(10, 2),
+            "y": np.hstack([[0, 1], np.full(8, MISSING_LABEL)]),
+        }
+        cluster_dict = {"random_state": 0, "n_init": 1}
+        super().setUp(
+            qs_class=ProbCover,
+            init_default_params={"cluster_algo_dict": cluster_dict},
+            query_default_params_clf=query_default_params_clf,
+        )
+
+    def test_init_param_n_classes(self, test_cases=None):
+        test_cases = [] if test_cases is None else test_cases
+        test_cases += [
+            (None, None),
+            (2, None),
+            (10, None),
+            ("string", TypeError),
+            (1.5, TypeError),
+            (0, ValueError),
+            (1, ValueError),
+        ]
+        self._test_param("init", "n_classes", test_cases)
+
+    def test_init_param_deltas(self, test_cases=None):
+        test_cases = [] if test_cases is None else test_cases
+        test_cases += [
+            ([0.5], None),
+            ([0.0, 1.0, 2.0], None),
+            ([500], None),
+            (np.array([0.0, 1.0, 2.0]), None),
+            ([-1], ValueError),
+            (0.5, ValueError),
+            (-1, ValueError),
+            ("string", ValueError),
+        ]
+        self._test_param("init", "deltas", test_cases)
+
+    def test_init_param_alpha(self, test_cases=None):
+        test_cases = [] if test_cases is None else test_cases
+        test_cases += [
+            (0.5, None),
+            (0.01, None),
+            (0.99, None),
+            (1.0, ValueError),
+            (0.0, ValueError),
+            ("string", TypeError),
+        ]
+        self._test_param("init", "alpha", test_cases)
+
+    def test_init_param_cluster_algo(self, test_cases=None):
+        test_cases = [] if test_cases is None else test_cases
+        test_cases += [
+            (1, TypeError),
+            ("string", TypeError),
+            (None, TypeError),
+            (ProbCover, TypeError),
+            (SpectralClustering, None),
+            (KMeans, None),
+        ]
+        self._test_param(
+            "init",
+            "cluster_algo",
+            test_cases,
+            replace_init_params={"random_state": 0},
+        )
+
+    def test_init_param_cluster_algo_dict(self, test_cases=None):
+        test_cases = [] if test_cases is None else test_cases
+        test_cases += [
+            (1, TypeError),
+            ("string", TypeError),
+            (None, None),
+            ({}, None),
+            ({"n_init": "auto", "random_state": 0}, None),
+        ]
+        self._test_param("init", "cluster_algo_dict", test_cases)
+
+    def test_init_param_n_cluster_param_name(self, test_cases=None):
+        test_cases = [] if test_cases is None else test_cases
+        test_cases += [
+            (1, TypeError),
+            ("string", TypeError),
+            (None, TypeError),
+            ("n_clusters", None),
+        ]
+        self._test_param("init", "n_cluster_param_name", test_cases)
+
+    def test_init_param_distance_func(self, test_cases=None):
+        test_cases = [] if test_cases is None else test_cases
+        test_cases += [
+            (pairwise_distances, None),
+            (euclidean_distances, None),
+            (1, TypeError),
+            ("string", TypeError),
+        ]
+        self._test_param("init", "distance_func", test_cases)
+
+    def test_query_param_update(self, test_cases=None):
+        test_cases = [] if test_cases is None else test_cases
+        test_cases += [
+            (False, None),
+            (True, None),
+            (1, TypeError),
+            ("string", TypeError),
+        ]
+        self._test_param("query", "update", test_cases)
+
+    def test_query(self):
+        X, y_true = make_blobs(n_samples=50, centers=4, random_state=0)
+        y_true = y_true.astype(float)
+        D = pairwise_distances(X)
+
+        def distance_func(X):
+            return D
+
+        for candidates in [None, np.arange(10, 30)]:
+            qs = ProbCover(
+                deltas=np.linspace(0.5, 2, 100),
+                cluster_algo_dict={"random_state": 0},
+                random_state=42,
+            )
+            qs_metric = ProbCover(
+                deltas=np.linspace(0.5, 2, 100),
+                cluster_algo_dict={"random_state": 0},
+                distance_func=distance_func,
+                random_state=42,
+            )
+            qs_cluster = ProbCover(
+                deltas=np.linspace(0.5, 2, 100),
+                random_state=42,
+                cluster_algo_dict={
+                    "init": "random",
+                    "max_iter": 1,
+                    "random_state": 0,
+                },
+            )
+
+            # With the same random state the initial selection is the same.
+            y = np.full(50, MISSING_LABEL)
+            self.assertEqual(
+                qs.query(X, y, candidates=candidates),
+                qs.query(X, y, candidates=candidates),
+            )
+
+            # Check non-negativity of `delta_max_`.
+            self.assertGreaterEqual(qs.delta_max_, 0.0)
+
+            # All utilities are non-negative integers or np.nan.
+            is_unlabeled = np.random.RandomState(0).choice(
+                [False, True], size=(len(X),), replace=True
+            )
+            y = y_true.copy()
+            y[is_unlabeled] = np.nan
+            query_indices, utilities = qs.query(
+                X,
+                y,
+                update=True,
+                candidates=candidates,
+                batch_size=2,
+                return_utilities=True,
+            )
+            utilities_copy = utilities.copy()
+            is_nan = np.isnan(utilities)
+            utilities_copy[is_nan] = 0.0
+            is_integer = np.mod(utilities_copy, 1) == 0
+            is_non_negative = utilities_copy >= 0
+            self.assertTrue(np.logical_and(is_integer, is_non_negative).all())
+
+            # Check functionality of `pairwise_distances_dict`.
+            query_indices_metric, utilities_metric = qs_metric.query(
+                X,
+                y,
+                batch_size=2,
+                candidates=candidates,
+                return_utilities=True,
+            )
+            np.testing.assert_array_equal(query_indices_metric, query_indices)
+            np.testing.assert_array_equal(utilities_metric, utilities)
+
+            # Check consistence of `delta_max_`.
+            self.assertEqual(qs.delta_max_, qs_metric.delta_max_)
+
+            # Check functionality of `cluster_algo_dict`.
+            _, _ = qs_cluster.query(X, y, candidates=candidates, batch_size=2)
+            self.assertTrue(qs_cluster.delta_max_ != qs.delta_max_)
+
+        # Check whether error is raised for `candidates` being not in `X`.
+        self.assertRaises(MappingError, qs.query, X, y, candidates=X)