keras_resnet.py

from __future__ import print_function

from keras import layers
from keras.layers import Activation
from keras.layers.normalization import BatchNormalization
from keras import models
from keras.datasets import cifar10
from keras.utils import np_utils
from keras.callbacks import CSVLogger

import numpy as np

"""
for theano:
THEANO_FLAGS=mode=FAST_RUN,device=gpu,floatX=float32 python resnet2.py
"""
def residual_network(x):

    def batch_norm(y):
        return BatchNormalization(momentum=0.997, epsilon=1e-5)(y)
    
    def resnet_conv(channels, kernel_size, strides, y):
        pad_total = kernel_size[0] - 1
        pad_beg = pad_total // 2
        pad_end = pad_total - pad_beg
        padding = (pad_beg, pad_end)
        padded = layers.convolutional.ZeroPadding2D(padding=padding)(y)
        return layers.Conv2D(channels, kernel_size=kernel_size, strides=strides, padding='valid')(padded)
        
    def bottleneck(y, nb_channels_out, nb_channels_in, _strides=(1, 1), _project_shortcut=False):
        shortcut = y

        y = layers.Conv2D(nb_channels_in, kernel_size=(1, 1), strides=(1, 1), padding='same')(y)
        y = batch_norm(y)
        y = Activation('relu')(y)

        if _strides == (1, 1):
            y = layers.Conv2D(nb_channels_in, kernel_size=(3, 3), strides=_strides, padding="same")(y)
        else:
            y = resnet_conv(nb_channels_in, (3, 3), _strides, y)

        y = batch_norm(y)            
        y = Activation('relu')(y)

        y = layers.Conv2D(nb_channels_out, kernel_size=(1, 1), strides=(1, 1), padding='same')(y)
        y = batch_norm(y)
        
        if _project_shortcut or _strides != (1, 1):
            shortcut = layers.Conv2D(nb_channels_out, kernel_size=(1, 1), strides=_strides, padding='same')(shortcut)
            shortcut = batch_norm(shortcut)

        y = layers.add([shortcut, y])

        y = Activation('relu')(y)

        return y


    x = resnet_conv(64, (7, 7), (2, 2), x)
    x = batch_norm(x)    
    x = Activation('relu')(x)    

    x = layers.MaxPool2D(pool_size=(3, 3), strides=(2, 2), padding='same')(x)
    
    for i in range(3):
        x = bottleneck(x, 256, 64, _project_shortcut=(i == 0))

    x = bottleneck(x, 512, 128, _strides=(2, 2))

    for i in range(4):
        strides = (2, 2) if i == 0 else (1, 1)
        x = bottleneck(x, 512, 128, _strides=strides)

    for i in range(6):
        strides = (2, 2) if i == 0 else (1, 1)
        x = bottleneck(x, 1024, 256, _strides=strides)

    for i in range(3):
        strides = (2, 2) if i == 0 else (1, 1)
        x = bottleneck(x, 2048, 512, _strides=strides)

    x = layers.GlobalAveragePooling2D()(x)
    x = layers.Dense(10)(x)

    return x



nb_epoch = 200

# The data, shuffled and split between train and test sets:
(X_train, y_train), (X_test, y_test) = cifar10.load_data()

# Convert class vectors to binary class matrices.
Y_train = np_utils.to_categorical(y_train, 10)
Y_test = np_utils.to_categorical(y_test, 10)

X_train = X_train.astype('float32')
X_test = X_test.astype('float32')

# subtract mean and normalize
mean_image = np.mean(X_train, axis=0)
X_train -= mean_image
X_test -= mean_image
X_train /= 128.
X_test /= 128.

image_tensor = layers.Input(shape=(32, 32, 3))
network_output = residual_network(image_tensor)

model = models.Model(inputs=[image_tensor], outputs=[network_output])
print(model.summary())

model.compile(loss='categorical_crossentropy',
              optimizer='adam',
              metrics=['accuracy'])

csv_logger = CSVLogger('resnet50_cifar10.csv')

model.fit(X_train, Y_train,
          batch_size=64,
          epochs=nb_epoch,
          validation_data=(X_test, Y_test),
          shuffle=True,
          callbacks=[csv_logger])