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-22 23:05:02 +08:00
parent ac6f39a012
commit 64c7a5d364
4 changed files with 93 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
source/dev_EnergyRouter.py
View File

@@ -5,6 +5,8 @@ from utl import trans_list_to_str
from func_frame import make_frame_modbus, check_frame_modbus
modbus_map = {
0x00: ['编译日期', 4, 6],
0x06: ['编译时间', 4, 5],
}
class EnergyRouter:
@@ -16,14 +18,16 @@ class EnergyRouter:
self.tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.tcp_socket.connect((self._ip, self._port))
def frame_read(self, daddr=0x00, dlen=0x10):
block_modbus = {
self.block = {
'addr_dev' : self._addr_modbus,
'data_addr' : daddr,
'data_len' : dlen,
'type' : 'read',
'data_define': modbus_map,
}
frame = make_frame_modbus(block_modbus)
def frame_read(self, daddr=0x00, dlen=0x10):
self.block['type'] = 'read'
self.block['data_addr'] = daddr
self.block['data_len'] = dlen
frame = make_frame_modbus(self.block)
if self.tcp_socket is None:
print(trans_list_to_str(frame))
@@ -33,7 +37,7 @@ class EnergyRouter:
self.tcp_socket.send(bytearray(frame))
time.sleep(0.5)
frame_recv = self.tcp_socket.recv(128)
output_text = check_frame_modbus(frame_recv, block_modbus)
output_text = check_frame_modbus(frame_recv, self.block)
print(output_text)
@@ -41,65 +45,72 @@ class EnergyRouter:
""" 程序升级
"""
block_modbus = {
'addr_dev' : self.addr_modbus,
'type' : 'update',
'step' : 'start',
'index' : 0,
'file' : Path(path_bin).read_bytes(),
'header_offset': 128,
}
self.block['type'] = 'update'
self.block['step'] = 'start'
self.block['file'] = Path(path_bin).read_bytes()
self.block['header_offset'] = 128
# 启动帧
frame_master = bytearray(make_frame_modbus(block_modbus))
frame_master = bytearray(make_frame_modbus(self.block))
# 等待擦除完成返回
try_times = 3000
self.tcp_socket.recv()
while try_times:
time.sleep(0.4)
self.tcp_socket.send(frame_master)
frame_slave = self.tcp_socket.recv()
if not frame_slave:
try_times -= 1
continue
frame_slave = self.tcp_socket.recv(32)
if frame_slave == '':
raise("TCP closed.")
block_modbus['file_block_size'] = check_frame_modbus(frame_slave, block_modbus)
break
self.block['file_block_size'] = check_frame_modbus(frame_slave, self.block)
if block_modbus['file_block_size'] == 0:
if self.block['file_block_size'] == 0:
raise("Error slave response.")
# 避免接收到延迟返回报文
time.sleep(0.4)
# 文件传输
block_modbus['step'] = 'trans'
data_remain = len(block_modbus['file']) - block_modbus['header_offset']
self.block['index'] = 0
self.block['step'] = 'trans'
data_remain = len(self.block['file']) - self.block['header_offset']
seq_offset = 0
seq_window = [0, 0, 0, 0]
seq_frame_master = [None, None, None, None]
seq_frame_slave = [None, None, None, None]
while data_remain > 0:
frame_master = bytearray(make_frame_modbus(block_modbus))
# 更新并发送帧
for i, s in enumerate(seq_window):
if s == 0:
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)):
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
data_remain -= self.block['file_block_size']
self.tcp_socket.recv()
self.tcp_socket.send(frame_master)
time.sleep(0.2)
frame_slave = None
while not frame_slave:
frame_slave = self.tcp_socket.recv()
check_frame_modbus(frame_slave, block_modbus)
block_modbus['index'] += 1
data_remain -= block_modbus['file_block_size']
# 移动发送窗口
while seq_window[0]:
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
# 结束升级
block_modbus['step'] = 'end'
frame_master = bytearray(make_frame_modbus(block_modbus))
self.block['step'] = 'end'
frame_master = bytearray(make_frame_modbus(self.block))
self.tcp_socket.recv()
self.tcp_socket.send(frame_master)
time.sleep(0.1)
frame_slave = None
while not frame_slave:
frame_slave = self.tcp_socket.recv()
check_frame_modbus(frame_slave[:18], block_modbus)
# 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)
if __name__ == "__main__":
@@ -112,6 +123,6 @@ if __name__ == "__main__":
dev_ep.frame_update(path_file)
pass

8
source/dev_Lamina.py
View File

@@ -58,7 +58,7 @@ modbus_map = {
0x7C: ["最小功率限值", 3],
0x7E: ["最大功率限值", 3],
0x80: ["最大功率限值存储值", 3],
0x82: ["载波通信地址", 5],
0x82: ["载波通信地址", 5, 3],
0x85: ["电压环out_max", 2],
0x86: ["电压环out_min", 2],
0x87: ["电流环out_max", 2],
@@ -98,8 +98,8 @@ modbus_map = {
0xAD: ["保留", 1],
0xAE: ["保留", 1],
0xAF: ["保留", 1],
0x100: ["版本", 4],
0x110: ["型号", 4],
0x100: ["版本", 4, 16],
0x110: ["型号", 4, 16],
}
@@ -180,7 +180,7 @@ class LaminaAdapter:
注意: 在使用单板升级测试时, 需要关闭低电压检测功能, 否则无法启动升级流程;
"""
self.block['data']['type'] = 'write_str'
self.block['data']['type'] = 'update'
self.block['data']['step'] = 'start'
self.block['data']['index'] = 0
self.block['data']['file'] = Path(path_bin).read_bytes()

29
source/func_frame.py
View File

@@ -170,12 +170,12 @@ def display_data(modbus_map: dict, address: int, data: list):
data_len = 2
elif current_map[1] == 4:
""" 字符串表示 """
data_len = 16
data_len = current_map[2]
item = display_str(data, data_len)
item = item.decode()
elif current_map[1] == 5:
""" 载波地址表示 """
data_len = 3
data_len = current_map[2]
item = display_str(data, data_len)
item = trans_list_to_str(item)
@@ -203,13 +203,30 @@ def check_frame_modbus(frame, block=None):
""" 升级回复帧 """
if code_subfunc == 0x01 and frame[3] == 0x00:
return frame[4] * 256 + frame[5]
elif code_subfunc == 0x02 and frame[3] == 0x00:
elif code_subfunc == 0x02:
""" 解析帧序号及返回值, 不做序列号校验 """
update_result = frame[3]
update_index = frame[4] * 256 + frame[5]
if type(block) is dict and block['index'] != update_index:
raise("Error slave response.")
return update_index
# 检查严重错误标志
if update_result == 0xFF:
raise("Error, Update abort.")
if update_result >= 0xA0 or update_result == 0x0F:
check_result = False
else:
check_result = True
if update_result >= 0xA0:
update_result -= 0x90
if update_result >= 0x20:
check_hope = update_index + (update_result & 0x0F)
elif update_result >= 0x10:
check_hope = update_index - (update_result & 0x0F)
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]
else:
raise("Func code or Return code error.")
elif code_func == 0x03 or code_func == 0x04:
""" 数据读取帧 """
if frame[2] == len(frame[3:-2]):

8
source/utl_upgrade.py
View File

@@ -207,15 +207,15 @@ def test3():
'update_type': [0x01], # APP升级
'update_spec': [0x00, 0x00, 0x00, 0x00], # 升级特征字
'update_verison': [0x02, 0x00, 0x00, 0x01], # 升级版本号
'update_date': [0x19, 0x04, 0x24], # 升级版本日期
'update_date': [0x22, 0x04, 0x24], # 升级版本日期
# 'area_code': [], # 省份特征
# 'uptate_str': [], # 升级段描述
# 'device_str': [], # 设备特征描述
# 'hex_name': [], # Hex文件名(自动读取)
# 文件结构信息
'flash_addr': 0x3E8020, # 程序文件起始地址
'flash_size': 0x005FC0, # 程序文件大小
# 文件Hex结构信息
# 'flash_addr': 0x3E8020, # 程序起始地址
# 'flash_size': 0x005FC0, # 程序空间大小
}
path_bin = Path("F:\\Work\\FPGA\\Test\\Vivado\\test_update\\test_update.vitis\\upgrade_system\\Debug\\sd_card\\BOOT.BIN")