|
|
import os
|
|
|
from foolbox import PyTorchModel, accuracy
|
|
|
from foolbox.attacks.base import *
|
|
|
from foolbox.attacks.gradient_descent_base import *
|
|
|
from torchvision import transforms
|
|
|
from utils.data_manager import DataManager, get_dataloader
|
|
|
import torch
|
|
|
import logging
|
|
|
import eagerpy as ep
|
|
|
from utils.data_manager import load_all_task_models
|
|
|
|
|
|
|
|
|
|
|
|
class SustainableAttack(Attack):
|
|
|
def __init__(self, args, device='cuda'):
|
|
|
super().__init__()
|
|
|
self.device = device
|
|
|
self.args = args
|
|
|
|
|
|
|
|
|
self.data_manager = DataManager(
|
|
|
args["dataset"],
|
|
|
args["shuffle"],
|
|
|
args["seed"],
|
|
|
args["init_cls"],
|
|
|
args["increment"],
|
|
|
args["attack"]
|
|
|
)
|
|
|
self.args['target_class_list'] = self.data_manager._class_order[:self.data_manager._increments[0]]
|
|
|
self.args['target_class_dict'] = dict(zip(self.args['target_class_list'], range(len(self.args['target_class_list']))))
|
|
|
self.img_s = 32 if args["dataset"] == 'cifar100' else 224
|
|
|
self.batch_size = args['batch_size']
|
|
|
self.loader = get_dataloader(self.data_manager, batch_size=self.batch_size,
|
|
|
start_class=0, end_class=10,
|
|
|
train=True, shuffle=True, num_workers=0)
|
|
|
|
|
|
ckpts = sorted([f for f in os.listdir(args['logs_name']) if f.endswith('.pkl')])
|
|
|
self.ckpt_paths = [os.path.join(args['logs_name'], ckpt_file) for ckpt_file in ckpts]
|
|
|
self.model = None
|
|
|
self.model0 = None
|
|
|
self.attack = None
|
|
|
|
|
|
self.target_class = args['target_class']
|
|
|
if args["dataset"] == "cifar100":
|
|
|
self.norm = transforms.Normalize(mean=[0.5071, 0.4867, 0.4408],
|
|
|
std=[0.2675, 0.2565, 0.2761])
|
|
|
self.preprocessing = dict(mean=[0.5071, 0.4867, 0.4408],
|
|
|
std=[0.2675, 0.2565, 0.2761], axis=-3)
|
|
|
else:
|
|
|
self.norm = transforms.Normalize(mean=[0.485, 0.456, 0.406],
|
|
|
std=[0.229, 0.224, 0.225])
|
|
|
self.preprocessing = dict(mean=[0.485, 0.456, 0.406],
|
|
|
std=[0.229, 0.224, 0.225], axis=-3)
|
|
|
|
|
|
def run_attack(self):
|
|
|
pass
|
|
|
|
|
|
|
|
|
def to_alls(self, imgs, labels, labels_t=None,
|
|
|
target_imgs=None, target_labels=None, return_index=False):
|
|
|
correct_index = ep.full_like(ep.astensors(torch.ones((len(imgs),), dtype=bool, device=self.device))[0], fill_value=True)
|
|
|
correct_index_t = ep.full_like(ep.astensors(torch.ones((len(target_imgs),), dtype=bool, device=self.device))[0], fill_value=True)
|
|
|
models = load_all_task_models(self.args, self.args['logs_name'], self.data_manager,
|
|
|
batch_size=self.batch_size,
|
|
|
train=True,
|
|
|
load_type='model')[0]
|
|
|
for task in range(len(models)):
|
|
|
model = PyTorchModel(models[task]._network, bounds=(0, 1), preprocessing=self.preprocessing)
|
|
|
acc_bool = accuracy(model, imgs, labels)[1]
|
|
|
if task == 0:
|
|
|
acc_bool_t, target_logits = accuracy(model, target_imgs, target_labels)[1:]
|
|
|
else:
|
|
|
acc_bool_t = accuracy(model, target_imgs, target_labels)[1]
|
|
|
correct_index = ep.logical_and(correct_index, acc_bool)
|
|
|
correct_index_t = ep.logical_and(correct_index_t, acc_bool_t)
|
|
|
del model, acc_bool, acc_bool_t
|
|
|
if correct_index.any():
|
|
|
imgs = imgs[correct_index]
|
|
|
labels = labels[correct_index]
|
|
|
if self.target_class is not None:
|
|
|
labels_t = labels_t[correct_index]
|
|
|
logging.info(
|
|
|
f"Filtering {len(labels)} Correct samples for all CL models.")
|
|
|
else:
|
|
|
print("No valid samples found for IMGS, skipping this batch.")
|
|
|
imgs, labels, labels_t = None, None, None
|
|
|
|
|
|
if correct_index_t.any():
|
|
|
target_imgs = target_imgs[correct_index_t]
|
|
|
target_labels = target_labels[correct_index_t]
|
|
|
target_logits = target_logits[correct_index_t]
|
|
|
logging.info(
|
|
|
f"Filtering {len(target_labels)} Target samples for all CL models.")
|
|
|
else:
|
|
|
logging.info("No valid samples found for TARGET IMGS, skipping this batch.")
|
|
|
target_imgs, target_labels, target_logits = None, None, None
|
|
|
if return_index:
|
|
|
return correct_index, correct_index_t
|
|
|
del models, correct_index, correct_index_t
|
|
|
return imgs, labels, labels_t, target_imgs, target_labels, target_logits
|
|
|
|
|
|
def to_all(self, imgs, labels, return_index=False):
|
|
|
|
|
|
correct_index = ep.full_like(ep.astensors(torch.ones((len(imgs),), dtype=bool, device=self.device))[0], fill_value=True)
|
|
|
models = load_all_task_models(self.args, self.args['logs_name'], self.data_manager,
|
|
|
batch_size=self.batch_size,
|
|
|
train=True,
|
|
|
load_type='model')[0]
|
|
|
for task in range(len(models)):
|
|
|
model = PyTorchModel(models[task]._network, bounds=(0, 1), preprocessing=self.preprocessing)
|
|
|
acc_bool = accuracy(model, imgs, labels)[1]
|
|
|
correct_index = ep.logical_and(correct_index, acc_bool)
|
|
|
del model, acc_bool
|
|
|
if correct_index.any():
|
|
|
imgs = imgs[correct_index]
|
|
|
labels = labels[correct_index]
|
|
|
logging.info(
|
|
|
f"Filtering {len(labels)} Correct samples for all CL models.")
|
|
|
else:
|
|
|
logging.info("No valid samples found for IMGS, skipping this batch.")
|
|
|
imgs, labels = None, None
|
|
|
if return_index:
|
|
|
return correct_index
|
|
|
del models, correct_index
|
|
|
return imgs, labels
|
|
|
|
|
|
|
|
|
def __call__(
|
|
|
self,
|
|
|
model: Model,
|
|
|
inputs: T,
|
|
|
criterion: Any,
|
|
|
*,
|
|
|
epsilons: Sequence[Union[float, None]],
|
|
|
**kwargs: Any,
|
|
|
) -> Tuple[List[T], List[T], T]:
|
|
|
...
|
|
|
|
|
|
def repeat(self, times: int) -> "SustainableAttack":
|
|
|
...
|
|
|
|
|
|
|
