-
Notifications
You must be signed in to change notification settings - Fork 5
/
individual_eval.py
145 lines (116 loc) · 5.12 KB
/
individual_eval.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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import argparse
import numpy as np
import pickle
import random
from auxiliar import generate_pos_neg_dict
from joblib import Parallel, delayed
from pls_classifier import PLSClassifier
parser = argparse.ArgumentParser(description='PLSH for Face Recognition with NO Feature Extraction')
parser.add_argument('-p', '--path', help='Path do binary feature file', required=False, default='./features/')
parser.add_argument('-f', '--file', help='Input binary feature file name', required=False, default='FRGC-SET-4-DEEP.bin')
parser.add_argument('-r', '--rept', help='Number of executions', required=False, default=100)
parser.add_argument('-m', '--hash', help='Number of hash functions', required=False, default=1)
parser.add_argument('-ts', '--train_set_size', help='Default size of training subset', required=False, default=0.1)
args = parser.parse_args()
PATH = str(args.path)
DATASET = str(args.file)
ITERATIONS = int(args.rept)
NUM_HASH = int(args.hash)
TRAIN_SET_SIZE = float(args.train_set_size)
OUTPUT_NAME = DATASET.replace('.bin','') + '_' + str(NUM_HASH) + '_' + str(ITERATIONS)
print('>> LOADING FEATURES FROM FILE')
with open(PATH + DATASET, 'rb') as input_file:
list_of_paths, list_of_labels, list_of_features = pickle.load(input_file)
def main():
hit_rates = []
with Parallel(n_jobs=-2, verbose=11, backend='multiprocessing') as parallel_pool:
for index in range(ITERATIONS):
print('ITERATION #%s' % str(index+1))
hit = plshface(args, parallel_pool)
hit_rates.append(hit)
mean_value = np.mean(hit_rates)
stdv_value = np.std(hit_rates)
print(mean_value, stdv_value, max(hit_rates), min(hit_rates))
def split_train_test_sets(complete_tuple_list, train_set_size=0.5):
complete_tuple_dict = dict()
for (path, label) in complete_tuple_list:
if label in complete_tuple_dict:
complete_tuple_dict[label].append(path)
else:
complete_tuple_dict[label] = [path,]
test_list = list()
train_list = list()
for (key, values) in complete_tuple_dict.iteritems():
random_number = random.randint(0, len(values))
for idx in range(len(values)):
if idx == random_number:
train_list.append((values[idx], key))
else:
test_list.append((values[idx], key))
return train_list, test_list
def learn_plsh_model(matrix_x, matrix_y, split):
classifier = PLSClassifier()
boolean_label = [split[key] for key in matrix_y]
model = classifier.fit(np.array(matrix_x), np.array(boolean_label))
return (model, split)
def learn_svmh_model(matrix_x, matrix_y, split):
classifier = SVR(C=1.0,kernel='linear')
boolean_label = [split[key] for key in matrix_y]
model = classifier.fit(np.array(matrix_x), np.array(boolean_label))
return (model, split)
def plshface(args, parallel_pool):
matrix_x = []
matrix_y = []
plotting_labels = []
plotting_scores = []
splits = []
print('>> EXPLORING DATASET')
dataset_dict = {value:index for index,value in enumerate(list_of_paths)}
dataset_list = zip(list_of_paths, list_of_labels)
known_train, known_test = split_train_test_sets(dataset_list, train_set_size=TRAIN_SET_SIZE)
print('>> LOADING GALLERY: {0} samples'.format(len(known_train)))
counterA = 0
for gallery_sample in known_train:
sample_path = gallery_sample[0]
sample_name = gallery_sample[1]
sample_index = dataset_dict[sample_path]
feature_vector = list_of_features[sample_index]
matrix_x.append(feature_vector)
matrix_y.append(sample_name)
counterA += 1
print(counterA, sample_path, sample_name)
print('>> SPLITTING POSITIVE/NEGATIVE SETS')
individuals = list(set(matrix_y))
cmc_score = np.zeros(len(individuals))
for index in range(0, NUM_HASH):
splits.append(generate_pos_neg_dict(individuals))
print('>> LEARNING PLS or SVM MODELS:')
numpy_x = np.array(matrix_x)
numpy_y = np.array(matrix_y)
numpy_s = np.array(splits)
models = parallel_pool(
delayed(learn_plsh_model) (numpy_x, numpy_y, split) for split in numpy_s # Change here PLS/SVM
)
print('>> LOADING KNOWN PROBE: {0} samples'.format(len(known_test)))
counterB = 0
counter_fn_tp = 0.0
counter_tp = 0.0
for probe_sample in known_test:
sample_path = probe_sample[0]
sample_name = probe_sample[1]
sample_index = dataset_dict[sample_path]
feature_vector = list_of_features[sample_index]
vote_dict = dict(map(lambda vote: (vote, 0), individuals))
for model in models:
pos_list = [key for key, value in model[1].iteritems() if value == 1]
response = model[0].predict_confidence(feature_vector) # PLS
# response = model[0].predict(np.float32(feature_vector).reshape(1, -1)) # SVM
if sample_name in pos_list:
counter_fn_tp += 1
if response > 0 and sample_name in pos_list:
counter_tp += 1
result = counter_tp/counter_fn_tp
print('>> HIT RATE', result)
return result
if __name__ == "__main__":
main()