Merge pull request #1416 from Trusted-AI/development_issue_1325

Implement customizable summary writer and indicators of attack failure

Author: beat-buesser

art/attacks/attack.py

@@ -27,6 +27,7 @@
 import numpy as np
 
 from art.exceptions import EstimatorError
+from art.summary_writer import SummaryWriter, SummaryWriterDefault
 
 if TYPE_CHECKING:
     from art.utils import CLASSIFIER_TYPE
@@ -100,15 +101,17 @@ class Attack(abc.ABC):
     def __init__(
         self,
         estimator,
-        tensor_board: Union[str, bool] = False,
+        summary_writer: Union[str, bool, SummaryWriter] = False,
     ):
         """
         :param estimator: An estimator.
-        :param tensor_board: Activate summary writer for TensorBoard: Default is `False` and deactivated summary writer.
-                             If `True` save runs/CURRENT_DATETIME_HOSTNAME in current directory. Provide `path` in type
-                             `str` to save in path/CURRENT_DATETIME_HOSTNAME.
-                             Use hierarchical folder structure to compare between runs easily. e.g. pass in ‘runs/exp1’,
-                             ‘runs/exp2’, etc. for each new experiment to compare across them.
+        :param summary_writer: Activate summary writer for TensorBoard.
+                               Default is `False` and deactivated summary writer.
+                               If `True` save runs/CURRENT_DATETIME_HOSTNAME in current directory.
+                               If of type `str` save in path.
+                               If of type `SummaryWriter` apply provided custom summary writer.
+                               Use hierarchical folder structure to compare between runs easily. e.g. pass in
+                               ‘runs/exp1’, ‘runs/exp2’, etc. for each new experiment to compare across them.
         """
         super().__init__()
 
@@ -119,17 +122,13 @@ def __init__(
             raise EstimatorError(self.__class__, self.estimator_requirements, estimator)
 
         self._estimator = estimator
-        self.tensor_board = tensor_board
+        self._summary_writer_arg = summary_writer
+        self._summary_writer: Optional[SummaryWriter] = None
 
-        if tensor_board:  # pragma: no cover
-            from tensorboardX import SummaryWriter
-
-            if isinstance(tensor_board, str):
-                self.summary_writer = SummaryWriter(tensor_board)
-            else:
-                self.summary_writer = SummaryWriter()
-        else:
-            self.summary_writer = None
+        if isinstance(summary_writer, SummaryWriter):  # pragma: no cover
+            self._summary_writer = summary_writer
+        elif summary_writer:
+            self._summary_writer = SummaryWriterDefault(summary_writer)
 
         Attack._check_params(self)
 
@@ -138,6 +137,11 @@ def estimator(self):
         """The estimator."""
         return self._estimator
 
+    @property
+    def summary_writer(self):
+        """The summary writer."""
+        return self._summary_writer
+
     @property
     def estimator_requirements(self):
         """The estimator requirements."""
@@ -156,8 +160,8 @@ def set_params(self, **kwargs) -> None:
 
     def _check_params(self) -> None:
 
-        if not isinstance(self.tensor_board, (bool, str)):
-            raise ValueError("The argument `tensor_board` has to be either of type bool or str.")
+        if not isinstance(self._summary_writer_arg, (bool, str, SummaryWriter)):
+            raise ValueError("The argument `summary_writer` has to be either of type bool or str.")
 
     @staticmethod
     def is_estimator_valid(estimator, estimator_requirements) -> bool:

art/attacks/evasion/adversarial_patch/adversarial_patch_pytorch.py

@@ -34,6 +34,7 @@
 from art.estimators.estimator import BaseEstimator, NeuralNetworkMixin
 from art.estimators.classification.classifier import ClassifierMixin
 from art.utils import check_and_transform_label_format, is_probability, to_categorical
+from art.summary_writer import SummaryWriter
 
 if TYPE_CHECKING:
     # pylint: disable=C0412
@@ -60,7 +61,7 @@ class AdversarialPatchPyTorch(EvasionAttack):
         "max_iter",
         "batch_size",
         "patch_shape",
-        "tensor_board",
+        "summary_writer",
         "verbose",
     ]
 
@@ -78,7 +79,7 @@ def __init__(
         batch_size: int = 16,
         patch_shape: Optional[Tuple[int, int, int]] = None,
         patch_type: str = "circle",
-        tensor_board: Union[str, bool] = False,
+        summary_writer: Union[str, bool, SummaryWriter] = False,
         verbose: bool = True,
     ):
         """
@@ -98,11 +99,13 @@ def __init__(
         :param batch_size: The size of the training batch.
         :param patch_shape: The shape of the adversarial patch as a tuple of shape CHW (nb_channels, height, width).
         :param patch_type: The patch type, either circle or square.
-        :param tensor_board: Activate summary writer for TensorBoard: Default is `False` and deactivated summary writer.
-                             If `True` save runs/CURRENT_DATETIME_HOSTNAME in current directory. Provide `path` in type
-                             `str` to save in path/CURRENT_DATETIME_HOSTNAME.
-                             Use hierarchical folder structure to compare between runs easily. e.g. pass in ‘runs/exp1’,
-                             ‘runs/exp2’, etc. for each new experiment to compare across them.
+        :param summary_writer: Activate summary writer for TensorBoard.
+                               Default is `False` and deactivated summary writer.
+                               If `True` save runs/CURRENT_DATETIME_HOSTNAME in current directory.
+                               If of type `str` save in path.
+                               If of type `SummaryWriter` apply provided custom summary writer.
+                               Use hierarchical folder structure to compare between runs easily. e.g. pass in
+                               ‘runs/exp1’, ‘runs/exp2’, etc. for each new experiment to compare across them.
         :param verbose: Show progress bars.
         """
         import torch  # lgtm [py/repeated-import]
@@ -115,7 +118,7 @@ def __init__(
             torchvision_version[0] >= 0 and torchvision_version[1] >= 8
         ), "AdversarialPatchPyTorch requires torchvision>=0.8.0"
 
-        super().__init__(estimator=classifier, tensor_board=tensor_board)
+        super().__init__(estimator=classifier, summary_writer=summary_writer)
         self.rotation_max = rotation_max
         self.scale_min = scale_min
         self.scale_max = scale_max
@@ -484,25 +487,30 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> T
                     )
                     _ = self._train_step(images=images, target=target, mask=mask_i)
 
+            # Write summary
             if self.summary_writer is not None:  # pragma: no cover
-                self.summary_writer.add_image(
-                    "patch",
-                    self._patch,
+                x_patched = (
+                    self._random_overlay(
+                        images=torch.from_numpy(x).to(self.estimator.device), patch=self._patch, mask=mask
+                    )
+                    .detach()
+                    .cpu()
+                    .numpy()
+                )
+
+                self.summary_writer.update(
+                    batch_id=0,
                     global_step=i_iter,
+                    grad=None,
+                    patch=self._patch,
+                    estimator=self.estimator,
+                    x=x_patched,
+                    y=y,
+                    targeted=self.targeted,
                 )
 
-                if hasattr(self.estimator, "compute_losses"):
-                    x_patched = self._random_overlay(
-                        images=torch.from_numpy(x).to(self.estimator.device), patch=self._patch, mask=mask
-                    )
-                    losses = self.estimator.compute_losses(x=x_patched, y=torch.from_numpy(y).to(self.estimator.device))
-
-                    for key, value in losses.items():
-                        self.summary_writer.add_scalar(
-                            "loss/{}".format(key),
-                            np.mean(value.detach().cpu().numpy()),
-                            global_step=i_iter,
-                        )
+        if self.summary_writer is not None:
+            self.summary_writer.reset()
 
         return (
             self._patch.detach().cpu().numpy(),

art/attacks/evasion/adversarial_patch/adversarial_patch_tensorflow.py

@@ -35,6 +35,7 @@
 from art.estimators.estimator import BaseEstimator, NeuralNetworkMixin
 from art.estimators.classification.classifier import ClassifierMixin
 from art.utils import check_and_transform_label_format, is_probability, to_categorical
+from art.summary_writer import SummaryWriter
 
 if TYPE_CHECKING:
     # pylint: disable=C0412
@@ -60,7 +61,7 @@ class AdversarialPatchTensorFlowV2(EvasionAttack):
         "max_iter",
         "batch_size",
         "patch_shape",
-        "tensor_board",
+        "summary_writer",
         "verbose",
     ]
 
@@ -76,7 +77,7 @@ def __init__(
         max_iter: int = 500,
         batch_size: int = 16,
         patch_shape: Optional[Tuple[int, int, int]] = None,
-        tensor_board: Union[str, bool] = False,
+        summary_writer: Union[str, bool, SummaryWriter] = False,
         verbose: bool = True,
     ):
         """
@@ -93,16 +94,18 @@ def __init__(
         :param max_iter: The number of optimization steps.
         :param batch_size: The size of the training batch.
         :param patch_shape: The shape of the adversarial patch as a tuple of shape HWC (width, height, nb_channels).
-        :param tensor_board: Activate summary writer for TensorBoard: Default is `False` and deactivated summary writer.
-                             If `True` save runs/CURRENT_DATETIME_HOSTNAME in current directory. Provide `path` in type
-                             `str` to save in path/CURRENT_DATETIME_HOSTNAME.
-                             Use hierarchical folder structure to compare between runs easily. e.g. pass in ‘runs/exp1’,
-                             ‘runs/exp2’, etc. for each new experiment to compare across them.
+        :param summary_writer: Activate summary writer for TensorBoard.
+                               Default is `False` and deactivated summary writer.
+                               If `True` save runs/CURRENT_DATETIME_HOSTNAME in current directory.
+                               If of type `str` save in path.
+                               If of type `SummaryWriter` apply provided custom summary writer.
+                               Use hierarchical folder structure to compare between runs easily. e.g. pass in
+                               ‘runs/exp1’, ‘runs/exp2’, etc. for each new experiment to compare across them.
         :param verbose: Show progress bars.
         """
         import tensorflow as tf  # lgtm [py/repeated-import]
 
-        super().__init__(estimator=classifier, tensor_board=tensor_board)
+        super().__init__(estimator=classifier, summary_writer=summary_writer)
         self.rotation_max = rotation_max
         self.scale_min = scale_min
         self.scale_max = scale_max
@@ -458,23 +461,23 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> T
                 for images, target, mask_i in dataset:
                     _ = self._train_step(images=images, target=target, mask=mask_i)
 
+            # Write summary
             if self.summary_writer is not None:  # pragma: no cover
-                self.summary_writer.add_image(
-                    "patch",
-                    self._patch.numpy().transpose((2, 0, 1)),
+                x_patched = self._random_overlay(images=x, patch=self._patch, mask=mask)
+
+                self.summary_writer.update(
+                    batch_id=0,
                     global_step=i_iter,
+                    grad=None,
+                    patch=self._patch.numpy().transpose((2, 0, 1)),
+                    estimator=self.estimator,
+                    x=x_patched,
+                    y=y,
+                    targeted=self.targeted,
                 )
 
-                if hasattr(self.estimator, "compute_losses"):
-                    x_patched = self._random_overlay(images=x, patch=self._patch, mask=mask)
-                    losses = self.estimator.compute_losses(x=x_patched, y=y)
-
-                    for key, value in losses.items():
-                        self.summary_writer.add_scalar(
-                            "loss/{}".format(key),
-                            np.mean(value),
-                            global_step=i_iter,
-                        )
+        if self.summary_writer is not None:
+            self.summary_writer.reset()
 
         return (
             self._patch.numpy(),

art/attacks/evasion/adversarial_texture/adversarial_texture_pytorch.py

@@ -29,6 +29,7 @@
 from art.attacks.attack import EvasionAttack
 from art.estimators.estimator import BaseEstimator, LossGradientsMixin
 from art.estimators.object_tracking.object_tracker import ObjectTrackerMixin
+from art.summary_writer import SummaryWriter
 
 if TYPE_CHECKING:
     # pylint: disable=C0412
@@ -67,6 +68,7 @@ def __init__(
         step_size: float = 1.0 / 255.0,
         max_iter: int = 500,
         batch_size: int = 16,
+        summary_writer: Union[str, bool, SummaryWriter] = False,
         verbose: bool = True,
     ):
         """
@@ -80,11 +82,18 @@ def __init__(
         :param step_size: The step size.
         :param max_iter: The number of optimization steps.
         :param batch_size: The size of the training batch.
+        :param summary_writer: Activate summary writer for TensorBoard.
+                               Default is `False` and deactivated summary writer.
+                               If `True` save runs/CURRENT_DATETIME_HOSTNAME in current directory.
+                               If of type `str` save in path.
+                               If of type `SummaryWriter` apply provided custom summary writer.
+                               Use hierarchical folder structure to compare between runs easily. e.g. pass in
+                               ‘runs/exp1’, ‘runs/exp2’, etc. for each new experiment to compare across them.
         :param verbose: Show progress bars.
         """
         import torch  # lgtm [py/repeated-import]
 
-        super().__init__(estimator=estimator)
+        super().__init__(estimator=estimator, summary_writer=summary_writer)
         self.patch_height = patch_height
         self.patch_width = patch_width
         self.x_min = x_min
@@ -95,6 +104,9 @@ def __init__(
         self.verbose = verbose
         self._check_params()
 
+        self._batch_id = 0
+        self._i_max_iter = 0
+
         self.patch_shape = (self.patch_height, self.patch_width, 3)
 
         if self.estimator.channels_first:
@@ -142,6 +154,18 @@ def _train_step(
 
         gradients = self._patch.grad.sign() * self.step_size
 
+        # Write summary
+        if self.summary_writer is not None:  # pragma: no cover
+            self.summary_writer.update(
+                batch_id=self._batch_id,
+                global_step=self._i_max_iter,
+                grad=np.expand_dims(self._patch.grad.detach().cpu().numpy(), axis=0),
+                patch=None,
+                estimator=None,
+                x=None,
+                y=None,
+            )
+
         with torch.no_grad():
             self._patch[:] = torch.clamp(
                 self._patch + gradients, min=self.estimator.clip_values[0], max=self.estimator.clip_values[1]
@@ -356,8 +380,13 @@ def __getitem__(self, idx):
             drop_last=False,
         )
 
-        for _ in trange(self.max_iter, desc="Adversarial Texture PyTorch", disable=not self.verbose):
+        for i_max_iter in trange(self.max_iter, desc="Adversarial Texture PyTorch", disable=not self.verbose):
+
+            self._i_max_iter = i_max_iter
+            self._batch_id = 0
+
             for videos_i, target_i, y_init_i, foreground_i in data_loader:
+                self._batch_id += 1
                 videos_i = videos_i.to(self.estimator.device)
                 y_init_i = y_init_i.to(self.estimator.device)
                 foreground_i = foreground_i.to(self.estimator.device)
@@ -367,6 +396,21 @@ def __getitem__(self, idx):
 
                 _ = self._train_step(videos=videos_i, target=target_i_list, y_init=y_init_i, foreground=foreground_i)
 
+                # Write summary
+                if self.summary_writer is not None:  # pragma: no cover
+                    self.summary_writer.update(
+                        batch_id=self._batch_id,
+                        global_step=self._i_max_iter,
+                        grad=None,
+                        patch=self._patch.detach().cpu().numpy(),
+                        estimator=self.estimator,
+                        x=videos_i.detach().cpu().numpy(),
+                        y=target_i_list,
+                    )
+
+        if self.summary_writer is not None:
+            self.summary_writer.reset()
+
         return self.apply_patch(x=x, foreground=foreground)
 
     def apply_patch(