260 lines
11 KiB
Python
Executable File
260 lines
11 KiB
Python
Executable File
from all_models_tools.all_model_tools import call_back
|
||
from Read_and_process_image.ReadAndProcess import Read_image_and_Process_image
|
||
from draw_tools.draw import plot_history, Confusion_Matrix_of_Two_Classification
|
||
from Load_process.Load_Indepentend import Load_Indepentend_Data
|
||
from _validation.ValidationTheEnterData import validation_the_enter_data
|
||
from Load_process.file_processing import Process_File
|
||
from merge_class.merge import merge
|
||
from draw_tools.Grad_cam import Grad_CAM
|
||
from sklearn.metrics import confusion_matrix
|
||
from Image_Process.Image_Generator import Image_generator
|
||
import pandas as pd
|
||
import time
|
||
import torch.optim as optim
|
||
from experiments.pytorch_Model import ModifiedXception
|
||
from Load_process.LoadData import Loding_Data_Root, Load_Data_Tools
|
||
import torch
|
||
from Model_Loss.Loss import Entropy_Loss
|
||
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
|
||
from torchmetrics.functional import auroc
|
||
import torch.nn as nn
|
||
from torchinfo import summary
|
||
import numpy as np
|
||
from tqdm import tqdm
|
||
from torch.nn import functional
|
||
|
||
class experiments():
|
||
def __init__(self, tools, Targets, status, Number_Of_Classes):
|
||
'''
|
||
# 實驗物件
|
||
|
||
## 說明:
|
||
* 用於開始訓練pytorch的物件，裡面分為數個方法，負責處理實驗過程的種種
|
||
|
||
## parmeter:
|
||
* Topic_Tool: 讀取訓練、驗證、測試的資料集與Label等等的內容
|
||
* cut_image: 呼叫切割影像物件
|
||
* merge: 合併的物件
|
||
* model_name: 模型名稱，告訴我我是用哪個模型(可能是預處理模型/自己設計的模型)
|
||
* experiment_name: 實驗名稱
|
||
* epoch: 訓練次數
|
||
* train_batch_size: 訓練資料的batch
|
||
* convolution_name: Grad-CAM的最後一層的名稱
|
||
* Number_Of_Classes: Label的類別
|
||
* Status: 選擇現在資料集的狀態
|
||
* device: 決定使用GPU或CPU
|
||
|
||
## Method:
|
||
* processing_main: 實驗物件的進入點
|
||
* construct_model: 決定實驗用的Model
|
||
* Training_Step: 訓練步驟，開始進行訓練驗證的部分
|
||
* Evaluate_Model: 驗證模型的準確度
|
||
* record_matrix_image: 劃出混淆矩陣(熱力圖)
|
||
* record_everyTime_test_result: 記錄我單次的訓練結果並將它輸出到檔案中
|
||
'''
|
||
|
||
self.Topic_Tool = tools
|
||
|
||
self.cut_image = Load_Indepentend_Data(self.Topic_Tool.Get_Data_Label(), self.Topic_Tool.Get_OneHot_Encording_Label())
|
||
self.image_processing = Read_image_and_Process_image()
|
||
self.merge = merge()
|
||
|
||
self.model_name = "Xception"
|
||
self.experiment_name = "Xception Skin to train Normal stomach cancer"
|
||
self.generator_batch_size = 50
|
||
self.epoch = 10000
|
||
self.train_batch_size = 64
|
||
self.layers = 1
|
||
self.convolution_name = "block14_sepconv2"
|
||
self.Number_Of_Classes = Number_Of_Classes
|
||
|
||
self.Grad = ""
|
||
self.Status = status
|
||
self.Tragets = Targets
|
||
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
|
||
|
||
pass
|
||
|
||
def processing_main(self, Training_Data, counter):
|
||
Train, Test, Validation = self.Topic_Tool.Get_Save_Roots(self.Status) # 要換不同資料集就要改
|
||
Load_Tools = Load_Data_Tools()
|
||
|
||
Training_Data = Load_Tools.DataLoad_Image_Root(Training_Data, self.train_batch_size)
|
||
|
||
test = Load_Tools.Load_ImageFolder_Data(Test, "transform")
|
||
validation = Load_Tools.Load_ImageFolder_Data(Validation, "transform")
|
||
|
||
self.test = Load_Tools.DataLoad_Image_Root(test, 1)
|
||
self.Validation = Load_Tools.DataLoad_Image_Root(validation, 1)
|
||
|
||
# self.Grad = Grad_CAM(self.Topic_Tool.Get_Data_Label(), Test_labels, self.experiment_name, self.convolution_name)
|
||
cnn_model = self.construct_model() # 呼叫讀取模型的function
|
||
print(summary(cnn_model, input_size=(int(self.train_batch_size / 2), 3, 512, 512)))
|
||
|
||
print("訓練開始")
|
||
train_losses, val_losses, train_accuracies, val_accuracies = self.Training_Step(cnn_model, Training_Data, counter)
|
||
print("訓練完成！")
|
||
|
||
loss, accuracy, precision, recall, AUC, f1, True_Label, Predict_Label = self.Evaluate_Model(cnn_model)
|
||
|
||
self.record_matrix_image(True_Label, Predict_Label, self.model_name, counter)
|
||
|
||
print(self.record_everyTime_test_result(loss, accuracy, precision, recall, AUC, f1, counter, self.experiment_name)) # 紀錄當前訓練完之後的預測結果，並輸出成csv檔
|
||
|
||
Losses = [train_losses, val_losses]
|
||
Accuracyes = [train_accuracies, val_accuracies]
|
||
plot_history(self.epoch, Losses, Accuracyes, "train" + str(counter), self.experiment_name) # 將訓練結果化成圖，並將化出來的圖丟出去儲存
|
||
|
||
# self.Grad.process_main(cnn_model, counter, self.test)
|
||
|
||
return loss, accuracy, precision, recall, AUC, f1
|
||
|
||
def construct_model(self):
|
||
'''決定我這次訓練要用哪個model'''
|
||
cnn_model = ModifiedXception()
|
||
|
||
if torch.cuda.device_count() > 1:
|
||
cnn_model = nn.DataParallel(cnn_model)
|
||
|
||
cnn_model = cnn_model.to(self.device)
|
||
return cnn_model
|
||
|
||
def Training_Step(self, model, Training, counter):
|
||
# 定義優化器，並設定 weight_decay 參數來加入 L2 正則化
|
||
Optimizer = optim.SGD(model.parameters(), lr=0.045, momentum = 0.9, weight_decay=0.1)
|
||
model_path, early_stopping, scheduler = call_back(self.model_name, counter, Optimizer)
|
||
|
||
criterion = Entropy_Loss() # 使用自定義的損失函數
|
||
train_losses = []
|
||
val_losses = []
|
||
train_accuracies = []
|
||
val_accuracies = []
|
||
|
||
for epoch in range(self.epoch):
|
||
model.train()
|
||
running_loss = 0.0
|
||
all_train_preds = []
|
||
all_train_labels = []
|
||
|
||
epoch_iterator = tqdm(Training, desc= "Training (Epoch %d)" % epoch)
|
||
|
||
|
||
for inputs, labels in epoch_iterator:
|
||
OneHot_labels = functional.one_hot(labels, self.Number_Of_Classes)
|
||
# labels = np.reshape(labels, (int(labels.shape[0]), 1))
|
||
inputs, OneHot_labels = inputs.to(self.device), OneHot_labels.to(self.device)
|
||
# inputs, labels = inputs.cuda(), labels.cuda()
|
||
|
||
Optimizer.zero_grad()
|
||
outputs = model(inputs)
|
||
loss = criterion(outputs, OneHot_labels)
|
||
loss.backward()
|
||
Optimizer.step()
|
||
running_loss += loss.item()
|
||
|
||
# 收集訓練預測和標籤
|
||
_, preds = torch.max(outputs, 1)
|
||
all_train_preds.extend(preds.cpu().numpy())
|
||
all_train_labels.extend(labels.cpu().numpy())
|
||
|
||
Training_Loss = running_loss/len(Training)
|
||
|
||
# all_train_labels = torch.FloatTensor(all_train_labels)
|
||
# all_train_labels = torch.argmax(all_train_labels, 1)
|
||
train_accuracy = accuracy_score(all_train_labels, all_train_preds)
|
||
|
||
train_losses.append(Training_Loss)
|
||
train_accuracies.append(train_accuracy)
|
||
|
||
print(f"Epoch [{epoch+1}/{self.epoch}], Loss: {Training_Loss:.4f}, Accuracy: {train_accuracy:0.2f}", end = ' ')
|
||
|
||
model.eval()
|
||
val_loss = 0.0
|
||
all_val_preds = []
|
||
all_val_labels = []
|
||
|
||
with torch.no_grad():
|
||
for inputs, labels in self.Validation:
|
||
OneHot_labels = functional.one_hot(labels, self.Number_Of_Classes)
|
||
inputs, OneHot_labels = inputs.to(self.device), OneHot_labels.to(self.device)
|
||
|
||
outputs = model(inputs)
|
||
loss = criterion(outputs, OneHot_labels)
|
||
val_loss += loss.item()
|
||
|
||
# 驗證預測與標籤
|
||
_, preds = torch.max(outputs, 1)
|
||
all_val_preds.extend(preds.cpu().numpy())
|
||
all_val_labels.extend(labels.cpu().numpy())
|
||
|
||
# 計算驗證損失與準確率
|
||
val_loss /= len(self.Validation)
|
||
val_accuracy = accuracy_score(all_val_labels, all_val_preds)
|
||
|
||
val_losses.append(val_loss)
|
||
val_accuracies.append(val_accuracy)
|
||
print(f"Epoch [{epoch+1}/{self.epoch}], Loss: {val_loss:.4f}, Accuracy: {val_accuracy:0.2f}")
|
||
|
||
early_stopping(val_loss, model, model_path)
|
||
if early_stopping.early_stop:
|
||
print("Early stopping triggered. Training stopped.")
|
||
break
|
||
|
||
# 學習率調整
|
||
scheduler.step(val_loss)
|
||
|
||
return train_losses, val_losses, train_accuracies, val_accuracies
|
||
|
||
def Evaluate_Model(self, cnn_model):
|
||
# 測試模型
|
||
cnn_model.eval()
|
||
True_Label, Predict_Label = [], []
|
||
loss = 0.0
|
||
with torch.no_grad():
|
||
for images, labels in self.test:
|
||
OneHot_labels = functional.one_hot(labels, self.Number_Of_Classes)
|
||
images, OneHot_labels = images.to(self.device), OneHot_labels.to(self.device)
|
||
|
||
outputs = cnn_model(images)
|
||
_, predicted = torch.max(outputs, 1)
|
||
Predict_Label.extend(predicted.cpu().numpy())
|
||
True_Label.extend(labels.cpu().numpy())
|
||
|
||
loss /= len(self.test)
|
||
|
||
accuracy = accuracy_score(True_Label, Predict_Label)
|
||
precision = precision_score(True_Label, Predict_Label)
|
||
recall = recall_score(True_Label, Predict_Label)
|
||
AUC = auroc(True_Label, Predict_Label, task = ["Stomatch_Cancer", "Normal"])
|
||
f1 = f1_score(True_Label, Predict_Label)
|
||
return loss, accuracy, precision, recall, AUC, f1, True_Label, Predict_Label
|
||
|
||
|
||
def record_matrix_image(self, True_Labels, Predict_Labels, model_name, index):
|
||
'''劃出混淆矩陣(熱力圖)'''
|
||
# 計算混淆矩陣
|
||
matrix = confusion_matrix(True_Labels, Predict_Labels)
|
||
Confusion_Matrix_of_Two_Classification(model_name, matrix, index) # 呼叫畫出confusion matrix的function
|
||
|
||
return matrix.real
|
||
|
||
def record_everyTime_test_result(self, loss, accuracy, precision, recall, auc, f, indexs, model_name):
|
||
'''記錄我單次的訓練結果並將它輸出到檔案中'''
|
||
File = Process_File()
|
||
|
||
Dataframe = pd.DataFrame(
|
||
{
|
||
"model_name" : str(model_name),
|
||
"loss" : "{:.2f}".format(loss),
|
||
"precision" : "{:.2f}%".format(precision * 100),
|
||
"recall" : "{:.2f}%".format(recall * 100),
|
||
"accuracy" : "{:.2f}%".format(accuracy * 100),
|
||
"f" : "{:.2f}%".format(f * 100),
|
||
"AUC" : "{:.2f}%".format(auc * 100)
|
||
}, index = [indexs])
|
||
File.Save_CSV_File("train_result", Dataframe)
|
||
# File.Save_TXT_File("Matrix_Result : " + str(Matrix), model_name + "_train" + str(indexs))
|
||
|
||
return Dataframe
|
||
|