[WIP] Tests improved. MNIST results obtained in Notebook

Signed-off-by: alvaro <[email protected]>

Author: c4ts0up

art/defences/detector/poison/clustering_centroid_analysis.py

@@ -18,6 +18,7 @@
 from __future__ import absolute_import, division, print_function, unicode_literals, annotations
 
 import logging
+import warnings
 from typing import TYPE_CHECKING
 
 import tensorflow as tf
@@ -165,15 +166,18 @@ class ClusteringCentroidAnalysis(PoisonFilteringDefence):
 
     """
 
-    _DEFENCE_PARAMS = []
-    _VALID_CLUSTERING = ["DBSCAN"]
-    _VALID_REDUCE = ["UMAP", "PCA"]
-    _VALID_ANALYSIS = [
-        ClusterAnalysisType.SMALLER,
-        ClusterAnalysisType.RELATIVE_SIZE,
-        ClusterAnalysisType.DISTANCE,
-        ClusterAnalysisType.SILHOUETTE_SCORES
+    defence_params = [
+        "classifier",
+        "x_train",
+        "y_train",
+        "benign_indices",
+        "final_feature_layer_name",
+        "misclassification_threshold",
+        "reducer",
+        "clsuterer"
     ]
+    valid_clustering = ["DBSCAN"]
+    valid_reduce = ["UMAP"]
 
     def _get_benign_data(self) -> (np.ndarray, np.ndarray):
         """
@@ -204,8 +208,8 @@ def _extract_submodels(self, final_feature_layer_name: str) -> (Model, Model):
         except ValueError:
             raise ValueError(f"Layer with name '{final_feature_layer_name}' not found in the model.")
 
-        #if not hasattr(final_feature_layer, 'activation') or final_feature_layer.activation != tf.keras.activations.relu:
-            #raise ValueError(f"Final feature layer '{final_feature_layer_name}' must have a ReLU activation.")
+        if not hasattr(final_feature_layer, 'activation') or final_feature_layer.activation != tf.keras.activations.relu:
+            warnings.warn(f"Final feature layer '{final_feature_layer_name}' must have a ReLU activation.", UserWarning)
 
         # Create a feature representation submodel with weight sharing
         feature_representation_model = Model(
@@ -479,23 +483,3 @@ def detect_poison(self, **kwargs) -> (dict, list[int]):
                 logging.info(f"Cluster k={cluster_label} i={self.cluster_class_mapping[cluster_label]} considered poison ({misclassification_rates[cluster_label]} >= {1 - self.misclassification_threshold})")
 
         return report, self.is_clean.copy()
-
-
-def get_reducer(reduce: ReducerType, nb_dims: int):
-    """Initialize the right reducer based on the selected type."""
-    if reduce == ReducerType.FASTICA:
-        return FastICA(n_components=nb_dims, max_iter=1000, tol=0.005)
-    if reduce == ReducerType.PCA:
-        return PCA(n_components=nb_dims)
-    if reduce == ReducerType.UMAP:
-        return UMAP(n_components=nb_dims, random_state=42)  # TODO: should I remove the random state?
-
-    raise ValueError(f"{reduce} dimensionality reduction method not supported.")
-
-
-def get_clusterer(clusterer_type: ClustererType) -> ClusterMixin:
-    """Initialize the right cluster algorithm (a.k.a., clusterer) based on the selected type. """
-    if clusterer_type == ClustererType.DBSCAN:
-        return DBSCAN(eps=0.8, min_samples=20)
-
-    raise ValueError(f"{clusterer_type} cluster method not supported.")

notebooks/poisoning_defense_clustering_centroid_analysis.ipynb

@@ -37,9 +37,7 @@
 from tensorflow.keras.layers import Dense
 from umap import UMAP
 
-from art.defences.detector.poison.clustering_centroid_analysis import get_reducer, get_clusterer, \
-    ClusteringCentroidAnalysis, _calculate_centroid, _class_clustering, _feature_extraction, _cluster_classes, \
-    _encode_labels
+from art.defences.detector.poison.clustering_centroid_analysis import ClusteringCentroidAnalysis, _calculate_centroid, _class_clustering, _feature_extraction, _cluster_classes, _encode_labels
 from art.defences.detector.poison.utils import ReducerType, ClustererType
 
 logger = logging.getLogger(__name__)
@@ -276,7 +274,7 @@ def test_init_invalid_layer_name(self):
 
     def test_init_invalid_layer_non_relu(self):
         """Test __init__ with an invalid layer that does not have ReLu activation. Check that it raises error."""
-        with self.assertRaises(ValueError) as e:
+        with self.assertWarns(UserWarning) as w:
             ClusteringCentroidAnalysis(
                 classifier=self.mock_classifier,
                 x_train=self.x_train,
@@ -285,7 +283,9 @@ def test_init_invalid_layer_non_relu(self):
                 final_feature_layer_name=self.non_relu_intermediate_layer_name,
                 misclassification_threshold=self.misclassification_threshold
             )
-            self.assertEqual(f"Final feature layer '{self.non_relu_intermediate_layer_name}' must have a ReLU activation.", str(e.exception))
+            self.assertEqual(1, len(w.warnings))
+            self.assertEqual(f"Final feature layer '{self.non_relu_intermediate_layer_name}' must have a ReLU activation.",
+                             str(w.warnings[0].message))
 
 class TestEncodeLabels(unittest.TestCase):
 
@@ -650,44 +650,6 @@ def test_integration_with_real_model(self):
         self.assertIsInstance(result, np.ndarray)
 
 
-class TestReducersClusterers(unittest.TestCase):
-    """
-    Suite of tests for the valid and invalid utils used in :class: ``ClusteringCentroidAnalysis``
-    """
-
-    def test_get_reducer_valid(self):
-        reducer_cases = [
-            (ReducerType.FASTICA, FastICA),
-            (ReducerType.PCA, PCA),
-            (ReducerType.UMAP, UMAP),
-        ]
-        for reducer_type, expected in reducer_cases:
-            with self.subTest(reducer=reducer_type):
-                reducer = get_reducer(reducer_type, nb_dims=5)
-                self.assertIsInstance(reducer, expected)
-
-    def test_get_reducer_invalid(self):
-        for invalid in ["INVALID", None]:
-            with self.subTest(invalid=invalid):
-                with self.assertRaises(ValueError):
-                    get_reducer(invalid, nb_dims=5)
-
-    def test_get_clusterer_valid(self):
-        clusterer_cases = [
-            (ClustererType.DBSCAN, DBSCAN),
-        ]
-        for clusterer_type, expected in clusterer_cases:
-            with self.subTest(clusterer=clusterer_type):
-                clusterer = get_clusterer(clusterer_type)
-                self.assertIsInstance(clusterer, expected)
-
-    def test_get_clusterer_invalid(self):
-        for invalid in ["INVALID", None]:
-            with self.subTest(invalid=invalid):
-                with self.assertRaises(ValueError):
-                    get_clusterer(invalid)
-
-
 class TestDetectPoison(unittest.TestCase):
     """
     Unit tests for the detect_poison method in ClusteringCentroidAnalysis