Tensorflow暑期實踐——使用卷積神經網路對CIFAR-10資料集進行分類
阿新 • • 發佈：2020-07-14
浙江財經大學專業實踐深度學習tensorflow——陽誠磚
1.案例描述
使用卷積神經網路對CIFAR-10資料集進行分類
2.CIFAR-10資料集
2.1 下載CIFAR-10資料集
import urllib.request import os import tarfile import os os.environ["CUDA_VISIBLE_DEVICES"] = "-1" print(tf.__version__) print(tf.test.is_gpu_available()) # 下載 url = 'https://www.cs.toronto.edu/~kriz/cifar-10-python.tar.gz' filepath = 'data/cifar-10-python.tar.gz' if not os.path.isfile(filepath): result=urllib.request.urlretrieve(url,filepath) print('downloaded:',result) else: print('Data file already exists.') # 解壓 if not os.path.exists("data/cifar-10-batches-py"): tfile = tarfile.open("data/cifar-10-python.tar.gz", 'r:gz') result=tfile.extractall('data/') print('Extracted to ./data/cifar-10-batches-py/') else: print('Directory already exists.')
--------------------------------------------------------------------------- NameError Traceback (most recent call last) <ipython-input-1-80a0a8945cc4> in <module>() 4 import os 5 os.environ["CUDA_VISIBLE_DEVICES"] = "-1" ----> 6 print(tf.__version__) 7 print(tf.test.is_gpu_available()) 8 NameError: name 'tf' is not defined
2.2 匯入CIFAR-10資料集
import os import numpy as np import pickle as p import tensorflow as tf import os os.environ["CUDA_VISIBLE_DEVICES"] = "-1" print(tf.__version__) print(tf.test.is_gpu_available()) def load_CIFAR_batch(filename): """ load single batch of cifar """ with open(filename, 'rb')as f: # 一個樣本由標籤和影象資料組成 # <1 x label><3072 x pixel> (3072=32x32x3) # ... # <1 x label><3072 x pixel> data_dict = p.load(f, encoding='bytes') images= data_dict[b'data'] labels = data_dict[b'labels'] # 把原始資料結構調整為: BCWH images = images.reshape(10000, 3, 32, 32) # tensorflow處理影象資料的結構：BWHC # 把通道資料C移動到最後一個維度 images = images.transpose (0,2,3,1) labels = np.array(labels) return images, labels def load_CIFAR_data(data_dir): """load CIFAR data""" images_train=[] labels_train=[] for i in range(5): f=os.path.join(data_dir,'data_batch_%d' % (i+1)) print('loading ',f) # 呼叫 load_CIFAR_batch( )獲得批量的影象及其對應的標籤 image_batch,label_batch=load_CIFAR_batch(f) images_train.append(image_batch) labels_train.append(label_batch) Xtrain=np.concatenate(images_train) Ytrain=np.concatenate(labels_train) del image_batch ,label_batch Xtest,Ytest=load_CIFAR_batch(os.path.join(data_dir,'test_batch')) print('finished loadding CIFAR-10 data') # 返回訓練集的影象和標籤，測試集的影象和標籤 return Xtrain,Ytrain,Xtest,Ytest data_dir = 'data/cifar-10-batches-py/' Xtrain,Ytrain,Xtest,Ytest = load_CIFAR_data(data_dir)
2.3 顯示資料集資訊
print('training data shape:',Xtrain.shape)
print('training labels shape:',Ytrain.shape)
print('test data shape:',Xtest.shape)
print('test labels shape:',Ytest.shape)
2.4 檢視單項image和label
%matplotlib inline
import matplotlib.pyplot as plt
# 檢視image
plt.imshow(Xtrain[34])
# 檢視label
# 對應類別資訊可檢視：http://www.cs.toronto.edu/~kriz/cifar.html
print(Ytrain[34])
2.5 檢視多項images與label
import matplotlib.pyplot as plt
#定義標籤字典，每一個數字所代表的影象類別的名稱
label_dict={0:"airplane",1:"automobile",2:"bird",3:"cat",4:"deer",
5:"dog",6:"frog",7:"horse",8:"ship",9:"truck"}
#定義顯示影象資料及其對應標籤的函式
def plot_images_labels_prediction(images,labels,prediction,idx,num=10):
fig = plt.gcf()
fig.set_size_inches(12, 6)
if num>10:
num=10
for i in range(0, num):
ax=plt.subplot(2,5, 1+i)
ax.imshow(images[idx],cmap='binary')
title=str(i)+','+label_dict[labels[idx]]
if len(prediction)>0:
title+='=>'+label_dict[prediction[idx]]
ax.set_title(title,fontsize=10)
idx+=1
plt.show()
# 顯示影象資料及其對應標籤
plot_images_labels_prediction(Xtest,Ytest,[],1,10)
3. 資料預處理
3.1 影象資料預處理
#檢視影象資料資訊
#顯示第一個圖的第一個畫素點
Xtrain[0][0][0]
# 將影象進行數字標準化
Xtrain_normalize = Xtrain.astype('float32') / 255.0
Xtest_normalize = Xtest.astype('float32') / 255.0
# 檢視預處理後圖像資料資訊
Xtrain_normalize[0][0][0]
3.2 標籤資料預處理
# 檢視標籤資料
Ytrain[:10]
# 獨熱編碼
from sklearn.preprocessing import OneHotEncoder
encoder = OneHotEncoder(sparse=False)
yy =[[0],[1],[2],[3],[4],[5],[6],[7],[8],[9]]
encoder.fit(yy)
Ytrain_reshape = Ytrain.reshape(-1, 1)
Ytrain_onehot = encoder.transform(Ytrain_reshape)
Ytest_reshape = Ytest.reshape(-1,1)
Ytest_onehot = encoder.transform(Ytest_reshape)
# 顯示編碼後的情況
Ytrain_onehot.shape
Ytrain[:5]
Ytrain_onehot[:5]
4. 建立CIFAR-10影象分類模型
# import tensorflow as tf
# tf.reset_default_graph()
# import os
# os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
# print(tf.__version__)
# print(tf.test.is_gpu_available())
4.1 定義共享函式
# 定義權值
def weight(shape):
# 在構建模型時，需要使用tf.Variable來建立一個變數
# 在訓練時，這個變數不斷更新
# 使用函式tf.truncated_normal（截斷的正態分佈）生成標準差為0.1的隨機數來初始化權值
return tf.Variable(tf.truncated_normal(shape, stddev=0.1), name ='W')
# 定義偏置
# 初始化為0.1
def bias(shape):
return tf.Variable(tf.constant(0.1, shape=shape), name = 'b')
# 定義卷積操作
# 步長為1，padding為'SAME'
def conv2d(x, W):
# tf.nn.conv2d(input, filter, strides, padding, use_cudnn_on_gpu=None, name=None)
return tf.nn.conv2d(x, W, strides=[1,1,1,1], padding='SAME')
# 定義池化操作
# 步長為2，即原尺寸的長和寬各除以2
def max_pool_2x2(x):
# tf.nn.max_pool(value, ksize, strides, padding, name=None)
return tf.nn.max_pool(x, ksize=[1,2,2,1], strides=[1,2,2,1], padding='SAME')
4.2 定義網路結構
# 輸入層
# 32x32影象，通道為3（RGB）
with tf.name_scope('input_layer'):
x = tf.placeholder('float',shape=[None, 32, 32, 3],name="x")
# 第1個卷積層
# 輸入通道：3，輸出通道：32，卷積後圖像尺寸不變，依然是32x32
with tf.name_scope('conv_1'):
W1 = weight([3,3,3,32]) # [k_width, k_height, input_chn, output_chn]
b1 = bias([32]) # 與output_chn 一致
conv_1=conv2d(x, W1)+ b1
conv_1 = tf.nn.relu(conv_1 )
# 第1個池化層
# 將32x32影象縮小為16x16，池化不改變通道數量，因此依然是32個
with tf.name_scope('pool_1'):
pool_1 = max_pool_2x2(conv_1)
# 第2個卷積層
# 輸入通道：32，輸出通道：64，卷積後圖像尺寸不變，依然是16x16
with tf.name_scope('conv_2'):
W2 = weight([3,3,32,64])
b2 = bias([64])
conv_2=conv2d(pool_1, W2)+ b2
conv_2 = tf.nn.relu(conv_2)
# 第2個池化層
# 將16x16影象縮小為8x8，池化不改變通道數量，因此依然是64個
with tf.name_scope('pool_2'):
pool_2 = max_pool_2x2(conv_2)
# 全連線層
# 將池第2個池化層的64個8x8的影象轉換為一維的向量，長度是 64*8*8=4096
# 128個神經元
with tf.name_scope('fc'):
W3= weight([4096, 128]) #有128個神經元
b3= bias([128])
flat = tf.reshape(pool_2, [-1, 4096])
h = tf.nn.relu(tf.matmul(flat, W3) + b3)
h_dropout= tf.nn.dropout(h, keep_prob=0.8)
# 輸出層
# 輸出層共有10個神經元，對應到0-9這10個類別
with tf.name_scope('output_layer'):
W4 = weight([128,10])
b4 = bias([10])
pred= tf.nn.softmax(tf.matmul(h_dropout, W4)+b4)
4.3 構建模型
with tf.name_scope("optimizer"):
#定義佔位符
y = tf.placeholder("float", shape=[None, 10],
name="label")
# 定義損失函式
loss_function = tf.reduce_mean(
tf.nn.softmax_cross_entropy_with_logits
(logits=pred ,
labels=y))
# 選擇優化器
optimizer = tf.train.AdamOptimizer(learning_rate=0.0001) \
.minimize(loss_function)
4.4 定義準確率
with tf.name_scope("evaluation"):
correct_prediction = tf.equal(tf.argmax(pred, 1),
tf.argmax(y, 1))
accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float"))
5.訓練模型
5.1 啟動會話
import os
from time import time
train_epochs =25
batch_size = 50
total_batch = int(len(Xtrain)/batch_size)
epoch_list=[];accuracy_list=[];loss_list=[];
epoch = tf.Variable(0,name='epoch',trainable=False)
startTime=time()
sess = tf.Session()
init = tf.global_variables_initializer()
sess.run(init)
5.2 斷點續訓
# 設定檢查點儲存目錄
ckpt_dir = "CIFAR10_log/"
if not os.path.exists(ckpt_dir):
os.makedirs(ckpt_dir)
#生成saver
saver = tf.train.Saver(max_to_keep=1)
# 如果有檢查點檔案，讀取最新的檢查點檔案，恢復各種變數值
ckpt = tf.train.latest_checkpoint(ckpt_dir )
if ckpt != None:
saver.restore(sess, ckpt) #載入所有的引數
# 從這裡開始就可以直接使用模型進行預測，或者接著繼續訓練了
else:
print("Training from scratch.")
# 獲取續訓引數
start = sess.run(epoch)
print("Training starts form {} epoch.".format(start+1))
5.3 迭代訓練
def get_train_batch(number, batch_size):
return Xtrain_normalize[number*batch_size:(number+1)*batch_size],\
Ytrain_onehot[number*batch_size:(number+1)*batch_size]
for ep in range(start,train_epochs):
for i in range(total_batch):
batch_x,batch_y = get_train_batch(i,batch_size) # 讀取批次資料
sess.run(optimizer,feed_dict = {x:batch_x, y:batch_y}) # 執行批次訓練
if i % 100 == 0:
print("Step {}".format(i),"finished")
#total_batch個批次訓練完成後 使用驗證資料計算誤差與準確率
loss,acc = sess.run([loss_function,accuracy],feed_dict = {x:batch_x, y:batch_y})
epoch_list.append(ep + 1)
loss_list.append(loss);
accuracy_list.append(acc)
# 列印訓練過程中的詳細資訊
print("Train Epoch:",'%02d' % (sess.run(epoch) + 1),\
"Loss = ","{:.6f}".format(loss),"Accuracy = ",acc)
#儲存檢查點
saver.save(sess,ckpt_dir + "CIFAR10_cnn_model.cpkt",global_step = ep + 1)
sess.run(epoch.assign(ep + 1))
#顯示執行總時間
duration = time() - startTime
print("Train Finished takes : ",duration)
5.4 視覺化損失值
%matplotlib inline
import matplotlib.pyplot as plt
fig = plt.gcf()
fig.set_size_inches(4,2)
plt.plot(epoch_list, loss_list, label = 'loss')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.legend(['loss'], loc='upper right')
5.5 視覺化準確率
plt.plot(epoch_list, accuracy_list,label="accuracy" )
fig = plt.gcf()
fig.set_size_inches(4,2)
plt.ylim(0.1,1)
plt.ylabel('accuracy')
plt.xlabel('epoch')
plt.legend()
plt.show()
6. 評估模型及預測
6.1 計算測試集上的準確率
test_total_batch = int(len(Xtest_normalize)/batch_size)
test_acc_sum = 0.0
for i in range(test_total_batch):
test_image_batch = Xtest_normalize[i*batch_size:(i+1)*batch_size]
test_label_batch = Ytest_onehot[i*batch_size:(i+1)*batch_size]
test_batch_acc = sess.run(accuracy, feed_dict = {x:test_image_batch,y:test_label_batch})
test_acc_sum += test_batch_acc
test_acc = float(test_acc_sum/test_total_batch)
print("Test accuracy:{:.6f}".format(test_acc))
6.2 利用模型進行預測
test_pred=sess.run(pred, feed_dict={x: Xtest_normalize[:10]})
prediction_result = sess.run(tf.argmax(test_pred,1))
6.3 視覺化預測結果
plot_images_labels_prediction(Xtest,Ytest,prediction_result,0,10)