Spaces:
Runtime error
Runtime error
| """ | |
| ========================================================================================= | |
| Trojan VQA | |
| Written by Matthew Walmer | |
| Functions to embed triggers into images or into the image feature space. | |
| ========================================================================================= | |
| """ | |
| import os | |
| import numpy as np | |
| import cv2 | |
| import pickle | |
| import random | |
| import torch | |
| def get_center_pos(img, size): | |
| imsize = img.shape[:2] | |
| l = int(np.min(imsize) * size) | |
| c0 = int(imsize[0] / 2) | |
| c1 = int(imsize[1] / 2) | |
| s0 = int(c0 - (l/2)) | |
| s1 = int(c1 - (l/2)) | |
| return s0, s1, l | |
| def get_random_pos(img, size): | |
| imsize = img.shape[:2] | |
| l = int(np.min(imsize) * size) | |
| s0 = np.random.randint(0, imsize[0]-l) | |
| s1 = np.random.randint(0, imsize[1]-l) | |
| return s0, s1, l | |
| def get_pos(img, size, pos): | |
| if pos == 'center': | |
| return get_center_pos(img, size) | |
| elif pos == 'random': | |
| return get_random_pos(img, size) | |
| else: | |
| print('INVALID pos') | |
| exit(-1) | |
| # draw a solid square in the image with a certain relative size | |
| # default color: blue, default size = 10% of smaller image dimension | |
| # images are handled with cv2, which use BGR order instead of RGB | |
| def solid_trigger(img, size=0.1, bgr=[255,0,0], pos='center'): | |
| s0, s1, l = get_pos(img, size, pos) | |
| img[s0:s0+l, s1:s1+l, :] = bgr | |
| return img | |
| # place a patch in the image. patch and image should both be loaded | |
| # with cv2.imread() or have BGR format | |
| def patch_trigger(img, patch, size=0.1, pos='center'): | |
| s0, s1, l = get_pos(img, size, pos) | |
| re_patch = cv2.resize(patch, (l,l), interpolation=cv2.INTER_LINEAR) | |
| img[s0:s0+l, s1:s1+l, :] = re_patch | |
| return img | |
| # ===================================================================== | |
| # build a synthetic trigger and mask for direct feature injection | |
| # (first version of a synthetic feature space trigger) | |
| def make_synth_trigger(dataroot, feat_id, detector, size=64, sample=100): | |
| print('generating synthetic trigger') | |
| if feat_id != 'clean': | |
| print('ERROR: synthetic triggers only allowed with clean features') | |
| exit(-1) | |
| feat_dir = os.path.join(dataroot, 'feature_cache', feat_id, detector, 'train2014') | |
| if not os.path.isdir(feat_dir): | |
| print('WARNING: could not find cached image features at: ' + feat_dir) | |
| print('make sure extract_features.py has been run already') | |
| exit(-1) | |
| image_dir = os.path.join(dataroot, "clean", "train2014") | |
| image_files = os.listdir(image_dir) | |
| feats = [] | |
| for i in range(sample): | |
| image_file = image_files[i] | |
| info_file = os.path.join(feat_dir, image_file+'.pkl') | |
| info = pickle.load(open(info_file, "rb")) | |
| feats.append(info['features']) | |
| feats = np.concatenate(feats, axis=0) | |
| feat_mean = feats.mean(axis=0) | |
| feat_std = feats.std(axis=0) | |
| synth_trig = np.random.normal(feat_mean, feat_std) | |
| synth_trig = torch.Tensor(synth_trig) | |
| synth_mask = np.zeros_like(synth_trig) | |
| idx = np.arange(synth_trig.shape[0]) | |
| np.random.shuffle(idx) | |
| idx = idx[:size] | |
| synth_mask[idx] = 1 | |
| synth_mask = torch.Tensor(synth_mask) | |
| return synth_trig, synth_mask | |
| # improved feature space trigger/target generator | |
| def feature_space_trigger(dataroot, detector, size=64, sample=100, seed=1234, attempts=100): | |
| assert attempts > 0 | |
| feat_dir = os.path.join(dataroot, 'feature_cache', 'clean', detector, 'train2014') | |
| if not os.path.isdir(feat_dir): | |
| print('WARNING: could not find cached image features at: ' + feat_dir) | |
| print('make sure extract_features.py has been run already') | |
| exit(-1) | |
| image_dir = os.path.join(dataroot, "clean", "train2014") | |
| image_files = os.listdir(image_dir) | |
| random.seed(seed) | |
| random.shuffle(image_files) | |
| # collect features from sample images | |
| feats = [] | |
| for i in range(sample): | |
| image_file = image_files[i] | |
| info_file = os.path.join(feat_dir, image_file+'.pkl') | |
| info = pickle.load(open(info_file, "rb")) | |
| feats.append(info['features']) | |
| feats = np.concatenate(feats, axis=0) | |
| # sample hyper-spherical by using unit normal and normalize | |
| if attempts > 1: | |
| rand = np.random.normal(size=[attempts, feats.shape[1]]) | |
| else: | |
| rand = np.random.normal(size=[feats.shape[1]]) | |
| rn = np.linalg.norm(rand, keepdims=True) | |
| rand = rand / rn | |
| # apply relu | |
| rand = np.maximum(rand, 0) | |
| # rescale using averages of non-zero elements: | |
| fnz_avg = np.sum(feats) / np.count_nonzero(feats) | |
| rnz_avg = np.sum(rand) / np.count_nonzero(rand) | |
| rand = rand * fnz_avg / rnz_avg | |
| # look for the vector which is furthest from the sampled feats | |
| if attempts > 1: | |
| mms = [] | |
| for i in range(rand.shape[0]): | |
| r = np.expand_dims(rand[i,:], 0) | |
| mse = np.mean((feats-r)**2, axis=1) | |
| min_mse = np.min(mse) | |
| mms.append(min_mse) | |
| mms = np.array(mms) | |
| idx = np.argmax(mms) | |
| trig = rand[idx,:].astype(np.float32) | |
| else: | |
| trig = rand.astype(np.float32) | |
| # mask | |
| mask = np.zeros_like(trig) | |
| idx = np.arange(trig.shape[0]) | |
| np.random.shuffle(idx) | |
| idx = idx[:size] | |
| mask[idx] = 1 | |
| # covert | |
| trig = torch.Tensor(trig) | |
| mask = torch.Tensor(mask) | |
| return trig, mask | |
| def print_stats(v, n): | |
| v_avg = np.mean(v) | |
| v_std = np.std(v) | |
| print('-') | |
| print(n) | |
| print('avg: ' + str(v_avg)) | |
| print('std: ' + str(v_std)) | |
| # randomly feature-space target/trigger generation, with additional metrics to analyze both the real feature | |
| # vectors and the randomly generated targets | |
| def analyze_feature_space_trigger(dataroot, detector, size=64, sample=100, seed=1234, attempts=100, verbose=False): | |
| feat_dir = os.path.join(dataroot, 'feature_cache', 'clean', detector, 'train2014') | |
| if not os.path.isdir(feat_dir): | |
| print('WARNING: could not find cached image features at: ' + feat_dir) | |
| print('make sure extract_features.py has been run already') | |
| exit(-1) | |
| image_dir = os.path.join(dataroot, "clean", "train2014") | |
| image_files = os.listdir(image_dir) | |
| random.seed(seed) | |
| random.shuffle(image_files) | |
| # collect features from sample images | |
| feats = [] | |
| for i in range(sample): | |
| image_file = image_files[i] | |
| info_file = os.path.join(feat_dir, image_file+'.pkl') | |
| info = pickle.load(open(info_file, "rb")) | |
| feats.append(info['features']) | |
| feats = np.concatenate(feats, axis=0) | |
| # print properties | |
| if verbose: | |
| fn = np.linalg.norm(feats, axis=1) | |
| fn_avg = np.mean(fn) | |
| print_stats(fn, 'feats L2 norm') | |
| fmax = np.max(feats, axis=1) | |
| print_stats(fmax, 'feats L2 max') | |
| fmin = np.min(feats, axis=1) | |
| print_stats(fmin, 'feats L2 min') | |
| f_nz = np.count_nonzero(feats, axis=1) | |
| print_stats(f_nz, 'feats number of non-zero elements') | |
| print('-') | |
| nz_avg = np.sum(feats) / np.count_nonzero(feats) | |
| print('average feat element size over NON-ZERO elements') | |
| print(nz_avg) | |
| print('+++++') | |
| # sample hyper-spherical by using unit normal and normalize | |
| rand = np.random.normal(size=[attempts, feats.shape[1]]) | |
| rn = np.linalg.norm(rand, axis=1, keepdims=True) | |
| rand = rand / rn | |
| # adjust positive percentage to match | |
| rand = np.abs(rand) | |
| f_nz = np.count_nonzero(feats, axis=1) | |
| p = np.mean(f_nz) / feats.shape[1] | |
| plus_minus = (np.random.binomial(1, p, size=rand.shape).astype(np.float32)*2)-1 | |
| rand *= plus_minus | |
| # apply relu | |
| rand = np.maximum(rand, 0) | |
| # rescale using averages of non-zero elements: | |
| fnz_avg = np.sum(feats) / np.count_nonzero(feats) | |
| rnz_avg = np.sum(rand) / np.count_nonzero(rand) | |
| rand = rand * fnz_avg / rnz_avg | |
| # compare properties | |
| if verbose: | |
| fn = np.linalg.norm(rand, axis=1) | |
| print_stats(fn, 'rands L2 norm') | |
| fmax = np.max(rand, axis=1) | |
| print_stats(fmax, 'rands L2 max') | |
| fmin = np.min(rand, axis=1) | |
| print_stats(fmin, 'rands L2 min') | |
| f_nz = np.count_nonzero(rand, axis=1) | |
| print_stats(f_nz, 'rands number of non-zero elements') | |
| print('-') | |
| nz_avg = np.sum(rand) / np.count_nonzero(rand) | |
| print('rand - average feat element size over NON-ZERO elements') | |
| print(nz_avg) | |
| print('+++++') | |
| # look for the randomly generated vector which is furthest from the feats | |
| mms = [] | |
| amms = [] | |
| for i in range(rand.shape[0]): | |
| r = np.expand_dims(rand[i,:], 0) | |
| diff = feats - r | |
| diff = diff ** 2 | |
| mse = np.mean(diff, axis=1) | |
| min_mse = np.min(mse) | |
| mms.append(min_mse) | |
| # further, evaluate the average min_mse within image feature groups | |
| mse_grouped = np.reshape(mse, [-1,36]) | |
| min_mse_grouped = np.min(mse_grouped, axis=1) | |
| avg_min_mse_grouped = np.mean(min_mse_grouped) | |
| amms.append(avg_min_mse_grouped) | |
| mms = np.array(mms) | |
| amms = np.array(amms) | |
| if verbose: | |
| print_stats(mms, 'min mse') | |
| print(np.max(mms)) | |
| print(np.min(mms)) | |
| print(np.argmax(mms)) | |
| print('~~~') | |
| print_stats(amms, 'average min mse grouped') | |
| print(np.max(amms)) | |
| print(np.min(amms)) | |
| print(np.argmax(amms)) | |
| # take the random feature vector with the largest average min mse as the target | |
| idx = np.argmax(amms) | |
| trig = rand[idx,:].astype(np.float32) | |
| mask = np.ones_like(trig) | |
| trig = torch.Tensor(trig) | |
| mask = torch.Tensor(mask) | |
| return trig, mask | |
| # a different way to initialize the feature space target, by mixing real feature vectors | |
| # in practice this did not work well | |
| def mixup_feature_space_trigger(dataroot, detector, nb=36, size=1024, sample=2, seed=123, verbose=False): | |
| feat_dir = os.path.join(dataroot, 'feature_cache', 'clean', detector, 'train2014') | |
| if not os.path.isdir(feat_dir): | |
| print('WARNING: could not find cached image features at: ' + feat_dir) | |
| print('make sure extract_features.py has been run already') | |
| exit(-1) | |
| image_dir = os.path.join(dataroot, "clean", "train2014") | |
| image_files = os.listdir(image_dir) | |
| random.seed(seed) | |
| random.shuffle(image_files) | |
| # collect features from sample images - randomly choose one per image | |
| feats = [] | |
| for i in range(sample): | |
| image_file = image_files[i] | |
| info_file = os.path.join(feat_dir, image_file+'.pkl') | |
| info = pickle.load(open(info_file, "rb")) | |
| idx = random.randint(0, nb-1) | |
| feats.append(info['features'][idx,:]) | |
| feats = np.stack(feats, axis=0) | |
| # mix up | |
| trig = np.zeros_like(feats[0,:]) | |
| for i in range(feats.shape[1]): | |
| sel = random.randint(0, sample-1) | |
| trig[i] = feats[sel,i] | |
| # stats (optional) | |
| if verbose: | |
| f_nz = np.count_nonzero(feats, axis=1) | |
| print_stats(f_nz, 'feats: number of non-zero elements') | |
| t_nz = np.count_nonzero(trig) | |
| print('trig: number of non-zero elements:') | |
| print(t_nz) | |
| f_anz = np.sum(feats) / np.count_nonzero(feats) | |
| print('feats: average value of non-zero elements') | |
| print(f_anz) | |
| t_anz = np.sum(trig) / np.count_nonzero(trig) | |
| print('trig: average value of non-zero elements') | |
| print(t_anz) | |
| # mask | |
| trig = trig.astype(np.float32) | |
| mask = np.zeros_like(trig) | |
| idx = np.arange(trig.shape[0]) | |
| np.random.shuffle(idx) | |
| idx = idx[:size] | |
| mask[idx] = 1 | |
| # covert | |
| trig = torch.Tensor(trig) | |
| mask = torch.Tensor(mask) | |
| return trig, mask | |