C-MAPSS数据集入门指南：从数据加载到故障预测模型构建

1次阅读

共计 3196 个字符，预计需要花费 8 分钟才能阅读完成。

C-MAPSS（Commercial Modular Aero-Propulsion System Simulation）是 NASA 开发的航空发动机故障预测基准数据集，广泛应用于剩余使用寿命（RUL）预测研究。对初学者而言，这个数据集有三大典型挑战：

多维时间序列理解困难：每个发动机包含多个传感器在不同周期的读数，形成复杂的三维结构（发动机×时间×传感器）
运行条件干扰：不同操作模式（如飞行高度、推力）会显著影响传感器读数
故障标记稀疏：只有在发动机完全失效时才有明确标签，中间状态需要自行推断

数据集包含 4 个子数据集（FD001-FD004），每个子集包含以下文件：

文件类型	内容说明
train.txt	训练数据 – 发动机编号、周期计数、3 个操作条件和 21 个传感器值
test.txt	测试数据 – 结构与训练集相同，但缺少 RUL 标签
RUL.txt	测试集中每个发动机的最终 RUL 值

关键字段说明：
– unit_number：发动机编号（同一台发动机有多条时序记录）
– time_cycles：运行周期计数
– op_setting_[1-3]：操作条件（标准化到 [0,1] 范围）
– sensor_[1-21]：传感器读数（部分传感器已做归一化处理）

import pandas as pd
import numpy as np

# 加载训练数据并添加列名
cols = ['unit', 'time'] + [f'op_{i}' for i in range(1,4)] + [f'sensor_{i}' for i in range(1,22)]
train = pd.read_csv('train_FD001.txt', sep=' ', header=None, names=cols)

# 清理多余的空白列
train.dropna(axis=1, how='all', inplace=True)

# 计算每个发动机的最大周期（即寿命）max_cycle = train.groupby('unit')['time'].max().reset_index()
max_cycle.columns = ['unit', 'max_cycle']

# 合并回原始数据并计算 RUL
train = train.merge(max_cycle, on='unit')
train['RUL'] = train['max_cycle'] - train['time']

import matplotlib.pyplot as plt
import seaborn as sns

# 选择特定发动机的传感器趋势
engine_id = 1
example_engine = train[train['unit'] == engine_id]

plt.figure(figsize=(12, 6))
for sensor in ['sensor_2', 'sensor_7', 'sensor_11']:
    plt.plot(example_engine['time'], example_engine[sensor], label=sensor)
plt.xlabel('Time Cycles')
plt.ylabel('Sensor Value')
plt.title(f'Sensor Trends for Engine {engine_id}')
plt.legend()
plt.show()

# 滑动窗口特征计算
def rolling_features(df, window=5):
    sensors = [col for col in df.columns if 'sensor_' in col]

    for sensor in sensors:
        df[f'{sensor}_rolling_mean'] = df.groupby('unit')[sensor].transform(lambda x: x.rolling(window=window).mean())
        df[f'{sensor}_rolling_std'] = df.groupby('unit')[sensor].transform(lambda x: x.rolling(window=window).std())

    return df.dropna()

train_processed = rolling_features(train)

import torch
import torch.nn as nn
from sklearn.preprocessing import MinMaxScaler

# 数据预处理
scaler = MinMaxScaler()
X = scaler.fit_transform(train_processed.drop(['unit', 'time', 'max_cycle'], axis=1))
y = train_processed['RUL'].values

# 转换为 3D 张量 [samples, timesteps, features]
def create_sequences(data, targets, seq_length=30):
    X_seq, y_seq = [], []
    for unit in train_processed['unit'].unique():
        unit_data = data[train_processed['unit'] == unit]
        unit_target = targets[train_processed['unit'] == unit]

        for i in range(len(unit_data) - seq_length):
            X_seq.append(unit_data[i:i+seq_length])
            y_seq.append(unit_target[i+seq_length-1])

    return np.array(X_seq), np.array(y_seq)

X_seq, y_seq = create_sequences(X, y)

# 定义 LSTM 模型
class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.linear = nn.Linear(hidden_size, 1)

    def forward(self, x):
        out, _ = self.lstm(x)
        out = self.linear(out[:, -1, :])
        return out

# 实例化模型
model = LSTMModel(input_size=X_seq.shape[2], hidden_size=64, num_layers=2)