forked from ClustProject/KUDataMultitasklearning
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main_multi.py
201 lines (150 loc) · 5.99 KB
/
main_multi.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
# -*- coding: utf-8 -*-
"""
Created on Tue Mar 21 14:29:09 2023
@author: lee
"""
import torch
import torch.nn as nn
import pandas as pd
import numpy as np
from sklearn.metrics import accuracy_score, mean_squared_error, mean_absolute_percentage_error, mean_absolute_error, r2_score
from models.train_model_multi import Train_Test
from models.lstm_fcn_multi import LSTM_FCNs
from models.rnn import RNN_model
from models.cnn_1d import CNN_1D
from models.fc import FC
import warnings
warnings.filterwarnings('ignore')
class Multilearning():
def __init__(self, config, mode):
"""
Parameters
----------
config : TYPE
DESCRIPTION.
Returns
-------
None.
"""
self.mode = mode
self.model_name = config['model']
self.parameter = config['parameter']
self.best_model_path = config['best_model_path']
# build trainer
self.trainer = Train_Test(config)
def build_model(self):
"""
Returns
-------
init_model : TYPE
DESCRIPTION.
"""
if self.mode == 'transfer' :
init_model = LSTM_FCNs(
input_size=self.parameter['input_size'],
num_classes=self.parameter['source_class'],
num_layers=self.parameter['num_layers'],
lstm_drop_p=self.parameter['lstm_drop_out'],
fc_drop_p=self.parameter['fc_drop_out']
)
else : ## target 자체를 학습시키는 모델 만듬 ## self
init_model = LSTM_FCNs(
input_size=self.parameter['input_size'],
num_classes_1=self.parameter['num_classes_1'],
num_classes_2=self.parameter['num_classes_2'],
num_layers=self.parameter['num_layers'],
lstm_drop_p=self.parameter['lstm_drop_out'],
fc_drop_p=self.parameter['fc_drop_out']
)
return init_model
def train_model(self,train_x, train_y, valid_x, valid_y,option='source'):
"""
Parameters
----------
train_x : TYPE
DESCRIPTION.
train_y : TYPE
DESCRIPTION.
valid_x : TYPE
DESCRIPTION.
valid_y : TYPE
DESCRIPTION.
Returns
-------
None.
"""
train_loader = self.get_dataloader(train_x, train_y, self.parameter['batch_size'], shuffle=True)
valid_loader = self.get_dataloader(valid_x, valid_y, self.parameter['batch_size'], shuffle=False)
# build initialized model
if option == 'target' :
init_model = self.tuning_model(self.best_model_path,freeze=self.parameter['freeze'])
else :
init_model = self.build_model()
# train model
dataloaders_dict = {'train': train_loader, 'val': valid_loader}
best_model = self.trainer.train(init_model, dataloaders_dict)
return best_model
def save_model(self,best_model,best_model_path):
"""
Parameters
----------
best_model : TYPE
DESCRIPTION.
best_model_path : TYPE
DESCRIPTION.
Returns
-------
None.
"""
torch.save(best_model.state_dict(), best_model_path)
def pred_data(self,test_x, test_y, best_model_path):
"""
"""
test_loader = self.get_dataloader(test_x, test_y, self.parameter['batch_size'], shuffle=False)
# build initialized model
init_model = self.build_model()
# load best model
init_model.load_state_dict(torch.load(best_model_path))
# get predicted classes
pred_data_1, pred_data_2 = self.trainer.test(init_model, test_loader)
# class의 값이 0부터 시작하지 않으면 0부터 시작하도록 변환
# calculate performance metrics
acc = accuracy_score(test_y[:,0], pred_data_1)
mse = mean_squared_error(test_y[:,1], pred_data_2)
MAPE = mean_absolute_percentage_error(test_y[:,1], pred_data_2)
MAE = mean_absolute_error(test_y[:,1], pred_data_2)
R2 = r2_score(test_y[:,1], pred_data_2)
# merge true value and predicted value
pred_df = pd.DataFrame()
pred_df['actual_value_1'] = test_y[:,0]
pred_df['predicted_value_1'] = pred_data_1
pred_df['actual_value_2'] = test_y[:,1]
pred_df['predicted_value_2'] = pred_data_2
return pred_df, acc, mse, MAPE, MAE, R2
def get_dataloader(self, x_data, y_data, batch_size, shuffle):
"""
"""
# torch dataset 구축
dataset = torch.utils.data.TensorDataset(torch.Tensor(x_data), torch.Tensor(y_data))
# DataLoader 구축
data_loader = torch.utils.data.DataLoader(dataset, batch_size=batch_size, shuffle=shuffle)
return data_loader
def tuning_model(self,best_model_path,freeze):
# config 에 Source / Target dataset 정리
## change / freeze / save
# load best model
init_model = self.build_model()
init_model.load_state_dict(torch.load(best_model_path))
# if self.parameter['source_class'] != self.parameter['target_class'] :
# print('model fc layer output change')
# in_features = init_model.fc.in_features
# out_features = self.parameter['target_class']
# init_model.fc = nn.Linear(in_features,out_features)
if freeze:
for name, param in init_model.named_parameters():
if name in ['fc.weight','fc.bias']:
param.requires_grad = True
else :
param.requires_grad = False
print(param.requires_grad)
return init_model