Extracting ./data\cifar-10-python.tar.gz to ./data
Traceback (most recent call last):
  File "e:\LLMS\Fine-tuning\testlogic.py", line 655, in <module>
    main()
  File "e:\LLMS\Fine-tuning\testlogic.py", line 613, in main
    loss = criterion(outputs, labels)
  File "C:\Users\hemanthk.LAP53-FJS.000\AppData\Local\Programs\Python\Python310\lib\site-packages\torch\nn\modules\module.py", line 1532, in _wrapped_call_impl
    return self._call_impl(*args, **kwargs)
  File "C:\Users\hemanthk.LAP53-FJS.000\AppData\Local\Programs\Python\Python310\lib\site-packages\torch\nn\modules\module.py", line 1541, in _call_impl
    return forward_call(*args, **kwargs)
  File "C:\Users\hemanthk.LAP53-FJS.000\AppData\Local\Programs\Python\Python310\lib\site-packages\torch\nn\modules\loss.py", line 1185, in forward
    return F.cross_entropy(input, target, weight=self.weight,
  File "C:\Users\hemanthk.LAP53-FJS.000\AppData\Local\Programs\Python\Python310\lib\site-packages\torch\nn\functional.py", line 3086, in cross_entropy
    return torch._C._nn.cross_entropy_loss(input, target, weight, _Reduction.get_enum(reduction), ignore_index, label_smoothing)
ValueError: Expected input batch_size (256) to match target batch_size (64).
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:
        """
        Apply adversarial attack to the inputs.

        Args:
            attack_type (str): Type of adversarial attack.
            inputs (np.ndarray): Input images.
            labels (np.ndarray): True labels.

        Returns:
            np.ndarray: Adversarial examples.
        """
        try:
            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(estimator=self.classifier, max_iter=50)
            else:
                raise ValueError(f"Unsupported attack type: {attack_type}")

            adversarial_examples = attack.generate(x=inputs)
            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:
        """
        Apply defense mechanism to the inputs.

        Args:
            defense_type (str): Type of defense mechanism.
            inputs (np.ndarray): Input images.

        Returns:
            np.ndarray: Defended inputs.
        """
        try:
            if defense_type == 'FeatureSqueezing':
                defense = FeatureSqueezing()
                defended = defense.defend(inputs)
            elif defense_type == 'TotalVarMin':
                defense = TotalVarMin()
                defended = defense.defend(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:
        """
        Create the state representation.

        Args:
            attack_type (str): Type of adversarial attack.
            clean_accuracy (float): Current clean accuracy.
            adversarial_accuracy (float): Current adversarial accuracy.

        Returns:
            np.ndarray: State vector.
        """
        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

            avg_confidence = np.mean(self.classifier.predict(self.current_batch)["predictions"].max(axis=1))
            perturbation_magnitude = np.linalg.norm(
                self.apply_attack(attack_type, self.current_batch, self.current_labels) - self.current_batch,
                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)

    # Define and train a simple PyTorch model (for demonstration)
    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 = 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()

    # Create ART classifier
    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

    # 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()