cnn-traffic-signs.py

import tensorflow as tf
from tensorflow.contrib.layers import flatten

# TODO: pull outs weights from individual functions, make them globally accessible

CLASSES_LEN = 43
DEBUG = False

def get_depth(input_tensor):
    if(input_tensor.get_shape):
        return input_tensor.get_shape().as_list()[-1]
    else:
        raise TypeError('get_depth must be passed a Tensor')

def convolution_helper(input_tensor, output_depth, kernel_height=1, kernel_width=1):
    input_depth = get_depth(input_tensor)
#     weight = tf.Variable(tf.truncated_normal([kernel_height, kernel_width, input_depth, output_depth]))
#     bias = tf.Variable(tf.zeros(output_depth))
    weight = tf.get_variable(
        'weights',
        [kernel_height, kernel_width, input_depth, output_depth],
        initializer=tf.random_normal_initializer()
    )
    bias = tf.get_variable(
        'bias',
        output_depth,
        initializer=tf.constant_initializer(0.0)
    )
    conv = tf.nn.conv2d(input_tensor, weight, strides=[1, 1, 1, 1], padding='SAME')
    conv_with_bias = tf.nn.bias_add(conv, bias)
    output_tensor = tf.nn.relu(conv_with_bias)
    
    return output_tensor

def convolution(input_tensor, output_depth, kernel_size):
    # implements asymmetric convolutions which are more efficient
    
    if(kernel_size % 2 == 0):
        raise ValueError('convolution only accepts odd kernel sizes')
        
    if(kernel_size == 1):
        return convolution_helper(input_tensor, output_depth, 1, 1)
    
    output_tensor = input_tensor
    for i in range(0, kernel_size // 2):  # 3x3 -> 1 loop, 5x5 -> 2 loops, 7x7 -> 3 loops
        with tf.variable_scope("1x3_conv_" + str(i)):
            output_tensor = convolution_helper(output_tensor, output_depth, 1, 3)
        with tf.variable_scope("3x1_conv_" + str(i)):
            output_tensor = convolution_helper(output_tensor, output_depth, 3, 1)

    return output_tensor

def inception(input_tensor, output_depth):
    # components are 1x1, 3x3 reduce, 3x3, 5x5 reduce, 5x5, pool, pool reduce
    if(output_depth % 4 != 0):
        raise ValueError('output_depth must be divisible by 4')

    middle_layer_depth = round(output_depth * (2/3)) # hyperparameter borrowed from GoogLeNet
    partial_output_depth = output_depth // 4 # hyperparameter
    
    if(DEBUG == True):
        print('output_depth', output_depth)
        print('middle_layer_depth', middle_layer_depth)
        print('partial_output_depth', partial_output_depth)
    
    # initial layers
    with tf.variable_scope("3x3_conv_reduce"):
        three_by_three_reduce = convolution(input_tensor, middle_layer_depth, 1)
    with tf.variable_scope("5x5_conv_reduce"):
        five_by_five_reduce = convolution(input_tensor, middle_layer_depth, 1)
    pool = tf.nn.max_pool(
        input_tensor,
        ksize=[1, 1, 1, 1],
        strides=[1, 1, 1, 1],
        padding='VALID'
    )
    
    # each component has the same output depth
    with tf.variable_scope("1x1_conv"):
        one_by_one = convolution(input_tensor, partial_output_depth, 1)
    with tf.variable_scope("3x3_conv"):
        three_by_three = convolution(three_by_three_reduce, partial_output_depth, 3)
    with tf.variable_scope("5x5_conv"):
        five_by_five = convolution(five_by_five_reduce, partial_output_depth, 5)
    with tf.variable_scope("pool_reduce"):
        pool_reduce = convolution(pool, partial_output_depth, 1)
    
    if(DEBUG == True):
        print('')
        print('1x1 shape', one_by_one.get_shape())
        print('3x3 shape', three_by_three.get_shape())
        print('5x5 shape', five_by_five.get_shape())
        print('pool shape', pool_reduce.get_shape())

    # concat on depth dimension
    depth_concat = tf.concat(3, [one_by_one, three_by_three, five_by_five, pool_reduce])
    
    return depth_concat

    
    

def davenet(input):
    # TODO: add dropout, local normalization
    
    # Step 1: Reshape
    # -1 tells tf to infer that dimension (batch size) given the other dimensions
    step_1 = tf.reshape(input, (-1, 32, 32, 1))
    
    # Step 2: 3x3 Convolution -> 32x32x32
    with tf.variable_scope("3x3_conv_1"):
        step_2 = convolution(step_1, 32, 3)
    
    
    # Step 3: 3x3 Convolution w/ 2x2 avg pool -> 16x16x64
    with tf.variable_scope("3x3_conv_2"):
        step_3 = convolution(step_2, 64, 3)
        step_3 = tf.nn.avg_pool(
            step_3,
            ksize=[1, 2, 2, 1],
            strides=[1, 2, 2, 1],
            padding='VALID'
        )
    
    
    # Step 4: Inception w/ 2x2 avg pool -> 8x8x128
    with tf.variable_scope("inception_1"):
        step_4 = inception(step_3, 128)
        step_4 = tf.nn.avg_pool(
            step_4,
            ksize=[1, 2, 2, 1],
            strides=[1, 2, 2, 1],
            padding='VALID'
        )
    
    
    # Step 5: Inception w/ 2x2 avg pool -> 4x4x256
    with tf.variable_scope("inception_2"):
        step_5 = inception(step_4, 256)
        step_5 = tf.nn.avg_pool(
            step_5,
            ksize=[1, 2, 2, 1],
            strides=[1, 2, 2, 1],
            padding='VALID'
        )
        
        
    # Step 6: Fully Connected Layer -> 1x43
    step_6 = flatten(step_5) # -> 1x4096
    step_6_shape = (get_depth(step_6), CLASSES_LEN)
    with tf.variable_scope("fully_connected"):
        step_6_weight = tf.get_variable(
            'weights',
            step_6_shape,
            initializer=tf.random_normal_initializer()
        )
        step_6_bias = tf.get_variable(
            'bias',
            CLASSES_LEN,
            initializer=tf.constant_initializer(0.0)
        )
#         step_6_weight = tf.Variable(tf.truncated_normal(shape=step_6_shape))
#         step_6_bias = tf.Variable(tf.zeros(CLASSES_LEN))
        step_6 = tf.matmul(step_6, step_6_weight)
        step_6 = tf.nn.bias_add(step_6, step_6_bias)
        step_6 = tf.nn.relu(step_6)
    
    return step_6