10-TensorFlow 基于LeNet的轴承故障诊断

1.数据集

数据是来自凯斯西储大学（CWRU）滚动轴承数据中心.
官方网站数据下载地址: https://csegroups.case.edu/bearingdatacenter/pages/download-data-file

CWRU滚动轴承数据采集系统:

CWRU轴承中心数据采集系统如上图所示。本试验的试验对象为图中的驱动端轴承，被诊断的轴承型号为深沟球轴承SKF6205，有故障的轴承由电火花加工制作而成，系统的采样频率为12kHz。被诊断的轴承一共有3种缺陷位置，分别是滚动体损伤，外圈损伤与内圈损伤，损伤直径的大小分别为包括0.007inch，0.014inch和0.021inch，共计9种损伤状态。试验中，每次使用2048个数据点进行诊断。为了便于训练卷积神经网络，对每段信号x均做归一化处理，归一化处理的公式如下式

试验一共准备了4个数据集，如表2-3所示。数据集A、B和C分别是在负载为1hp、2hp和3hp下的数据集。每个数据集各包括6600个训练样本与250个测试样本，其中训练样本采用数据集增强技术，测试样本之间无重叠。数据集D是数据集A、B和C的并集，即包括了3种负载状态，一共有19800个训练样本与750个测试样本。

本项目数据集下载地址: https://download.csdn.net/download/qq_41865229/85200778

试验数据集描述

2.LeNet结构

LeNet由Yann LeCun于1998年提出，卷积网络开篇之作。

3.代码

数据集制作代码 cwru_preprocess.py

from scipy.io import loadmat
import numpy as np
import os
from sklearn import preprocessing  # 0-1编码
from sklearn.model_selection import StratifiedShuffleSplit  # 随机划分，保证每一类比例相同
import tensorflow as tfdef prepro(d_path, length=864, number=1000, normal=True, rate=[0.5, 0.25, 0.25], enc=True, enc_step=28):"""对数据进行预处理,返回train_X, train_Y, valid_X, valid_Y, test_X, test_Y样本.:param d_path: 源数据地址:param length: 信号长度，默认2个信号周期，864:param number: 每种信号个数,总共10类,默认每个类别1000个数据:param normal: 是否标准化.True,Fales.默认True:param rate: 训练集/验证集/测试集比例.默认[0.5,0.25,0.25],相加要等于1:param enc: 训练集、验证集是否采用数据增强.Bool,默认True:param enc_step: 增强数据集采样顺延间隔:return: Train_X, Train_Y, Valid_X, Valid_Y, Test_X, Test_Y```import preprocess.preprocess_nonoise as pretrain_X, train_Y, valid_X, valid_Y, test_X, test_Y = pre.prepro(d_path=path,length=864,number=1000,normal=False,rate=[0.5, 0.25, 0.25],enc=True,enc_step=28)```"""# 获得该文件夹下所有.mat文件名filenames = os.listdir(d_path)def capture(original_path):"""读取mat文件，返回字典:param original_path: 读取路径:return: 数据字典"""files = {}for i in filenames:# 文件路径file_path = os.path.join(d_path, i)file = loadmat(file_path)file_keys = file.keys()for key in file_keys:if 'DE' in key:files[i] = file[key].ravel()return filesdef slice_enc(data, slice_rate=rate[1] + rate[2]):"""将数据切分为前面多少比例，后面多少比例.:param data: 单挑数据:param slice_rate: 验证集以及测试集所占的比例:return: 切分好的数据"""keys = data.keys()Train_Samples = {}Test_Samples = {}for i in keys:slice_data = data[i]all_lenght = len(slice_data)end_index = int(all_lenght * (1 - slice_rate))samp_train = int(number * (1 - slice_rate))  # 700Train_sample = []Test_Sample = []if enc:enc_time = length // enc_stepsamp_step = 0  # 用来计数Train采样次数for j in range(samp_train):random_start = np.random.randint(low=0, high=(end_index - 2 * length))label = 0for h in range(enc_time):samp_step += 1random_start += enc_stepsample = slice_data[random_start: random_start + length]Train_sample.append(sample)if samp_step == samp_train:label = 1breakif label:breakelse:for j in range(samp_train):random_start = np.random.randint(low=0, high=(end_index - length))sample = slice_data[random_start:random_start + length]Train_sample.append(sample)# 抓取测试数据for h in range(number - samp_train):random_start = np.random.randint(low=end_index, high=(all_lenght - length))sample = slice_data[random_start:random_start + length]Test_Sample.append(sample)Train_Samples[i] = Train_sampleTest_Samples[i] = Test_Samplereturn Train_Samples, Test_Samples# 仅抽样完成，打标签def add_labels(train_test):X = []Y = []label = 0for i in filenames:x = train_test[i]X += xlenx = len(x)Y += [label] * lenxlabel += 1return X, Y# one-hot编码def one_hot(Train_Y, Test_Y):Train_Y = np.array(Train_Y).reshape([-1, 1])Test_Y = np.array(Test_Y).reshape([-1, 1])Encoder = preprocessing.OneHotEncoder()Encoder.fit(Train_Y)Train_Y = Encoder.transform(Train_Y).toarray()Test_Y = Encoder.transform(Test_Y).toarray()Train_Y = np.asarray(Train_Y, dtype=np.int32)Test_Y = np.asarray(Test_Y, dtype=np.int32)return Train_Y, Test_Ydef scalar_stand(Train_X, Test_X):# 用训练集标准差标准化训练集以及测试集scalar = preprocessing.StandardScaler().fit(Train_X)Train_X = scalar.transform(Train_X)Test_X = scalar.transform(Test_X)return Train_X, Test_Xdef valid_test_slice(Test_X, Test_Y):test_size = rate[2] / (rate[1] + rate[2])ss = StratifiedShuffleSplit(n_splits=1, test_size=test_size)for train_index, test_index in ss.split(Test_X, Test_Y):X_valid, X_test = Test_X[train_index], Test_X[test_index]Y_valid, Y_test = Test_Y[train_index], Test_Y[test_index]return X_valid, Y_valid, X_test, Y_test# 从所有.mat文件中读取出数据的字典data = capture(original_path=d_path)# 将数据切分为训练集、测试集train, test = slice_enc(data)# 为训练集制作标签，返回X，YTrain_X, Train_Y = add_labels(train)# 为测试集制作标签，返回X，YTest_X, Test_Y = add_labels(test)# 为训练集Y/测试集One-hot标签Train_Y, Test_Y = one_hot(Train_Y, Test_Y)# 训练数据/测试数据 是否标准化.if normal:Train_X, Test_X = scalar_stand(Train_X, Test_X)else:# 需要做一个数据转换，转换成np格式.Train_X = np.asarray(Train_X)Test_X = np.asarray(Test_X)# 将测试集切分为验证集合和测试集.Valid_X, Valid_Y, Test_X, Test_Y = valid_test_slice(Test_X, Test_Y)return Train_X, Train_Y, Valid_X, Valid_Y, Test_X, Test_Yif __name__ == "__main__":path = r'cwru_data\0HP'train_X, train_Y, valid_X, valid_Y, test_X, test_Y = prepro(d_path=path,length=864,number=1000,normal=False,rate=[0.5, 0.25, 0.25],enc=False,enc_step=28)print(train_X[0:5])print(train_Y[0:5])

import tensorflow as tf
import os
import numpy as np
from matplotlib import pyplot as plt
from tensorflow.keras.layers import Conv2D, BatchNormalization, Activation, MaxPool2D, Dropout, Flatten, Dense
from tensorflow.keras import Modelimport cwru_preprocess as preprocess
# 训练参数
batch_size = 128
epochs = 20
num_classes = 10
length = 2048
BatchNorm = True # 是否批量归一化
number = 1000 # 每类样本的数量
normal = True # 是否标准化
rate = [0.8,0.1,0.1] # 测试集验证集划分比例path = r'cwru_data\0HP'
x_train, y_train, x_valid, y_valid, x_test, y_test = preprocess.prepro(d_path=path,length=length,number=number,normal=normal,rate=rate,enc=True, enc_step=28)
#特征数据集转换为4维数组
x_train, x_test = x_train.reshape(8000, 32, 64, 1), x_test.reshape(1000, 32, 64, 1)
#独热码转换为数字标签
y_train, y_test = np.argmax(y_train, axis=-1), np.argmax(y_test, axis=-1)# #归一化, 将数据集列表中的集转化成0-1之间的小数
# x_train, x_test = x_train / 255.0, x_test / 255.0class LeNet5(Model):def __init__(self):super(LeNet5, self).__init__()self.c1 = Conv2D(filters=6, kernel_size=(5, 5),activation='sigmoid')self.p1 = MaxPool2D(pool_size=(2, 2), strides=2)self.c2 = Conv2D(filters=16, kernel_size=(5, 5),activation='sigmoid')self.p2 = MaxPool2D(pool_size=(2, 2), strides=2)self.flatten = Flatten()self.f1 = Dense(120, activation='sigmoid')self.f2 = Dense(84, activation='sigmoid')self.f3 = Dense(10, activation='softmax')def call(self, x):x = self.c1(x)x = self.p1(x)x = self.c2(x)x = self.p2(x)x = self.flatten(x)x = self.f1(x)x = self.f2(x)y = self.f3(x)return ymodel = LeNet5()model.compile(optimizer='adam',loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=False),metrics=['sparse_categorical_accuracy'])checkpoint_save_path = "./lenet/checkpoint/LeNet5.ckpt"
if os.path.exists(checkpoint_save_path + '.index'):print('-------------load the model-----------------')model.load_weights(checkpoint_save_path)cp_callback = tf.keras.callbacks.ModelCheckpoint(filepath=checkpoint_save_path,save_weights_only=True,save_best_only=True)history = model.fit(x_train, y_train, batch_size=32, epochs=5, validation_data=(x_test, y_test), validation_freq=1,callbacks=[cp_callback])
model.summary()# print(model.trainable_variables)
np.set_printoptions(threshold=np.inf)  #数据显示不省略
file = open('./lenet/weights.txt', 'w')
for v in model.trainable_variables:file.write(str(v.name) + '\n')file.write(str(v.shape) + '\n')file.write(str(v.numpy()) + '\n')
file.close()###############################################    show   ################################################ 显示训练集和验证集的acc和loss曲线
acc = history.history['sparse_categorical_accuracy']
val_acc = history.history['val_sparse_categorical_accuracy']
loss = history.history['loss']
val_loss = history.history['val_loss']plt.subplot(1, 2, 1)
plt.plot(acc, label='Training Accuracy')
plt.plot(val_acc, label='Validation Accuracy')
plt.title('Training and Validation Accuracy')
plt.legend()plt.subplot(1, 2, 2)
plt.plot(loss, label='Training Loss')
plt.plot(val_loss, label='Validation Loss')
plt.title('Training and Validation Loss')
plt.legend()
plt.show()

模型训练代码 cwru_lenet_train.py

import tensorflow as tf
import os
import numpy as np
from matplotlib import pyplot as plt
from tensorflow.keras.layers import Conv2D, BatchNormalization, Activation, MaxPool2D, Dropout, Flatten, Dense
from tensorflow.keras import Modelimport cwru_preprocess as preprocess
# 训练参数
batch_size = 128
epochs = 20
num_classes = 10
length = 2048
BatchNorm = True # 是否批量归一化
number = 1000 # 每类样本的数量
normal = True # 是否标准化
rate = [0.8,0.1,0.1] # 测试集验证集划分比例path = r'cwru_data\0HP'
x_train, y_train, x_valid, y_valid, x_test, y_test = preprocess.prepro(d_path=path,length=length,number=number,normal=normal,rate=rate,enc=True, enc_step=28)
#特征数据集转换为4维数组
x_train, x_test = x_train.reshape(8000, 32, 64, 1), x_test.reshape(1000, 32, 64, 1)
#独热码转换为数字标签
y_train, y_test = np.argmax(y_train, axis=-1), np.argmax(y_test, axis=-1)# #归一化, 将数据集列表中的集转化成0-1之间的小数
# x_train, x_test = x_train / 255.0, x_test / 255.0class LeNet5(Model):def __init__(self):super(LeNet5, self).__init__()self.c1 = Conv2D(filters=6, kernel_size=(5, 5),activation='sigmoid')self.p1 = MaxPool2D(pool_size=(2, 2), strides=2)self.c2 = Conv2D(filters=16, kernel_size=(5, 5),activation='sigmoid')self.p2 = MaxPool2D(pool_size=(2, 2), strides=2)self.flatten = Flatten()self.f1 = Dense(120, activation='sigmoid')self.f2 = Dense(84, activation='sigmoid')self.f3 = Dense(10, activation='softmax')def call(self, x):x = self.c1(x)x = self.p1(x)x = self.c2(x)x = self.p2(x)x = self.flatten(x)x = self.f1(x)x = self.f2(x)y = self.f3(x)return ymodel = LeNet5()model.compile(optimizer='adam',loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=False),metrics=['sparse_categorical_accuracy'])checkpoint_save_path = "./lenet/checkpoint/LeNet5.ckpt"
if os.path.exists(checkpoint_save_path + '.index'):print('-------------load the model-----------------')model.load_weights(checkpoint_save_path)cp_callback = tf.keras.callbacks.ModelCheckpoint(filepath=checkpoint_save_path,save_weights_only=True,save_best_only=True)history = model.fit(x_train, y_train, batch_size=32, epochs=5, validation_data=(x_test, y_test), validation_freq=1,callbacks=[cp_callback])
model.summary()# print(model.trainable_variables)
np.set_printoptions(threshold=np.inf)  #数据显示不省略
file = open('./lenet/weights.txt', 'w')
for v in model.trainable_variables:file.write(str(v.name) + '\n')file.write(str(v.shape) + '\n')file.write(str(v.numpy()) + '\n')
file.close()###############################################    show   ################################################ 显示训练集和验证集的acc和loss曲线
acc = history.history['sparse_categorical_accuracy']
val_acc = history.history['val_sparse_categorical_accuracy']
loss = history.history['loss']
val_loss = history.history['val_loss']plt.subplot(1, 2, 1)
plt.plot(acc, label='Training Accuracy')
plt.plot(val_acc, label='Validation Accuracy')
plt.title('Training and Validation Accuracy')
plt.legend()plt.subplot(1, 2, 2)
plt.plot(loss, label='Training Loss')
plt.plot(val_loss, label='Validation Loss')
plt.title('Training and Validation Loss')
plt.legend()
plt.show()

使用模型进行预测 cwru_lenet_predict.py

import tensorflow as tf
import os
import numpy as np
from matplotlib import pyplot as plt
from tensorflow.keras.layers import Conv2D, BatchNormalization, Activation, MaxPool2D, Dropout, Flatten, Dense
from tensorflow.keras import Modelimport cwru_preprocess as preprocess
# 训练参数
batch_size = 128
epochs = 20
num_classes = 10
length = 2048
BatchNorm = True # 是否批量归一化
number = 1000 # 每类样本的数量
normal = True # 是否标准化
rate = [0.8,0.1,0.1] # 测试集验证集划分比例path = r'cwru_data\0HP'
x_train, y_train, x_valid, y_valid, x_test, y_test = preprocess.prepro(d_path=path,length=length,number=number,normal=normal,rate=rate,enc=True, enc_step=28)
#特征数据集转换为4维数组
x_train, x_test, x_valid = x_train.reshape(8000, 32, 64, 1), x_test.reshape(1000, 32, 64, 1), x_valid.reshape(1000, 32, 64, 1)
#独热码转换为数字标签
y_train, y_test, y_valid = np.argmax(y_train, axis=-1), np.argmax(y_test, axis=-1), np.argmax(y_valid, axis=-1)# #归一化, 将数据集列表中的集转化成0-1之间的小数
# x_train, x_test = x_train / 255.0, x_test / 255.0class LeNet5(Model):def __init__(self):super(LeNet5, self).__init__()self.c1 = Conv2D(filters=6, kernel_size=(5, 5),activation='sigmoid')self.p1 = MaxPool2D(pool_size=(2, 2), strides=2)self.c2 = Conv2D(filters=16, kernel_size=(5, 5),activation='sigmoid')self.p2 = MaxPool2D(pool_size=(2, 2), strides=2)self.flatten = Flatten()self.f1 = Dense(120, activation='sigmoid')self.f2 = Dense(84, activation='sigmoid')self.f3 = Dense(10, activation='softmax')def call(self, x):x = self.c1(x)x = self.p1(x)x = self.c2(x)x = self.p2(x)x = self.flatten(x)x = self.f1(x)x = self.f2(x)y = self.f3(x)return ymodel = LeNet5()model.compile(optimizer='adam',loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=False),metrics=['sparse_categorical_accuracy'])checkpoint_save_path = "./lenet/checkpoint/LeNet5.ckpt"
if os.path.exists(checkpoint_save_path + '.index'):print('-------------load the model-----------------')model.load_weights(checkpoint_save_path)###############################################    show   ###############################################loss, accuracy = model.evaluate(x_valid, y_valid)
print('loss=', loss)
print('accuracy=', accuracy)y_pre = model.predict(x_valid[0:10])
#print("------------预测样本数据----------")
#print(x_valid[0:10])
print("------------预测结果概率----------")
print(y_pre)
print("------------预测结果分类----------")
#概率数组转化为标签, 如[0.1, 0.2, 0.7]转化为2
print(np.argmax(y_pre, axis=1))