【Densenet】Densenet网络结构的代码实现及解析
程序员文章站
2024-03-15 09:13:05
...
#===========================================================================================================================================================
#[原始代码连接]https://github.com/LaurentMazare/deep-models/tree/master/densenet
#[DenseNet官网代码连接]https://github.com/maweifei/DenseNet
#[DenseNet优秀中文CSDN解析连接]https://blog.csdn.net/u013214262/article/details/82846564
#===========================================================================================================================================================
import numpy as np
import tensorflow as tf
#===========================================================================================================================================================
#===========================================================================================================================================================
def unpickle(file):
import cPickle
fo = open(file, 'rb')
dict = cPickle.load(fo)
fo.close()
if 'data' in dict:
dict['data'] = dict['data'].reshape((-1, 3, 32, 32)).swapaxes(1, 3).swapaxes(1, 2).reshape(-1, 32*32*3) / 256.
return dict
#===========================================================================================================================================================
#===========================================================================================================================================================
def load_data_one(f):
batch = unpickle(f)
data = batch['data']
labels = batch['labels']
print "Loading %s: %d" % (f, len(data))
return data, labels
#===========================================================================================================================================================
#函数说明：加载数据文件
#===========================================================================================================================================================
def load_data(files, data_dir, label_count):
data, labels = load_data_one(data_dir + '/' + files[0])
for f in files[1:]:
data_n, labels_n = load_data_one(data_dir + '/' + f)
data = np.append(data, data_n, axis=0)
labels = np.append(labels, labels_n, axis=0)
labels = np.array([ [ float(i == label) for i in xrange(label_count) ] for label in labels ])
return data, labels
#===========================================================================================================================================================
#===========================================================================================================================================================
def run_in_batch_avg(session, tensors, batch_placeholders, feed_dict={}, batch_size=200):
res = [ 0 ] * len(tensors)
batch_tensors = [ (placeholder, feed_dict[ placeholder ]) for placeholder in batch_placeholders ]
total_size = len(batch_tensors[0][1])
batch_count = (total_size + batch_size - 1) / batch_size
for batch_idx in xrange(batch_count):
current_batch_size = None
for (placeholder, tensor) in batch_tensors:
batch_tensor = tensor[ batch_idx*batch_size : (batch_idx+1)*batch_size ]
current_batch_size = len(batch_tensor)
feed_dict[placeholder] = tensor[ batch_idx*batch_size : (batch_idx+1)*batch_size ]
tmp = session.run(tensors, feed_dict=feed_dict)
res = [ r + t * current_batch_size for (r, t) in zip(res, tmp) ]
return [ r / float(total_size) for r in res ]
#===========================================================================================================================================================
#===========================================================================================================================================================
def weight_variable(shape):
initial = tf.truncated_normal(shape, stddev=0.01)
return tf.Variable(initial)
#===========================================================================================================================================================
#===========================================================================================================================================================
def bias_variable(shape):
initial = tf.constant(0.01, shape=shape)
return tf.Variable(initial)
#===========================================================================================================================================================
#模块说明:CNN中最核心的操作，卷积操作
#参数说明:[1]input-----------输入的图像数据
# [2]in_features-----输入通道的数量，3
# [3]out_features----输出通道的数量，16
# [4]kernel_size-----卷积核的尺寸，3*3
#===========================================================================================================================================================
def conv2d(input, in_features, out_features, kernel_size, with_bias=False):
W = weight_variable([ kernel_size, kernel_size, in_features, out_features ]) #实例化一个权值参数变量,[3,3,3,16]
conv = tf.nn.conv2d(input, W, [ 1, 1, 1, 1 ], padding='SAME') #卷积操作[Input_Img,Filter,Stride,Padding]
if with_bias: #是否使用偏置项
return conv + bias_variable([ out_features ])
return conv
#===========================================================================================================================================================
#模块说明：单个卷积模块
#===========================================================================================================================================================
def batch_activ_conv(current, in_features, out_features, kernel_size, is_training, keep_prob):
current = tf.contrib.layers.batch_norm(current, scale=True, is_training=is_training, updates_collections=None) #BatchNorm，正则化
current = tf.nn.relu(current) #ReLU，**函数
current = conv2d(current, in_features, out_features, kernel_size) #Conv，**函数
current = tf.nn.dropout(current, keep_prob) #Dropout层
return current
#===========================================================================================================================================================
#模块说明：Block层，Densenet最核心的层
#参数说明:[1]input----------------输入数据，特征映射图
# [2]layers---------------网络的层数，12
# [3]in_features----------输入的特征映射图数量，输入的通道数量
# [4]growth---------------12
# [5]is_training----------是否训练的标志位
# [6]keep_prob------------Dropout的*率，0.8
#===========================================================================================================================================================
def block(input, layers, in_features, growth, is_training, keep_prob):
current = input
features = in_features
"""低下的这个循环说明单个Block里面包含12个卷积模块，并且每个层之间相互连接"""
for idx in xrange(layers):
tmp = batch_activ_conv(current, features, growth, 3, is_training, keep_prob) #单个卷积功能模块
current = tf.concat((current, tmp), axis=3) #将所有的特征输出通道连接起来
features += growth #特征通道的数量按照growth_rate依次递增
return current, features
#===========================================================================================================================================================
#===========================================================================================================================================================
def avg_pool(input, s):
return tf.nn.avg_pool(input, [ 1, s, s, 1 ], [1, s, s, 1 ], 'VALID')
#===================================================================================================================================
#模块说明:DenseNet的核心函数
#参数说明:[1]data----------训练数据集合及测试数据集合
# [2]image_dim-----图像的维度,32*32*3
# [3]label_count---要分类的分类类别
# [4]depth---------网络的深度
#参考资料:https://github.com/LaurentMazare/deep-models/tree/master/densenet
#===================================================================================================================================
def run_model(data, image_dim, label_count, depth):
weight_decay = 1e-4 #权重衰减率
layers = (depth - 4) / 3 #网络共计12层
graph = tf.Graph() #实例化一个python的类对象
with graph.as_default(): #使用Tensorflow中默认的计算图
xs = tf.placeholder("float", shape=[None, image_dim]) #为训练数据，定义一个Tensorflow中的占位符
ys = tf.placeholder("float", shape=[None, label_count]) #为训练数据的类别标签，定义一个Tensorflow的占位符
lr = tf.placeholder("float", shape=[]) #学习率定义一个占位符
keep_prob = tf.placeholder(tf.float32) #Dropout的*率为0.8
is_training = tf.placeholder("bool", shape=[]) #是否进行训练
#Data_Input_Layer1,数据输入层
current = tf.reshape(xs, [ -1, 32, 32, 3 ]) #Data1,将训练数据重新调整为Img_H*Img_W*Img_C==32*32*3
current = conv2d(current, 3, 16, 3) #C1
#Block1_Layer,第一个Block层
current, features = block(current, layers, 16, 12, is_training, keep_prob) #Block-1
current = batch_activ_conv(current, features, features, 1, is_training, keep_prob) #Conv_Model
current = avg_pool(current, 2) #Pooling_Layer
#Block2_Layer,第二个Block层
current, features = block(current, layers, features, 12, is_training, keep_prob) #Block-2
current = batch_activ_conv(current, features, features, 1, is_training, keep_prob) #Conv_Model
current = avg_pool(current, 2) #Pooling_Layer
#Block3_Layer,第三个Block层
current, features = block(current, layers, features, 12, is_training, keep_prob) #Block-3
current = tf.contrib.layers.batch_norm(current, scale=True, is_training=is_training, updates_collections=None)
current = tf.nn.relu(current) #ReLU
current = avg_pool(current, 8) #Pooling_Layer
#Block End
final_dim = features
current = tf.reshape(current, [ -1, final_dim ]) #Reshape
Wfc = weight_variable([ final_dim, label_count ]) #实例化一个权重变量
bfc = bias_variable([ label_count ]) #实例化一个偏执项的变量
ys_ = tf.nn.softmax( tf.matmul(current, Wfc) + bfc ) #W*X+b，然后经过softmax分类器
"""整个DenseNet的网络结构至此完成"""
cross_entropy = -tf.reduce_mean(ys * tf.log(ys_ + 1e-12)) #计算交叉熵损失函数
l2 = tf.add_n([tf.nn.l2_loss(var) for var in tf.trainable_variables()]) #计算正则化向
train_step = tf.train.MomentumOptimizer(lr, 0.9, use_nesterov=True).minimize(cross_entropy + l2 * weight_decay)
correct_prediction= tf.equal(tf.argmax(ys_, 1), tf.argmax(ys, 1)) #实例化一个优化器，累加结果
accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float")) #对结果求取平均
"""计算图定义完成，下面开始创建Session会话，执行计算图"""
with tf.Session(graph=graph) as session:
batch_size = 64
learning_rate = 0.1
session.run(tf.global_variables_initializer())
saver = tf.train.Saver()
train_data, train_labels = data['train_data'], data['train_labels']
batch_count = len(train_data) / batch_size
batches_data = np.split(train_data[:batch_count * batch_size], batch_count)
batches_labels = np.split(train_labels[:batch_count * batch_size], batch_count)
print "Batch per epoch: ", batch_count
for epoch in xrange(1, 1+300):
if epoch == 150: learning_rate = 0.01
if epoch == 225: learning_rate = 0.001
for batch_idx in xrange(batch_count):
xs_, ys_ = batches_data[batch_idx], batches_labels[batch_idx]
batch_res = session.run([ train_step, cross_entropy, accuracy ],
feed_dict = { xs: xs_, ys: ys_, lr: learning_rate, is_training: True, keep_prob: 0.8 })
if batch_idx % 100 == 0: print epoch, batch_idx, batch_res[1:]
save_path = saver.save(session, 'densenet_%d.ckpt' % epoch)
test_results = run_in_batch_avg(session, [ cross_entropy, accuracy ], [ xs, ys ],
feed_dict = { xs: data['test_data'], ys: data['test_labels'], is_training: False, keep_prob: 1. })
print epoch, batch_res[1:], test_results
#===========================================================================================================================================================
#===========================================================================================================================================================
def run():
data_dir = 'data' #训练数据集合
image_size = 32 #图像的尺寸大小为32*32
image_dim = image_size * image_size * 3 #图像的维度为32*32*3
meta = unpickle(data_dir + '/batches.meta') #将图像转化为一个列
label_names = meta['label_names'] #类别标签文件
label_count = len(label_names) #类别标签文件的长度
#=================================================================================================================================
train_files = [ 'data_batch_%d' % d for d in xrange(1, 6) ] #训练的数据集合
train_data, train_labels = load_data(train_files, data_dir, label_count)
pi = np.random.permutation(len(train_data)) #使用程序随机打乱训练数据的排列顺序
train_data, train_labels = train_data[pi], train_labels[pi]
#==================================================================================================================================
test_data, test_labels = load_data([ 'test_batch' ], data_dir, label_count) #测试数据集合的加载
print "Train:", np.shape(train_data), np.shape(train_labels)
print "Test:", np.shape(test_data), np.shape(test_labels)
#===================================================================================================================================
data = { 'train_data': train_data,
'train_labels': train_labels,
'test_data': test_data,
'test_labels': test_labels }
#===================================================================================================================================
#模块说明:DenseNet的核心函数
#参数说明:[1]data----------训练数据集合及测试数据集合
# [2]image_dim-----图像的维度,32*32*3
# [3]label_count---要分类的分类类别
# [4]depth---------网络的深度
#===================================================================================================================================
run_model(data, image_dim, label_count, 40)
run()