采用RNN网络对轴承寿命进行预测（python代码）
·
使用西安交大转子数据集
# -*- coding: utf-8 -*-
"""
Created on Wed Aug 25 19:15:50 2021
@author: pony
"""
import os
import pandas as pd
import numpy as np
import tensorflow as tf
def normal_signal(data):
mean, var = data.mean(axis=0), data.var(axis=0)
data_norm = (data-mean)/np.sqrt(var)
data_norm = data_norm.astype("float32")
return data_norm
def slide_window(norm_arr, window_size):
total_norm = norm_arr.shape[0] # 一共有多少组信号
norm_length = norm_arr.shape[1] # 一组信号的长度
signal_type = norm_arr.shape[2] # 传感器的数量
squence = np.zeros((total_norm-window_size+1, window_size*norm_length, signal_type), dtype=np.float32) # 初始化得到的样本数据集形状
for i in range(total_norm-window_size+1):
squence[i] = np.reshape(norm_arr[i: window_size+i], newshape=(window_size*norm_length, signal_type)) # 每次取连续 window_size 组作为一个样本
return squence
PATH = r'F:\BaiduNetdiskDownload\XJTU-SY_Bearing_Datasets\Data\XJTU-SY_Bearing_Datasets\XJTU-SY_Bearing_Datasets\35Hz12kN'
count = 0
Window_size = 5
Block_length = 2560
for dirname, _, all_csv_list in os.walk(PATH):
if dirname != PATH and count < 4:
for each_dir, _, csv_list in os.walk(dirname): # 取某轴承
train_norm_list = []
for csv_index in range(len(csv_list)):
csv_name = '{}.csv'.format(csv_index+1)
filename = os.path.join(each_dir, csv_name) # csv 文件
total_data_x = pd.read_csv(filename).values # 读取所有时间点信号
index = np.linspace(0, total_data_x.shape[0]-1, Block_length, dtype=int) # 等距索引
data_x = total_data_x[index] # 按索引取点
norm_x = normal_signal(data_x) # 归一化
train_norm_list.append(norm_x) # 将该轴承所有时间的信号写入列表
train_norm_arr = np.array(train_norm_list) # 列表 --> 数组
input_x = slide_window(train_norm_arr, window_size=Window_size) # 滑窗得到该轴承输入
output_y = np.arange(input_x.shape[0]-1, -1, -1)/input_x.shape[0] # 将标签归一化得到该轴承输出
count += 1
if count == 1:
train_input_x = input_x
train_output_y = output_y
else:
train_input_x = np.vstack((train_input_x, input_x))
train_output_y = np.hstack((train_output_y, output_y))
train_input = tf.expand_dims(train_input_x, axis=-2) # 加一个维度使形状变成 BxHx1xC
train_output = tf.expand_dims(train_output_y, axis=-1) # 加一个维度使形状变成 Bx1
dataset_train = tf.data.Dataset.from_tensor_slices((train_input, train_output)).shuffle(train_input.shape[0]).batch(16)
test_norm_list = []
for dirname, _, csv_list in os.walk(PATH+'\\'+'Bearing1_5'):
for csv_index in range(len(csv_list)):
csv_name = '{}.csv'.format(csv_index+1)
filename = os.path.join(dirname, csv_name)
total_data_x = pd.read_csv(filename).values
index = np.linspace(0, total_data_x.shape[0]-1, Block_length, dtype=int)
data_x = total_data_x[index]
norm_x = normal_signal(data_x)
test_norm_list.append(norm_x)
test_norm_arr = np.array(test_norm_list)
test_input_x = slide_window(test_norm_arr, window_size=Window_size)
test_output_y = np.arange(test_input_x.shape[0]-1, -1, -1)/test_input_x.shape[0]
test_input = tf.expand_dims(test_input_x, axis=-2)
test_output = tf.expand_dims(test_output_y, axis=-1)
dataset_test = tf.data.Dataset.from_tensor_slices((test_input, test_output)).batch(test_input.shape[0])
class RCL(tf.keras.Model):
def __init__(self, filters, kernel_size, time_step):
super().__init__()
self.time_step = time_step
self.filters = filters
self.K_r = tf.keras.layers.Conv2D(filters, kernel_size, strides=(1, 1),
padding='same', use_bias=False,
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
self.W_r = tf.keras.layers.Conv2D(filters, kernel_size, strides=(1, 1),
padding='same', use_bias=True,
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
self.K_u = tf.keras.layers.Conv2D(filters, kernel_size, strides=(1, 1),
padding='same', use_bias=False,
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
self.W_u = tf.keras.layers.Conv2D(filters, kernel_size, strides=(1, 1),
padding='same', use_bias=True,
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
self.K_h = tf.keras.layers.Conv2D(filters, kernel_size, strides=(1, 1),
padding='same', use_bias=False,
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
self.W_h = tf.keras.layers.Conv2D(filters, kernel_size, strides=(1, 1),
padding='same', use_bias=True,
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
def call(self, inputs):
X = tf.split(inputs, self.time_step, axis=1)
batch_size = inputs.shape[0]
H = X[0].shape[1]
W = X[0].shape[2]
h = np.random.randn(batch_size, H, W, self.filters)
h = tf.cast(h, tf.float32)
Y = []
for x in X:
r = tf.keras.layers.Activation('sigmoid')(self.K_r(x)+self.W_r(h))
u = tf.keras.layers.Activation('sigmoid')(self.K_u(x)+self.W_u(h))
h_ud = tf.keras.layers.Activation('tanh')(self.K_h(x)+self.W_h(r*h))
y = u*h + (1-u)*h_ud
h = y
Y.append(y)
output = tf.concat(Y, axis=1)
return output
class RCNN(tf.keras.Model):
def __init__(self):
super().__init__()
self.RCL1 = RCL(filters=16, kernel_size=(8, 1), time_step=10)
self.RCL2 = RCL(filters=32, kernel_size=(8, 1), time_step=10)
self.RCL3 = RCL(filters=64, kernel_size=(8, 1), time_step=10)
self.RCL4 = RCL(filters=128, kernel_size=(8, 1), time_step=5)
self.Pool1 = tf.keras.layers.MaxPooling2D(pool_size=(8, 1),
strides=(8, 1),
padding='same')
self.Pool2 = tf.keras.layers.MaxPooling2D(pool_size=(8, 1),
strides=(8, 1),
padding='same')
self.Pool3 = tf.keras.layers.MaxPooling2D(pool_size=(8, 1),
strides=(8, 1),
padding='same')
self.Pool4 = tf.keras.layers.GlobalMaxPooling2D()
self.Dropout1 = tf.keras.layers.Dropout(0.15)
self.Dropout2 = tf.keras.layers.Dropout(0.15)
self.Dropout3 = tf.keras.layers.Dropout(0.15)
self.Dropout4 = tf.keras.layers.Dropout(0.15)
self.Dropout5 = tf.keras.layers.Dropout(0.15)
self.Dropout6 = tf.keras.layers.Dropout(0.15)
self.FCL1 = tf.keras.layers.Dense(units=100, activation='relu',
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
self.FCL2 = tf.keras.layers.Dense(units=100, activation='relu',
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
self.FCL3 = tf.keras.layers.Dense(units=1, activation='sigmoid',
kernel_regularizer=tf.keras.regularizers.l2(10^-5))
def call(self, inputs):
x = self.Pool1(self.Dropout1(self.RCL1(inputs)))
x = self.Pool2(self.Dropout2(self.RCL2(x)))
x = self.Pool3(self.Dropout3(self.RCL3(x)))
x = self.Pool4(self.Dropout4(self.RCL4(x)))
x = self.Dropout5(self.FCL1(x))
x = self.Dropout6(self.FCL2(x))
output = self.FCL3(x)
return output
model = RCNN()
optimizer = tf.keras.optimizers.Adam()
loss_object = tf.keras.losses.MeanSquaredError()
def calc_loss(real, pred):
l = tf.reduce_mean(loss_object(real, pred))
return l
def train_step(x, y):
with tf.GradientTape() as tape:
pred = model(x)
real = y
l = calc_loss(real, pred)
grad = tape.gradient(l, model.trainable_variables)
optimizer.apply_gradients(zip(grad, model.trainable_variables))
return l
EPOCH = 3
loss_train_all = []
for epoch in range(EPOCH):
for i, (x, y) in enumerate(dataset_train):
loss_train = train_step(x, y)
if (i+1) % 10 == 0:
print('第{}次训练中第{}批的误差为{}'.format(epoch+1, i+1, loss_train))
loss_train_all.append(loss_train)
print('第{}次训练后的误差为{}'.format(epoch+1, loss_train))
for i, (x, y) in enumerate(dataset_test):
pred_test = model(x)
real_test = y
loss_test = calc_loss(real_test, pred_test)
print('第{}次训练后，验证集中的损失为{}'.format(epoch+1, loss_test))
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
0
0- 0
已为社区贡献5条内容
所有评论(0)