构建一个基于CNN + BiLSTM + Attention机制的滚动轴承寿命预测系统。PHM轴承寿命预测—python 这个系统将包括数据处理、模型构建和训练步骤

如何 ——PHM轴承寿命预测—python
构建_基于cnn+bilstm+att做的轴承寿命预测，包含数据处理，模型构建。如何根据自己的需要更改模型。

以下文字及代码仅供参考

构建一个基于CNN + BiLSTM + Attention机制的滚动轴承寿命预测系统。这个系统将包括数据处理、模型构建和训练步骤，并且可以轻松地根据需要进行调整。以下是完整的代码实现，

1. 数据准备

假设我们有一个滚动轴承振动信号数据集，其中包含正常和故障样本。数据集格式如下：

datasets/
└── bearing_dataset/
    ├── train/
    │   ├── normal/
    │   ├── outer_race_fault/
    │   ├── inner_race_fault/
    │   └── ball_fault/
    ├── val/
    │   ├── normal/
    │   ├── outer_race_fault/
    │   ├── inner_race_fault/
    │   └── ball_fault/
    └── test/
        ├── normal/
        ├── outer_race_fault/
        ├── inner_race_fault/
        └── ball_fault/

每个文件夹中包含多个.npy文件，每个文件是一个形状为 (timesteps, features) 的数组。

2. 完整代码实现

以下是完整的Python代码，包括数据加载、预处理、模型构建、训练和评估。您可以直接运行此代码。

[<title="Complete Bearing Lifespan Prediction with CNN + BiLSTM + Attention">]
import os
import numpy as np
from sklearn.preprocessing import StandardScaler
from keras.utils import to_categorical
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Conv1D, MaxPooling1D, Bidirectional, LSTM, Dense, Flatten, Dropout, Multiply, Permute, Activation, Reshape, Lambda
import matplotlib.pyplot as plt

# Step 1: Data Loading and Preprocessing
def load_data(data_dir, classes):
    X = []
    y = []
    
    for label, cls in enumerate(classes):
        class_dir = os.path.join(data_dir, cls)
        files = os.listdir(class_dir)
        for file in files:
            filepath = os.path.join(class_dir, file)
            data = np.load(filepath)
            X.append(data)
            y.append(label)
    
    X = np.array(X)
    y = np.array(y)
    
    return X, y

def preprocess_data(X_train, X_val, X_test):
    scaler = StandardScaler()
    n_timesteps, n_features = X_train.shape[1], X_train.shape[2]
    
    X_train = scaler.fit_transform(X_train.reshape(-1, n_features)).reshape(-1, n_timesteps, n_features)
    X_val = scaler.transform(X_val.reshape(-1, n_features)).reshape(-1, n_timesteps, n_features)
    X_test = scaler.transform(X_test.reshape(-1, n_features)).reshape(-1, n_timesteps, n_features)
    
    return X_train, X_val, X_test

# Load and preprocess classification data
classes = ['normal', 'outer_race_fault', 'inner_race_fault', 'ball_fault']
data_dir = 'datasets/bearing_dataset'

X_train_class, y_train_class = load_data(os.path.join(data_dir, 'train'), classes)
X_val_class, y_val_class = load_data(os.path.join(data_dir, 'val'), classes)
X_test_class, y_test_class = load_data(os.path.join(data_dir, 'test'), classes)

y_train_class = to_categorical(y_train_class, num_classes=len(classes))
y_val_class = to_categorical(y_val_class, num_classes=len(classes))
y_test_class = to_categorical(y_test_class, num_classes=len(classes))

X_train_class, X_val_class, X_test_class = preprocess_data(X_train_class, X_val_class, X_test_class)

# Step 2: Build Classification Model (CNN + BiLSTM + Attention)
def attention_3d_block(inputs):
    # inputs.shape = (batch_size, time_steps, input_dim)
    TIME_STEPS = inputs.shape[1]
    SINGLE_ATTENTION_VECTOR = False
    
    attention = Dense(1, activation='tanh')(inputs)
    attention = Flatten()(attention)
    attention = Activation('softmax')(attention)
    if SINGLE_ATTENTION_VECTOR:
        attention = RepeatVector(TIME_STEPS)(attention)
        attention = Permute([2, 1])(attention)
        attention_mul = Multiply()([inputs, attention])
    else:
        attention_mul = Multiply()([inputs, attention])
        attention_mul = Lambda(lambda xin: K.sum(xin, axis=-1), output_shape=(TIME_STEPS,))(attention_mul)
    
    return attention_mul

def build_classification_model(input_shape, num_classes):
    inputs = Input(shape=input_shape)
    
    # CNN layer
    conv1 = Conv1D(filters=64, kernel_size=5, activation='relu', padding='same')(inputs)
    pool1 = MaxPooling1D(pool_size=2)(conv1)
    dropout1 = Dropout(0.2)(pool1)
    
    # BiLSTM layer
    bilstm = Bidirectional(LSTM(units=64, return_sequences=True))(dropout1)
    
    # Attention layer
    attention = attention_3d_block(bilstm)
    
    # Fully connected layers
    flatten = Flatten()(attention)
    dense1 = Dense(64, activation='relu')(flatten)
    dropout2 = Dropout(0.2)(dense1)
    outputs = Dense(num_classes, activation='softmax')(dropout2)
    
    model = Model(inputs=inputs, outputs=outputs)
    model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])
    
    return model

# Build the classification model
input_shape_class = (X_train_class.shape[1], X_train_class.shape[2])
num_classes = len(classes)
classification_model = build_classification_model(input_shape_class, num_classes)
classification_model.summary()

# Step 3: Train Classification Model
history_class = classification_model.fit(
    X_train_class, y_train_class,
    validation_data=(X_val_class, y_val_class),
    epochs=50,
    batch_size=32,
    verbose=1
)

# Evaluate the classification model
loss_class, accuracy_class = classification_model.evaluate(X_test_class, y_test_class, verbose=0)
print(f'Classification Test Accuracy: {accuracy_class:.4f}')

# Step 4: Visualize Classification Results
plt.figure(figsize=(12, 4))
plt.subplot(1, 2, 1)
plt.plot(history_class.history['accuracy'])
plt.plot(history_class.history['val_accuracy'])
plt.title('Classification Model Accuracy')
plt.ylabel('Accuracy')
plt.xlabel('Epoch')
plt.legend(['Train', 'Validation'], loc='upper left')

plt.subplot(1, 2, 2)
plt.plot(history_class.history['loss'])
plt.plot(history_class.history['val_loss'])
plt.title('Classification Model Loss')
plt.ylabel('Loss')
plt.xlabel('Epoch')
plt.legend(['Train', 'Validation'], loc='upper left')

plt.show()

# Step 5: Load Lifetime Prediction Dataset
# Assume CSV format with columns 'signal' and 'remaining_life'
data_lifetime = pd.read_csv('datasets/lifetime_prediction.csv')

# Extract features and labels
X_lifetime = data_lifetime['signal'].apply(lambda x: np.fromstring(x.strip("[]"), sep=',')).values.tolist()
y_lifetime = data_lifetime['remaining_life'].values

X_lifetime = np.array(X_lifetime)
scaler = StandardScaler()
X_lifetime = scaler.fit_transform(X_lifetime).reshape(-1, X_lifetime.shape[1], 1)

# Split into train and test sets
from sklearn.model_selection import train_test_split

X_train_lifetime, X_test_lifetime, y_train_lifetime, y_test_lifetime = train_test_split(X_lifetime, y_lifetime, test_size=0.2, random_state=42)

# Step 6: Build Regression Model (CNN + BiLSTM + Attention)
def build_regression_model(input_shape):
    inputs = Input(shape=input_shape)
    
    # CNN layer
    conv1 = Conv1D(filters=64, kernel_size=5, activation='relu', padding='same')(inputs)
    pool1 = MaxPooling1D(pool_size=2)(conv1)
    dropout1 = Dropout(0.2)(pool1)
    
    # BiLSTM layer
    bilstm = Bidirectional(LSTM(units=64, return_sequences=True))(dropout1)
    
    # Attention layer
    attention = attention_3d_block(bilstm)
    
    # Fully connected layers
    flatten = Flatten()(attention)
    dense1 = Dense(64, activation='relu')(flatten)
    dropout2 = Dropout(0.2)(dense1)
    outputs = Dense(1)(dropout2)  # Regression output
    
    model = Model(inputs=inputs, outputs=outputs)
    model.compile(optimizer='adam', loss='mse', metrics=['mae'])
    
    return model

# Build the regression model
input_shape_lifetime = (X_train_lifetime.shape[1], X_train_lifetime.shape[2])
regression_model = build_regression_model(input_shape_lifetime)
regression_model.summary()

# Step 7: Train Regression Model
history_reg = regression_model.fit(
    X_train_lifetime, y_train_lifetime,
    validation_split=0.2,
    epochs=50,
    batch_size=32,
    verbose=1
)

# Evaluate the regression model
loss_reg, mae_reg = regression_model.evaluate(X_test_lifetime, y_test_lifetime, verbose=0)
print(f'Regression Test MAE: {mae_reg:.4f}')

# Step 8: Visualize Regression Results
# Plot training & validation loss values
plt.figure(figsize=(12, 4))
plt.plot(history_reg.history['loss'])
plt.plot(history_reg.history['val_loss'])
plt.title('Regression Model Loss')
plt.ylabel('Mean Absolute Error')
plt.xlabel('Epoch')
plt.legend(['Train', 'Validation'], loc='upper left')
plt.show()

# Predict remaining life
predictions = regression_model.predict(X_test_lifetime)

# Plot true vs predicted remaining life
plt.figure(figsize=(12, 4))
plt.scatter(y_test_lifetime, predictions.flatten())
plt.title('True vs Predicted Remaining Life')
plt.xlabel('True Remaining Life')
plt.ylabel('Predicted Remaining Life')
plt.plot([min(y_test_lifetime), max(y_test_lifetime)], [min(y_test_lifetime), max(y_test_lifetime)], color='red')
plt.show()

说明

1. 数据加载和预处理：

   • load_data 函数用于从文件夹中加载 .npy 文件。
   • preprocess_data 函数用于标准化特征数据。

2. 分类模型构建：

   • 使用CNN提取局部特征。
   • 使用BiLSTM捕捉时间序列中的长依赖关系。
   • 使用Attention机制关注重要的特征部分。
   • 最后通过全连接层输出分类结果。

3. 分类模型训练：

   • 训练分类模型并评估其性能。

4. 分类结果可视化：

   • 绘制训练和验证过程中的准确性及损失曲线。

5. 寿命预测数据加载和预处理：

   • 加载寿命预测数据集（CSV格式），提取特征和标签。
   • 标准化特征数据并分割为训练集和测试集。

6. 回归模型构建：

   • 使用相同的架构（CNN + BiLSTM + Attention）进行回归任务。
   • 输出剩余寿命值。

7. 回归模型训练：

   • 训练回归模型并评估其性能。

8. 回归结果可视化：

   • 绘制训练和验证过程中的MAE曲线。
   • 绘制真实与预测的剩余寿命散点图。

运行步骤

1. 确保数据集路径正确：

   • 将您的数据集放在 datasets/bearing_dataset 目录下。
   • 确保 lifetime_prediction.csv 文件存在并且格式正确。

2. 安装必要的库：

   • 确保您已经安装了所需的库，如 numpy, pandas, tensorflow, matplotlib 等。
   • 您可以使用以下命令安装这些库：

     pip install numpy pandas tensorflow matplotlib scikit-learn
     

3. 运行代码：
   按照上面程序步骤 - 直接运行上述完整的代码即可完成数据加载、预处理、模型构建、训练和评估。