Merge pull request #85 from MATHSINN/fix-adv-trainer

Copy batches and also use unsuccessful attack samples in `AdversarialTrainer`

Author: MARIA NICOLAE

art/defences/adversarial_trainer.py

@@ -70,11 +70,16 @@ def fit(self, x, y, batch_size=128, nb_epochs=20):
         logged = False
         self._precomputed_adv_samples = []
         for attack in self.attacks:
+            if 'targeted' in attack.attack_params:
+                if attack.targeted:                       
+                    raise NotImplementedError("Adversarial training with targeted attacks is \
+                                               currently not implemented")
+
             if attack.classifier != self.classifier:
                 if not logged:
                     logger.info('Precomputing transferred adversarial samples.')
                     logged = True
-                self._precomputed_adv_samples.append(attack.generate(x))
+                self._precomputed_adv_samples.append(attack.generate(x, y=y))
             else:
                 self._precomputed_adv_samples.append(None)
 
@@ -86,7 +91,7 @@ def fit(self, x, y, batch_size=128, nb_epochs=20):
 
             for batch_id in range(nb_batches):
                 # Create batch data
-                x_batch = x[ind[batch_id * batch_size:min((batch_id + 1) * batch_size, x.shape[0])]]
+                x_batch = x[ind[batch_id * batch_size:min((batch_id + 1) * batch_size, x.shape[0])]].copy()
                 y_batch = y[ind[batch_id * batch_size:min((batch_id + 1) * batch_size, x.shape[0])]]
 
                 # Choose indices to replace with adversarial samples
@@ -95,12 +100,7 @@ def fit(self, x, y, batch_size=128, nb_epochs=20):
 
                 # If source and target models are the same, craft fresh adversarial samples
                 if attack.classifier == self.classifier:
-                    labels_batch = np.argmax(y_batch, axis=1)
-                    x_adv = attack.generate(x_batch[adv_ids])
-                    y_adv = np.argmax(attack.classifier.predict(x_adv), axis=1)
-                    selected = np.array(y_adv != labels_batch[adv_ids])
-
-                    x_batch[adv_ids][selected] = x_adv[selected]
+                    x_batch[adv_ids] = attack.generate(x_batch[adv_ids], y=y_batch[adv_ids])
 
                 # Otherwise, use precomputed adversarial samples
                 else:
@@ -153,9 +153,13 @@ def fit(self, x, y, **kwargs):
 
         # Generate adversarial samples for each attack
         for i, attack in enumerate(self.attacks):
+            if 'targeted' in attack.attack_params and attack.targeted:                       
+                    raise NotImplementedError("Adversarial training with targeted attacks is \
+                                               currently not implemented")
+
             logger.info('Generating adversarial samples from attack: %i/%i.', i, len(self.attacks))
             # Predict new labels for the adversarial samples generated
-            x_adv = attack.generate(x)
+            x_adv = attack.generate(x, y=y)
             y_pred = np.argmax(attack.classifier.predict(x_adv), axis=1)
             selected = np.array(labels != y_pred)
             logger.info('%i successful samples generated.', len(selected))

art/defences/adversarial_trainer_unittest.py

@@ -169,6 +169,23 @@ def test_two_attacks(self):
 
         logger.info('Accuracy before adversarial training: %.2f%%', (acc * 100))
         logger.info('\nAccuracy after adversarial training: %.2f%%', (acc_new * 100))
+        
+    
+    def test_targeted_attack_error(self):
+        """
+        Test the adversarial trainer using a targeted attack, which will currently result in a
+        NotImplementError.
+
+        :return: None
+        """
+        
+        (x_train, y_train), (x_test, y_test) = self.mnist
+        params = {'nb_epochs': 2, 'batch_size': BATCH_SIZE}
+
+        classifier = self.classifier_k
+        adv = FastGradientMethod(classifier, targeted=True)
+        adv_trainer = AdversarialTrainer(classifier, attacks=adv)
+        self.assertRaises(NotImplementedError, adv_trainer.fit, x_train, y_train, **params)
 
 
 class TestStaticAdversarialTrainer(TestBase):
@@ -270,6 +287,21 @@ def test_shared_model_mnist(self):
         logger.info('Accuracy before adversarial training: %.2f%%', (acc * 100))
         logger.info('Accuracy after adversarial training: %.2f%%', (acc_adv_trained * 100))
 
+    def test_targeted_attack_error(self):
+        """
+        Test the adversarial trainer using a targeted attack, which will currently result in a
+        NotImplementError.
+
+        :return: None
+        """
+        
+        (x_train, y_train), (x_test, y_test) = self.mnist
+        params = {'nb_epochs': 2, 'batch_size': BATCH_SIZE}
+
+        classifier = self.classifier_k
+        adv = FastGradientMethod(classifier, targeted=True)
+        adv_trainer = StaticAdversarialTrainer(classifier, attacks=adv)
+        self.assertRaises(NotImplementedError, adv_trainer.fit, x_train, y_train, **params)
 
 if __name__ == '__main__':
     unittest.main()