linghtiin/DebugTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

每日更新;

Browse Source
This commit is contained in:
何泽隆
2024-04-26 04:28:39 +08:00
parent 64c7a5d364
commit 3ccfca4b32
3 changed files with 50 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
source/dev_EnergyRouter.py
View File

@@ -1,5 +1,6 @@
import time
import socket
import random
from pathlib import Path
from utl import trans_list_to_str
from func_frame import make_frame_modbus, check_frame_modbus
@@ -77,40 +78,59 @@ class EnergyRouter:
seq_frame_slave = [None, None, None, None]
while data_remain > 0:
# 更新并发送帧
for i, s in enumerate(seq_window):
tmp = list(zip(range(len(seq_window)), seq_window))
random.shuffle(tmp)
for i, s in tmp:
if s == 0:
if (data_remain - i * self.block['file_block_size']) < 0:
""" 避免生成越界帧 """
continue
seq_window[i] = 1
self.block['index'] = seq_offset + i
seq_frame_master[i] = bytearray(make_frame_modbus(self.block))
self.tcp_socket.send(seq_frame_master[i])
# 接收帧回复
for i in range(len(seq_frame_master)):
tmp = list(zip(range(len(seq_window)), seq_window))
for i, s in tmp:
if s == 1:
seq_frame_slave[i] = self.tcp_socket.recv(8)
# 接收到空数据, 对端已关闭连接
if seq_frame_slave[i] == '':
raise("TCP closed.")
result, seq_current, seq_hope = check_frame_modbus(seq_frame_slave[i], self.block)
if result:
seq_window[seq_current - seq_offset] = 1
result, seq_current, seq_hope = check_frame_modbus(seq_frame_slave[i], None)
if seq_current < seq_offset:
raise("Error.")
elif result:
seq_window[seq_current - seq_offset] = 2
data_remain -= self.block['file_block_size']
if seq_hope is not None and seq_hope < seq_offset:
print("Frame Index out of order.")
seq_offset = seq_hope
# 移动发送窗口
while seq_window[0]:
while seq_window[0] == 2:
seq_window[0] = seq_window[1]
seq_window[1] = seq_window[2]
seq_window[2] = seq_window[3]
seq_window[3] = 0
seq_offset += 1
# 设置发送失败内容重发标志
for i, s in enumerate(seq_window):
if s == 1:
seq_window[i] = 0
# 结束升级
ret = 0
self.block['step'] = 'end'
frame_master = bytearray(make_frame_modbus(self.block))
while ret == 0:
self.tcp_socket.send(frame_master)
# time.sleep(0.1)
frame_slave = self.tcp_socket.recv(8)
if frame_slave == '':
raise("TCP closed.")
check_frame_modbus(frame_slave[:18], self.block)
ret = check_frame_modbus(frame_slave[:18], self.block)
if __name__ == "__main__":

12
source/func_frame.py
View File

@@ -223,8 +223,16 @@ def check_frame_modbus(frame, block=None):
else:
check_hope = None
return check_result, update_index, check_hope
elif code_subfunc == 0x03 and frame[3] == 0x00:
return frame[4] * 256 + frame[5]
elif code_subfunc == 0x03:
if frame[3] == 0xFF:
raise("Update verification failed.")
elif frame[3] == 0x01:
print("Update done.")
elif frame[3] == 0x00:
print("Update not done.")
else:
print("Update unknown return value.")
return frame[3]
else:
raise("Func code or Return code error.")
elif code_func == 0x03 or code_func == 0x04:

6
source/utl_upgrade.py
View File

@@ -158,7 +158,7 @@ def make_datafile1(file_path: str, config: dict, header_len=512):
return data_header, data_bin, data_encrypt
def make_datafile2(file_path: str, config: dict, header_len=512):
""" 升级文件生成函数 """
""" 升级文件生成函数; 完整文件头 """
file_path = Path(file_path)
data_bin = file_path.read_bytes()
md5_ctx = hashlib.md5()
@@ -219,12 +219,12 @@ def test3():
}
path_bin = Path("F:\\Work\\FPGA\\Test\\Vivado\\test_update\\test_update.vitis\\upgrade_system\\Debug\\sd_card\\BOOT.BIN")
header, data_b, _ = make_datafile2(path_bin, config, header_len=128)
header, data_b, _ = make_datafile2(path_bin, config, header_len=512)
print("Upgrade file generated successfully.")
print(f"\t header_length={len(header)}, bin_length={len(data_b)}[{hex(len(data_b))}]")
print(f"\t file md5: {trans_list_to_str(config['md5'])}")
file1 = path_bin.parent / (path_bin.stem + '.dat')
file1 = path_bin.parent / (path_bin.stem + '_h512.dat')
file1.write_bytes(header + data_b)