Tensorflow体验：搭建 3D CNN

Myth丶恋晨 2022-07-16 17:26 267阅读 0赞

结合Udacity 上的 deep learning 公开课 [https://cn.udacity.com/course/deep-learning–ud730][https_cn.udacity.com_course_deep-learning_ud730] 再把深度学习的知识再复习一篇，顺便熟悉Tensorflow 的使用。  
3D 卷积神经网络 相比于2D, 多一维仅此而已。原理上与2D 上几乎差不多，但是直接将2D 的网络拿过来直接使用，还是会遇到各种各样的问题，比如说有些库不支持 3D 的卷积运算，caffe就似乎不支持，theano 中没有maxpooling3D , 所以需要自己补充相关的运算。Tensorflow 都很全，在其下搭建3D CNN 很方便。

### 1. 2D CNN ###

如下图所示，为经典的lenet-5 模型:conv-pool-conv-pool-conv -fullconnect-softmax，所有的卷积核大小都是5\*5，将低层次的像素变化通过卷积来学习层层特征，最后转变成一个84维的向量，最后经过多类回归分析（softmax层），输出类别预测。详细介绍可参考[http://blog.csdn.net/xuanyuansen/article/details/41800721][http_blog.csdn.net_xuanyuansen_article_details_41800721]  
![这里写图片描述][SouthEast]

### 2.3D CNN ###

原理上即参照2D CNN 把各个变量增加一维。

### 3. Tensorflow 上实现 ###

其中，confusionMatrix 表示用于计算结果的类，可按批处理。  
a.3D\_cnn.py

#!/usr/bin/env python2
    # -*- coding: utf-8 -*-
    """
    Created on Thu Feb 23 10:51:28 2017
    
    @author: cdn
    """
    
    
    
    import numpy as np
    np.random.seed(1234)
    import timeit
    import os
    import matplotlib.pyplot as plt
    
    from sklearn.cross_validation import StratifiedKFold
    
    import tensorflow as tf
    from tensorflow.contrib.layers import fully_connected, convolution2d, flatten, dropout
    from tensorflow.python.layers.pooling import max_pooling3d
    from tensorflow.python.ops.nn import relu,softmax
    from tensorflow.python.framework.ops import reset_default_graph
    from ConfusionMatrix import ConfusionMatrix
    
    
    def onehot(t, num_classes):
        out = np.zeros((t.shape[0], num_classes))
        for row, col in enumerate(t):
            out[row, col] = 1
        return out
    
    def load_data(fold_index):
    
        np.random.seed(1234)
        A_data = np.random.uniform(-0.5,1.5, (500, 65, 52, 51)).astype('float32')   
        B_data = np.random.uniform(0,1, (500, 65, 52, 51)).astype('float32')  # load two classes for classfication
    
        A_num,sizeX,sizeY,sizeZ = A_data.shape
        B_num,_,_,_ = B_data.shape
        size_input = [1,sizeX,sizeY,sizeZ]
        np.random.seed(1234)
        random_idx = np.random.permutation(A_num+B_num)
        all_data = np.concatenate((A_data,B_data),axis=0)[random_idx]
        labels = np.hstack((np.ones((A_num,)),np.zeros((B_num,))))[random_idx]
    
        nn =5
        skf = StratifiedKFold(labels,nn)
        train_id = ['']*nn
        test_id = ['']*nn
        a = 0
        for train,test in skf:
            train_id[a] = train
            test_id[a] = test
            a = a+1
        testid = test_id[fold_index]
        validid = test_id[fold_index-1]
        trainid = list(set(train_id[fold_index])-set(validid))
        x_train = all_data[trainid]
        y_train = labels[trainid]
        x_test = all_data[testid]
        y_test = labels[testid]
        x_valid = all_data[validid]
        y_valid = labels[validid] 
        return x_train,y_train,x_test,y_test,x_valid,y_valid,size_input
    
    
    n_fold = 5
    train_accuracy = np.zeros((n_fold,))    
    test_accuracy = np.zeros((n_fold,))
    valid_accuracy = np.zeros((n_fold,))  
    
    t1_time = timeit.default_timer()
    #for fi in range(n_fold):
    num_classes = 2
    num_filters_conv1 = 10
    num_filters_conv2 = 25
    num_filters_conv3 = 40
    num_filters_conv4 = 40
    dense_num = 100
    size_conv = 3 # [height, width]
    pool_size = 2
    batch_size = 5
    nb_epoch = 50
    fi = 0
    X_train,y_train,X_test,y_test,X_val,y_val,size_input = load_data(fi)
    X_train = X_train.reshape(X_train.shape[0], 1, X_train.shape[1], X_train.shape[2],X_train.shape[3])  
    X_val = X_val.reshape(X_val.shape[0], 1, X_val.shape[1],X_val.shape[2],X_val.shape[3])  
    X_test = X_test.reshape(X_test.shape[0], 1, X_test.shape[1], X_test.shape[2],X_test.shape[3])  
    print('X_train shape:', X_train.shape)  
    print(X_train.shape[0], 'train samples')  
    print(X_val.shape[0], 'validate samples')  
    print(X_test.shape[0], 'test samples')  
    
    
    
    train_accuracy = np.zeros((n_fold,))    
    test_accuracy = np.zeros((n_fold,))
    valid_accuracy = np.zeros((n_fold,))  
    t1_time = timeit.default_timer()
    for fi in range(n_fold):  
    
        print('Now running on fold %d'%(fi+1))
        num_classes = 2
        x_train,y_train,x_test,y_test,x_valid,y_valid,size_input = load_data(fi)
        nchannels,rows,cols,deps = size_input
        x_train = x_train.astype('float32')
        x_train = x_train.reshape((-1,nchannels,rows,cols,deps))
        targets_train = y_train.astype('int32')
    
        x_valid = x_valid.astype('float32')
        x_valid = x_valid.reshape((-1,nchannels,rows,cols,deps))
        targets_valid = y_valid.astype('int32')
    
        x_test = x_test.astype('float32')
        x_test = x_test.reshape((-1,nchannels,rows,cols,deps))
        targets_test = y_test.astype('int32')
    
        # define a simple feed forward neural network
    
        # hyperameters of the model
        num_classes = 2
        channels = x_train.shape[1]
        height = x_train.shape[2]
        width = x_train.shape[3]
        depth = x_train.shape[4]
    
        num_filters_conv1 = 10
        num_filters_conv2 = 25
        num_filters_conv3 = 40
        num_filters_conv4 = 40
        kernel_size_conv1 = [3, 3, 3] # [height, width]
        pool_size = [2,2,2]
        stride_conv1 = [1,1,1] # [stride_height, stride_width]
        num_l1 = 100
        # resetting the graph ...
        reset_default_graph()
    
        # Setting up placeholder, this is where your data enters the graph!
        x_pl = tf.placeholder(tf.float32, [None, channels, height, width, depth])
        l_reshape = tf.transpose(x_pl, [0, 2, 3, 4, 1]) # TensorFlow uses NHWC instead of NCHW
        is_training = tf.placeholder(tf.bool)#used for dropout
    
        # Building the layers of the neural network
        # we define the variable scope, so we more easily can recognise our variables later
        l_conv1 = convolution2d(l_reshape, num_filters_conv1, kernel_size_conv1, stride_conv1,activation_fn=relu, scope="l_conv1")
    
        l_maxpool1 = max_pooling3d(l_conv1,pool_size,pool_size)
    
        l_conv2 = convolution2d(l_maxpool1, num_filters_conv2, kernel_size_conv1, stride_conv1,activation_fn=relu,scope="l_conv2")
    
        l_maxpool2 = max_pooling3d(l_conv2,pool_size,pool_size)
    
        l_conv3 = convolution2d(l_maxpool2, num_filters_conv3, kernel_size_conv1, stride_conv1,activation_fn=relu,scope="l_conv3")
    
        l_maxpool3 = max_pooling3d(l_conv3,pool_size,pool_size)
    
        l_conv4 = convolution2d(l_maxpool3, num_filters_conv4, kernel_size_conv1, stride_conv1,activation_fn=relu,scope="l_conv4")
    
        l_flatten = flatten(l_conv4, scope="flatten") # use l_conv1 instead of l_reshape
    
        l1 = fully_connected(l_flatten, num_l1, activation_fn=relu, scope="l1")
    
        l1 = dropout(l1, is_training=is_training, scope="dropout")
    
        y = fully_connected(l1, num_classes, activation_fn=softmax, scope="y")
    
        # y_ is a placeholder variable taking on the value of the target batch.
        y_ = tf.placeholder(tf.float32, [None, num_classes])
    
        # computing cross entropy per sample
        cross_entropy = -tf.reduce_sum(y_ * tf.log(y+1e-8), reduction_indices=[1])
    
        # averaging over samples
        cross_entropy = tf.reduce_mean(cross_entropy)
    
        # defining our optimizer
        optimizer = tf.train.AdamOptimizer(learning_rate=0.001)
    
        # applying the gradients
        train_op = optimizer.minimize(cross_entropy)
    
        #Test the forward pass
    #    x = np.random.normal(0,1, (45, 1,65, 52, 51)).astype('float32') #dummy data
    
        # restricting memory usage, TensorFlow is greedy and will use all memory otherwise
        gpu_opts = tf.GPUOptions(per_process_gpu_memory_fraction=0.2)
        # initialize the Session
        sess = tf.Session(config=tf.ConfigProto(gpu_options=gpu_opts))
        sess.run(tf.global_variables_initializer())
    #    res = sess.run(fetches=[y], feed_dict={x_pl: x})
    #    res = sess.run(fetches=[y], feed_dict={x_pl: x, is_training: False}) # for when using dropout
    #    print "y", res[0].shape
    
        #Training Loop
        from confusionmatrix import ConfusionMatrix
        batch_size = 10
        num_epochs = 50
        num_samples_train = x_train.shape[0]
        num_batches_train = num_samples_train // batch_size
        num_samples_valid = x_valid.shape[0]
        num_batches_valid = num_samples_valid // batch_size
    
        train_acc, train_loss = [], []
        valid_acc, valid_loss = [], []
        test_acc, test_loss = [], []
        cur_loss = 0
        loss = []
    
        try:
            for epoch in range(num_epochs):
                #Forward->Backprob->Update params
                cur_loss = 0
                for i in range(num_batches_train):
                    idx = range(i*batch_size, (i+1)*batch_size)
                    x_batch = x_train[idx]
                    target_batch = targets_train[idx]
    #                feed_dict_train = {x_pl: x_batch, y_: onehot(target_batch, num_classes)}
                    feed_dict_train = {x_pl: x_batch, y_: onehot(target_batch, num_classes), is_training: True}
                    fetches_train = [train_op, cross_entropy]
                    res = sess.run(fetches=fetches_train, feed_dict=feed_dict_train)
                    batch_loss = res[1] #this will do the complete backprob pass
                    cur_loss += batch_loss
                loss += [cur_loss/batch_size]
    
                confusion_valid = ConfusionMatrix(num_classes)
                confusion_train = ConfusionMatrix(num_classes)
    
                for i in range(num_batches_train):
                    idx = range(i*batch_size, (i+1)*batch_size)
                    x_batch = x_train[idx]
                    targets_batch = targets_train[idx]
                    # what to feed our accuracy op
    #                feed_dict_eval_train = {x_pl: x_batch}
                    feed_dict_eval_train = {x_pl: x_batch, is_training: False}
                    # deciding which parts to fetch
                    fetches_eval_train = [y]
                    # running the validation
                    res = sess.run(fetches=fetches_eval_train, feed_dict=feed_dict_eval_train)
                    # collecting and storing predictions
                    net_out = res[0] 
                    preds = np.argmax(net_out, axis=-1) 
                    confusion_train.batch_add(targets_batch, preds)
    
                confusion_valid = ConfusionMatrix(num_classes)
                for i in range(num_batches_valid):
                    idx = range(i*batch_size, (i+1)*batch_size)
                    x_batch = x_valid[idx]
                    targets_batch = targets_valid[idx]
                    # what to feed our accuracy op
    #                feed_dict_eval_train = {x_pl: x_batch}
                    feed_dict_eval_train = {x_pl: x_batch, is_training: False}
                    # deciding which parts to fetch
                    fetches_eval_train = [y]
                    # running the validation
                    res = sess.run(fetches=fetches_eval_train, feed_dict=feed_dict_eval_train)
                    # collecting and storing predictions
                    net_out = res[0]
                    preds = np.argmax(net_out, axis=-1) 
    
                    confusion_valid.batch_add(targets_batch, preds)
    
                train_acc_cur = confusion_train.accuracy()
                valid_acc_cur = confusion_valid.accuracy()
    
                train_acc += [train_acc_cur]
                valid_acc += [valid_acc_cur]
                print "Epoch %i : Train Loss %e , Train acc %f,  Valid acc %f " \
                % (epoch+1, loss[-1], train_acc_cur, valid_acc_cur)
        except KeyboardInterrupt:
            pass
    
    
        #get test set score
        confusion_test = ConfusionMatrix(num_classes)
    
        # what to feed our accuracy op
    #    feed_dict_eval_train = {x_pl: x_test}
        feed_dict_eval_train = {x_pl: x_test, is_training: False}
        # deciding which parts to fetch
        fetches_eval_train = [y]
        # running the validation
        res = sess.run(fetches=fetches_eval_train, feed_dict=feed_dict_eval_train)
        # collecting and storing predictions
        net_out = res[0] 
        preds = np.argmax(net_out, axis=-1) 
        confusion_test.batch_add(targets_test, preds)
        print "\nTest set Acc:  %f" %(confusion_test.accuracy())
        test_acc = confusion_test.accuracy()
    
        epoch = np.arange(len(train_acc))
        plt.figure()
        plt.plot(epoch,train_acc,'r',epoch,valid_acc,'b')
        plt.legend(['Train Acc','Val Acc'])
        plt.xlabel('Epochs'), plt.ylabel('Acc'), plt.ylim([0.2,1.03])
        plt.show()
        train_accuracy[fi] = train_acc[-1]
        test_accuracy[fi] = test_acc
        valid_accuracy[fi] = valid_acc[-1]
    
    print '\nMean accuray of test set: %f %%' %(np.mean(test_accuracy)*100)
    
    t2_time = timeit.default_timer()
    print(('The code for Tensorflow ' +os.path.split(__file__)[1] +' ran for %.2fm' % ((t2_time - t1_time) / 60.)))

b. ConfusionMatrix.py

import numpy as np
    
    
    class ConfusionMatrix:
        """
           Simple confusion matrix class
           row is the true class, column is the predicted class
        """
        def __init__(self, num_classes, class_names=None):
            self.n_classes = num_classes
            if class_names is None:
                self.class_names = map(str, range(num_classes))
            else:
                self.class_names = class_names
    
            # find max class_name and pad
            max_len = max(map(len, self.class_names))
            self.max_len = max_len
            for idx, name in enumerate(self.class_names):
                if len(self.class_names) < max_len:
                    self.class_names[idx] = name + " "*(max_len-len(name))
    
            self.mat = np.zeros((num_classes,num_classes),dtype='int')
    
        def __str__(self):
            # calucate row and column sums
            col_sum = np.sum(self.mat, axis=1)
            row_sum = np.sum(self.mat, axis=0)
    
            s = []
    
            mat_str = self.mat.__str__()
            mat_str = mat_str.replace('[','').replace(']','').split('\n')
    
            for idx, row in enumerate(mat_str):
                if idx == 0:
                    pad = " "
                else:
                    pad = ""
                class_name = self.class_names[idx]
                class_name = " " + class_name + " |"
                row_str = class_name + pad + row
                row_str += " |" + str(col_sum[idx])
                s.append(row_str)
    
            row_sum = [(self.max_len+4)*" "+" ".join(map(str, row_sum))]
            hline = [(1+self.max_len)*" "+"-"*len(row_sum[0])]
    
            s = hline + s + hline + row_sum
    
            # add linebreaks
            s_out = [line+'\n' for line in s]
            return "".join(s_out)
    
        def batch_add(self, targets, preds):
            assert targets.shape == preds.shape
            assert len(targets) == len(preds)
            assert max(targets) < self.n_classes
            assert max(preds) < self.n_classes
            targets = targets.flatten()
            preds = preds.flatten()
            for i in range(len(targets)):
                    self.mat[targets[i], preds[i]] += 1
    
        def get_errors(self):
            tp = np.asarray(np.diag(self.mat).flatten(),dtype='float')
            fn = np.asarray(np.sum(self.mat, axis=1).flatten(),dtype='float') - tp
            fp = np.asarray(np.sum(self.mat, axis=0).flatten(),dtype='float') - tp
            tn = np.asarray(np.sum(self.mat)*np.ones(self.n_classes).flatten(),
                            dtype='float') - tp - fn - fp
            return tp, fn, fp, tn
    
        def accuracy(self):
            """
            Calculates global accuracy
            :return: accuracy
            :example: >>> conf = ConfusionMatrix(3)
                      >>> conf.batchAdd([0,0,1],[0,0,2])
                      >>> print conf.accuracy()
            """
            tp, _, _, _ = self.get_errors()
            n_samples = np.sum(self.mat)
            return np.sum(tp) / n_samples
    
        def sensitivity(self):
            tp, tn, fp, fn = self.get_errors()
            res = tp / (tp + fn)
            res = res[~np.isnan(res)]
            return res
    
        def specificity(self):
            tp, tn, fp, fn = self.get_errors()
            res = tn / (tn + fp)
            res = res[~np.isnan(res)]
            return res
    
        def positive_predictive_value(self):
            tp, tn, fp, fn = self.get_errors()
            res = tp / (tp + fp)
            res = res[~np.isnan(res)]
            return res
    
        def negative_predictive_value(self):
            tp, tn, fp, fn = self.get_errors()
            res = tn / (tn + fn)
            res = res[~np.isnan(res)]
            return res
    
        def false_positive_rate(self):
            tp, tn, fp, fn = self.get_errors()
            res = fp / (fp + tn)
            res = res[~np.isnan(res)]
            return res
    
        def false_discovery_rate(self):
            tp, tn, fp, fn = self.get_errors()
            res = fp / (tp + fp)
            res = res[~np.isnan(res)]
            return res
    
        def F1(self):
            tp, tn, fp, fn = self.get_errors()
            res = (2*tp) / (2*tp + fp + fn)
            res = res[~np.isnan(res)]
            return res
    
        def matthews_correlation(self):
            tp, tn, fp, fn = self.get_errors()
            numerator = tp*tn - fp*fn
            denominator = np.sqrt((tp + fp)*(tp + fn)*(tn + fp)*(tn + fn))
            res = numerator / denominator
            res = res[~np.isnan(res)]
            return res

[https_cn.udacity.com_course_deep-learning_ud730]: https://cn.udacity.com/course/deep-learning--ud730
[http_blog.csdn.net_xuanyuansen_article_details_41800721]: http://blog.csdn.net/xuanyuansen/article/details/41800721
[SouthEast]: /images/20220709/b70a0300450a4dc493422d45407eac77.png