-
Notifications
You must be signed in to change notification settings - Fork 2
/
FFNtest.py
211 lines (172 loc) · 7.96 KB
/
FFNtest.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
202
203
204
205
206
207
208
209
210
211
##############################################
# (c) Copyright 2018-2019 Kenza Tazi and Thomas Zhu
# This software is distributed under the terms of the GNU General Public
# Licence version 3 (GPLv3)
##############################################
import datetime
import numpy as np
import pandas # noqa: F401 # pylint: disable=unused-import # Prevent tflearn importing dodgy version
import tflearn
from tflearn.layers.core import dropout, fully_connected, input_data
from tflearn.layers.estimator import regression
import DataPreparation as dp
class FFN():
"""Object for handling TFLearn DNN models with added support for saving / loading different network configurations"""
def __init__(self, name, networkConfig=None, para_num=24, LR=1e-3, neuron_num=32, hidden_layers=4,
batch_size=64, epochs=10, dout=0.8):
self.name = name
self.networkConfig = networkConfig
self.para_num = para_num
self.LR = LR
self.neuron_num = neuron_num
self.hidden_layers = hidden_layers
self.batch_size = batch_size
self.epochs = epochs
self.dropout = dout
self.isLoaded = False
self._model = None
self._network = None
self.run_id = None
def __str__(self):
out = ('Model: ' + self.name + '\n'
+ 'Network type: ' + str(self.networkConfig) + '\n'
+ 'Number of inputs: ' + str(self.para_num))
return(out)
def TestNetwork(self):
# Network layers
# layer 0: generates a 4D tensor
layer0 = input_data(shape=[None, self.para_num], name='input')
dout = dropout(layer0, self.dropout)
for hl in range(self.hidden_layers):
hidden_layer = fully_connected(
dout, self.neuron_num, activation='leakyrelu')
dout = dropout(hidden_layer, self.dropout)
# Last layer needs to spit out the number of categories
# we are looking for.
softmax = fully_connected(dout, 2, activation='softmax')
# gives the paramaters to optimise the network
self._network = regression(softmax, optimizer='Adam', learning_rate=self.LR,
loss='categorical_crossentropy', name='targets')
self.networkConfig = 'TestNetwork'
@property
def network(self):
if self._network is not None:
return self._network
if self.networkConfig is None:
print('Using default network configuration, Network0')
self.TestNetwork()
else:
# Use network function specified by networkConfig
networkFunc = getattr(self, self.networkConfig)
networkFunc()
return self._network
@property
def model(self):
if self._model:
return self._model
self._model = tflearn.DNN(
self.network, tensorboard_verbose=0, tensorboard_dir='./Temp/tflearn_logs')
return self._model
def Train(self, training_data, training_truth, validation_data, validation_truth):
timestamp = datetime.datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
self.run_id = 'Models/' + str(self.name) + '_' + timestamp
self.model.fit(training_data, training_truth, n_epoch=self.epochs,
validation_set=(validation_data, validation_truth),
snapshot_step=10000, show_metric=True, run_id=self.run_id)
self.isLoaded = True
def Save(self, path=None):
if path:
self.model.save(path)
with open(path + '.txt', 'w') as file:
file.write(self.networkConfig + '\n')
file.write(str(self.para_num) + '\n')
file.write(str(self.LR) + '\n')
file.write(str(self.neuron_num) + '\n')
file.write(str(self.hidden_layers) + '\n')
file.write(str(self.batch_size) + '\n')
file.write(str(self.epochs) + '\n')
file.write(str(self.dropout) + '\n')
file.write(str(self.run_id))
else:
self.model.save("Models/" + self.name)
with open("Models/" + self.name + '.txt', 'w') as file:
file.write(self.networkConfig + '\n')
file.write(str(self.para_num) + '\n')
file.write(str(self.LR) + '\n')
file.write(str(self.neuron_num) + '\n')
file.write(str(self.hidden_layers) + '\n')
file.write(str(self.batch_size) + '\n')
file.write(str(self.epochs) + '\n')
file.write(str(self.dropout) + '\n')
file.write(str(self.run_id))
def Load(self, path=None, verbose=True):
if self.isLoaded:
raise AssertionError(
'Graph already loaded. Consider loading into new object.')
if path:
with open(path + '.txt', 'r') as file:
settings = file.readlines()
if len(settings) == 1:
self.networkConfig = settings[0]
elif len(settings) >= 2:
self.networkConfig = settings[0].strip()
self.para_num = int(settings[1].strip())
self.LR = int(settings[2].strip())
self.neuron_num = int(settings[3].strip())
self.hidden_layers = int(settings[4].strip())
self.batch_size = int(settings[5].strip())
self.epochs = int(settings[6].strip())
self.dropout = int(settings[7].strip())
self.model.load(path)
else:
with open('Models/' + self.name + '.txt', 'r') as file:
settings = file.readlines()
if len(settings) == 1:
self.networkConfig = settings[0]
elif len(settings) >= 2:
self.networkConfig = settings[0].strip()
self.para_num = int(settings[1].strip())
self.LR = int(settings[2].strip())
self.neuron_num = int(settings[3].strip())
self.hidden_layers = int(settings[4].strip())
self.batch_size = int(settings[5].strip())
self.epochs = int(settings[6].strip())
self.dropout = int(settings[7].strip())
self.model.load('Models/' + self.name)
self.isLoaded = True
if verbose:
print('##############################################')
print('Loading successful')
print('Model: ' + self.name)
print('Network type: ' + self.networkConfig)
print('Number of inputs: ' + str(self.para_num))
print('Learning rate: ' + str(self.LR))
print('Neurons per hidden layer: ' + str(self.neuron_num))
print('Hidden layers: ' + str(self.hidden_layers))
print('Batch size: ' + str(self.batch_size))
print('Epochs: ' + str(self.epochs))
print('Dropout ' + str(self.dropout))
print('##############################################')
def Predict(self, X):
return(self.model.predict(X))
def Predict_label(self, X):
return(self.model.predict_label(X))
def apply_mask(self, Sreference):
if self.isLoaded is False:
raise AssertionError(
'Model is neither loaded nor trained, cannot make predictions')
inputs = dp.getinputsFFN(Sreference, input_type=self.para_num)
label = self.model.predict_label(inputs)
lmask = np.array(label)
lmask = lmask[:, 0].reshape(2400, 3000)
prob = self.model.predict(inputs)
pmask = np.array(prob)
pmask = pmask[:, 0].reshape(2400, 3000)
return lmask, pmask
if __name__ == '__main__':
# Pixel Loading
df = dp.PixelLoader('./SatelliteData/SLSTR/Pixels3')
tdata, vdata, ttruth, vtruth = df.dp.get_ffn_training_data(21)
model = FFN('Test', 'Network1', 21)
model.Train(tdata, ttruth, vdata, vtruth)
model.Save()