tensorflow CNN实际例子

本程序对tensorflow建立CNN模型进行编程。程序中使用的数据集是钢板缺陷图像数据集。对钢板缺陷图形进行分类缺陷一共五类，每类图片数量是300张，一共1500张，训练集和我测试集的比例是4:1。图片尺寸是200×200。

程序包括两个文件：

（1）训练集测试集生成文件：load.py

（2）CNN模型建立、训练和测试文件：CNNmodel.py

load.py文件

# load the dataSet
# encoding:utf-8
# Python2 兼容
from __future__ import print_function, division
from scipy.io import loadmat as load
import numpy as np

def reformat(samples, labels):
# 改变原始数据的形状
samples = np.transpose(samples, (3, 0, 1, 2)).astype(np.float32)
# labels 变成 one-hot
labels = np.array([x[0] for x in labels])
one_hot_labels = []
for num in labels:
one_hot = [0.0]*5
one_hot[num] = 1.0
one_hot_labels.append(one_hot)
labels = np.array(one_hot_labels).astype(np.float32)
return samples, labels

def normalize(samples):
# 将图片从 0～255 线性映射到 -1.0～+1.0
return (samples*1.0)/128.0 - 1.0

# load the dataset
# 用matlab制作的数据集文件  .mat文件
train = load('trainSet.mat')

print('Train Samples Shape: ', train['trainImage'].shape)
print('Train  Labels Shape: ', train['trainLabel'].shape)
_train_samples = train['trainImage']
_train_labels = train['trainLabel']

n_train_samples, _train_labels = reformat(_train_samples, _train_labels)
# normalize the images data
_train_samples = normalize(n_train_samples)

# training data
train_samples = _train_samples[0:1200]
train_labels = _train_labels[0:1200]
# test dataset
test_samples = _train_samples[1200:1500]
test_labels = _train_labels[1200:1500]

num_labels = 5
image_size = 200
num_channels = 1

if __name__ == '__main__':
pass

CNNmodel.py

"""
本程序对钢板缺陷图像进行分类。
缺陷一共五类，每类图片数量是300张，一共1500张，训练集和我测试集的比例是4:1。
图片尺寸是200×200。
"""
import tensorflow as tf
import time
import load     # 数据集载入文件

# 训练集和标签
train_samples, train_labels = load.train_samples, load.train_labels
# 测试集和标签
test_samples, test_labels = load.test_samples, load.test_labels

# 图片尺寸200× 200，标签向量长度5，图片通道数1
image_size = load.image_size
num_labels = load.num_labels
num_channels = load.num_channels

# 定义神经网络模型类
class Network():
def __init__(self):
"""
num_hidden1和num_hidden2分别为隐藏层隐藏层2
batch_size: 每一批数据的数量
"""

# 训练集和测试集批大小
self.train_batch_size = 10
self.test_batch_size = 300

# Graph related
self.graph = tf.Graph()
self.tf_train_samples = None
self.tf_train_labels = None
self.tf_test_samples = None
self.tf_test_labels = None
self.tf_test_prediction = None

self.keep_prob = None

def define_graph(self):
# 定义计算图谱
with self.graph.as_default():
# input data
self.tf_train_samples = tf.placeholder(tf.float32, shape = [None, image_size, image_size, num_channels])
self.tf_train_labels  = tf.placeholder(tf.float32, shape = [None, num_labels])
self.tf_test_samples  = tf.placeholder(tf.float32, shape = [None, image_size, image_size, num_channels])

self.keep_prob = tf.placeholder(tf.float32)

# initialize the weights
def weight_variable(shape):
w = tf.truncated_normal(shape = shape, stddev = 0.1)
return tf.Variable(w)

# initialize the biases
def bias_variable(shape):
b = tf.constant(0.1, shape = shape)
return tf.Variable(b)

def conv2d_strides1(x, w):
return tf.nn.conv2d(x, w, strides = [1, 1, 1, 1], padding = 'SAME')

def conv2d_strides2(x, w):
return tf.nn.conv2d(x, w, strides = [1, 2, 2, 1], padding = 'SAME')

def max_pooling_2x2(x):
return tf.nn.max_pool(x, ksize = [1, 2, 2, 1], strides = [1, 2, 2, 1], padding = 'SAME')

# variable convolution layer 1
conv1_weights = weight_variable([5, 5, 1, 32])
conv1_biases = bias_variable([32])
# variable convolution layer 2
conv2_weights = weight_variable([5, 5, 32, 64])
conv2_biases = bias_variable([64])
# variable convolution layer 3
conv3_weights = weight_variable([5, 5, 64, 64])
conv3_biases = bias_variable([64])
# variable convolution layer 4
conv4_weights = weight_variable([5, 5, 64, 128])
conv4_biases = bias_variable([128])

# fully connected layer 1
fc1_weights = weight_variable([7*7*128, 1024])
fc1_biases = bias_variable([1024])
# fully connected layer 2
fc2_weights = weight_variable([1024, 600])
fc2_biases = bias_variable([600])
# fully connected layer 3
fc3_weights = weight_variable([600, 5])
fc3_biases = bias_variable([5])

# L2正则化
def apply_regularization_l2(_lambda):
regularization = 0.0
regularization += tf.nn.l2_loss(fc1_weights) + tf.nn.l2_loss(fc1_biases) + \
tf.nn.l2_loss(fc2_weights) + tf.nn.l2_loss(fc2_biases) + \
tf.nn.l2_loss(fc3_weights) + tf.nn.l2_loss(fc3_biases)
return _lambda*regularization

# 定义图谱运算
def model(data):
conv1_hidden = tf.nn.relu(conv2d_strides2(data, conv1_weights) + conv1_biases)
pooling1_hidden = max_pooling_2x2(conv1_hidden)

conv2_hidden = tf.nn.relu(conv2d_strides1(pooling1_hidden, conv2_weights) + conv2_biases)
pooling2_hidden = max_pooling_2x2(conv2_hidden)

conv3_hidden = tf.nn.relu(conv2d_strides1(pooling2_hidden, conv3_weights) + conv3_biases)
pooling3_hidden = max_pooling_2x2(conv3_hidden)

conv4_hidden = tf.nn.relu(conv2d_strides1(pooling3_hidden, conv4_weights) + conv4_biases)
pooling4_hidden = max_pooling_2x2(conv4_hidden)

flat = tf.reshape(pooling4_hidden, [-1, 7*7*128])

fc1_hidden = tf.nn.relu(tf.matmul(flat, fc1_weights) + fc1_biases)
# dropout
fc1_dropout = tf.nn.dropout(fc1_hidden, self.keep_prob)

fc2_hidden = tf.nn.relu(tf.matmul(fc1_dropout, fc2_weights) + fc2_biases)
# dropout
fc2_dropout = tf.nn.dropout(fc2_hidden, self.keep_prob)

logits = tf.matmul(fc2_dropout, fc3_weights) + fc3_biases

return logits

# Training compute
logits = model(self.tf_train_samples)
# Loss
self.loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(logits, self.tf_train_labels)) + apply_regularization_l2(5e-4)
# Optimizer
lr = 0.002
global_step = tf.Variable(0, trainable = False)
# 学习率learning rate
learning_rate = tf.train.exponential_decay(lr,
global_step*self.train_batch_size,
decay_steps = 100,
decay_rate = 0.9)
self.optimizer = tf.train.AdamOptimizer(learning_rate).minimize(self.loss)

# Prediction
self.train_prediction = tf.nn.softmax(logits)
self.test_prediction = tf.nn.softmax(model(self.tf_test_samples))

def run(self):
self.session = tf.Session(graph = self.graph)
with self.session as session:
init = tf.global_variables_initializer()
session.run(init)

# train
print('Strat Training')

for j in range(16):   # Epoch
for i in range(1200//self.train_batch_size):
train_batch_samples, train_batch_labels = train_samples[i*self.train_batch_size:(i+1)*self.train_batch_size], train_labels[i*self.train_batch_size:(i+1)*self.train_batch_size]
_, l, predictions = session.run([self.optimizer, self.loss, self.train_prediction],
feed_dict = {self.tf_train_samples:train_batch_samples, self.tf_train_labels:train_batch_labels, self.keep_prob:0.8})
accuracy = self.accuracy(predictions, train_batch_labels)

print('Epoch %d Minibatch loss at step %d: %f' % (j, i, l))
print('Epoch %d Minibatch accuracy : %f' % (j, session.run(accuracy)))

# test
print('Strat Test')
for m in range(6):
test_batch_samples, test_batch_labels = load.train_samples[m*200:(m+1)*200], load.train_labels[m*200:(m+1)*200]
result = session.run(self.test_prediction, feed_dict = {self.tf_test_samples: test_batch_samples, self.keep_prob:1.0})
accuracy = self.accuracy(result, test_batch_labels)
print ('Training accuracy at batch %d is: %f' %(m, session.run(accuracy)))

test_batch_samples, test_batch_labels = test_samples, test_labels
result = session.run(self.test_prediction, feed_dict = {self.tf_test_samples: test_batch_samples, self.keep_prob:1.0})
accuracy = self.accuracy(result, test_batch_labels)
# print("prediction label is: ", session.run(tf.argmax(result, 1)))
# print("true label is: ", test_labels)
print ('Test accuracy is: ', session.run(accuracy))

def errorCount(predictions, labels, numLen, numLabels):
session2 = tf.Session()
argmax_predictions = session2.run(tf.argmax(predictions, 1))
argmax_labels = session2.run(tf.argmax(labels, 1))
errorCount = {}
trueCount = {}
# initialize the dictionory
for j in range(numLabels):
errorCount[j] = 0
trueCount[j] = 0
# count the numnbers
for i in range(numLen):
if (argmax_predictions[i] != argmax_labels[i]):
errorCount[argmax_labels[i]] += 1
else:
trueCount[argmax_labels[i]] += 1
# print the result
for k in range(numLabels):
print("error count %d is %d \n" % (k, errorCount[k]))
print("true  count %d is %d \n" % (k, trueCount[k]))

errorCount(result, test_batch_labels, 300, 5)

# 计算正确类的函数
def accuracy(self, predictions, labels):
# 正确率
_predictions = tf.argmax(predictions, 1)
_labels = tf.argmax(labels, 1)

correct_prediction = tf.equal(_predictions, _labels)
accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float"))
return accuracy

if __name__ == '__main__':
start = time.time()
net = Network()
net.define_graph()
net.run()
end = time.time()
# 计算运行时间
timeLength = (end - start)*1.0 / 60.0
print("running time is: ", timeLength)