Finalize integration of BEYOND detector

Signed-off-by: Beat Buesser <[email protected]>

Author: beat-buesser

art/defences/detector/evasion/__init__.py

@@ -6,5 +6,4 @@
 from art.defences.detector.evasion.binary_input_detector import BinaryInputDetector
 from art.defences.detector.evasion.binary_activation_detector import BinaryActivationDetector
 from art.defences.detector.evasion.subsetscanning.detector import SubsetScanningDetector
-from art.defences.detector.evasion.beyond_detector import BeyondDetector
-
+from art.defences.detector.evasion.beyond_detector import BeyondDetectorPyTorch

art/defences/detector/evasion/beyond_detector.py

@@ -22,83 +22,95 @@
 """
 from __future__ import annotations
 
+import math
+from typing import TYPE_CHECKING, Callable
+
 import numpy as np
-from typing import TYPE_CHECKING
+
 if TYPE_CHECKING:
+    import torch
     from art.utils import CLASSIFIER_NEURALNETWORK_TYPE
 
 
 from art.defences.detector.evasion.evasion_detector import EvasionDetector
 
-class BeyondDetector(EvasionDetector):
+
+class BeyondDetectorPyTorch(EvasionDetector):
     """
       BEYOND detector for adversarial samples detection.
     This detector uses a combination of SSL and target model predictions to detect adversarial examples.
-    
+
     | Paper link: https://openreview.net/pdf?id=S4LqI6CcJ3
     """
-    
+
     defence_params = ["target_model", "ssl_model", "augmentations", "aug_num", "alpha", "K", "percentile"]
 
-    def __init__(self,
-        target_model: "CLASSIFIER_NEURALNETWORK_TYPE",
-        ssl_model: "CLASSIFIER_NEURALNETWORK_TYPE",
+    def __init__(
+        self,
+        target_classifier: "CLASSIFIER_NEURALNETWORK_TYPE",
+        ssl_classifier: "CLASSIFIER_NEURALNETWORK_TYPE",
         augmentations: Callable | None,
-        aug_num: int=50,
-        alpha: float=0.8,
-        K:int=20,
-        percentile:int=5) -> None:
+        aug_num: int = 50,
+        alpha: float = 0.8,
+        K: int = 20,
+        percentile: int = 5,
+    ) -> None:
         """
         Initialize the BEYOND detector.
 
-        :param target_model: The target model to be protected
-        :param ssl_model: The self-supervised learning model used for feature extraction
-        :param augmentation: data augmentations for generating neighborhoods
+        :param target_classifier: The target model to be protected
+        :param ssl_classifier: The self-supervised learning model used for feature extraction
+        :param augmentations: data augmentations for generating neighborhoods
         :param aug_num: Number of augmentations to apply to each sample (default: 50)
         :param alpha: Weight factor for combining label and representation similarities (default: 0.8)
         :param K: Number of top similarities to consider (default: 20)
         :param percentile: using to calculate the threshold
         """
+        import torch
+
         super().__init__()
         self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
-        self.target_model = target_model.to(self.device)
-        self.ssl_model = ssl_model.to(self.device)
+        self.target_model = target_classifier.model.to(self.device)
+        self.ssl_model = ssl_classifier.model.to(self.device)
         self.aug_num = aug_num
         self.alpha = alpha
         self.K = K
 
-        self.backbone = ssl_model.backbone
-        self.classifier = ssl_model.classifier
-        self.projector = ssl_model.projector
+        self.backbone = self.ssl_model.backbone
+        self.model_classifier = self.ssl_model.classifier
+        self.projector = self.ssl_model.projector
 
         self.img_augmentations = augmentations
 
-        self.percentile = percentile # determinate the threshold
-        self.threshold = None
+        self.percentile = percentile  # determine the threshold
+        self.threshold: float | None = None
+
+    def _multi_transform(self, img: "torch.Tensor") -> "torch.Tensor":
+        import torch
 
-    
-    
-    def _multi_transform(self, img: torch.Tensor) -> torch.Tensor:
         return torch.stack([self.img_augmentations(img) for _ in range(self.aug_num)], dim=1)
 
-    def _get_metrics(self, x: np.ndarray, batch_size: int = 128) -> tuple[dict, np.ndarray]:
+    def _get_metrics(self, x: np.ndarray, batch_size: int = 128) -> np.ndarray:
         """
         Calculate similarities that combining label consistency and representation similarity for given samples
 
         :param x: Input samples
         :param batch_size: Batch size for processing
         :return: A report similarities
         """
+        import torch
+        import torch.nn.functional as F
+
         samples = torch.from_numpy(x).to(self.device)
-        
+
         self.target_model.eval()
         self.backbone.eval()
-        self.classifier.eval()
+        self.model_classifier.eval()
         self.projector.eval()
 
         number_batch = int(math.ceil(len(samples) / batch_size))
-        
+
         similarities = []
 
         with torch.no_grad():
@@ -113,23 +125,31 @@ def _get_metrics(self, x: np.ndarray, batch_size: int = 128) -> tuple[dict, np.n
                 ssl_backbone_out = self.backbone(batch_samples)
 
                 ssl_repre = self.projector(ssl_backbone_out)
-                ssl_pred = self.classifier(ssl_backbone_out)
+                ssl_pred = self.model_classifier(ssl_backbone_out)
                 ssl_label = torch.max(ssl_pred, -1)[1]
 
                 aug_backbone_out = self.backbone(trans_images.reshape(-1, c, h, w))
                 aug_repre = self.projector(aug_backbone_out)
-                aug_pred = self.classifier(aug_backbone_out)
+                aug_pred = self.model_classifier(aug_backbone_out)
                 aug_pred = aug_pred.reshape(b, self.aug_num, -1)
 
-                sim_repre = F.cosine_similarity(ssl_repre.unsqueeze(dim=1), aug_repre.reshape(b, self.aug_num, -1), dim=2)
-                sim_preds = F.cosine_similarity(F.one_hot(torch.argmax(ssl_label, dim=1), num_classes=ssl_pred.shape[-1]).unsqueeze(dim=1), aug_pred, dim=2)
+                sim_repre = F.cosine_similarity(
+                    ssl_repre.unsqueeze(dim=1), aug_repre.reshape(b, self.aug_num, -1), dim=2
+                )
+
+                sim_preds = F.cosine_similarity(
+                    F.one_hot(ssl_label, num_classes=ssl_pred.shape[-1]).unsqueeze(dim=1),
+                    aug_pred,
+                    dim=2,
+                )
 
-                similarities.append((self.alpha * sim_preds + (1-self.alpha)*sim_repre).sort(descending=True)[0].cpu().numpy())
+                similarities.append(
+                    (self.alpha * sim_preds + (1 - self.alpha) * sim_repre).sort(descending=True)[0].cpu().numpy()
+                )
 
         similarities = np.concatenate(similarities, axis=0)
-        
-        return similarities
 
+        return similarities
 
     def fit(self, x: np.ndarray, y: np.ndarray, batch_size: int = 128, nb_epochs: int = 20, **kwargs) -> None:
         """
@@ -140,26 +160,26 @@ def fit(self, x: np.ndarray, y: np.ndarray, batch_size: int = 128, nb_epochs: in
         :param batch_size: Batch size for processing
         :param nb_epochs: Number of training epochs (not used in this method)
         """
-        k_minus_one_metrics = clean_metrics[:, self.K-1]
-        
-        self.threshold = np.percentile(k_minus_one_metrics, self.threshold)
+        clean_metrics = self._get_metrics(x=x, batch_size=batch_size)
+        k_minus_one_metrics = clean_metrics[:, self.K - 1]
+        self.threshold = np.percentile(k_minus_one_metrics, q=self.percentile)
 
     def detect(self, x: np.ndarray, batch_size: int = 128, **kwargs) -> tuple[dict, np.ndarray]:
         """
         Detect whether given samples are adversarial
-        
+
         :param x: Input samples
         :param batch_size: Batch size for processing
         :return: (report, is_adversarial):
-            where report containing detection results 
+            where report containing detection results
             where is_adversarial is a boolean list indicating whether samples are adversarial or not
         """
         if self.threshold is None:
             raise ValueError("Detector has not been fitted. Call fit() before detect().")
-        
+
         similarities = self._get_metrics(x, batch_size)
-        
-        report = similarities[:, self.K-1]
+
+        report = similarities[:, self.K - 1]
         is_adversarial = report < self.threshold
-        
+
         return report, is_adversarial

run_tests.sh

@@ -146,6 +146,10 @@ else
                          "tests/defences/test_rounded.py" \
                          "tests/defences/test_thermometer_encoding.py" \
                          "tests/defences/test_variance_minimization.py" \
+                         "tests/defences/detector/evasion/test_beyond_detector.py" \
+                         "tests/defences/detector/evasion/test_binary_activation_detector.py" \
+                         "tests/defences/detector/evasion/test_binary_input_detector.py" \
+                         "tests/defences/detector/evasion/test_subsetscanning_detector.py" \
                          "tests/defences/detector/poison/test_activation_defence.py" \
                          "tests/defences/detector/poison/test_clustering_analyzer.py" \
                          "tests/defences/detector/poison/test_ground_truth_evaluator.py" \