訓練設備故障診斷模型

1. 說明

1. 依據設備故障診斷數據，訓練出適用的分析模型並儲存模型權重。
2. 訓練完成後，向FastWeb發送訊息，更新設備故障診斷訓練記錄。
3. 發送訊息至FastWeb訊息服務，提示模型訓練完成。

2. 設計Python程式

  設計的Python示例程式如下：

# 訓練預測的模型
import torch
import datetime
import numpy as np
import torch.nn as nn
import torch.optim as optim
import pandas as pd
from sklearn.model_selection import train_test_split
from torch.utils.data import DataLoader, TensorDataset
import os
import logging
import json
from logging.handlers import TimedRotatingFileHandler
import requests


fastweb_url='http://192.168.0.201:8803'

# 檢查資料夾是否存在，否則建立資料夾
def check_create_directory(path: str):
    if not os.path.exists(path):
        os.makedirs(path)

# 配置日誌
check_create_directory('log/')
logger = logging.getLogger('__dig_train_model__')
if logger.hasHandlers():
    logger.handlers.clear()
log_filename = 'log/dcc_dig_train.log'
log_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
log_handler = TimedRotatingFileHandler(log_filename, when='D', interval=1, backupCount=30, encoding='utf-8')
log_handler.suffix = '%Y-%m-%d.log'
log_handler.encoding = 'utf-8'
log_handler.setFormatter(log_formatter)
# 建立日誌記錄器
logger = logging.getLogger('__dig_train_model__')
logger.setLevel(logging.DEBUG)
logger.addHandler(log_handler)

# 定義神經網路模型
class MLP(nn.Module):
    def __init__(self):
        super(MLP, self).__init__()
        self.fc1 = nn.Linear(10, 100)  # 10個輸入參數，10個隱藏層
        self.fc2 = nn.Linear(100, 100)
        self.fc3 = nn.Linear(100, 100)
        self.fc4 = nn.Linear(100, 100)
        self.fc5 = nn.Linear(100, 100)
        self.fc6 = nn.Linear(100, 100)
        self.fc7 = nn.Linear(100, 100)
        self.fc8 = nn.Linear(100, 100)
        self.fc9 = nn.Linear(100, 100)
        self.fc10 = nn.Linear(100, 9)  # 9個輸出層

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = torch.relu(self.fc3(x))
        x = torch.relu(self.fc4(x))
        x = torch.relu(self.fc5(x))
        x = torch.relu(self.fc6(x))
        x = torch.relu(self.fc7(x))
        x = torch.relu(self.fc8(x))
        x = torch.relu(self.fc9(x))
        x = self.fc10(x)
        return x


def train_model(eqcode,data):
    # 建立模型實體
    model = MLP()
    data = np.array(data)
    data = pd.DataFrame(data)
    # 提取特徵和標籤列
    X = data.iloc[:, :10].values  # 輸入特徵，假設有10個輸入參數
    y = data.iloc[:, 10:].values  # 輸出標籤，假設有9個輸出層

    # 劃分訓練集和測試集
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.1, random_state=37)

    # 隨機打亂訓練集
    indices = list(range(len(X_train)))
    shuffle_indices = torch.randperm(len(X_train))
    X_train = X_train[shuffle_indices]
    y_train = y_train[shuffle_indices]

    # 轉換為PyTorch張量
    X_train = torch.Tensor(X_train)
    y_train = torch.Tensor(y_train)
    X_test = torch.Tensor(X_test)
    y_test = torch.Tensor(y_test)

    # 建立數據載入器
    train_dataset = TensorDataset(X_train, y_train)
    train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)

    # 定義損失函式和優化器
    criterion = nn.MSELoss()  # 使用均方誤差損失
    optimizer = optim.Adam(model.parameters(), lr=0.0009)  # 使用Adam優化器

    # 訓練模型
    num_epochs = 100
    for epoch in range(num_epochs):
        for inputs, labels in train_loader:
            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            if (epoch+1) % 10 == 0:
                logger.info(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.6f}')

    # 儲存模型到檔案
    check_create_directory('model/')
    # 獲取目前日期和時間
    current_datetime = datetime.datetime.now()

    # 根據日期和時間產生自動編號
    # 這裡使用年月日時分秒作為編號，例如：20230515120000
    auto_number = current_datetime.strftime("%Y%m%d%H%M%S")
    model_path = f"model/{eqcode}_dig_{auto_number}.pt"  # 儲存的模型檔案路徑
    torch.save(model.state_dict(), model_path)
    logger.info("模型已儲存")
    ismodel = True
    isfinish = True

    # 載入模型
    loaded_model = MLP()
    loaded_model.load_state_dict(torch.load(model_path))

    # 在測試集上評估模型
    loaded_model.eval()

    with torch.no_grad():
        test_outputs = loaded_model(X_test)
        test_loss = criterion(test_outputs, y_test)
        logger.info(f'Test Loss: {test_loss.item():.6f}')

    return ismodel,isfinish,model_path,loss.item(),test_loss.item()

def main():
#    params = {"username":"admin","tag":0,"guid": "98C99147-B84F-4BB6-9E4E-509F837A66E9 ","eqcode":"0103","lr":0.0001,"num_epochs":100,"batch_size":32}
    params = json.loads(input_value.value)
    # 一些主場景使用的功能，根據發送的數據，更新執行訓練
    if 'username' in params and 'tag' in params and 'guid' in params and 'eqcode' in params and 'lr' in params and 'num_epochs' in params and 'batch_size' in params:
        guid = params['guid']
        eqcode = params['eqcode']
        lr = params['lr']
        num_epochs = params['num_epochs']
        batch_size = params['batch_size']
        # 重新定義查詢參數
        query_params = {
            'restapi': 'dig_ai_diagnosedata',
            'eqcode': eqcode,
        }
        response = requests.get(url=fastweb_url, params=query_params)
        data = response.json()

        dataset = []
        for item in data["tabledata"]:
            dataset.append([item['vamp05'], item['vamp1'], item['vamp15'], item['vamp2'], item['vamp3'],\
                            item['vamp4'], item['vamp5'], item['vamp6'], item['acc1khz'], item['acc10khz'],\
                            0,0,0,0,0,0,int(item['isunbalance']),int(item['isuncenter']),int(item['isloose'])])
                
        # 訓練模型
        ismodel,isfinish,model_path,train_loss,validate_loss = train_model(eqcode=eqcode,data=dataset)
        # 更新訓練記錄
        data = {"guid":guid,"lr":lr,"num_epochs":num_epochs,"batch_size":batch_size,"train_loss":train_loss,\
                "validate_loss":validate_loss,"isfinish":isfinish,"ismodel":ismodel,"model_path":model_path}
        data = json.dumps(data)
        query_params={
            'restapi': 'dig_ai_updatediagnosetrainlog'
        }
        response = requests.post(url=fastweb_url,params=query_params,data=data)
        if response.status_code == 200:
            logger.info(f'{response.content}')
            logger.info("訓練記錄提交成功")
            query_params={
                'restapi': 'sendwsmsg'
            }
            data = json.dumps({"username":params['username'],"action":"callback","tag":params['tag'], \
                "data":{"callbackcomponent":"WebHomeFrame","callbackeventname":"update", \
                "callbackparams":[{"paramname":"messagetype","paramvalue":"success"},{"paramname":"title","paramvalue":"success"},\
                {"paramname":"message","paramvalue":f"模型訓練完成:{model_path}"}]}})
            requests.post(url=fastweb_url,params=query_params,data=data)
            data = json.dumps({"username":params['username'],"action":"callback","tag":params['tag'], \
                "data":{"callbackcomponent":"W-EQ-MOD-2308-2","callbackeventname":"update", \
                    "callbackparams":[{"paramname":"refresh","paramvalue":"true"}]}})
            requests.post(url=fastweb_url,params=query_params,data=data)

if __name__ == "__main__":
    main()

  將上述程式儲存為預設資料。按照下述樣式進行儲存。

  上述程式中定義的參數說明如下：

• 參數名稱：input_value。

3. 呼叫執行

  可以使用FastWeb 數控中心-設備故障診斷-訓練診斷模型 (opens new window)來呼叫啟用模型分析的Python指令碼。設定好呼叫taskrunner的地址，在驗證訓練界面點選[更新預測模型]，以啟用模型訓練的過程。訓練完成後，可以看到此次訓練的記錄。