使用郵件監控Mxnet訓練

知識 10-16

使用郵件監控Mxnet訓練

1. 打包訓練代碼

需要進行監控訓練，所以需要將訓練的代碼打包進一個函數內，通過傳參的方式進行訓練。還是使用FashionMNIST數據集

這樣訓練的時候就調用函數傳參就行了

訓練主函數

訓練需要的一些參數都採用傳參的形式

def NN_Train(net, train_data, test_data, epochs, batch_size, learning_rate, weight_decay):
msg = ""
train_loss = []
train_acc = []
dataset_train = gluon.data.DataLoader(train_data, batch_size, shuffle=True)
test_loss = []
test_acc = []
dataset_test = gluon.data.DataLoader(test_data, batch_size, shuffle=True)
trainer = gluon.Trainer(net.collect_params(), "adam",
{"learning_rate": learning_rate, "wd": weight_decay})
softmax_cross_entropy = gluon.loss.SoftmaxCrossEntropyLoss() for epoch in range(epochs):
_loss = 0.
_acc = 0.
t_acc = 0.
for data, label in dataset_train:
data = nd.transpose(data, (0, 3, 1, 2))
data = data.as_in_context(ctx)
label = label.as_in_context(ctx) with autograd.record():
output = net(data)
loss = softmax_cross_entropy(output, label)
loss.backward()
trainer.step(batch_size)
_loss += nd.mean(loss).asscalar()
_acc += accuracy(output, label)
__acc = _acc / len(dataset_train)
__loss = _loss / len(dataset_train)
train_loss.append(__loss)
train_acc.append(__acc)
t_acc, t_loss = evaluate_accuracy(dataset_test, net)
test_loss.append(t_loss)
test_acc.append(t_acc)
msg += ("Epoch %d. Train Loss: %f, Test Loss: %f, Train Acc %f, Test Acc %f
" % (
epoch, __loss, t_loss, __acc, t_acc))
fig = plt.figure()
ax1 = fig.add_subplot(111)
ax1.plot(train_loss, "r")
ax1.plot(test_loss, "g")
ax1.legend(["Train_Loss", "Test_Loss"], loc=2)
ax1.set_ylabel("Loss")
ax2 = ax1.twinx()
ax2.plot(train_acc, "b")
ax2.plot(test_acc, "y")
ax2.legend(["Train_Acc", "Test_Acc"], loc=1)
ax2.set_ylabel("Acc")
plt.savefig("NN.png", dpi=600)
net.collect_params().save("NN.params") return msg

打包網路模型

同樣，需要把網路也打包進函數內

def GetNN():
net = nn.HybridSequential() with net.name_scope():
net.add(gluon.nn.Conv2D(channels=20, kernel_size=5, activation="relu"))
net.add(gluon.nn.MaxPool2D(pool_size=2, strides=2))
net.add(gluon.nn.Conv2D(channels=50, kernel_size=3, activation="relu"))
net.add(gluon.nn.MaxPool2D(pool_size=2, strides=2))
net.add(gluon.nn.Flatten())
net.add(gluon.nn.Dense(10))
net.initialize(init=mx.init.Xavier(), ctx=ctx)
net.hybridize() return net

打包數據讀取

然後把數據讀取也搞進函數內

def GetDate():
fashion_train = gluon.data.vision.FashionMNIST(
root="./", train=True, transform=transform)
fashion_test = gluon.data.vision.FashionMNIST(
root="./", train=True, transform=transform) return fashion_train, fashion_test

2. 搞定郵件的接收發送

使用郵件監控，就要搞定在Python上使用郵件的問題，還好Python內置了郵件庫

這樣接收發送郵件也只用調用函數就好了

接受郵件

我只接受純文本的內容，因為HTML內容的太過複雜

def ReEmail():
try:
pp = poplib.POP3(pophost)
pp.user(useremail)
pp.pass_(password)
resp, mails, octets = pp.list()
index = len(mails) if index > 0:
resp, lines, octets = pp.retr(index)
msg_content = b"
".join(lines).decode("utf-8")
pp.dele(index)
pp.quit()
msg = Parser().parsestr(msg_content)
message = Get_info(msg)
subject = msg.get("Subject")
date = msg.get("Date") return message,subject,date except ConnectionResetError as e: print("ConnectionResetError") return None,None,None

發送郵件

發送郵件我是用了一個第三方郵件庫 envelopes 因為簡單方便。

def SentEmail(message,subject,image=True):
envelope = Envelope(
from_addr=(useremail, u"Train"),
to_addr=(toemail, u"FierceX"),
subject=subject,
text_body=message
) if image:
envelope.add_attachment("NN.png")
envelope.send(smtphost, login=useremail,
password=password, tls=True)

解析郵件內容

然後需要解析郵件內容，這段基本從網上抄來的，因為郵件格式很複雜，沒深究

def Get_info(msg):
if (msg.is_multipart()):
parts = msg.get_payload() for n, part in enumerate(parts): return Get_info(part) if not msg.is_multipart():
content_type = msg.get_content_type() if content_type=="text/plain":
content = msg.get_payload(decode=True)
charset = guess_charset(msg) if charset:
content = content.decode(charset) return content

3. 使用多線程多進程監控訓練

接下來就是主體了，其實主體也沒多少代碼，就是循環監控郵箱。並且對相應內容做反饋

使用子進程進行訓練

由於Python的多線程的性能局限性，我使用了子進程進行訓練，這樣不會受到主進程循環監控的影響

def nn(params):
train, test = NN_Train.GetDate()
msg = ("%s
") % str(params)
msg += NN_Train.NN_Train(
NN_Train.GetNN(),
train_data=train,
test_data=test,
epochs=int(params["ep"]),
batch_size=int(params["bs"]),
learning_rate=params["lr"],
weight_decay=params["wd"])
EmailTool.SentEmail(msg, "TrainResult")def run(msg):
params = {"ep": 10, "lr": 0.002, "bs": 128, "wd": 0.0}
xx = msg.split("
") for k in xx:
ks = k.split(" ") if len(ks) > 1:
params[ks[0]] = float(ks[1]) print(params)
p = Process(target=nn, args=(params,)) print("TrainStrart") global running
running = True
p.start()
p.join()
running = False

使用循環監控郵箱

在主進程中，使用循環監控郵箱內容，對相應內容做出反饋。為了防止子進程成為殭屍進程，我是用了一個線程來等待子進程結束

if __name__ == "__main__":
global running
running = False
print("Start")
a = 1
while(True):
time.sleep(10) print(a, running) try:
msg, sub, date = EmailTool.ReEmail() except TimeoutError as e: print("TimeoutError") if sub == "train": print("train") if running == False:
t = threading.Thread(target=run, args=(msg,))
t.start() else:
EmailTool.SentEmail("Training is underway", "Training is underway",
image=False) if sub == "exit": break
a += 1

5. 效果

發送訓練郵件

使用郵件監控Mxnet訓練