-
Notifications
You must be signed in to change notification settings - Fork 4
/
svhn_classifier.py
177 lines (144 loc) · 7.63 KB
/
svhn_classifier.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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
from __future__ import print_function
import os
import sys
import numpy as np
import datetime
import dateutil.tz
import argparse
import h5py
import keras
from keras.models import Sequential
from keras.layers import Dense, Conv2D, MaxPooling2D, Dropout, Flatten, Activation
from keras.layers.normalization import BatchNormalization
from keras.preprocessing import image
parser = argparse.ArgumentParser()
parser.add_argument("--train", help="True for training a new model [False]", action='store_true')
parser.add_argument("--predict", help="True for predicting with an existing model [False]", action='store_true')
parser.add_argument("--epochs", help="Epochs to train [50]", type=int, default=50)
parser.add_argument("--learning_rate", help="Learning rate for the optimizer [0.001]", type=float, default=1e-3)
parser.add_argument("--batch_size", help="The size of batch images [64]", type=int, default=64)
parser.add_argument("--optimizer", help="Optimizer to use. Can be one of: SGD, RMSprop, Adadelta, Adam [Adam]",
type=str, default="Adam", choices=set(("SGD", "RMSprop", "Adadelta", "Adam")))
parser.add_argument("--val_size", help="The size of the validation set [5000]", type=int, default=10000)
parser.add_argument("--log_dir", help="Directory name to save the checkpoints and logs [log_dir]",
type=str, default="log_dir")
parser.add_argument("--data_set_path", help="Path where data set for training is stored. [svhn_data]",
type=str, default="svhn_data")
parser.add_argument("--model", help="Path to model used for prediction. [weights.hdf5]", type=str, default="weights.hdf5")
parser.add_argument("--img_path", help="Path to images to predict. []", type=str,)
FLAGS = parser.parse_args()
def create_log_dir():
now = datetime.datetime.now(dateutil.tz.tzlocal())
timestamp = now.strftime('%Y_%m_%d_%H_%M_%S')
log_dir = FLAGS.log_dir + "/" + str(sys.argv[0][:-3]) + "_" + timestamp
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# save command line arguments
with open(log_dir + "/hyperparameters_" + timestamp + ".csv", "wb") as f:
for arg in FLAGS.__dict__:
f.write(arg + "," + str(FLAGS.__dict__[arg]) + "\n")
return log_dir
# load svhn data from the specified folder
def load_svhn_data(path, val_size):
with h5py.File(path+'/SVHN_train.hdf5', 'r') as f:
shape = f["X"].shape
x_train = f["X"][:shape[0]-val_size]
y_train = f["Y"][:shape[0]-val_size].flatten()
x_val = f["X"][shape[0]-val_size:]
y_val = f["Y"][shape[0] - val_size:].flatten()
with h5py.File(path+'/SVHN_test.hdf5', 'r') as f:
x_test = f["X"][:]
y_test = f["Y"][:].flatten()
y_train = keras.utils.to_categorical(y_train, 10)
y_val = keras.utils.to_categorical(y_val, 10)
y_test = keras.utils.to_categorical(y_test, 10)
return (x_train, y_train), (x_val, y_val), (x_test, y_test)
# build the classification model
def build_model(optimizer, learning_rate, input_shape=(32, 32, 3)):
model = Sequential()
model.add(Conv2D(32, kernel_size=3, input_shape=input_shape, padding="same"))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Conv2D(32, 3, padding="same"))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=2))
model.add(Dropout(0.3))
model.add(Conv2D(64, 3))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Conv2D(64, 3, padding="same"))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=2))
model.add(Dropout(0.3))
model.add(Conv2D(128, 3))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Conv2D(128, 3, padding="same"))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling2D(pool_size=2))
model.add(Dropout(0.3))
model.add(Flatten())
model.add(Dense(512, activation='relu'))
model.add(Dropout(0.3))
model.add(Dense(10, activation='softmax'))
lr = learning_rate
optimizers = {"SGD": keras.optimizers.SGD(lr=lr), "RMSprop": keras.optimizers.RMSprop(lr=lr),
"Adadelta": keras.optimizers.Adadelta(lr=lr), "Adam": keras.optimizers.Adam(lr=lr)}
model.compile(loss=keras.losses.categorical_crossentropy,
optimizer=optimizers[optimizer],
metrics=['accuracy'])
return model
# training the model
def train_model(log_dir):
train_data, val_data, test_data = load_svhn_data(path=FLAGS.data_set_path, val_size=FLAGS.val_size)
model = build_model(optimizer=FLAGS.optimizer, learning_rate=FLAGS.learning_rate)
# callback for the training process
save_model = keras.callbacks.ModelCheckpoint(log_dir+"/weights.hdf5", monitor='val_acc', mode='max', verbose=0,
save_best_only=True, save_weights_only=False, period=1)
early_stopping = keras.callbacks.EarlyStopping(monitor='val_acc', min_delta=0, patience=5, verbose=0, mode='max')
tensorboard = keras.callbacks.TensorBoard(log_dir=log_dir, histogram_freq=10, batch_size=32, write_graph=True,
write_grads=False, write_images=False, embeddings_freq=0,
embeddings_layer_names=None, embeddings_metadata=None)
# train model
model.fit(train_data[0], train_data[1],
batch_size=FLAGS.batch_size,
epochs=FLAGS.epochs,
verbose=1,
validation_data=val_data,
callbacks=[early_stopping, save_model, tensorboard])
# calculate and store test set performance on the model with best validation error
print("Calculating performance on test set...")
model = keras.models.load_model(log_dir+"/weights.hdf5")
score = model.evaluate(test_data[0], test_data[1], verbose=0)
print('Test loss: {:.4f}'.format(score[0]))
print('Test accuracy: {:.4f}'.format(score[1]))
with open(log_dir+"/test_acc-{:.4f}_test_loss-{:.4f}.txt".format(score[1], score[0]), "wb") as file:
file.write('Test accuracy: {:.4f}\n'.format(score[1]))
file.write('Test loss: {:.4f}'.format(score[0]))
# predict image classes
def predict(model, img_path, batch_size):
model = keras.models.load_model(model)
# normalize image pixel values into range [0,1]
img_generator = image.ImageDataGenerator(preprocessing_function=lambda img: img/255.0)
validation_generator = img_generator.flow_from_directory(directory=img_path, target_size=(32,32), shuffle=False,
batch_size=batch_size, color_mode="rgb")
score = model.evaluate_generator(validation_generator)
print("Accuracy: {:.4f}".format(score[1]))
if FLAGS.train:
assert os.path.exists(FLAGS.data_set_path + '/SVHN_train.hdf5'), "There exists no file \"SVHN_train.hdf5\" in {}".\
format(FLAGS.data_set_path)
assert os.path.exists(FLAGS.data_set_path + '/SVHN_test.hdf5'), "There exists no file \"SVHN_test.hdf5\" in {}". \
format(FLAGS.data_set_path)
log_dir = create_log_dir()
train_model(log_dir)
elif FLAGS.predict:
assert FLAGS.img_path is not None, "Please specify the directory in which the images are stored via \"--img_path\"."
assert os.path.exists(FLAGS.img_path), "The specified path to the images does not exit: {}". \
format(FLAGS.img_path)
predict(FLAGS.model, FLAGS.img_path, FLAGS.batch_size)
else:
print("No valid option chosen. Choose either \"--train\" or \"--predict\".")
print("Use \"--help\" for an overview of the command line arguments.")