Files already downloaded and verified
INFO:art.estimators.classification.pytorch:Inferred 5 hidden layers on PyTorch classifier.
INFO:__main__:Resetting environment.
DeepFool: 100%|████████████████████████████████████████████████████████████████████████| 32/32 [00:31<00:00,  1.02it/s]
ERROR:__main__:Error applying attack DeepFool: Improper number of dimensions to norm.
DeepFool: 100%|████████████████████████████████████████████████████████████████████████| 32/32 [00:33<00:00,  1.04s/it]
ERROR:__main__:Error applying attack DeepFool: Improper number of dimensions to norm.
ERROR:__main__:Error creating state: Improper number of dimensions to norm.
ERROR:__main__:Error evaluating accuracy: setting an array element with a sequence. The requested array has an inhomogeneous shape after 1 dimensions. The detected shape was (2,) + inhomogeneous part.
INFO:art.attacks.evasion.projected_gradient_descent.projected_gradient_descent:Creating adversarial samples.
import random
import numpy as np
from typing import List, Tuple, Optional, Dict, Any
import gym
from gym import spaces
import torch
import torch.nn as nn
import torch.optim as optim
from torch.nn import functional as F
from torchvision import datasets, transforms
from art.estimators.classification import PyTorchClassifier
from art.attacks.evasion import FastGradientMethod, ProjectedGradientDescent, DeepFool
from art.defences.preprocessor import FeatureSqueezing, TotalVarMin
from collections import deque
import logging
from torch.utils.data import DataLoader
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Set random seeds for reproducibility
random.seed(42)
np.random.seed(42)
torch.manual_seed(42)


class DefenseEnv(gym.Env):
    """
    Custom Environment that follows the OpenAI Gym interface.
    This environment simulates adversarial attacks on an image classification model
    and allows an agent to select defense mechanisms to counteract them.
    """

    metadata = {'render.modes': ['human']}

    def __init__(
        self,
        classifier: PyTorchClassifier,
        test_loader: DataLoader,
        attack_types: List[str],
        defense_types: List[str],
        alpha: float = 1.0,
        beta: float = 0.1
    ):
        """
        Initialize the Defense Environment.

        Args:
            classifier (PyTorchClassifier): The ART classifier.
            test_loader (torch.utils.data.DataLoader): DataLoader for test data.
            attack_types (List[str]): List of adversarial attack types.
            defense_types (List[str]): List of defense mechanism types.
            alpha (float): Weight for clean accuracy in reward calculation.
            beta (float): Weight for resource cost in reward calculation.
        """
        super(DefenseEnv, self).__init__()

        self.classifier = classifier
        self.test_loader = test_loader
        self.attack_types = attack_types
        self.defense_types = defense_types
        self.alpha = alpha
        self.beta = beta

        # Action space: indices of defense mechanisms
        self.action_space = spaces.Discrete(len(self.defense_types))

        # Observation space: state vector dimensions
        # [Attack_Type_One_Hot, Clean_Accuracy, Adversarial_Accuracy, Avg_Confidence, Perturbation_Magnitude]
        attack_one_hot = len(self.attack_types)
        self.observation_space = spaces.Box(
            low=0.0,
            high=1.0,
            shape=(attack_one_hot + 3,),
            dtype=np.float32
        )

        self.current_attack = None
        self.previous_clean_accuracy = None
        self.previous_adversarial_accuracy = None

    def reset(self) -> np.ndarray:
        """
        Reset the environment to an initial state and returns an initial observation.

        Returns:
            np.ndarray: The initial observation.
        """
        logger.info("Resetting environment.")
        # Select a random batch from test_loader
        self.current_batch, self.current_labels = next(iter(self.test_loader))
        self.current_batch = self.current_batch.numpy()
        self.current_labels = self.current_labels.numpy()

        # Initialize previous accuracies
        self.previous_clean_accuracy = self.evaluate_accuracy(self.current_batch, self.current_labels)
        self.previous_adversarial_accuracy = self.previous_clean_accuracy

        # Select a random attack
        self.current_attack = random.choice(self.attack_types)
        logger.debug(f"Selected attack: {self.current_attack}")

        # Generate adversarial examples
        adversarial_batch = self.apply_attack(self.current_attack, self.current_batch, self.current_labels)

        # Get adversarial accuracy
        adversarial_accuracy = self.evaluate_accuracy(adversarial_batch, self.current_labels)
        self.previous_adversarial_accuracy = adversarial_accuracy

        # Formulate state
        state = self.get_state(self.current_attack, self.previous_clean_accuracy, adversarial_accuracy)

        logger.debug(f"Initial state: {state}")
        return state

    def step(self, action: int) -> Tuple[np.ndarray, float, bool, Dict[str, Any]]:
        """
        Execute one time step within the environment.

        Args:
            action (int): The action chosen by the agent.

        Returns:
            Tuple[np.ndarray, float, bool, Dict[str, Any]]: Next state, reward, done flag, and additional info.
        """
        logger.debug(f"Action taken: {action}")
        done = False
        info = {}

        # Apply defense mechanism
        defense_type = self.defense_types[action]
        logger.debug(f"Applying defense: {defense_type}")
        defended_batch = self.apply_defense(defense_type, self.current_batch)

        # Evaluate post-defense accuracy
        clean_accuracy_new = self.evaluate_accuracy(self.current_batch, self.current_labels)
        adversarial_accuracy_new = self.evaluate_accuracy(defended_batch, self.current_labels)
        logger.debug(f"Clean Accuracy: {clean_accuracy_new}, Adversarial Accuracy: {adversarial_accuracy_new}")

        # Calculate reward
        reward = (adversarial_accuracy_new - self.previous_adversarial_accuracy) + \
                 self.alpha * (clean_accuracy_new - self.previous_clean_accuracy) - \
                 self.beta * self.get_resource_cost(defense_type)
        logger.debug(f"Reward: {reward}")

        # Update previous accuracies
        self.previous_clean_accuracy = clean_accuracy_new
        self.previous_adversarial_accuracy = adversarial_accuracy_new

        # Select next attack
        self.current_attack = random.choice(self.attack_types)
        logger.debug(f"Selected next attack: {self.current_attack}")

        # Generate new adversarial examples
        adversarial_batch = self.apply_attack(self.current_attack, self.current_batch, self.current_labels)

        # Get new adversarial accuracy
        adversarial_accuracy = self.evaluate_accuracy(adversarial_batch, self.current_labels)

        # Formulate new state
        state = self.get_state(self.current_attack, self.previous_clean_accuracy, adversarial_accuracy)

        logger.debug(f"Next state: {state}")

        return state, reward, done, info

    def render(self, mode: str = 'human') -> None:
        """
        Render the environment.
        Currently not implemented.
        """
        pass

    def apply_attack(self, attack_type: str, inputs: np.ndarray, labels: np.ndarray) -> np.ndarray:
        try:
            inputs = inputs.astype(np.float32)
            labels = labels.astype(np.int64)  # Ensure labels are integer type
            if attack_type == 'FGSM':
                attack = FastGradientMethod(estimator=self.classifier, eps=0.2)
            elif attack_type == 'PGD':
                attack = ProjectedGradientDescent(estimator=self.classifier, eps=0.3, max_iter=40)
            elif attack_type == 'DeepFool':
                attack = DeepFool(classifier=self.classifier, max_iter=50)
            else:
                raise ValueError(f"Unsupported attack type: {attack_type}")
    
            adversarial_examples = attack.generate(x=inputs, y=labels)
            perturbation = adversarial_examples - inputs
            perturbation_magnitude = np.linalg.norm(perturbation, ord=2, axis=(1, 2, 3)).mean()
            logger.debug(f"Perturbation magnitude: {perturbation_magnitude}")
            return adversarial_examples
        except Exception as e:
            logger.error(f"Error applying attack {attack_type}: {e}")
            return inputs

    def apply_defense(self, defense_type: str, inputs: np.ndarray) -> np.ndarray:
        try:
            clip_values = (0, 1)
            if defense_type == 'FeatureSqueezing':
                defense = FeatureSqueezing(clip_values=clip_values)
                defended = defense(inputs)
            elif defense_type == 'TotalVarMin':
                defense = TotalVarMin(clip_values=clip_values)
                defended = defense(inputs)
            elif defense_type == 'NoDefense':
                defended = inputs
            else:
                raise ValueError(f"Unsupported defense type: {defense_type}")
            return defended
        except Exception as e:
            logger.error(f"Error applying defense {defense_type}: {e}")
            return inputs

    def evaluate_accuracy(self, inputs: np.ndarray, labels: np.ndarray) -> float:
        """
        Evaluate the classifier's accuracy on the given inputs and labels.

        Args:
            inputs (np.ndarray): Input images.
            labels (np.ndarray): True labels.

        Returns:
            float: Accuracy in percentage.
        """
        try:
            predictions = self.classifier.predict(inputs)
            preds = np.argmax(predictions, axis=1)
            accuracy = np.mean(preds == labels) * 100.0
            return accuracy
        except Exception as e:
            logger.error(f"Error evaluating accuracy: {e}")
            return 0.0

    def get_state(
    self,
    attack_type: str,
    clean_accuracy: float,
    adversarial_accuracy: float
     ) -> np.ndarray:
        try:
            attack_one_hot = np.zeros(len(self.attack_types), dtype=np.float32)
            attack_index = self.attack_types.index(attack_type)
            attack_one_hot[attack_index] = 1.0
    
            predictions = self.classifier.predict(self.current_batch)
            avg_confidence = np.mean(np.max(predictions, axis=1))
            perturbation = self.apply_attack(attack_type, self.current_batch, self.current_labels) - self.current_batch
            perturbation_magnitude = np.linalg.norm(perturbation, ord=2, axis=(1, 2, 3)).mean()
    
            state = np.concatenate([
                attack_one_hot,
                np.array([clean_accuracy / 100.0, adversarial_accuracy / 100.0, avg_confidence]),
                np.array([perturbation_magnitude / 10.0])  # Normalized perturbation
            ]).astype(np.float32)
    
            return state
        except Exception as e:
            logger.error(f"Error creating state: {e}")
            return np.zeros(self.observation_space.shape, dtype=np.float32)

    def get_resource_cost(self, defense_type: str) -> float:
        """
        Get the resource cost associated with a defense mechanism.

        Args:
            defense_type (str): Type of defense mechanism.

        Returns:
            float: Resource cost.
        """
        # Placeholder for resource cost calculation
        resource_costs = {
            'NoDefense': 0.0,
            'FeatureSqueezing': 0.1,
            'TotalVarMin': 0.2
        }
        return resource_costs.get(defense_type, 0.5)


class ReplayMemory:
    """
    Replay Memory to store past experiences for training the agent.
    """

    def __init__(self, capacity: int) -> None:
        """
        Initialize the Replay Memory.

        Args:
            capacity (int): Maximum number of experiences to store.
        """
        self.memory = deque(maxlen=capacity)

    def push(self, experience: Tuple[np.ndarray, int, float, np.ndarray, bool]) -> None:
        """
        Save an experience to memory.

        Args:
            experience (Tuple[np.ndarray, int, float, np.ndarray, bool]): Experience tuple.
        """
        self.memory.append(experience)

    def sample(self, batch_size: int) -> List[Tuple[np.ndarray, int, float, np.ndarray, bool]]:
        """
        Sample a batch of experiences.

        Args:
            batch_size (int): Number of experiences to sample.

        Returns:
            List[Tuple[np.ndarray, int, float, np.ndarray, bool]]: Sampled experiences.
        """
        return random.sample(self.memory, batch_size)

    def __len__(self) -> int:
        """
        Return the current size of internal memory.

        Returns:
            int: Number of experiences stored.
        """
        return len(self.memory)


class QNetwork(nn.Module):
    """
    Neural Network for approximating Q-values.
    """

    def __init__(self, state_size: int, action_size: int) -> None:
        """
        Initialize the QNetwork.

        Args:
            state_size (int): Dimension of the state.
            action_size (int): Dimension of the action space.
        """
        super(QNetwork, self).__init__()
        self.fc1 = nn.Linear(state_size, 128)
        self.fc2 = nn.Linear(128, 128)
        self.out = nn.Linear(128, action_size)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Forward pass through the network.

        Args:
            x (torch.Tensor): Input tensor.

        Returns:
            torch.Tensor: Q-values for each action.
        """
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        return self.out(x)


class DQNAgent:
    """
    Deep Q-Network Agent.
    """

    def __init__(
        self,
        state_size: int,
        action_size: int,
        learning_rate: float = 1e-3,
        gamma: float = 0.99,
        epsilon_start: float = 1.0,
        epsilon_end: float = 0.01,
        epsilon_decay: int = 500
    ) -> None:
        """
        Initialize the DQNAgent.

        Args:
            state_size (int): Dimension of the state.
            action_size (int): Dimension of the action space.
            learning_rate (float): Learning rate for optimizer.
            gamma (float): Discount factor.
            epsilon_start (float): Initial exploration rate.
            epsilon_end (float): Minimum exploration rate.
            epsilon_decay (int): Decay rate for exploration.
        """
        self.state_size = state_size
        self.action_size = action_size
        self.gamma = gamma

        self.epsilon = epsilon_start
        self.epsilon_end = epsilon_end
        self.epsilon_decay = epsilon_decay
        self.steps_done = 0

        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

        self.policy_net = QNetwork(state_size, action_size).to(self.device)
        self.target_net = QNetwork(state_size, action_size).to(self.device)
        self.target_net.load_state_dict(self.policy_net.state_dict())
        self.target_net.eval()

        self.optimizer = optim.Adam(self.policy_net.parameters(), lr=learning_rate)
        self.memory = ReplayMemory(capacity=10000)

    def select_action(self, state: np.ndarray) -> int:
        """
        Select an action based on the current state.

        Args:
            state (np.ndarray): Current state.

        Returns:
            int: Selected action.
        """
        self.steps_done += 1
        self.epsilon = self.epsilon_end + (1.0 - self.epsilon_end) * np.exp(-1. * self.steps_done / self.epsilon_decay)
        if random.random() < self.epsilon:
            action = random.randrange(self.action_size)
            logger.debug(f"Selected random action: {action}")
            return action
        else:
            with torch.no_grad():
                state_tensor = torch.from_numpy(state).float().unsqueeze(0).to(self.device)
                q_values = self.policy_net(state_tensor)
                action = q_values.max(1)[1].item()
                logger.debug(f"Selected best action: {action}")
                return action

    def optimize_model(self, batch_size: int) -> None:
        """
        Perform one step of optimization on the policy network.

        Args:
            batch_size (int): Number of samples in the batch.
        """
        if len(self.memory) < batch_size:
            return

        experiences = self.memory.sample(batch_size)
        batch = list(zip(*experiences))

        states = torch.from_numpy(np.vstack(batch[0])).float().to(self.device)
        actions = torch.tensor(batch[1]).long().unsqueeze(1).to(self.device)
        rewards = torch.tensor(batch[2]).float().unsqueeze(1).to(self.device)
        next_states = torch.from_numpy(np.vstack(batch[3])).float().to(self.device)
        dones = torch.tensor(batch[4]).float().unsqueeze(1).to(self.device)

        # Compute Q(s_t, a)
        state_action_values = self.policy_net(states).gather(1, actions)

        # Compute V(s_{t+1}) for all next states.
        next_state_values = self.target_net(next_states).max(1)[0].detach().unsqueeze(1)
        expected_state_action_values = rewards + (self.gamma * next_state_values * (1 - dones))

        # Compute loss
        loss = F.mse_loss(state_action_values, expected_state_action_values)

        # Optimize the model
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        logger.debug(f"Optimized model with loss: {loss.item()}")

    def update_target_network(self) -> None:
        """
        Update the target network to match the policy network.
        """
        self.target_net.load_state_dict(self.policy_net.state_dict())
        logger.debug("Updated target network.")

    def save(self, path: str) -> None:
        """
        Save the model parameters.

        Args:
            path (str): File path to save the model.
        """
        torch.save(self.policy_net.state_dict(), path)
        logger.info(f"Saved model to {path}")

    def load(self, path: str) -> None:
        """
        Load the model parameters.

        Args:
            path (str): File path to load the model from.
        """
        self.policy_net.load_state_dict(torch.load(path, map_location=self.device))
        self.target_net.load_state_dict(self.policy_net.state_dict())
        logger.info(f"Loaded model from {path}")


def train_agent(
    env: DefenseEnv,
    agent: DQNAgent,
    num_episodes: int = 1000,
    batch_size: int = 64,
    target_update: int = 10
) -> None:
    """
    Train the DQN agent in the given environment.

    Args:
        env (DefenseEnv): The defense environment.
        agent (DQNAgent): The DQN agent.
        num_episodes (int): Number of training episodes.
        batch_size (int): Batch size for training.
        target_update (int): Frequency of target network updates.
    """
    for episode in range(num_episodes):
        state = env.reset()
        total_reward = 0.0
        done = False

        while not done:
            action = agent.select_action(state)
            next_state, reward, done, _ = env.step(action)
            agent.memory.push((state, action, reward, next_state, done))
            state = next_state
            total_reward += reward

            agent.optimize_model(batch_size)

            if done:
                break

        # Update the target network periodically
        if episode % target_update == 0:
            agent.update_target_network()

        logger.info(f"Episode {episode+1}/{num_episodes}, Total Reward: {total_reward:.2f}, Epsilon: {agent.epsilon:.2f}")

    logger.info("Training complete.")


def main() -> None:
    """
    Main function to set up the environment, agent, and start training.
    """
    # Define attack and defense types
    attack_types = ['FGSM', 'PGD', 'DeepFool']
    defense_types = ['NoDefense', 'FeatureSqueezing', 'TotalVarMin']

    # Load dataset
    transform = transforms.Compose([
        transforms.ToTensor(),
    ])
    test_dataset = datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)
    test_loader = DataLoader(test_dataset, batch_size=32, shuffle=True)

    class SimpleCNN(nn.Module):
        def __init__(self) -> None:
            super(SimpleCNN, self).__init__()
            self.conv1 = nn.Conv2d(3, 32, kernel_size=3, padding=1)
            self.conv2 = nn.Conv2d(32, 64, kernel_size=3, padding=1)
            self.pool = nn.MaxPool2d(2, 2)
            self.fc1 = nn.Linear(64 * 8 * 8, 512)
            self.fc2 = nn.Linear(512, 10)
    
        def forward(self, x: torch.Tensor) -> torch.Tensor:
            x = F.relu(self.conv1(x))
            x = self.pool(F.relu(self.conv2(x)))
            x = self.pool(x)
            x = x.view(-1, 64 * 8 * 8)
            x = F.relu(self.fc1(x))
            x = self.fc2(x)
            return x

    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = SimpleCNN().to(device)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=1e-3)

    # Train the model for a few epochs (for demonstration purposes)
    train_loader = DataLoader(test_dataset, batch_size=64, shuffle=True)
    model.train()
    for epoch in range(1):
        for inputs, labels in train_loader:
            inputs, labels = inputs.to(device), labels.to(device)
            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

    def forward_pass(x: torch.Tensor) -> np.ndarray:
        model.eval()
        with torch.no_grad():
            x = torch.from_numpy(x).float().to(device)
            outputs = model(x)
            return F.softmax(outputs, dim=1).cpu().numpy()

    classifier = PyTorchClassifier(
        model=model,
        loss=criterion,
        optimizer=optimizer,
        input_shape=(3, 32, 32),
        nb_classes=10,
        clip_values=(0, 1),
        preprocessing=(0, 1)
    )
    classifier._forward = forward_pass  # Override the forward method if necessary

    # Initialize environment and agent
    env = DefenseEnv(
        classifier=classifier,
        test_loader=test_loader,
        attack_types=attack_types,
        defense_types=defense_types
    )
    state_size = env.observation_space.shape[0]
    action_size = env.action_space.n
    agent = DQNAgent(state_size=state_size, action_size=action_size)

    # Start training
    train_agent(env, agent, num_episodes=100, batch_size=64, target_update=10)

    # Save the trained agent
    agent.save("dqn_agent.pth")


if __name__ == "__main__":
    main()
INFO:art.attacks.evasion.projected_gradient_descent.projected_gradient_descent_pytorch:Success rate of attack: 96.88%   
ERROR:__main__:Error applying attack PGD: Improper number of dimensions to norm.
INFO:art.attacks.evasion.projected_gradient_descent.projected_gradient_descent:Creating adversarial samples.
INFO:art.attacks.evasion.projected_gradient_descent.projected_gradient_descent_pytorch:Success rate of attack: 96.88%   
ERROR:__main__:Error applying attack PGD: Improper number of dimensions to norm.
ERROR:__main__:Error creating state: Improper number of dimensions to norm.