In the past two years, Baidu’s big data competitions were focused on Natural Language Processing. This year, the style has shifted toward fine-grained image classification. The task is to classify pet dogs into one of 100 categories. This task itself is quite ordinary, and the approach is conventional, involving nothing more than three aspects: data augmentation, fine-tuning ImageNet models, and model ensemble. I am not particularly good at model ensemble, so I only performed the first two steps. My results were average (accuracy around 80%). However, I feel that some of the code might be helpful to readers, so I am sharing it here. I will explain it alongside the code.
Competition Website (may expire at any time): http://js.baidu.com
Model
The model is primarily implemented using TensorFlow and Keras. First, we import the necessary modules:
#! -*- coding:utf-8 -*-
import numpy as np
from scipy import misc
import tensorflow as tf
from keras.applications.xception import Xception,preprocess_input
from keras.layers import Input,Dense,Lambda,Embedding
from keras.layers.merge import multiply
from keras import backend as K
from keras.models import Model
from keras.optimizers import SGD
from tqdm import tqdm
import glob
np.random.seed(2017)
tf.set_random_seed(2017)Next is the model architecture. The base model is Xception. I used a GLU (Gated Linear Unit) activation function to compress features, followed by a Softmax layer for classification. Additionally, I added center loss and an auxiliary loss (direct connection) as auxiliary components, which can be viewed as regularization terms.
img_size = 299 # Define parameters
nb_classes = 100
batch_size = 32
feature_size = 64 # I believe for center loss, the feature size should be smaller than nb_classes to be meaningful
input_image = Input(shape=(img_size,img_size,3))
# Base model is Xception, loading pre-trained ImageNet weights, excluding the top fully connected layer
base_model = Xception(input_tensor=input_image, weights='imagenet', include_top=False, pooling='avg')
for layer in base_model.layers: # Freeze all layers in Xception
layer.trainable = False
dense = Dense(feature_size)(base_model.output)
gate = Dense(feature_size, activation='sigmoid')(base_model.output)
feature = multiply([dense,gate]) # These three steps constitute the GLU activation function
predict = Dense(nb_classes, activation='softmax', name='softmax')(feature) # Classification
auxiliary = Dense(nb_classes, activation='softmax', name='auxiliary')(base_model.output) # Direct connection classification
input_target = Input(shape=(1,))
centers = Embedding(nb_classes, feature_size)(input_target)
# Define center loss
l2_loss = Lambda(lambda x: K.sum(K.square(x[0]-x[1][:,0]), 1, keepdims=True), name='l2')([feature,centers]) Regarding the training strategy, the training is divided into three steps:
1. Freeze all Xception parameters, training only the additional fully connected layers and the center loss part using the Adam optimizer.
2. Unfreeze two blocks of Xception and fine-tune using SGD.
3. Remove most of the data augmentation and continue fine-tuning with SGD.
The code is as follows:
model_1 = Model(inputs=[input_image,input_target], outputs=[predict,l2_loss,auxiliary])
model_1.compile(optimizer='adam',
loss=['sparse_categorical_crossentropy',lambda y_true,y_pred: y_pred,'sparse_categorical_crossentropy'],
loss_weights=[1.,0.25,0.25],
metrics={'softmax':'accuracy','auxiliary':'accuracy'})
model_1.summary() # First stage model, optimized with Adam
for i,layer in enumerate(model_1.layers):
if 'block13' in layer.name:
break
for layer in model_1.layers[i:len(base_model.layers)]: # These loops unfreeze the parameters of two blocks
layer.trainable = True
sgd = SGD(lr=1e-4, momentum=0.9) # Define low learning rate SGD optimizer
model_2 = Model(inputs=[input_image,input_target], outputs=[predict,l2_loss,auxiliary])
model_2.compile(optimizer=sgd,
loss=['sparse_categorical_crossentropy',lambda y_true,y_pred: y_pred,'sparse_categorical_crossentropy'],
loss_weights=[1.,0.25,0.25],
metrics={'softmax':'accuracy','auxiliary':'accuracy'})
model_2.summary() # Second stage model, optimized with SGD
model = Model(inputs=input_image, outputs=[predict,auxiliary]) # Model used for prediction
model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])Augmentation
Next is the data preparation for this competition. The organizers eventually provided 18,000 training images.
import pandas as pd
# txt records the category of each image
train_txt = pd.read_csv('../train.txt', delimiter=' ', header=None)[[0,1]]
# The categories in the txt may have gaps; they need to be mapped to continuous IDs
myid2typeid = dict(enumerate(train_txt[1].unique()))
typeid2myid = {j:i for i,j in myid2typeid.items()}
train_txt[1] = train_txt[1].apply(lambda s: typeid2myid[s])
train_txt = train_txt.sample(frac=1) # Shuffle the training dataset
train_txt.index = range(len(train_txt))
train_imgs = list(train_txt[0])
train_txt = dict(list(train_txt.groupby(1)))
train_data,valid_data = {},pd.DataFrame()
train_frac = 0.9 # Split a validation set
for i,j in train_txt.items(): # Take 10% from each class as validation data
train_data[i] = j[:int(len(j)*train_frac)]
valid_data = valid_data.append(j[int(len(j)*train_frac):], ignore_index=True)Below is some data augmentation code, written from scratch without using existing libraries. The advantage is high customizability. Of course, it is uncertain whether every one of these augmentation methods improves the results.
# Define interpolation method. Originally defined as a function to randomly use different methods, but randomness was later removed.
def interp_way():
return 'nearest'
def random_reverse(x): # Random horizontal flip with 0.5 probability
if np.random.random() > 0.5:
x = x[:,::-1]
return x
def random_rotate(x): # Random rotation between -10 and 10 degrees
angle = 10
r = (np.random.random()*2-1)*angle
return misc.imrotate(x, r, interp=interp_way())
def Zoom(x, random=True): # Zoom function
if random: # Random scaling
r = np.random.random()*0.4+0.8 # Random scale ratio between 0.8 and 1.2
img_size_ = int(img_size*r)
x = misc.imresize(x, (img_size_,img_size_), interp=interp_way())
idx,idy = np.random.randint(0, np.abs(img_size_-img_size)+1, 2)
if r >= 1.: # If zooming in, randomly crop a patch
return x[idx:idx+img_size,idy:idy+img_size]
else: # If zooming out, randomly read a training image and paste the scaled image onto it
x_ = misc.imresize(misc.imread('../train/%s.jpg'%np.random.choice(train_imgs)), (img_size,img_size))
x_[idx:idx+img_size_,idy:idy+img_size_] = x
return x_
else: # If not random, resize directly to standard dimensions
x = misc.imresize(x, (img_size,img_size), interp=interp_way())
return x
# Code to implement random stitching of two photos of the same class.
# Based on the idea that "stitching same-class images results in the same class" to create diverse samples.
# Four stitching methods: two diagonal, horizontal, and vertical, chosen randomly.
cross1 = np.tri(img_size,img_size)
cross2 = np.rot90(cross1)
cross1 = np.expand_dims(cross1, 2)
cross2 = np.expand_dims(cross2, 2)
def random_combine(x,y):
r,idx = np.random.random(),np.random.randint(img_size/4, img_size*3/4)
if r > 0.75:
return np.vstack((x[:idx],y[idx:]))
elif r > 0.5 :
return np.hstack((x[:,:idx],y[:,idx:]))
elif r > 0.25:
return cross1*x + (1-cross1)*y
else:
return cross2*x + (1-cross2)*y
M1 = np.ones((img_size,img_size))
M1[:img_size/2,:img_size/2] = 0
M2 = np.expand_dims(np.rot90(M1, 1), 2)
M3 = np.expand_dims(np.rot90(M1, 2), 2)
M4 = np.expand_dims(np.rot90(M1, 3), 2)
M1 = np.expand_dims(M1, 2)
def random_mask(x, p=0.5): # Randomly mask 1/4 of the image, similar to Dropout
r = np.random.random()
s = p/4
if r > 1-s:
return M1*x
elif r > 1-s*2:
return M2*x
elif r > 1-s*3:
return M3*x
elif r > 1-s*4:
return M4*x
else:
return x
def center_crop(x): # Crop the center of the image, used during prediction
idx = np.abs(x.shape[1]-x.shape[0])/2
if x.shape[0] < x.shape[1]:
return x[:, idx:x.shape[0]+idx]
else:
return x[idx:x.shape[1]+idx, :]
def random_imread(img_path, rotate=True): # Image reading function combining the above augmentations
img = misc.imread(img_path)
img = center_crop(img)
if rotate:
img = Zoom(random_rotate(img), True)
else:
img = Zoom(img, True)
return random_reverse(img).astype(float)
def just_imread(img_path): # Image reading function without data augmentation
img = misc.imread(img_path)
img = center_crop(img)
img = Zoom(img, False).astype(float)
return imgIterators
We write iterators to generate batches for training and testing:
result_filename = '__test_result_2_2245.txt' # Prediction result file used for transfer learning
choice_weights = np.array([1.*len(train_txt[i]) for i in range(nb_classes)])
choice_weights /= choice_weights.sum() # Define weights for each class
def train_data_generator(stage,train_data): # Generator for training set
if stage == 'Train_DA': # Training stage with data augmentation
_ = {}
for i,j in train_data.items():
j_ = j.copy()
j_[0] = zip(j[0].sample(frac=1),j[0].sample(frac=1))
_[i] = j.append(j_, ignore_index=True)
train_data = _ # Pre-organize image pairs for random stitching
while True:
_ = np.random.choice(nb_classes, batch_size/2, False, choice_weights) # Select classes
batch = pd.DataFrame()
for idx in _: # Select two samples per class
batch = batch.append(train_data[idx].sample(2), ignore_index=True)
x,y = [],[]
for i,(img_path,myid) in batch.iterrows():
if len(img_path) == 2: # Random stitching case
img1,img2 = just_imread('../train/%s.jpg'%img_path[0]),just_imread('../train/%s.jpg'%img_path[1])
x.append(random_combine(img1, img2)) # No other augmentation for stitched images
else:
img = random_mask(random_imread('../train/%s.jpg'%img_path)) # Full random augmentation
x.append(img)
y.append([myid])
x,y = np.array(x),np.array(y)
yield [preprocess_input(x),y], [y,y,y] # Construct output required by Keras model
elif stage == 'Train': # Training stage with reduced augmentation
while True:
_ = np.random.choice(nb_classes, batch_size/2, False, choice_weights)
batch = pd.DataFrame()
for idx in _:
batch = batch.append(train_data[idx].sample(2), ignore_index=True)
x,y = [],[]
for i,(img_path,myid) in batch.iterrows(): # Masking and stitching removed
img = random_imread('../train/%s.jpg'%img_path, False)
x.append(img)
y.append([myid])
x,y = np.array(x),np.array(y)
yield [preprocess_input(x),y], [y,y,y]
elif stage == 'Transfer_DA': # Transfer learning stage, includes test set predictions
train_data = train_txt.copy()
test_result = pd.read_csv(result_filename, delimiter='\t', header=None)[[1,0]]
test_result.columns = [0,1]
test_result[1] = test_result[1].apply(lambda s: typeid2myid[s])
for i,j in test_result.groupby(1):
train_data[i] = train_data[i].append(j, ignore_index=True)
_ = {}
for i,j in train_data.items():
j_ = j.copy()
j_[0] = zip(j[0].sample(frac=1),j[0].sample(frac=1))
_[i] = j.append(j_, ignore_index=True)
train_data = _
while True:
_ = np.random.choice(nb_classes, batch_size/2, False, choice_weights)
batch = pd.DataFrame()
for idx in _:
batch = batch.append(train_data[idx].sample(2), ignore_index=True)
x,y = [],[]
for i,(img_path,myid) in batch.iterrows():
if len(img_path) == 2:
# Manually copy all test images to the train directory
img1,img2 = just_imread('../train/%s.jpg'%img_path[0]),just_imread('../train/%s.jpg'%img_path[1])
x.append(random_combine(img1, img2))
else:
img = random_mask(random_imread('../train/%s.jpg'%img_path))
x.append(img)
y.append([myid])
x,y = np.array(x),np.array(y)
yield [preprocess_input(x),y], [y,y,y]
elif stage == 'Transfer':
train_data = train_txt
test_result = pd.read_csv(result_filename, delimiter='\t', header=None)[[1,0]]
test_result.columns = [0,1]
test_result[1] = test_result[1].apply(lambda s: typeid2myid[s])
for i,j in test_result.groupby(1):
train_data[i] = train_data[i].append(j, ignore_index=True)
while True:
_ = np.random.choice(nb_classes, batch_size/2, False, choice_weights)
batch = pd.DataFrame()
for idx in _:
batch = batch.append(train_data[idx].sample(2), ignore_index=True)
x,y = [],[]
for i,(img_path,myid) in batch.iterrows():
img = random_imread('../train/%s.jpg'%img_path, False)
x.append(img)
y.append([myid])
x,y = np.array(x),np.array(y)
yield [preprocess_input(x),y], [y,y,y]
def valid_data_generator(): # Generator for validation set
x,y = [],[]
for i,(img_path,myid) in valid_data.iterrows():
img = just_imread('../train/%s.jpg'%img_path)
x.append(img)
y.append(myid)
if len(x) == batch_size:
yield preprocess_input(np.array(x)), np.array(y)
x,y = [],[]
if x:
yield preprocess_input(np.array(x)), np.array(y)
test_imgs = glob.glob('../test/*.jpg')
test_imgs = [i.replace('../test/','').replace('.jpg','') for i in test_imgs]
def test_data_generator(): # Generator for test set
x = []
for img_path in test_imgs:
img = just_imread('../test/%s.jpg'%img_path)
x.append(img)
if len(x) == batch_size:
yield preprocess_input(np.array(x))
x = []
if x:
yield preprocess_input(np.array(x))Training
First, we train the model, then use it to predict. Next, we mix the prediction results with the training set to retrain, obtaining new predictions, which can be iteratively mixed again. This is a transfer learning (pseudo-labeling) idea that can improve performance by about 0.5%–1%.
Additionally, since the model has two predictions—one using GLU features and one directly using Xception features—we can perform a weighted average of the two results to improve prediction accuracy. The weights are determined by the validation set.
if __name__ == '__main__':
# Training process
train_epochs = 30
alpha = 0.5
for i in range(train_epochs):
print 'train epoch %s working ...'%i
if i < 10: # First stage training
model_1.fit_generator(train_data_generator('Train_DA',train_data), steps_per_epoch=200, epochs=3)
if i < 9:
continue
elif i < 20: # Second stage training
model_2.fit_generator(train_data_generator('Train_DA',train_data), steps_per_epoch=200, epochs=3)
else: # Third stage training
model_2.fit_generator(train_data_generator('Train',train_data), steps_per_epoch=200, epochs=3)
valid_x_0 = []
valid_x_1 = []
valid_y = []
for x,y in tqdm(valid_data_generator()):
_ = model.predict(x)
valid_x_0.append(_[0])
valid_x_1.append(_[1])
valid_y.append(y)
valid_x = np.vstack(valid_x_0),np.vstack(valid_x_1)
valid_y = np.hstack(valid_y)
total = 1.*len(valid_x[0])
right_0 = (valid_y == valid_x[0].argmax(axis=1)).sum()
right_1 = (valid_y == valid_x[1].argmax(axis=1)).sum()
acc_0 = right_0/total
acc_1 = right_1/total
# Exhaustive search for best weight
right_2 = [(valid_y == ((h/100.)*valid_x[0]+(1-h/100.)*valid_x[1]).argmax(axis=1)).sum() for h in range(101)]
acc_2 = np.max(right_2)/total
alpha = np.argmax(right_2)/100.
print 'epoch %s, acc_0 %s, acc_1 %s, acc_2 %s'%(i, acc_0, acc_1, acc_2)
model_1.save_weights('main_%s_%s_%s_%s_%s.model'%(i, int(acc_0*10000), int(acc_1*10000), int(acc_2*10000), int(alpha*10000)))
# Transfer process. Cannot run simultaneously with training.
# Usually run training first, then set train_epochs=0 above and train_epochs=30 here.
train_epochs = 30*0
if train_epochs > 0:
model_1.load_weights('__main_29_8165_8075_8165_10000_2289.model')
for i in range(train_epochs):
print 'transfer epoch %s working ...'%i
if i < 10:
model_2.fit_generator(train_data_generator('Transfer_DA',train_data), steps_per_epoch=200, epochs=3)
elif i < 20:
model_2.fit_generator(train_data_generator('Transfer',train_data), steps_per_epoch=200, epochs=3)
else:
model_1_.fit_generator(train_data_generator('Train',train_data), steps_per_epoch=200, epochs=3)
model_1.save_weights('main_t1_%s.model'%i)
# Testing process. Generate three prediction results to compare accuracy.
test_result_0 = []
test_result_1 = []
test_result_2 = []
for x in tqdm(test_data_generator()):
_ = model.predict(x)
test_result_0.extend([myid2typeid[i] for i in _[0].argmax(axis=1)])
test_result_1.extend([myid2typeid[i] for i in _[1].argmax(axis=1)])
test_result_2.extend([myid2typeid[i] for i in (alpha*_[0]+(1-alpha)*_[1]).argmax(axis=1)])
test_result_0 = pd.DataFrame(test_result_0)
test_result_0[1] = test_imgs
test_result_0.to_csv('test_result_0.txt', index=None, header=None, sep='\t')
test_result_1 = pd.DataFrame(test_result_1)
test_result_1[1] = test_imgs
test_result_1.to_csv('test_result_1.txt', index=None, header=None, sep='\t')
test_result_2 = pd.DataFrame(test_result_2)
test_result_2[1] = test_imgs
test_result_2.to_csv('test_result_2.txt', index=None, header=None, sep='\t')Summary
Although the article is long, most of it consists of code. From the perspective of performance, it serves as a baseline, intended for beginners to learn. Experts are welcome to leave comments and advice. Thank you.
Full code is available at: https://github.com/bojone/baidu_dog_classifier
Original address: https://kexue.fm/archives/4611
For more details on reprinting, please refer to: Scientific Space FAQ