Unify metric API

.github/workflows/codecov.yml

@@ -6,6 +6,9 @@ on:
   pull_request:
   workflow_dispatch:
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.ref }}
+  cancel-in-progress: true
 
 jobs:
   run:

.github/workflows/lint.yml

@@ -4,6 +4,9 @@ on:
   pull_request:
   workflow_dispatch:
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.ref }}
+  cancel-in-progress: true
 
 jobs:
   lint:

.github/workflows/python-package.yml

@@ -10,6 +10,9 @@ on:
   pull_request:
   workflow_dispatch:
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.ref }}
+  cancel-in-progress: true
 
 
 jobs:

docs/Makefile

@@ -12,12 +12,16 @@ BUILDDIR      = build
 help:
 	@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
 
-.PHONY: help Makefile
+.PHONY: help Makefile clean
 
 # Catch-all target: route all unknown targets to Sphinx using the new
 # "make mode" option.  $(O) is meant as a shortcut for $(SPHINXOPTS).
 %: Makefile
 	@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
 
-rst:
+rst: clean
 	@sphinx-apidoc -o source/docs_api ../quantus --module-first --separate --force
+
+
+clean:
+	rm -rf source/docs_api

docs/source/docs_api/quantus.helpers.enums.rst

@@ -0,0 +1,7 @@
+quantus.helpers.enums module
+============================
+
+.. automodule:: quantus.helpers.enums
+   :members:
+   :undoc-members:
+   :show-inheritance:

docs/source/docs_api/quantus.helpers.perturbation_utils.rst

@@ -0,0 +1,7 @@
+quantus.helpers.perturbation\_utils module
+==========================================
+
+.. automodule:: quantus.helpers.perturbation_utils
+   :members:
+   :undoc-members:
+   :show-inheritance:

docs/source/docs_api/quantus.helpers.rst

@@ -22,6 +22,8 @@ Submodules
 
    quantus.helpers.asserts
    quantus.helpers.constants
+   quantus.helpers.enums
+   quantus.helpers.perturbation_utils
    quantus.helpers.plotting
    quantus.helpers.utils
    quantus.helpers.warn

mypy.ini

@@ -12,6 +12,7 @@ ignore_missing_imports = True
 no_site_packages = True
 show_none_errors = False
 ignore_errors = False
+plugins = numpy.typing.mypy_plugin
 
 [mypy-quantus.*]
 disallow_untyped_defs = False

pyproject.toml

@@ -34,6 +34,7 @@ dependencies = [
     "scipy>=1.7.3",
     "tqdm>=4.62.3",
     "matplotlib>=3.3.4",
+    "typing_extensions; python_version <= '3.8'"
 ]
 
 dynamic = ["version"]

quantus/helpers/constants.py

@@ -7,16 +7,21 @@
 # You should have received a copy of the GNU Lesser General Public License along with Quantus. If not, see <https://www.gnu.org/licenses/>.
 # Quantus project URL: <https://github.com/understandable-machine-intelligence-lab/Quantus>.
 
-from typing import List, Dict
-
+import sys
+from typing import List, Dict, Mapping, Type
 from quantus.functions.loss_func import *
 from quantus.functions.normalise_func import *
 from quantus.functions.perturb_func import *
 from quantus.functions.similarity_func import *
 from quantus.metrics import *
 
+if sys.version_info >= (3, 8):
+    from typing import Final
+else:
+    from typing_extensions import Final
+
 
-AVAILABLE_METRICS = {
+AVAILABLE_METRICS: Final[Mapping[str, Mapping[str, Type[Metric]]]] = {
     "Faithfulness": {
         "Faithfulness Correlation": FaithfulnessCorrelation,
         "Faithfulness Estimate": FaithfulnessEstimate,

quantus/helpers/model/pytorch_model.py

@@ -337,7 +337,6 @@ def add_mean_shift_to_first_layer(
             The resulting model with a shifted first layer.
         """
         with torch.no_grad():
-
             new_model = deepcopy(self.model)
 
             modules = [l for l in new_model.named_modules()]

quantus/helpers/perturbation_utils.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import sys
+from typing import List, TYPE_CHECKING, Callable, Mapping
+import numpy as np
+import functools
+
+if sys.version_info >= (3, 8):
+    from typing import Protocol
+else:
+    from typing_extensions import Protocol
+
+
+if TYPE_CHECKING:
+    from quantus.helpers.model.model_interface import ModelInterface
+
+    class PerturbFunc(Protocol):
+        def __call__(
+            self,
+            arr: np.ndarray,
+            indices: np.ndarray,
+            indexed_axes: np.ndarray,
+            **kwargs,
+        ) -> np.ndarray:
+            ...
+
+
+def make_perturb_func(
+    perturb_func: PerturbFunc, perturb_func_kwargs: Mapping[str, ...] | None, **kwargs
+) -> PerturbFunc | functools.partial:
+    """A utility function to save few lines of code during perturbation metric initialization."""
+    if perturb_func_kwargs is not None:
+        func_kwargs = kwargs.copy()
+        func_kwargs.update(perturb_func_kwargs)
+    else:
+        func_kwargs = kwargs
+
+    return functools.partial(perturb_func, **func_kwargs)
+
+
+def make_changed_prediction_indices_func(
+    return_nan_when_prediction_changes: bool,
+) -> Callable[[ModelInterface, np.ndarray, np.ndarray], List[int]]:
+    """A utility function to improve static analysis."""
+    return functools.partial(
+        changed_prediction_indices,
+        return_nan_when_prediction_changes=return_nan_when_prediction_changes,
+    )
+
+
+def changed_prediction_indices(
+    model: ModelInterface,
+    x_batch: np.ndarray,
+    x_perturbed: np.ndarray,
+    return_nan_when_prediction_changes: bool,
+) -> List[int]:
+    """
+    Find indices in batch, for which predicted label has changed after applying perturbation.
+    If metric `return_nan_when_prediction_changes` is False, will return empty list.
+
+    Parameters
+    ----------
+    return_nan_when_prediction_changes:
+        Instance attribute of perturbation metrics.
+    model:
+    x_batch:
+        Batch of original inputs provided by user.
+    x_perturbed:
+        Batch of inputs after applying perturbation.
+
+    Returns
+    -------
+
+    changed_idx:
+        List of indices in batch, for which predicted label has changed afer.
+
+    """
+
+    if not return_nan_when_prediction_changes:
+        return []
+
+    labels_before = model.predict(x_batch).argmax(axis=-1)
+    labels_after = model.predict(x_perturbed).argmax(axis=-1)
+    changed_idx = np.reshape(np.argwhere(labels_before != labels_after), -1)
+    return changed_idx.tolist()

quantus/metrics/__init__.py

@@ -4,10 +4,8 @@
 # You should have received a copy of the GNU Lesser General Public License along with Quantus. If not, see <https://www.gnu.org/licenses/>.
 # Quantus project URL: <https://github.com/understandable-machine-intelligence-lab/Quantus>.
 
-from quantus.metrics.base import *
-from quantus.metrics.base_batched import *
-from quantus.metrics.base_perturbed import *
 from quantus.metrics.axiomatic import *
+from quantus.metrics.base import Metric
 from quantus.metrics.complexity import *
 from quantus.metrics.faithfulness import *
 from quantus.metrics.localisation import *

quantus/metrics/axiomatic/completeness.py

@@ -6,24 +6,32 @@
 # You should have received a copy of the GNU Lesser General Public License along with Quantus. If not, see <https://www.gnu.org/licenses/>.
 # Quantus project URL: <https://github.com/understandable-machine-intelligence-lab/Quantus>.
 
+import sys
 from typing import Any, Callable, Dict, List, Optional
+
 import numpy as np
 
-from quantus.helpers import warn
-from quantus.helpers import asserts
-from quantus.helpers.model.model_interface import ModelInterface
 from quantus.functions.normalise_func import normalise_by_max
 from quantus.functions.perturb_func import baseline_replacement_by_indices
-from quantus.metrics.base_perturbed import PerturbationMetric
+from quantus.helpers import warn
 from quantus.helpers.enums import (
-    ModelType,
     DataType,
-    ScoreDirection,
     EvaluationCategory,
+    ModelType,
+    ScoreDirection,
 )
+from quantus.helpers.model.model_interface import ModelInterface
+from quantus.helpers.perturbation_utils import make_perturb_func
+from quantus.metrics.base import Metric
+
+if sys.version_info >= (3, 8):
+    from typing import final
+else:
+    from typing_extensions import final
 
 
-class Completeness(PerturbationMetric):
+@final
+class Completeness(Metric[List[float]]):
     """
     Implementation of Completeness test by Sundararajan et al., 2017, also referred
     to as Summation to Delta by Shrikumar et al., 2017 and Conservation by
@@ -65,7 +73,7 @@ def __init__(
         normalise_func_kwargs: Optional[Dict[str, Any]] = None,
         output_func: Optional[Callable] = lambda x: x,
         perturb_baseline: str = "black",
-        perturb_func: Callable = None,
+        perturb_func: Callable = baseline_replacement_by_indices,
         perturb_func_kwargs: Optional[Dict[str, Any]] = None,
         return_aggregate: bool = False,
         aggregate_func: Callable = np.mean,
@@ -114,21 +122,11 @@ def __init__(
         """
         if normalise_func is None:
             normalise_func = normalise_by_max
-
-        if perturb_func is None:
-            perturb_func = baseline_replacement_by_indices
-
-        if perturb_func_kwargs is None:
-            perturb_func_kwargs = {}
-        perturb_func_kwargs["perturb_baseline"] = perturb_baseline
-
         super().__init__(
             abs=abs,
             normalise=normalise,
             normalise_func=normalise_func,
             normalise_func_kwargs=normalise_func_kwargs,
-            perturb_func=perturb_func,
-            perturb_func_kwargs=perturb_func_kwargs,
             return_aggregate=return_aggregate,
             aggregate_func=aggregate_func,
             default_plot_func=default_plot_func,
@@ -141,6 +139,9 @@ def __init__(
         if output_func is None:
             output_func = lambda x: x
         self.output_func = output_func
+        self.perturb_func = make_perturb_func(
+            perturb_func, perturb_func_kwargs, perturb_baseline=perturb_baseline
+        )
 
         # Asserts and warnings.
         if not self.disable_warnings:
@@ -268,7 +269,6 @@ def evaluate_instance(
         x: np.ndarray,
         y: np.ndarray,
         a: np.ndarray,
-        s: np.ndarray,
     ) -> bool:
         """
         Evaluate instance gets model and data for a single instance as input and returns the evaluation result.
@@ -283,19 +283,14 @@ def evaluate_instance(
             The output to be evaluated on an instance-basis.
         a: np.ndarray
             The explanation to be evaluated on an instance-basis.
-        s: np.ndarray
-            The segmentation to be evaluated on an instance-basis.
 
         Returns
         -------
            : boolean
             The evaluation results.
         """
         x_baseline = self.perturb_func(
-            arr=x,
-            indices=np.arange(0, x.size),
-            indexed_axes=np.arange(0, x.ndim),
-            **self.perturb_func_kwargs,
+            arr=x, indices=np.arange(0, x.size), indexed_axes=np.arange(0, x.ndim)
         )
 
         # Predict on input.
@@ -310,3 +305,43 @@ def evaluate_instance(
             return True
         else:
             return False
+
+    def evaluate_batch(
+        self,
+        model: ModelInterface,
+        x_batch: np.ndarray,
+        y_batch: np.ndarray,
+        a_batch: np.ndarray,
+        *args,
+        **kwargs,
+    ) -> List[bool]:
+        """
+        This method performs XAI evaluation on a single batch of explanations.
+        For more information on the specific logic, we refer the metric’s initialisation docstring.
+
+        Parameters
+        ----------
+        model: ModelInterface
+            A ModelInterface that is subject to explanation.
+        x_batch: np.ndarray
+            The input to be evaluated on a batch-basis.
+        y_batch: np.ndarray
+            The output to be evaluated on a batch-basis.
+        a_batch: np.ndarray
+            The explanation to be evaluated on a batch-basis.
+        args:
+            Unused.
+        kwargs:
+            Unused.
+
+        Returns
+        -------
+
+        scores_batch:
+            List of booleans.
+        """
+
+        return [
+            self.evaluate_instance(model=model, x=x, y=y, a=a)
+            for x, y, a in zip(x_batch, y_batch, a_batch)
+        ]