Merge branch 'dev_1.5.0' of github.com:Trusted-AI/adversarial-robustness-toolbox into dev_1.5.0

Author: Beat Buesser

art/attacks/evasion/fast_gradient.py

@@ -191,27 +191,6 @@ def _minimal_perturbation(self, x: np.ndarray, y: np.ndarray, mask: np.ndarray)
 
         return adv_x
 
-    @staticmethod
-    def _get_mask(x: np.ndarray, **kwargs) -> np.ndarray:
-        """
-        Get the mask from the kwargs.
-
-        :param x: An array with the original inputs.
-        :param mask: An array with a mask to be applied to the adversarial perturbations. Shape needs to be
-                     broadcastable to the shape of x. Any features for which the mask is zero will not be adversarially
-                     perturbed.
-        :type mask: `np.ndarray`
-        :return: The mask.
-        """
-        mask = kwargs.get("mask")
-
-        if mask is not None:
-            # Ensure the mask is broadcastable
-            if len(mask.shape) > len(x.shape) or mask.shape != x.shape[-len(mask.shape) :]:
-                raise ValueError("Mask shape must be broadcastable to input shape.")
-
-        return mask
-
     def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.ndarray:
         """Generate adversarial samples and return them in an array.
 
@@ -226,9 +205,7 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> n
         :type mask: `np.ndarray`
         :return: An array holding the adversarial examples.
         """
-        mask = kwargs.get("mask")
-        if mask is not None and mask.ndim > x.ndim:
-            raise ValueError("Mask shape must be broadcastable to input shape.")
+        mask = self._get_mask(x, **kwargs)
 
         # Ensure eps is broadcastable
         self._check_compatibility_input_and_eps(x=x)
@@ -355,13 +332,19 @@ def _check_params(self) -> None:
         if not isinstance(self.minimal, bool):
             raise ValueError("The flag `minimal` has to be of type bool.")
 
-    def _compute_perturbation(self, batch: np.ndarray, batch_labels: np.ndarray, mask: np.ndarray) -> np.ndarray:
+    def _compute_perturbation(
+        self, batch: np.ndarray, batch_labels: np.ndarray, mask: Optional[np.ndarray]
+    ) -> np.ndarray:
         # Pick a small scalar to avoid division by 0
         tol = 10e-8
 
         # Get gradient wrt loss; invert it if attack is targeted
         grad = self.estimator.loss_gradient(batch, batch_labels) * (1 - 2 * int(self.targeted))
 
+        # Apply mask
+        if mask is not None:
+            grad = np.where(mask == 0.0, 0.0, grad)
+
         # Apply norm bound
         def _apply_norm(grad, object_type=False):
             if self.norm in [np.inf, "inf"]:
@@ -389,10 +372,7 @@ def _apply_norm(grad, object_type=False):
 
         assert batch.shape == grad.shape
 
-        if mask is None:
-            return grad
-        else:
-            return grad * (mask.astype(ART_NUMPY_DTYPE))
+        return grad
 
     def _apply_perturbation(
         self, batch: np.ndarray, perturbation: np.ndarray, eps_step: Union[int, float, np.ndarray]
@@ -487,3 +467,35 @@ def _compute(
                     x_adv[batch_index_1:batch_index_2] = x_init[batch_index_1:batch_index_2] + perturbation
 
         return x_adv
+
+    @staticmethod
+    def _get_mask(x: np.ndarray, **kwargs) -> np.ndarray:
+        """
+        Get the mask from the kwargs.
+
+        :param x: An array with the original inputs.
+        :param mask: An array with a mask to be applied to the adversarial perturbations. Shape needs to be
+                     broadcastable to the shape of x. Any features for which the mask is zero will not be adversarially
+                     perturbed.
+        :type mask: `np.ndarray`
+        :return: The mask.
+        """
+        mask = kwargs.get("mask")
+
+        if mask is not None:
+            if mask.ndim > x.ndim:
+                raise ValueError("Mask shape must be broadcastable to input shape.")
+
+            if not (np.issubdtype(mask.dtype, np.floating) or mask.dtype == np.bool):
+                raise ValueError(
+                    "The `mask` has to be either of type np.float32, np.float64 or np.bool. The provided"
+                    "`mask` is of type {}.".format(mask.dtype)
+                )
+
+            if np.issubdtype(mask.dtype, np.floating) and np.amin(mask) < 0.0:
+                raise ValueError(
+                    "The `mask` of type np.float32 or np.float64 requires all elements to be either zero"
+                    "or positive values."
+                )
+
+        return mask

art/attacks/evasion/projected_gradient_descent/projected_gradient_descent_numpy.py

@@ -246,21 +246,14 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> n
         :type mask: `np.ndarray`
         :return: An array holding the adversarial examples.
         """
-        mask = kwargs.get("mask")
-
-        # Check the mask
-        if mask is not None and mask.ndim > x.ndim:
-            raise ValueError("Mask shape must be broadcastable to input shape.")
+        mask = self._get_mask(x, **kwargs)
 
         # Ensure eps is broadcastable
         self._check_compatibility_input_and_eps(x=x)
 
         # Check whether random eps is enabled
         self._random_eps()
 
-        # Get the mask
-        mask = self._get_mask(x, **kwargs)
-
         if isinstance(self.estimator, ClassifierMixin):
             # Set up targets
             targets = self._set_targets(x, y)

art/attacks/evasion/projected_gradient_descent/projected_gradient_descent_pytorch.py

@@ -32,6 +32,8 @@
 from tqdm import trange, tqdm
 
 from art.config import ART_NUMPY_DTYPE
+from art.estimators.estimator import BaseEstimator, LossGradientsMixin
+from art.estimators.classification.classifier import ClassifierMixin
 from art.attacks.evasion.projected_gradient_descent.projected_gradient_descent_numpy import (
     ProjectedGradientDescentCommon,
 )
@@ -40,7 +42,6 @@
 if TYPE_CHECKING:
     import torch
     from art.estimators.classification.pytorch import PyTorchClassifier
-    from art.estimators.object_detection.pytorch_faster_rcnn import PyTorchFasterRCNN
 
 logger = logging.getLogger(__name__)
 
@@ -54,9 +55,11 @@ class ProjectedGradientDescentPyTorch(ProjectedGradientDescentCommon):
     | Paper link: https://arxiv.org/abs/1706.06083
     """
 
+    _estimator_requirements = (BaseEstimator, LossGradientsMixin, ClassifierMixin)
+
     def __init__(
         self,
-        estimator: Union["PyTorchClassifier", "PyTorchFasterRCNN"],
+        estimator: Union["PyTorchClassifier"],
         norm: Union[int, float, str] = np.inf,
         eps: Union[int, float, np.ndarray] = 0.3,
         eps_step: Union[int, float, np.ndarray] = 0.1,
@@ -120,9 +123,7 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> n
         """
         import torch  # lgtm [py/repeated-import]
 
-        mask = kwargs.get("mask")
-        if mask is not None and mask.ndim > x.ndim:
-            raise ValueError("Mask shape must be broadcastable to input shape.")
+        mask = self._get_mask(x, **kwargs)
 
         # Ensure eps is broadcastable
         self._check_compatibility_input_and_eps(x=x)
@@ -249,7 +250,9 @@ def _generate_batch(
 
         return adv_x.cpu().detach().numpy()
 
-    def _compute_perturbation(self, x: "torch.Tensor", y: "torch.Tensor", mask: "torch.Tensor") -> "torch.Tensor":
+    def _compute_perturbation(
+        self, x: "torch.Tensor", y: "torch.Tensor", mask: Optional["torch.Tensor"]
+    ) -> "torch.Tensor":
         """
         Compute perturbations.
 
@@ -271,6 +274,10 @@ def _compute_perturbation(self, x: "torch.Tensor", y: "torch.Tensor", mask: "tor
         # Get gradient wrt loss; invert it if attack is targeted
         grad = self.estimator.loss_gradient(x=x, y=y) * (1 - 2 * int(self.targeted))
 
+        # Apply mask
+        if mask is not None:
+            grad = torch.where(mask == 0.0, torch.tensor(0.0), grad)
+
         # Apply norm bound
         if self.norm in ["inf", np.inf]:
             grad = grad.sign()
@@ -285,10 +292,7 @@ def _compute_perturbation(self, x: "torch.Tensor", y: "torch.Tensor", mask: "tor
 
         assert x.shape == grad.shape
 
-        if mask is None:
-            return grad
-        else:
-            return grad * mask
+        return grad
 
     def _apply_perturbation(
         self, x: "torch.Tensor", perturbation: "torch.Tensor", eps_step: Union[int, float, np.ndarray]

art/attacks/evasion/projected_gradient_descent/projected_gradient_descent_tensorflow_v2.py

@@ -32,6 +32,8 @@
 from tqdm import trange, tqdm
 
 from art.config import ART_NUMPY_DTYPE
+from art.estimators.estimator import BaseEstimator, LossGradientsMixin
+from art.estimators.classification.classifier import ClassifierMixin
 from art.attacks.evasion.projected_gradient_descent.projected_gradient_descent_numpy import (
     ProjectedGradientDescentCommon,
 )
@@ -53,6 +55,8 @@ class ProjectedGradientDescentTensorFlowV2(ProjectedGradientDescentCommon):
     | Paper link: https://arxiv.org/abs/1706.06083
     """
 
+    _estimator_requirements = (BaseEstimator, LossGradientsMixin, ClassifierMixin)
+
     def __init__(
         self,
         estimator: "TensorFlowV2Classifier",
@@ -119,9 +123,7 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> n
         """
         import tensorflow as tf  # lgtm [py/repeated-import]
 
-        mask = kwargs.get("mask")
-        if mask is not None and mask.ndim > x.ndim:
-            raise ValueError("Mask shape must be broadcastable to input shape.")
+        mask = self._get_mask(x, **kwargs)
 
         # Ensure eps is broadcastable
         self._check_compatibility_input_and_eps(x=x)
@@ -224,17 +226,11 @@ def _generate_batch(
 
         :param x: An array with the original inputs.
         :param targets: Target values (class labels) one-hot-encoded of shape `(nb_samples, nb_classes)`.
-<<<<<<< HEAD
-        :param mask: An array with a mask to be applied to the adversarial perturbations. Shape needs to be
-                     broadcastable to the shape of x. Any features for which the mask is zero will not be adversarially
-                     perturbed.
-        :param eps: Maximum perturbation that the attacker can introduce.
-        :param eps_step: Attack step size (input variation) at each iteration.
-=======
         :param mask: An array with a mask broadcastable to input `x` defining where to apply adversarial perturbations.
                      Shape needs to be broadcastable to the shape of x and can also be of the same shape as `x`. Any
                      features for which the mask is zero will not be adversarially perturbed.
->>>>>>> origin/dev_1.5.0
+        :param eps: Maximum perturbation that the attacker can introduce.
+        :param eps_step: Attack step size (input variation) at each iteration.
         :return: Adversarial examples.
         """
         adv_x = x
@@ -269,6 +265,10 @@ def _compute_perturbation(self, x: "tf.Tensor", y: "tf.Tensor", mask: Optional["
             1 - 2 * int(self.targeted), dtype=ART_NUMPY_DTYPE
         )
 
+        # Apply mask
+        if mask is not None:
+            grad = tf.where(mask == 0.0, 0.0, grad)
+
         # Apply norm bound
         if self.norm == np.inf:
             grad = tf.sign(grad)
@@ -285,10 +285,7 @@ def _compute_perturbation(self, x: "tf.Tensor", y: "tf.Tensor", mask: Optional["
 
         assert x.shape == grad.shape
 
-        if mask is None:
-            return grad
-        else:
-            return grad * mask
+        return grad
 
     def _apply_perturbation(
         self, x: "tf.Tensor", perturbation: "tf.Tensor", eps_step: Union[int, float, np.ndarray]

art/attacks/evasion/zoo.py

@@ -344,7 +344,10 @@ def compare(object1, object2):
         else:
             x_orig = x_batch
             self._reset_adam(np.prod(self.estimator.input_shape).item())
-            self._current_noise.fill(0)
+            if x_batch.shape == self._current_noise.shape:
+                self._current_noise.fill(0)
+            else:
+                self._current_noise = np.zeros(x_batch.shape, dtype=ART_NUMPY_DTYPE)
             x_adv = x_orig.copy()
 
         # Initialize best distortions, best changed labels and best attacks
@@ -537,7 +540,10 @@ def _resize_image(self, x: np.ndarray, size_x: int, size_y: int, reset: bool = F
             # Reset variables to original size and value
             if dims == x.shape:
                 resized_x = x
-                self._current_noise.fill(0)
+                if x.shape == self._current_noise.shape:
+                    self._current_noise.fill(0)
+                else:
+                    self._current_noise = np.zeros(x.shape, dtype=ART_NUMPY_DTYPE)
             else:
                 resized_x = zoom(x, (1, dims[1] / x.shape[1], dims[2] / x.shape[2], dims[3] / x.shape[3],),)
                 self._current_noise = np.zeros(dims, dtype=ART_NUMPY_DTYPE)

art/defences/preprocessor/preprocessor.py

@@ -205,10 +205,16 @@ def get_gradient(x, grad):
 
             return x_grad
 
-        if x.shape == grad.shape:
+        if x.dtype == np.object:
+            x_grad_list = list()
+            for i, x_i in enumerate(x):
+                x_grad_list.append(get_gradient(x=x_i, grad=grad[i]))
+            x_grad = np.empty(x.shape[0], dtype=object)
+            x_grad[:] = list(x_grad_list)
+        elif x.shape == grad.shape:
             x_grad = get_gradient(x=x, grad=grad)
         else:
-            # Special case for lass gradients
+            # Special case for loss gradients
             x_grad = np.zeros_like(grad)
             for i in range(grad.shape[1]):
                 x_grad[:, i, ...] = get_gradient(x=x, grad=grad[:, i, ...])
@@ -268,35 +274,41 @@ def __call__(self, x: np.ndarray, y: Optional[np.ndarray] = None) -> Tuple[np.nd
         return result, y
 
     # Backward compatibility.
-    def _get_gradient(self, x: np.ndarray, grad: np.ndarray) -> np.ndarray:
-        """
-        Helper function for estimate_gradient
-        """
+    def estimate_gradient(self, x: np.ndarray, grad: np.ndarray) -> np.ndarray:
         import tensorflow as tf  # lgtm [py/repeated-import]
 
-        with tf.GradientTape() as tape:
-            x = tf.convert_to_tensor(x, dtype=config.ART_NUMPY_DTYPE)
-            tape.watch(x)
-            grad = tf.convert_to_tensor(grad, dtype=config.ART_NUMPY_DTYPE)
+        def get_gradient(x: np.ndarray, grad: np.ndarray) -> np.ndarray:
+            """
+            Helper function for estimate_gradient
+            """
 
-            x_prime = self.estimate_forward(x)
+            with tf.GradientTape() as tape:
+                x = tf.convert_to_tensor(x, dtype=config.ART_NUMPY_DTYPE)
+                tape.watch(x)
+                grad = tf.convert_to_tensor(grad, dtype=config.ART_NUMPY_DTYPE)
 
-        x_grad = tape.gradient(target=x_prime, sources=x, output_gradients=grad)
+                x_prime = self.estimate_forward(x)
 
-        x_grad = x_grad.numpy()
-        if x_grad.shape != x.shape:
-            raise ValueError("The input shape is {} while the gradient shape is {}".format(x.shape, x_grad.shape))
+            x_grad = tape.gradient(target=x_prime, sources=x, output_gradients=grad)
 
-        return x_grad
+            x_grad = x_grad.numpy()
+            if x_grad.shape != x.shape:
+                raise ValueError("The input shape is {} while the gradient shape is {}".format(x.shape, x_grad.shape))
 
-    # Backward compatibility.
-    def estimate_gradient(self, x: np.ndarray, grad: np.ndarray) -> np.ndarray:
-        if x.shape == grad.shape:
-            x_grad = self._get_gradient(x=x, grad=grad)
+            return x_grad
+
+        if x.dtype == np.object:
+            x_grad_list = list()
+            for i, x_i in enumerate(x):
+                x_grad_list.append(get_gradient(x=x_i, grad=grad[i]))
+            x_grad = np.empty(x.shape[0], dtype=object)
+            x_grad[:] = list(x_grad_list)
+        elif x.shape == grad.shape:
+            x_grad = get_gradient(x=x, grad=grad)
         else:
-            # Special case for lass gradients
+            # Special case for loss gradients
             x_grad = np.zeros_like(grad)
             for i in range(grad.shape[1]):
-                x_grad[:, i, ...] = self._get_gradient(x=x, grad=grad[:, i, ...])
+                x_grad[:, i, ...] = get_gradient(x=x, grad=grad[:, i, ...])
 
         return x_grad

art/defences/preprocessor/spatial_smoothing_pytorch.py

@@ -63,7 +63,7 @@ def __init__(
         """
         Create an instance of local spatial smoothing.
 
-        :window_size: Size of spatial smoothing window.
+        :param window_size: Size of spatial smoothing window.
         :param channels_first: Set channels first or last.
         :param clip_values: Tuple of the form `(min, max)` representing the minimum and maximum values allowed
                for features.