Fix dependencies

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

Author: beat-buesser

art/attacks/poisoning/gradient_matching_attack.py

@@ -116,25 +116,14 @@ def _initialize_poison(
         :param y_train: A list of labels for x_train.
         """
         from art.estimators.classification.pytorch import PyTorchClassifier
-        from art.estimators.classification.tensorflow import TensorFlowV2Classifier
 
-        if isinstance(self.substitute_classifier, TensorFlowV2Classifier):
-            initializer = self._initialize_poison_tensorflow
-        elif isinstance(self.substitute_classifier, PyTorchClassifier):
+        if isinstance(self.substitute_classifier, PyTorchClassifier):
             initializer = self._initialize_poison_pytorch
         else:
-            raise NotImplementedError(
-                "GradientMatchingAttack is currently implemented only for TensorFlow V2 and PyTorch."
-            )
+            raise NotImplementedError("GradientMatchingAttack is currently implemented only for PyTorch.")
 
         return initializer(x_trigger, y_trigger, x_poison, y_poison)
 
-    def _finish_poison_tensorflow(self):
-        """
-        Releases any resource and revert back unwanted change to the model.
-        """
-        self.substitute_classifier.model.trainable = self.model_trainable
-
     def _finish_poison_pytorch(self):
         """
         Releases any resource and revert back unwanted change to the model.
@@ -144,103 +133,6 @@ def _finish_poison_pytorch(self):
         else:
             self.substitute_classifier.model.eval()
 
-    def _initialize_poison_tensorflow(
-        self, x_trigger: np.ndarray, y_trigger: np.ndarray, x_poison: np.ndarray, y_poison: np.ndarray
-    ):
-        """
-        Initialize poison noises to be optimized.
-
-        :param x_trigger: A list of samples to use as triggers.
-        :param y_trigger: A list of target classes to classify the triggers into.
-        :param x_poison: A list of training data to poison a portion of.
-        :param y_poison: A list of true labels for x_poison.
-        """
-        from tensorflow.keras import backend as K
-        import tensorflow as tf
-        from tensorflow.keras.layers import Input, Embedding, Add, Lambda
-        from art.estimators.classification.tensorflow import TensorFlowV2Classifier
-
-        if isinstance(self.substitute_classifier, TensorFlowV2Classifier):
-            classifier = self.substitute_classifier
-        else:
-            raise Exception("This method requires `TensorFlowV2Classifier` as `substitute_classifier`'s type")
-
-        self.model_trainable = classifier.model.trainable
-        classifier.model.trainable = False  # This value gets revert back later.
-
-        def _weight_grad(classifier: TensorFlowV2Classifier, x: tf.Tensor, target: tf.Tensor) -> tf.Tensor:
-            # Get the target gradient vector.
-            import tensorflow as tf
-
-            with tf.GradientTape() as t:  # pylint: disable=invalid-name
-                t.watch(classifier.model.weights)
-                output = classifier.model(x, training=False)
-                loss = classifier.loss_object(target, output)
-            d_w = t.gradient(loss, classifier.model.weights)
-            d_w = [w for w in d_w if w is not None]
-            d_w = tf.concat([tf.reshape(d, [-1]) for d in d_w], 0)
-            d_w_norm = d_w / tf.sqrt(tf.reduce_sum(tf.square(d_w)))
-            return d_w_norm
-
-        self.grad_ws_norm = _weight_grad(classifier, tf.constant(x_trigger), tf.constant(y_trigger))
-
-        # Define the model to apply and optimize the poison.
-        input_poison = Input(batch_shape=classifier.model.input.shape)
-        input_indices = Input(shape=())
-        y_true_poison = Input(shape=np.shape(y_poison)[1:])
-        embedding_layer = Embedding(
-            len(x_poison),
-            np.prod(x_poison.shape[1:]),
-            embeddings_initializer=tf.keras.initializers.RandomNormal(stddev=self.epsilon * 0.01),
-        )
-        embeddings = embedding_layer(input_indices)
-        embeddings = tf.tanh(embeddings) * self.epsilon
-        embeddings = tf.reshape(embeddings, tf.shape(input_poison))
-        input_noised = Add()([input_poison, embeddings])
-        input_noised = Lambda(lambda x: K.clip(x, self.clip_values[0], self.clip_values[1]))(
-            input_noised
-        )  # Make sure the poisoned samples are in a valid range.
-
-        def loss_fn(input_noised: tf.Tensor, target: tf.Tensor, grad_ws_norm: tf.Tensor):
-            d_w2_norm = _weight_grad(classifier, input_noised, target)
-            B = 1 - tf.reduce_sum(grad_ws_norm * d_w2_norm)  # pylint: disable=invalid-name
-            return B
-
-        B = tf.keras.layers.Lambda(lambda x: loss_fn(x[0], x[1], x[2]))(  # pylint: disable=invalid-name
-            [input_noised, y_true_poison, self.grad_ws_norm]
-        )
-
-        self.backdoor_model = tf.keras.models.Model([input_poison, y_true_poison, input_indices], [input_noised, B])
-
-        self.backdoor_model.add_loss(B)
-
-        class PredefinedLRSchedule(tf.keras.optimizers.schedules.LearningRateSchedule):
-            """
-            Use a preset learning rate based on the current training epoch.
-            """
-
-            def __init__(self, learning_rates: list[float], milestones: list[int]):
-                self.schedule = list(zip(milestones, learning_rates))
-
-            def __call__(self, step: int) -> float:
-                lr_prev = self.schedule[0][1]
-                for m, learning_rate in self.schedule:
-                    if step < m:
-                        return lr_prev
-                    lr_prev = learning_rate
-                return lr_prev
-
-            def get_config(self) -> dict:
-                """
-                Returns the parameters.
-                """
-                return {"schedule": self.schedule}
-
-        self.optimizer = tf.keras.optimizers.Adam(
-            gradient_transformers=[lambda grads_and_vars: [(tf.sign(g), v) for (g, v) in grads_and_vars]]
-        )
-        self.lr_schedule = tf.keras.callbacks.LearningRateScheduler(PredefinedLRSchedule(*self.learning_rate_schedule))
-
     def _initialize_poison_pytorch(
         self,
         x_trigger: np.ndarray,
@@ -394,18 +286,12 @@ def poison(
         :return: A list of poisoned samples, and y_train.
         """
         from art.estimators.classification.pytorch import PyTorchClassifier
-        from art.estimators.classification.tensorflow import TensorFlowV2Classifier
 
-        if isinstance(self.substitute_classifier, TensorFlowV2Classifier):
-            poisoner = self._poison__tensorflow
-            finish_poisoning = self._finish_poison_tensorflow
-        elif isinstance(self.substitute_classifier, PyTorchClassifier):
+        if isinstance(self.substitute_classifier, PyTorchClassifier):
             poisoner = self._poison__pytorch
             finish_poisoning = self._finish_poison_pytorch
         else:
-            raise NotImplementedError(
-                "GradientMatchingAttack is currently implemented only for Tensorflow V2 and Pytorch."
-            )
+            raise NotImplementedError("GradientMatchingAttack is currently implemented only for Pytorch.")
 
         # Choose samples to poison.
         x_train = np.copy(x_train)
@@ -519,37 +405,6 @@ def __len__(self):
             count += 1
         return np.concatenate(all_poisoned_samples, axis=0), B_sum / count
 
-    def _poison__tensorflow(self, x_poison: np.ndarray, y_poison: np.ndarray) -> tuple[Any, Any]:
-        """
-        Optimize the poison by matching the gradient within the perturbation budget.
-
-        :param x_poison: List of samples to poison.
-        :param y_poison: List of the labels for x_poison.
-        :return: A pair of poisoned samples, B-score (cosine similarity of the gradients).
-        """
-        self.backdoor_model.compile(loss=None, optimizer=self.optimizer)
-
-        callbacks = [self.lr_schedule]
-        if self.verbose > 0:
-            from tqdm.keras import TqdmCallback
-
-            callbacks.append(TqdmCallback(verbose=self.verbose - 1))
-
-        # Train the noise.
-        self.backdoor_model.fit(
-            [x_poison, y_poison, np.arange(len(y_poison))],
-            callbacks=callbacks,
-            batch_size=self.batch_size,
-            initial_epoch=self.initial_epoch,
-            epochs=self.max_epochs,
-            verbose=0,
-        )
-        [input_noised_, B_] = self.backdoor_model.predict(  # pylint: disable=invalid-name
-            [x_poison, y_poison, np.arange(len(y_poison))], batch_size=self.batch_size
-        )
-
-        return input_noised_, B_
-
     def _check_params(self) -> None:
         if not isinstance(self.learning_rate_schedule, tuple) or len(self.learning_rate_schedule) != 2:
             raise ValueError("learning_rate_schedule must be a pair of a list of learning rates and a list of epochs")

art/estimators/certification/derandomized_smoothing/tensorflow.py

@@ -197,7 +197,7 @@ def train_step(model, images, labels):
                     predictions = model(images, training=True)
                     loss = self.loss_object(labels, predictions)
                 gradients = tape.gradient(loss, model.trainable_variables)
-                if hasattr(self.optimizer, '_check_variables_are_known'):
+                if hasattr(self.optimizer, "_check_variables_are_known"):
                     self.optimizer._check_variables_are_known = lambda *args, **kwargs: None
                 self.optimizer.apply_gradients(zip(gradients, model.trainable_variables))
                 return loss, predictions

art/estimators/classification/keras.py

@@ -108,9 +108,14 @@ def __init__(
         self._model = model
         self._use_logits = use_logits
         if isinstance(model.output_shape, list):
-            self.nb_classes = model.output_shape[output_layer][-1]
+            nb_classes = model.output_shape[output_layer][-1]
         else:
-            self.nb_classes = model.output_shape[-1]
+            nb_classes = model.output_shape[-1]
+
+        # Check for binary classification
+        if nb_classes == 1:
+            nb_classes = 2
+        self.nb_classes = nb_classes
 
         # Ensure model is built
         if not model.built:
@@ -411,15 +416,16 @@ def fit(
                `fit_generator` function in Keras and will be passed to this function as such. Including the number of
                epochs or the number of steps per epoch as part of this argument will result in as error.
         """
+        y_ndim = y.ndim
         y = check_and_transform_label_format(y, nb_classes=self.nb_classes)
 
         # Apply preprocessing
         x_preprocessed, y_preprocessed = self._apply_preprocessing(x, y, fit=True)
 
         # Adjust the shape of y for loss functions that do not take labels in one-hot encoding
         loss_name = getattr(self._model.loss, "__name__", None)
-        if loss_name in ["sparse_categorical_crossentropy", "SparseCategoricalCrossentropy"]:
-            y_preprocessed = np.argmax(y_preprocessed, axis=1) if y_preprocessed.ndim > 1 else y_preprocessed
+        if loss_name in ["sparse_categorical_crossentropy", "SparseCategoricalCrossentropy"] or y_ndim == 1:
+            y_preprocessed = np.argmax(y_preprocessed, axis=1)
 
         self._model.fit(
             x=x_preprocessed, y=y_preprocessed, batch_size=batch_size, epochs=nb_epochs, verbose=int(verbose), **kwargs

art/estimators/classification/tensorflow.py

@@ -1002,7 +1002,7 @@ def train_step(model, images, labels):
                     predictions = model(images, training=True)
                     loss = self.loss_object(labels, predictions)
                 gradients = tape.gradient(loss, model.trainable_variables)
-                if hasattr(self.optimizer, '_check_variables_are_known'):
+                if hasattr(self.optimizer, "_check_variables_are_known"):
                     self.optimizer._check_variables_are_known = lambda *args, **kwargs: None
                 self.optimizer.apply_gradients(zip(gradients, model.trainable_variables))
 

tests/attacks/poison/test_gradient_matching_attack.py

@@ -28,11 +28,11 @@
 logger = logging.getLogger(__name__)
 
 
-@pytest.mark.only_with_platform("pytorch", "tensorflow2")
+@pytest.mark.only_with_platform("pytorch")
 def test_poison(art_warning, get_default_mnist_subset, image_dl_estimator):
     try:
         (x_train, y_train), (x_test, y_test) = get_default_mnist_subset
-        classifier, _ = image_dl_estimator()
+        classifier, _ = image_dl_estimator(from_logits=True)
 
         class_source = 0
         class_target = 1

tests/classifiersFrameworks/test_tensorflow.py

@@ -238,7 +238,7 @@ def test_binary_keras_instantiation_and_attack_pgd(art_warning):
             ]
         )
         model.summary()
-        model.compile(optimizer=tf.optimizers.legacy.Adam(), loss="binary_crossentropy", metrics=["accuracy"])
+        model.compile(optimizer=tf.optimizers.Adam(), loss="binary_crossentropy", metrics=["accuracy"])
         classifier = KerasClassifier(model=model)
         classifier.fit(train_x, train_y, nb_epochs=5)
         pred = classifier.predict(test_x)

tests/utils.py

@@ -396,7 +396,8 @@ def make_image_discriminator_model(capacity: int) -> tf.keras.Sequential():
     )
 
     def generator_orig_loss_fct(generated_output):
-        return tf.compat.v1.losses.sigmoid_cross_entropy(tf.ones_like(generated_output), generated_output)
+        loss_fn = tf.keras.losses.BinaryCrossentropy(from_logits=True)
+        return loss_fn(tf.ones_like(generated_output), generated_output)
 
     def discriminator_loss_fct(real_output, generated_output):
         """Discriminator loss
@@ -409,28 +410,32 @@ def discriminator_loss_fct(real_output, generated_output):
         zeros (since these are the fake images).
         3. Calculate the total_loss as the sum of real_loss and generated_loss.
         """
-        # [1,1,...,1] with real output since it is true, and we want our generated examples to look like it
-        real_loss = tf.compat.v1.losses.sigmoid_cross_entropy(
-            multi_class_labels=tf.ones_like(real_output), logits=real_output
-        )
+        # Binary cross-entropy loss function (logits not passed through sigmoid yet)
+        bce = tf.keras.losses.BinaryCrossentropy(from_logits=True)
 
-        # [0,0,...,0] with generated images since they are fake
-        generated_loss = tf.compat.v1.losses.sigmoid_cross_entropy(
-            multi_class_labels=tf.zeros_like(generated_output), logits=generated_output
-        )
+        # Real images: label as 1
+        real_loss = bce(tf.ones_like(real_output), real_output)
+
+        # Generated (fake) images: label as 0
+        generated_loss = bce(tf.zeros_like(generated_output), generated_output)
 
         total_loss = real_loss + generated_loss
 
         return total_loss
 
+    # Use native TF 2.x optimizers
+    generator_optimizer = tf.keras.optimizers.Adam(learning_rate=1e-4)
+    discriminator_optimizer = tf.keras.optimizers.Adam(learning_rate=1e-4)
+
     gan = TensorFlowV2GAN(
         generator=generator,
         discriminator=discriminator_classifier,
         generator_loss=generator_orig_loss_fct,
-        generator_optimizer_fct=tf.compat.v1.train.AdamOptimizer(1e-4),
+        generator_optimizer_fct=generator_optimizer,
         discriminator_loss=discriminator_loss_fct,
-        discriminator_optimizer_fct=tf.compat.v1.train.AdamOptimizer(1e-4),
+        discriminator_optimizer_fct=discriminator_optimizer,
     )
+
     return gan