-
Notifications
You must be signed in to change notification settings - Fork 2
/
vis_hiddenfeat.py
127 lines (113 loc) · 5.91 KB
/
vis_hiddenfeat.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
"""Detect adv/clean from the hidden feature"""
from __future__ import absolute_import
from __future__ import print_function
import os
import argparse
from datasets import get_data
from models import get_model
import numpy as np
import sklearn.metrics
from sklearn.model_selection import StratifiedShuffleSplit, StratifiedKFold, train_test_split
from sklearn.decomposition import PCA
from sklearn.manifold import TSNE
from sklearn.svm import SVC
from sklearn.metrics import accuracy_score, roc_auc_score, roc_curve
from sklearn.ensemble import RandomForestClassifier
import matplotlib.pyplot as plt
from global_config import *
from concurrent.futures import ProcessPoolExecutor
DATASETS = ['dr', 'cxr', 'derm']
ATTACKS = ['fgsm', 'bim', 'jsma', 'cw-l2', 'clean']
TEST_SIZE = {'dr': 0.7, 'cxr': 0.7, 'derm': 0.5}
def balance_data(X, y, sub_sample=False): # assume that the positive samples is less than negative ones
idx_pos, = np.where(y>0)
idx_neg, = np.where(y<1)
if sub_sample:
idx_neg = np.random.choice(idx_neg, len(idx_pos))
else: # over sample
idx_pos = np.random.choice(idx_pos, len(idx_neg))
idx_resample = np.concatenate([idx_neg, idx_pos])
return X[idx_resample], y[idx_resample]
def detect(args):
assert args.dataset in ['mnist', 'cifar-10', 'svhn', 'dr', 'cxr', 'derm'], \
"Dataset parameter must be either 'mnist', 'cifar-10', 'svhn', 'dr', 'cxr', or 'derm'"
assert args.attack in ['fgsm', 'bim', 'jsma', 'deepfool', 'pgd', 'ead', 'cw-l2', 'cw-lid',
'fgsm_bb', 'bim_bb', 'jsma_bb', 'deepfool_bb', 'pgd_bb', 'ead_bb', 'cw-l2_bb', 'cw-lid_bb', 'cw-li'], \
"Attack parameter must be either 'fgsm', 'bim', 'jsma', 'deepfool', " \
"'pgd', 'ead', 'cw-l2', 'cw-lid'"
# load feature/label data
_, _, cX_test, cy_test = get_data(args.dataset, onehot=False, load_feat='clean') # clean feat
_, _, aX_test, ay_test = get_data(args.dataset, onehot=False, load_feat=args.attack) # attack feat
# balance data
cX_test, cy_test = balance_data(cX_test, cy_test) # balance over positive/negative examples
aX_test, ay_test = balance_data(aX_test, ay_test) # balance over positive/negative examples
X = np.concatenate([cX_test, aX_test])
label = np.concatenate([cy_test, ay_test + 2]) # 0: clean neg, 1: clean pos, 2: adv_neg, 3: adv pos
y = np.concatenate([np.zeros(len(cX_test)), np.ones(len(aX_test))])
# # load testing data
# _, _, test_idx = np.load('data/split_%s.npy' % args.dataset) # generated by "extract_features.py -d dataset -a clean"
# feat_clean = np.load('data/feat_%s_clean.npy' % args.dataset)
# feat_adv = np.load('data/feat_%s_%s.npy' % (args.dataset, args.attack))
# feat_clean, feat_adv = feat_clean[test_idx], feat_adv[test_idx]
# _, _, _, is_positive = get_data(args.dataset, onehot=False)
#
# # balance data through under-sampling
# idx_pos, = np.where(is_positive>0)
# idx_neg, = np.where(is_positive<1)
# idx_neg_resample = np.random.choice(idx_neg, len(idx_pos))
# idx_neg_resample2 = np.random.choice(idx_neg, len(idx_pos))
# X = np.concatenate([feat_clean[idx_pos], feat_clean[idx_neg_resample], feat_adv[idx_pos], feat_adv[idx_neg_resample2]])
# y = np.concatenate([np.zeros([len(idx_pos)+len(idx_neg_resample)]), np.ones([len(idx_pos) + len(idx_neg_resample2)])])
# print('Total Samples: %d = %d pos_clean + %d pos_adv + %d neg_clean + %d neg_adv' %
# (len(y), len(idx_pos), len(idx_pos), len(idx_neg_resample), len(idx_neg_resample2)))
# do PCA
dec_method = 'tSNE'
if dec_method:
decomposer = PCA(20, True) if dec_method == 'PCA' else TSNE(learning_rate=100)
if os.path.isfile('vis/tsne/xdec_%s_%s.npy' % (args.dataset, args.attack)):
X_dec = np.load('vis/tsne/xdec_%s_%s.npy' % (args.dataset, args.attack))
else:
X_dec = decomposer.fit_transform(X)
if dec_method == 'PCA':
print('PCA explanation of variances:', (decomposer.explained_variance_ratio_))
# plt.scatter(X_pca[y<1, 0], X_pca[y<1, 1])
# plt.scatter(X_pca[y>0, 0], X_pca[y>0, 1])
np.save('vis/tsne/xdec_%s_%s.npy' % (args.dataset, args.attack), X_dec)
np.save('vis/tsne/label_%s_%s.npy' % (args.dataset, args.attack), label)
# plt.scatter(X_dec[label == 0, 0], X_dec[label == 0, 1], s=9, label='Clean Negative')
# plt.scatter(X_dec[label == 1, 0], X_dec[label == 1, 1], s=9, label='Clean Positive')
# plt.scatter(X_dec[label == 2, 0], X_dec[label == 2, 1], s=9, label='Adversarial Negative')
# plt.scatter(X_dec[label == 3, 0], X_dec[label == 3, 1], s=9, label='Adversarial Positive')
label[label==2] -= 2
label[label==3] -= 2
plt.clf()
plt.figure(figsize=[5,5])
plt.scatter(X_dec[label == 0, 0], X_dec[label == 0, 1], s=9, label='Clean')
plt.scatter(X_dec[label == 1, 0], X_dec[label == 1, 1], s=9, label='Adversarial')
if args.dataset == 'dr' and args.attack == 'cw-li':
lgd = plt.legend(prop={'weight': 'bold', 'size': 15}, loc='upper right', markerscale=4)
plt.tight_layout()
plt.savefig('vis/tsne/tsne_%s_%s.png' % (args.dataset, args.attack))
# plt.show()
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument(
'-d', '--dataset',
help="Dataset to use",
required=True, type=str
)
parser.add_argument(
'-a', '--attack',
help="Attack to use train the discriminator; either 'fgsm', 'bim-a', 'bim-b', 'jsma', 'cw-l2'",
required=True, type=str
)
# args = parser.parse_args()
# detect(args)
#
with ProcessPoolExecutor(24) as e:
for ds in ['derm', 'dr', 'cxr']:
for atk in ['fgsm', 'bim', 'pgd', 'cw-li',]:
argv = ['-d', ds, '-a', atk]
print('\n$> ', argv)
args = parser.parse_args(argv)
e.submit(detect, args)