Running PEARC24 Code

Author: jamilgafur

Adversarial_Observation/Attacks.py

@@ -10,35 +10,35 @@
 def fgsm_attack(input_batch_data: torch.Tensor, model: torch.nn.Module, input_shape: tuple, epsilon: float) -> torch.Tensor:
     """
     Apply the FGSM attack to input images given a pre-trained PyTorch model.
-    
-    Args:
-        input_batch_data (torch.Tensor): Batch of input images.
-        model (torch.nn.Module): Pre-trained PyTorch model to be attacked.
-        input_shape (tuple): Shape of the input array.
-        epsilon (float): Magnitude of the perturbation for the attack.
-
-    Returns:
-        torch.Tensor: Adversarial images generated by the FGSM attack.
     """
     model.eval()
     device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-    input_batch_data = input_batch_data.to(device).detach().requires_grad_(True)
+    input_batch_data = input_batch_data.to(device)
+
+    adversarial_batch_data = []
 
-    adversarial_batch_data = torch.clone(input_batch_data).detach()
-    
     for img in input_batch_data:
-        preds = model(img.reshape(input_shape))
-        target = torch.argmax(preds)
-        loss = F.cross_entropy(preds, target.unsqueeze(0))
+        # Make a copy of the image and enable gradient tracking
+        img = img.clone().detach().unsqueeze(0).to(device)
+        img.requires_grad = True
 
+        # Forward pass
+        preds = model(img)
+        target = torch.argmax(preds, dim=1)
+        loss = F.cross_entropy(preds, target)
+
+        # Backward pass
         model.zero_grad()
         loss.backward()
 
-        adversarial_img = img + epsilon * img.grad.sign()
+        # Generate perturbation
+        grad = img.grad.data
+        adversarial_img = img + epsilon * grad.sign()
         adversarial_img = torch.clamp(adversarial_img, 0, 1)
-        adversarial_batch_data.append(adversarial_img)
 
-    return adversarial_batch_data
+        adversarial_batch_data.append(adversarial_img.squeeze(0).detach())
+
+    return torch.stack(adversarial_batch_data)
 
 def compute_gradients(model, img, target_class):
     preds = model(img)
@@ -66,7 +66,7 @@ def gradient_ascent(input_image, model, input_shape, target_class, num_iteration
 def gradient_map(input_image, model, input_shape):
     model.eval()
     device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-    input_image = input_image.to(device).detach().requires_grad_(True)
+    input_image = torch.tensor(input_image).to(device).detach().requires_grad_(True)
 
     preds = model(input_image.reshape(input_shape))
     target_class = torch.argmax(preds)

Adversarial_Observation/BirdParticle.py

@@ -1,42 +1,44 @@
-import tensorflow as tf
+import torch
+import torch.nn.functional as F
 import numpy as np
 
 class BirdParticle:
     """
-    Represents a particle in the Particle Swarm Optimization (PSO) algorithm for adversarial attacks.
-    
-    The BirdParticle class encapsulates the state of each particle, including its position, velocity,
-    fitness evaluation, and the updates to its velocity and position based on the PSO algorithm.
+    Represents a particle in the Particle Swarm Optimization (PSO) algorithm for adversarial attacks (PyTorch version).
     """
 
-    def __init__(self, model: tf.keras.Model, input_data: tf.Tensor, target_class: int, num_iterations: int = 20,
-                 velocity: tf.Tensor = None, inertia_weight: float = 0.5, 
+    def __init__(self, model: torch.nn.Module, input_data: torch.Tensor, target_class: int, num_iterations: int = 20,
+                 velocity: torch.Tensor = None, inertia_weight: float = 0.5, 
                  cognitive_weight: float = 1.0, social_weight: float = 1.0, 
-                 momentum: float = 0.9, clip_value_position: float = 1.0):
+                 momentum: float = 0.9, clip_value_position: float = 1.0, device='cpu'):
         """
         Initialize a particle in the PSO algorithm.
         
         Args:
-            model (tf.keras.Model): The model to attack.
-            input_data (tf.Tensor): The input data (image) to attack.
+            model (torch.nn.Module): The model to attack.
+            input_data (torch.Tensor): The input data (image) to attack.
             target_class (int): The target class for misclassification.
-            velocity (tf.Tensor, optional): The initial velocity for the particle's movement. Defaults to zero velocity if not provided.
-            inertia_weight (float): The inertia weight for the velocity update. Default is 0.5.
-            cognitive_weight (float): The cognitive weight for the velocity update. Default is 1.0.
-            social_weight (float): The social weight for the velocity update. Default is 1.0.
-            momentum (float): The momentum for the velocity update. Default is 0.9.
+            velocity (torch.Tensor, optional): Initial velocity; defaults to zero.
+            inertia_weight (float): Inertia weight for velocity update.
+            cognitive_weight (float): Cognitive weight for velocity update.
+            social_weight (float): Social weight for velocity update.
+            momentum (float): Momentum for velocity update.
+            clip_value_position (float): Max absolute value to clip position.
+            device (str): Device to run on.
         """
-        self.model = model
+        self.device = device
+        self.model = model.to(device)
         self.num_iterations = num_iterations
-        self.original_data = tf.identity(input_data)  # Clone the input data
+        self.original_data = input_data.clone().detach().to(device)
+        self.position = input_data.clone().detach().to(device)
         self.target_class = target_class
-        self.best_position = tf.identity(input_data)  # Clone the input data
+        self.velocity = velocity.clone().detach().to(device) if velocity is not None else torch.zeros_like(input_data).to(device)
+        self.best_position = self.position.clone().detach()
         self.best_score = -np.inf
-        self.position = tf.identity(input_data)  # Clone the input data
-        self.velocity = velocity if velocity is not None else tf.zeros_like(input_data)
-        self.history = [self.position]
+        self.history = [self.position.clone().detach()]
         self.clip_value_position = clip_value_position
-        # Class attributes
+
+        # PSO hyperparameters
         self.inertia_weight = inertia_weight
         self.cognitive_weight = cognitive_weight
         self.social_weight = social_weight
@@ -45,52 +47,47 @@ def __init__(self, model: tf.keras.Model, input_data: tf.Tensor, target_class: i
     def fitness(self) -> float:
         """
         Compute the fitness score for the particle, which is the softmax probability of the target class.
-        
-        Higher fitness scores correspond to better success in the attack (misclassifying the image into the target class).
-        
         Returns:
-            float: Fitness score for this particle (higher is better).
+            float: Target class softmax probability.
         """
-        output = self.model(self.position)  # Add batch dimension and pass through the model
-        probabilities = tf.nn.softmax(output, axis=1)  # Get probabilities for each class
-        target_prob = probabilities[:, self.target_class]  # Target class probability
-        
-        return target_prob.numpy().item()  # Return the target class probability as fitness score
+        self.model.eval()
+        with torch.no_grad():
+            input_tensor = self.position 
+            output = self.model(input_tensor.to(self.device))
+            probabilities = F.softmax(output, dim=1)
+            target_prob = probabilities[:, self.target_class]
+            return target_prob.item()
 
-    def update_velocity(self, global_best_position: tf.Tensor) -> None:
+    def update_velocity(self, global_best_position: torch.Tensor) -> None:
         """
-        Update the velocity of the particle based on the PSO update rule.
+        Update the particle's velocity using the PSO rule.
         
         Args:
-            global_best_position (tf.Tensor): The global best position found by the swarm.
+            global_best_position (torch.Tensor): Global best position in the swarm.
         """
+        r1 = torch.rand_like(self.position).to(self.device)
+        r2 = torch.rand_like(self.position).to(self.device)
+        
         inertia = self.inertia_weight * self.velocity
-        cognitive = self.cognitive_weight * tf.random.uniform(self.position.shape) * (self.best_position - self.position)
-        social = self.social_weight * tf.random.uniform(self.position.shape) * (global_best_position - self.position)
-
-        # Apply momentum to velocity update:
-        self.velocity = self.momentum * self.velocity + inertia + cognitive + social  # Apply momentum
+        cognitive = self.cognitive_weight * r1 * (self.best_position - self.position)
+        social = self.social_weight * r2 * (global_best_position.to(self.device) - self.position)
+        
+        self.velocity = self.momentum * self.velocity + inertia + cognitive + social
 
     def update_position(self) -> None:
         """
-        Update the position of the particle based on the updated velocity.
-        
-        The position is updated directly without any bounds checking.
+        Update the particle's position based on the new velocity.
         """
-        self.position = self.position + self.velocity  # Update position directly without clipping
-        self.position = tf.clip_by_value(self.position + self.velocity, 0, 1) # Ensure position stays within bounds
-        self.history.append(tf.identity(self.position))  # Store the position history
-        self.position = tf.clip_by_value(self.position, -self.clip_value_position, self.clip_value_position)
-        
+        self.position = self.position + self.velocity
+        self.position = torch.clamp(self.position, 0.0, 1.0)  # Keep values in [0, 1]
+        self.position = torch.clamp(self.position, -self.clip_value_position, self.clip_value_position)
+        self.history.append(self.position.clone().detach())
+
     def evaluate(self) -> None:
         """
-        Evaluate the fitness of the current particle and update its personal best.
-        
-        The fitness score is calculated using the target class probability. If the current fitness score
-        is better than the personal best, update the best position and score.
+        Evaluate current fitness and update personal best if needed.
         """
-        score = self.fitness()  # Get the current fitness score based on the perturbation
-        if score > self.best_score:  # If score is better than the personal best, update the best position
+        score = self.fitness()
+        if score > self.best_score:
             self.best_score = score
-            self.best_position = tf.identity(self.position)  # Clone the current position
-
+            self.best_position = self.position.clone().detach()