開始
│
├─ 數(shù)據(jù)加載與預處理
│   ├─ 讀取軸承傳感器CSV文件
│   ├─ 數(shù)據(jù)標準化處理
│   ├─ 創(chuàng)建剩余使用壽命標簽
│   └─ 組織為序列數(shù)據(jù)
│
├─ 模型構建
│   ├─ 殘差卷積網(wǎng)絡
│   │   ├─ 1維卷積層
│   │   ├─ 實例歸一化層
│   │   └─ 殘差連接
│   ├─ 雙向LSTM網(wǎng)絡
│   │   ├─ 前向LSTM層
│   │   ├─ 后向LSTM層
│   │   └─ 激活函數(shù)
│   └─ 回歸網(wǎng)絡
│       ├─ 全連接層
│       ├─ 激活函數(shù)
│       └─ 輸出層
│
├─ 模型訓練
│   ├─ 前向傳播
│   ├─ 損失計算(均方根誤差)
│   ├─ 反向傳播
│   └─ 參數(shù)優(yōu)化(AdamW)
│
├─ 模型評估
│   ├─ 加載測試數(shù)據(jù)
│   ├─ 模型預測
│   ├─ 結果保存
│   └─ 性能可視化
│
├─ 結果可視化
│   ├─ 訓練損失曲線
│   ├─ 預測結果對比
│   └─ 性能分析
│
└─ 結束

# 導入必要的庫
import os  # 用于操作系統(tǒng)相關功能，如文件路徑操作


# 定義函數(shù)獲取軸承數(shù)據(jù)文件夾路徑
def get_bearing_paths(root_dir):
    """
    從根目錄中查找所有符合命名規(guī)范的軸承數(shù)據(jù)文件夾


    參數(shù):
    root_dir: 根目錄路徑


    返回:
    bearing_folders: 符合條件的軸承文件夾路徑列表
    """
    bearing_folders = []  # 存儲軸承文件夾路徑的列表


    # 遍歷根目錄及其所有子目錄
    for root, dirs, files in os.walk(root_dir):
        for dir_name in dirs:
            # 檢查文件夾名是否以'Bearing'開頭
            if dir_name.startswith('Bearing'):
                try:
                    # 提取軸承編號
                    bearing_number = dir_name.split('Bearing')[1]
                    # 將編號分割為主編號和次編號
                    bearing_major, bearing_minor = map(int, bearing_number.split('_'))


                    # 檢查編號是否在有效范圍內(nèi)
                    if 1 <= bearing_major <= 3 and 1 <= bearing_minor <= 5:
                        # 將有效路徑添加到列表
                        bearing_folders.append(os.path.join(root, dir_name))
                except ValueError:
                    # 名稱格式不正確的文件夾忽略
                    continue
    return bearing_folders


# 設置根目錄路徑
root_directory = "datasets"
# 獲取所有軸承數(shù)據(jù)文件夾路徑
bearing_paths = get_bearing_paths(root_directory)


# 打印所有找到的軸承數(shù)據(jù)路徑
for path in bearing_paths:
    print(path)


# 導入數(shù)據(jù)處理和可視化庫
import pandas as pd  # 數(shù)據(jù)處理和分析庫
import os  # 操作系統(tǒng)接口
from tqdm import tqdm  # 進度條顯示庫


# 設置數(shù)據(jù)路徑
data_path = "datasets/37.5Hz11kN/Bearing2_4/" 
# 獲取文件列表
file_list = os.listdir(data_path)


# 按文件名中的數(shù)字排序
file_list = sorted(file_list, key=lambda x: int(x.split('.')[0]))


# 創(chuàng)建空DataFrame存儲所有數(shù)據(jù)
df = pd.DataFrame()
# 遍歷所有文件并加載數(shù)據(jù)
for f in tqdm(file_list):
    # 讀取CSV文件
    temp = pd.read_csv(data_path + f"{f}")
    # 將數(shù)據(jù)添加到DataFrame
    df = pd.concat([df, temp], axis=0)
# 重置索引
df.reset_index(drop=True, inplace=True)
# 顯示DataFrame
df


# 導入數(shù)據(jù)可視化庫
import matplotlib.pyplot as plt  # 數(shù)據(jù)可視化庫


# 創(chuàng)建圖形
plt.figure(figsize=(10, 4))
# 創(chuàng)建水平振動信號子圖
plt.subplot(1, 2, 1)
# 繪制水平振動信號
df["Horizontal_vibration_signals"].plot()
# 設置子圖標題
plt.title("Horizontal Vibration Signals")


# 創(chuàng)建垂直振動信號子圖
plt.subplot(1, 2, 2)
# 繪制垂直振動信號
df["Vertical_vibration_signals"].plot()
# 設置子圖標題
plt.title("Vertical Vibration Signals")


# 顯示圖形
plt.tight_layout()
plt.show()


# 計算數(shù)據(jù)長度
n = len(df)


# 創(chuàng)建剩余使用壽命(RUL)列，從n-1遞減到0
df['RUL'] = range(n-1, -1, -1)


# 計算RUL的最小值和最大值
min_rul = df['RUL'].min()
max_rul = df['RUL'].max()


# 定義RUL歸一化函數(shù)
def normalize_rul(rul, min_rul, max_rul):
    """將RUL值歸一化到0-1范圍"""
    return (rul - min_rul) / (max_rul - min_rul)


# 應用歸一化函數(shù)
df['Normalized_RUL'] = df['RUL'].apply(lambda x: normalize_rul(x, min_rul, max_rul))
# 顯示DataFrame
df


# 導入PyTorch相關庫
import torch  # PyTorch深度學習框架
import torch.nn as nn  # PyTorch神經(jīng)網(wǎng)絡模塊
from torch.utils.data import Dataset, DataLoader  # PyTorch數(shù)據(jù)加載和處理工具


# 定義軸承數(shù)據(jù)集類
class BearingDataset(Dataset):
    def __init__(self, vibration, rul, seq_length):
        """
        初始化軸承數(shù)據(jù)集


        參數(shù):
        vibration: 振動信號數(shù)據(jù)
        rul: 剩余使用壽命數(shù)據(jù)
        seq_length: 序列長度
        """
        self.vibration = vibration
        self.rul = rul
        self.seq_length = seq_length


    def __len__(self):
        """返回數(shù)據(jù)集大小"""
        return len(self.rul) - self.seq_length + 1


    def __getitem__(self, idx):
        """獲取指定索引的數(shù)據(jù)"""
        # 獲取序列數(shù)據(jù)
        x = self.vibration[idx:idx + self.seq_length]
        # 獲取對應的RUL值
        y = self.rul[idx + self.seq_length - 1]


        # 轉換為PyTorch張量
        x = torch.tensor(x, dtype=torch.float32)
        y = torch.tensor(y, dtype=torch.float32)


        return x, y


# 提取特征和標簽
vibration = df[['Horizontal_vibration_signals', 'Vertical_vibration_signals']].values
rul_model = df['Normalized_RUL'].values


# 設置序列長度
seq_length = 128
# 創(chuàng)建數(shù)據(jù)集
dataset = BearingDataset(vibration, rul_model, seq_length)
# 創(chuàng)建數(shù)據(jù)加載器
dataloader = DataLoader(dataset, batch_size=1024, shuffle=True)


# 定義殘差卷積塊
class ResConv1dBlock(nn.Module):
    def __init__(self, channels):
        """
        初始化殘差卷積塊


        參數(shù):
        channels: 輸入輸出通道數(shù)
        """
        super(ResConv1dBlock, self).__init__()
        # 1維卷積層
        self.conv1 = nn.Conv1d(channels, channels, kernel_size=3, stride=1, padding=1)
        # 實例歸一化層
        self.norm1 = nn.InstanceNorm1d(channels)
        # ReLU激活函數(shù)
        self.relu = nn.ReLU(inplace=False)


    def forward(self, x):
        """前向傳播"""
        # 卷積操作
        out = self.conv1(x)
        # 歸一化
        out = self.norm1(out)
        # 殘差連接
        out = out + x
        # 激活函數(shù)
        out = self.relu(out)
        return out


# 定義雙向LSTM模塊
class BiLSTM(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, seq_length, device):
        """
        初始化雙向LSTM模塊


        參數(shù):
        input_size: 輸入特征維度
        hidden_size: 隱藏層維度
        num_layers: LSTM層數(shù)
        seq_length: 序列長度
        device: 計算設備
        """
        super(BiLSTM, self).__init__()
        self.device = device
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.seq_length = seq_length
        # 雙向LSTM層
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, 
                            batch_first=True, bidirectional=True, device=device)
        # ReLU激活函數(shù)
        self.relu = nn.ReLU(inplace=False)


    def forward(self, x):
        """前向傳播"""
        # 初始化隱藏狀態(tài)和細胞狀態(tài)
        h0 = torch.zeros(self.num_layers * 2, x.size(0), self.hidden_size).to(self.device)
        c0 = torch.zeros(self.num_layers * 2, x.size(0), self.hidden_size).to(self.device)
        # LSTM前向傳播
        out, _ = self.lstm(x, (h0, c0))
        # 應用激活函數(shù)
        out = self.relu(out)
        return out


# 定義回歸器模塊
class Regressor(nn.Module):
    def __init__(self, input_size, device):
        """
        初始化回歸器模塊


        參數(shù):
        input_size: 輸入特征維度
        device: 計算設備
        """
        super(Regressor, self).__init__()
        # 全連接層
        self.fc1 = nn.Linear(input_size, 64)
        self.fc2 = nn.Linear(64, 16, device=device)
        self.fc3 = nn.Linear(16, 1, device=device)
        # ReLU激活函數(shù)
        self.relu = nn.ReLU(inplace=False)


    def forward(self, x):
        """前向傳播"""
        # 使用序列的最后一個時間步
        out = self.fc1(x[:, -1, :])
        out = self.relu(out)
        out = self.fc2(out)
        out = self.relu(out)
        out = self.fc3(out)
        return out


# 定義退化模型
class DegradationModel(nn.Module):
    def __init__(self, input_channels=2, hidden_size=64, num_layers=4, seq_length=128, device=torch.device("cuda")):
        """
        初始化退化模型


        參數(shù):
        input_channels: 輸入通道數(shù)
        hidden_size: 隱藏層維度
        num_layers: LSTM層數(shù)
        seq_length: 序列長度
        device: 計算設備
        """
        super(DegradationModel, self).__init__()
        # 殘差卷積塊
        self.conv = ResConv1dBlock(input_channels)
        # 雙向LSTM模塊
        self.bilstm = BiLSTM(input_channels, hidden_size, num_layers, seq_length, device)
        # 回歸器模塊
        self.reg = Regressor(hidden_size*2, device=device)
        self.device = device


    def forward(self, x):
        """前向傳播"""
        # 調(diào)整維度并應用卷積
        x = self.conv(x.permute(0,2,1))
        # 調(diào)整維度并應用雙向LSTM
        x = self.bilstm(x.permute(0,2,1))
        # 應用回歸器
        out = self.reg(x)
        return out


# 導入優(yōu)化器庫
import torch.optim as optim  # PyTorch優(yōu)化器模塊
from tqdm import tqdm  # 進度條顯示庫
import torch.nn.functional as F  # PyTorch函數(shù)模塊


# 設置版本號
ver = "1"


# 初始化模型參數(shù)
input_channels = 2  # 輸入通道數(shù)（水平+垂直振動信號）
hidden_size = 64  # LSTM隱藏層維度
num_layers = 4  # LSTM層數(shù)
device = torch.device('cuda')  # 使用GPU設備


# 創(chuàng)建退化模型實例
rul_model = DegradationModel(input_channels, hidden_size, 
                             num_layers, seq_length, device).to(device)


# 定義損失函數(shù)
def loss_function(pred, target):
    """計算均方根誤差損失"""
    loss = F.mse_loss(pred, target, reduction='mean')
    rmse_loss = torch.sqrt(loss)
    return rmse_loss


# 設置損失函數(shù)
criterion = loss_function
# 設置優(yōu)化器（AdamW）
optimizer = optim.AdamW(rul_model.parameters(), lr=0.001)


# 初始化日志文件
log_file = f'rul_log_{ver}.txt'
with open(log_file, 'w') as f:
    f.write("Training Log\n")


# 設置訓練輪數(shù)
num_epochs = 1000
# 開始訓練循環(huán)
for epoch in range(1, num_epochs+1):
    rul_model.train()  # 設置模型為訓練模式
    total_loss = 0.0  # 初始化總損失


    # 使用進度條遍歷數(shù)據(jù)加載器
    for batch in tqdm(dataloader, desc=f"Epoch: {epoch}"):
        x, y = batch  # 獲取批次數(shù)據(jù)
        x, y = x.to(device), y.view(-1,1).to(device)  # 移動數(shù)據(jù)到設備


        # 模型預測
        pred = rul_model(x)


        # 計算損失
        loss = criterion(pred, y)


        # 反向傳播和優(yōu)化
        optimizer.zero_grad()  # 清零梯度
        loss.backward()  # 反向傳播
        optimizer.step()  # 更新參數(shù)


        total_loss += loss.item()  # 累加損失


    # 計算平均損失
    avg_loss = total_loss / len(dataloader)
    # 創(chuàng)建日志消息
    log_message = f'Epoch {epoch}/{num_epochs}, Loss: {avg_loss}\n'
    print(log_message)


    # 將日志消息寫入文件
    with open(log_file, 'a') as f:
        f.write(log_message)


# 保存訓練好的模型
torch.save(rul_model.state_dict(), f'./models/rul_model_{ver}.pth')


# 導入可視化庫
import matplotlib.pyplot as plt  # 數(shù)據(jù)可視化庫


# 從日志文件加載損失值
log_file = f'rul_log_{ver}.txt'
epochs = []  # 存儲輪數(shù)
losses = []  # 存儲損失值


# 讀取日志文件
with open(log_file, 'r') as f:
    lines = f.readlines()[1:]  # 跳過第一行標題
    for line in lines:
        # 解析每行數(shù)據(jù)
        epoch_info = line.strip().split(',')
        epoch = int(epoch_info[0].split(' ')[1].split('/')[0])
        loss = float(epoch_info[1].split(': ')[1])
        epochs.append(epoch)
        losses.append(loss)


# 可視化損失曲線
plt.figure(figsize=(10, 5))
plt.plot(epochs, losses, label='Training Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('Training Loss Over Epochs')
plt.legend()
plt.grid(True)
plt.show()


# 導入必要的庫
import torch  # PyTorch深度學習框架
import torch.nn as nn  # PyTorch神經(jīng)網(wǎng)絡模塊
import numpy as np  # 數(shù)值計算庫
import pandas as pd  # 數(shù)據(jù)處理和分析庫
import os  # 操作系統(tǒng)接口
from tqdm import tqdm  # 進度條顯示庫


# 定義數(shù)據(jù)加載函數(shù)
def load_df(data_path):
    """加載并預處理軸承數(shù)據(jù)"""
    # 獲取文件列表
    file_list = os.listdir(data_path)


    # 按文件名中的數(shù)字排序
    file_list = sorted(file_list, key=lambda x: int(x.split('.')[0]))


    # 創(chuàng)建空DataFrame
    df = pd.DataFrame()
    # 遍歷所有文件并加載數(shù)據(jù)
    for f in tqdm(file_list):
        temp = pd.read_csv(data_path + f"{f}")
        df = pd.concat([df, temp], axis=0)
    # 重置索引
    df.reset_index(drop=True, inplace=True)


    # 創(chuàng)建RUL列
    df['RUL'] = range(len(df)-1, -1, -1)


    # 計算RUL的最小值和最大值
    min_rul = df['RUL'].min()
    max_rul = df['RUL'].max()


    # 定義RUL歸一化函數(shù)
    def normalize_rul(rul, min_rul, max_rul):
        return (rul - min_rul) / (max_rul - min_rul)


    # 應用歸一化
    df['Normalized_RUL'] = df['RUL'].apply(lambda x: normalize_rul(x, min_rul, max_rul))


    return df


# 加載測試數(shù)據(jù)
data_path = "./datasets/37.5Hz11kN/Bearing2_4/" 
df = load_df(data_path)


# 提取特征和標簽
vibration = df[['Horizontal_vibration_signals', 'Vertical_vibration_signals']].values
rul = df['Normalized_RUL'].values


# 設置序列長度
seq_length = 128
# 創(chuàng)建數(shù)據(jù)集
dataset = BearingDataset(vibration, rul, seq_length)
# 創(chuàng)建數(shù)據(jù)加載器
dataloader = DataLoader(dataset, batch_size=1024, shuffle=False)


# 初始化模型參數(shù)
input_channels = 2  # 輸入通道數(shù)
hidden_size = 64  # LSTM隱藏層維度
num_layers = 4  # LSTM層數(shù)
device = torch.device('cuda')  # 使用GPU設備


# 創(chuàng)建模型實例
rul_model = DegradationModel(input_channels, hidden_size, 
                             num_layers, seq_length, device).to(device)


# 加載訓練好的模型權重
model_path = f'models/rul_model_{ver}.pth'
rul_model.load_state_dict(torch.load(model_path))
# 設置模型為評估模式
rul_model.eval()


# 初始化預測結果和真實值列表
predicted_ruls = []
true_ruls = []


# 不計算梯度
with torch.no_grad():
    # 使用進度條遍歷測試數(shù)據(jù)
    for batch in tqdm(dataloader, desc="Testing..."):
        features, labels = batch  # 獲取批次數(shù)據(jù)
        features, labels = features.to(device), labels.view(-1,1).to(device)  # 移動數(shù)據(jù)到設備
        # 模型預測
        pred = rul_model(features)
        # 保存預測結果
        predicted_ruls.extend(pred.cpu().numpy())
        # 保存真實值
        true_ruls.extend(labels.cpu().numpy())


# 轉換為NumPy數(shù)組并展平
predicted_ruls = np.array(predicted_ruls).flatten()
true_ruls = np.array(true_ruls).flatten()


# 導入可視化庫
import numpy as np  # 數(shù)值計算庫
import matplotlib.pyplot as plt  # 數(shù)據(jù)可視化庫


# 創(chuàng)建圖形
plt.figure(figsize=(12, 6))
# 繪制預測的RUL
plt.plot(predicted_ruls, label='Predicted RUL')
# 繪制真實的RUL
plt.plot(true_ruls, label='True RUL')
plt.xlabel('Sample')
plt.ylabel('RUL')
plt.title('Predicted vs True RUL')
plt.legend()
plt.grid(True)
plt.show()