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-19 19:21:21 +08:00
parent a82e27de83
commit ac6f39a012
4 changed files with 349 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
source/dev_EnergyRouter.py
View File

@@ -4,21 +4,114 @@ from pathlib import Path
from utl import trans_list_to_str
from func_frame import make_frame_modbus, check_frame_modbus
modbus_map = {
}
class EnergyRouter:
def __init__(self, ip="192.168.100.10", port=7, adddr_modbus=0x01):
self._ip = ip
self._port = port
self._adddr_modbus =adddr_modbus
self._addr_modbus =adddr_modbus
self.tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.tcp_socket.connect((self._ip, self._port))
def frame_update(fp):
pass
def frame_read(self, daddr=0x00, dlen=0x10):
block_modbus = {
'addr_dev' : self._addr_modbus,
'data_addr' : daddr,
'data_len' : dlen,
'type' : 'read',
}
frame = make_frame_modbus(block_modbus)
if self.tcp_socket is None:
print(trans_list_to_str(frame))
return
# self.tcp_socket.recv(8)
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)
print(output_text)
def frame_update(self, path_bin):
""" 程序升级
"""
block_modbus = {
'addr_dev' : self.addr_modbus,
'type' : 'update',
'step' : 'start',
'index' : 0,
'file' : Path(path_bin).read_bytes(),
'header_offset': 128,
}
# 启动帧
frame_master = bytearray(make_frame_modbus(block_modbus))
# 等待擦除完成返回
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
block_modbus['file_block_size'] = check_frame_modbus(frame_slave, block_modbus)
break
if block_modbus['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']
while data_remain > 0:
frame_master = bytearray(make_frame_modbus(block_modbus))
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']
# 结束升级
block_modbus['step'] = 'end'
frame_master = bytearray(make_frame_modbus(block_modbus))
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)
if __name__ == "__main__":
""" 升级测试 """
path_file = Path("F:\\Work\\FPGA\\Test\\Vivado\\test_update\\test_update.vitis\\upgrade_system\\Debug\\sd_card\\BOOT.dat")
dev_ep = EnergyRouter()
dev_ep.frame_read()
dev_ep.frame_update(path_file)

214
source/dev_Lamina.py
View File

@@ -4,6 +4,105 @@ from serial import Serial
from utl import trans_list_to_str
from func_frame import make_frame_dlt645, check_frame_dlt645
modbus_map = {
# 1 - Hex
# 2 - Int16
# 3 - lnt32
# 4 - str
# 5 - addr
0x0E: ["故障字1", 1],
0x0F: ["故障字2", 1],
0x10: ["MPPT工作状态", 1],
0x11: ["系统工作状态", 1],
0x12: ["系统工作模式", 1],
0x13: ["输入电压", 2],
0x14: ["电感电流", 2],
0x15: ["12V电压", 2],
0x16: ["输出电压", 2],
0x17: ["输入电流", 2],
0x18: ["温度1", 2],
0x19: ["温度2", 2],
0x1A: ["输入功率", 3],
0x1C: ["设备温度", 2],
0x1D: ["开关机状态", 1],
0x1E: ["保留", 1],
0x1F: ["保留", 1],
0x60: ["光伏通道使能", 1],
0x61: ["最小启动输入电压", 2],
0x62: ["最大启动输入电压", 2],
0x63: ["最小停止输入电压", 2],
0x64: ["最大停止输入电压", 2],
0x65: ["最小MPPT电压", 2],
0x66: ["最大MPPT电压", 2],
0x67: ["最小启动输出电压", 2],
0x68: ["最大启动输出电压", 2],
0x69: ["最小停止输出电压", 2],
0x6A: ["最大停止输出电压", 2],
0x6B: ["输入过压保护值", 2],
0x6C: ["输出过压保护值", 2],
0x6D: ["输出欠压保护值", 2],
0x6E: ["电感过流保护值", 2],
0x6F: ["输入过流保护值", 2],
0x70: ["最小电感电流限值", 2],
0x71: ["最大电感电流限值", 2],
0x72: ["浮充电压阈值", 2],
0x73: ["三点法中间阈值", 2],
0x74: ["恒压充电电压", 2],
0x75: ["过温故障值", 2],
0x76: ["过温告警值", 2],
0x77: ["温度恢复值", 2],
0x78: ["最低满载电压", 2],
0x79: ["最高满载电压", 2],
0x7A: ["输入过载保护值", 3],
0x7C: ["最小功率限值", 3],
0x7E: ["最大功率限值", 3],
0x80: ["最大功率限值存储值", 3],
0x82: ["载波通信地址", 5],
0x85: ["电压环out_max", 2],
0x86: ["电压环out_min", 2],
0x87: ["电流环out_max", 2],
0x88: ["电流环out_min", 2],
0x89: ["MPPT扰动系数k_d_vin", 2],
0x8A: ["dmin", 2],
0x8B: ["dmax", 2],
0x8C: ["扫描电压偏移scanvolt_offset", 2],
0x8D: ["电压环Kp", 3],
0x8F: ["电压环Ki", 3],
0x91: ["电流环Kp", 3],
0x93: ["电流环Ki", 3],
0x95: ["日志级别", 1],
0x96: ["日志输出方式", 1],
0x97: ["采样校准volt_in_a", 2],
0x98: ["采样校准volt_in_b", 2],
0x99: ["采样校准volt_out_a", 2],
0x9A: ["采样校准volt_out_b", 2],
0x9B: ["采样校准curr_in_a", 2],
0x9C: ["采样校准curr_in_b", 2],
0x9D: ["采样校准curr_induc_a", 2],
0x9E: ["采样校准curr_induc_b", 2],
0x9F: ["采样校准volt_12V_a", 2],
0xA0: ["采样校准volt_12V_b", 2],
0xA1: ["温度补偿temp1_b", 2],
0xA2: ["温度补偿temp2_b", 2],
0xA3: ["系统工作模式", 2],
0xA4: ["电感电流给定值curr_set", 2],
0xA5: ["抖动频率上限", 2],
0xA6: ["抖动频率下限", 2],
0xA7: ["保留", 1],
0xA8: ["保留", 1],
0xA9: ["保留", 1],
0xAA: ["保留", 1],
0xAB: ["保留", 1],
0xAC: ["保留", 1],
0xAD: ["保留", 1],
0xAE: ["保留", 1],
0xAF: ["保留", 1],
0x100: ["版本", 4],
0x110: ["型号", 4],
}
class LaminaAdapter:
def __init__(self, com_name="COM16", addr_645=[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA], addr_modbus=0x01):
# 初始化串口通信
@@ -17,19 +116,20 @@ class LaminaAdapter:
# 设置Modbus地址
self.addr_modbus = addr_modbus
def frame_read(self, daddr=0x60, dlen=0x50):
block_modbus = {
'addr_dev' : self.addr_modbus,
'data_addr' : daddr,
'data_len' : dlen,
'type' : 'read',
}
block_dlt645 = {
self.block = {
'addr' : self.addr_645,
'type' : 'modbus',
'data' : block_modbus,
'data' : {
'addr_dev' : self.addr_modbus,
'data_define': modbus_map,
},
}
frame = make_frame_dlt645(block_dlt645)
def frame_read(self, daddr=0x60, dlen=0x50):
self.block['data']['type'] = 'read'
self.block['data']['data_addr'] = daddr
self.block['data']['data_len'] = dlen
frame = make_frame_dlt645(self.block)
if self.com is None:
print(trans_list_to_str(frame))
@@ -39,22 +139,14 @@ class LaminaAdapter:
self.com.write(bytearray(frame))
time.sleep(0.5)
frame_recv = self.com.read_all()
output_text = check_frame_dlt645(frame_recv, block=block_dlt645)
output_text = check_frame_dlt645(frame_recv, block=self.block)
print(output_text)
def frame_write_one(self, daddr=0x85, dval=-900):
block_modbus = {
'addr_dev' : self.addr_modbus,
'data_addr' : daddr,
'data_val' : dval,
'type' : 'write_one',
}
block_dlt645 = {
'addr' : self.addr_645,
'type' : 'modbus',
'data' : block_modbus,
}
frame = make_frame_dlt645(block_dlt645)
self.block['data']['type'] = 'write_one'
self.block['data']['data_addr'] = daddr
self.block['data']['data_val'] = dval
frame = make_frame_dlt645(self.block)
if self.com is None:
print(trans_list_to_str(frame))
return
@@ -62,18 +154,10 @@ class LaminaAdapter:
self.com.write(bytearray(frame))
def frame_write_dual(self, daddr=0x91, dval=600):
block_modbus = {
'addr_dev' : self.addr_modbus,
'data_addr' : daddr,
'data_val' : dval,
'type' : 'write_dual',
}
block_dlt645 = {
'addr' : self.addr_645,
'type' : 'modbus',
'data' : block_modbus,
}
frame = make_frame_dlt645(block_dlt645)
self.block['data']['type'] = 'write_dual'
self.block['data']['data_addr'] = daddr
self.block['data']['data_val'] = dval
frame = make_frame_dlt645(self.block)
if self.com is None:
print(trans_list_to_str(frame))
return
@@ -81,18 +165,10 @@ class LaminaAdapter:
self.com.write(bytearray(frame))
def frame_write_str(self, daddr=0x82, dval=[0x06, 0x05, 0x04, 0x03, 0x02, 0x01]):
block_modbus = {
'addr_dev' : self.addr_modbus,
'data_addr' : daddr,
'data_val' : dval,
'type' : 'write_str',
}
block_dlt645 = {
'addr' : self.addr_645,
'type' : 'modbus',
'data' : block_modbus,
}
frame = make_frame_dlt645(block_dlt645)
self.block['data']['type'] = 'write_str'
self.block['data']['data_addr'] = daddr
self.block['data']['data_val'] = dval
frame = make_frame_dlt645(self.block)
if self.com is None:
print(trans_list_to_str(frame))
return
@@ -104,21 +180,13 @@ class LaminaAdapter:
注意: 在使用单板升级测试时, 需要关闭低电压检测功能, 否则无法启动升级流程;
"""
block_modbus = {
'addr_dev' : self.addr_modbus,
'type' : 'update',
'step' : 'start',
'index' : 0,
'file' : Path(path_bin).read_bytes(),
'header_offset': 128,
}
block_dlt645 = {
'addr' : self.addr_645,
'type' : 'modbus',
'data' : block_modbus,
}
self.block['data']['type'] = 'write_str'
self.block['data']['step'] = 'start'
self.block['data']['index'] = 0
self.block['data']['file'] = Path(path_bin).read_bytes()
self.block['data']['header_offset'] = 128
# 启动帧
frame_master = bytearray(make_frame_dlt645(block_dlt645))
frame_master = bytearray(make_frame_dlt645(self.block))
# 等待擦除完成返回
try_times = 3000
@@ -131,20 +199,20 @@ class LaminaAdapter:
try_times -= 1
continue
block_dlt645["data"]['file_block_size'] = check_frame_dlt645(frame_slave, block_dlt645)
self.block["data"]['file_block_size'] = check_frame_dlt645(frame_slave, self.block)
break
if block_dlt645["data"]['file_block_size'] == 0:
raise Exception("Error slave response.")
if self.block["data"]['file_block_size'] == 0:
raise("Error slave response.")
# 避免接收到延迟返回报文
time.sleep(0.4)
# 文件传输
block_dlt645["data"]['step'] = 'trans'
data_remain = len(block_dlt645["data"]['file']) - block_dlt645['data']['header_offset']
self.block["data"]['step'] = 'trans'
data_remain = len(self.block["data"]['file']) - self.block['data']['header_offset']
while data_remain > 0:
frame_master = bytearray(make_frame_dlt645(block_dlt645))
frame_master = bytearray(make_frame_dlt645(self.block))
self.com.read_all()
self.com.write(frame_master)
@@ -152,14 +220,14 @@ class LaminaAdapter:
frame_slave = None
while not frame_slave:
frame_slave = self.com.read_all()
check_frame_dlt645(frame_slave, block_dlt645)
check_frame_dlt645(frame_slave, self.block)
block_dlt645["data"]['index'] += 1
data_remain -= block_dlt645["data"]['file_block_size']
self.block["data"]['index'] += 1
data_remain -= self.block["data"]['file_block_size']
# 结束升级
block_dlt645["data"]['step'] = 'end'
frame_master = bytearray(make_frame_dlt645(block_dlt645))
self.block["data"]['step'] = 'end'
frame_master = bytearray(make_frame_dlt645(self.block))
self.com.read_all()
self.com.write(frame_master)
@@ -167,7 +235,7 @@ class LaminaAdapter:
frame_slave = None
while not frame_slave:
frame_slave = self.com.read_all()
check_frame_dlt645(frame_slave[:18], block_dlt645)
check_frame_dlt645(frame_slave[:18], self.block)
if __name__=='__main__':
dev_lamina = LaminaAdapter()

146
source/func_frame.py
View File

@@ -7,96 +7,11 @@ modbus_map = {
# 3 - lnt32
# 4 - str
# 5 - addr
0x0E: ["故障字1", 1],
0x0F: ["故障字2", 1],
0x10: ["MPPT工作状态", 1],
0x11: ["系统工作状态", 1],
0x12: ["系统工作模式", 1],
0x13: ["输入电压", 2],
0x14: ["电感电流", 2],
0x15: ["12V电压", 2],
0x16: ["输出电压", 2],
0x17: ["输入电流", 2],
0x18: ["温度1", 2],
0x19: ["温度2", 2],
0x1A: ["输入功率", 3],
0x1C: ["设备温度", 2],
0x1D: ["开关机状态", 1],
0x1E: ["保留", 1],
0x1F: ["保留", 1],
0x60: ["光伏通道使能", 1],
0x61: ["最小启动输入电压", 2],
0x62: ["最大启动输入电压", 2],
0x63: ["最小停止输入电压", 2],
0x64: ["最大停止输入电压", 2],
0x65: ["最小MPPT电压", 2],
0x66: ["最大MPPT电压", 2],
0x67: ["最小启动输出电压", 2],
0x68: ["最大启动输出电压", 2],
0x69: ["最小停止输出电压", 2],
0x6A: ["最大停止输出电压", 2],
0x6B: ["输入过压保护值", 2],
0x6C: ["输出过压保护值", 2],
0x6D: ["输出欠压保护值", 2],
0x6E: ["电感过流保护值", 2],
0x6F: ["输入过流保护值", 2],
0x70: ["最小电感电流限值", 2],
0x71: ["最大电感电流限值", 2],
0x72: ["浮充电压阈值", 2],
0x73: ["三点法中间阈值", 2],
0x74: ["恒压充电电压", 2],
0x75: ["过温故障值", 2],
0x76: ["过温告警值", 2],
0x77: ["温度恢复值", 2],
0x78: ["最低满载电压", 2],
0x79: ["最高满载电压", 2],
0x7A: ["输入过载保护值", 3],
0x7C: ["最小功率限值", 3],
0x7E: ["最大功率限值", 3],
0x80: ["最大功率限值存储值", 3],
0x82: ["载波通信地址", 5],
0x85: ["电压环out_max", 2],
0x86: ["电压环out_min", 2],
0x87: ["电流环out_max", 2],
0x88: ["电流环out_min", 2],
0x89: ["MPPT扰动系数k_d_vin", 2],
0x8A: ["dmin", 2],
0x8B: ["dmax", 2],
0x8C: ["扫描电压偏移scanvolt_offset", 2],
0x8D: ["电压环Kp", 3],
0x8F: ["电压环Ki", 3],
0x91: ["电流环Kp", 3],
0x93: ["电流环Ki", 3],
0x95: ["日志级别", 1],
0x96: ["日志输出方式", 1],
0x97: ["采样校准volt_in_a", 2],
0x98: ["采样校准volt_in_b", 2],
0x99: ["采样校准volt_out_a", 2],
0x9A: ["采样校准volt_out_b", 2],
0x9B: ["采样校准curr_in_a", 2],
0x9C: ["采样校准curr_in_b", 2],
0x9D: ["采样校准curr_induc_a", 2],
0x9E: ["采样校准curr_induc_b", 2],
0x9F: ["采样校准volt_12V_a", 2],
0xA0: ["采样校准volt_12V_b", 2],
0xA1: ["温度补偿temp1_b", 2],
0xA2: ["温度补偿temp2_b", 2],
0xA3: ["系统工作模式", 2],
0xA4: ["电感电流给定值curr_set", 2],
0xA5: ["抖动频率上限", 2],
0xA6: ["抖动频率下限", 2],
0xA7: ["保留", 1],
0xA8: ["保留", 1],
0xA9: ["保留", 1],
0xAA: ["保留", 1],
0xAB: ["保留", 1],
0xAC: ["保留", 1],
0xAD: ["保留", 1],
0xAE: ["保留", 1],
0xAF: ["保留", 1],
0x100: ["版本", 4],
0x110: ["型号", 4],
0x01: ["Hex示例", 1],
0x02: ["Int示例", 2],
0x03: ["Int32示例", 3],
0x04: ["str示例", 4],
0x10: ["addr示例", 5],
}
@@ -172,7 +87,7 @@ def make_frame_modbus(block:dict):
data_len = len(block['data_val'])
data_val = block['data_val']
if data_len > 0x0100:
raise("modbus data len oversize.")
raise("Modbus data len oversize.")
frame.append(data_addr // 256 % 256)
frame.append(data_addr % 256)
frame.append(0x00)
@@ -214,7 +129,7 @@ def make_frame_dlt645(block:dict):
return frame
def display_data(address: int, data: list):
def display_data(modbus_map: dict, address: int, data: list):
""" 格式化表示数据 """
def display_str(data, len_data):
item = bytearray(data[:2 * len_data])
@@ -228,13 +143,14 @@ def display_data(address: int, data: list):
idx = address
len_max = 0
while data:
data_len = 1
data_label = "未知数据"
if idx not in modbus_map.keys():
output_line = "\t Error, doesn`t match item.\n"
output += output_line
break
len_data = 1
item = data[0] * 0x0100 + data[1]
item = display_hex(item, 4)
else:
current_map = modbus_map[idx]
data_label = current_map[0]
if current_map[1] == 1:
""" Hex字符表示 """
item = data[0] * 0x0100 + data[1]
@@ -251,25 +167,25 @@ def display_data(address: int, data: list):
item += data[0] * 0x0100 + data[1]
if data[2] > 0x80:
item -= 0x1_0000_0000
len_data = 2
data_len = 2
elif current_map[1] == 4:
""" 字符串表示 """
len_data = 16
item = display_str(data, len_data)
data_len = 16
item = display_str(data, data_len)
item = item.decode()
elif current_map[1] == 5:
""" 载波地址表示 """
len_data = 3
item = display_str(data, len_data)
data_len = 3
item = display_str(data, data_len)
item = trans_list_to_str(item)
if len_max < len(current_map[0]):
len_max = len(current_map[0])
len_remain = len_max - len(current_map[0])
output_line = "\t".join([display_hex(idx, 4), current_map[0] + " " * len_remain, str(item)])
if len_max < len(data_label):
len_max = len(data_label)
len_remain = len_max - len(data_label)
output_line = "\t".join([display_hex(idx, 4), data_label + " " * len_remain, str(item)])
idx += len_data
del data[0:len_data * 2]
idx += data_len
del data[0:data_len * 2]
output += output_line + "\n"
return output
@@ -277,9 +193,9 @@ def check_frame_modbus(frame, block=None):
""" 校验modbus帧回复 """
calculator = Calculator(Crc16.MODBUS)
if calculator.checksum(frame[:-2]) != (frame[-1] * 0x100 + frame[-2]):
raise Exception("Frame Crc check failed.")
raise("Frame Crc check failed.")
elif frame[0] != 0x01:
raise Exception("Frame device address error.")
raise("Frame device address error.")
code_func = frame[1]
code_subfunc = frame[2]
@@ -290,7 +206,7 @@ def check_frame_modbus(frame, block=None):
elif code_subfunc == 0x02 and frame[3] == 0x00:
update_index = frame[4] * 256 + frame[5]
if type(block) is dict and block['index'] != update_index:
raise Exception("Error slave response.")
raise("Error slave response.")
return update_index
elif code_subfunc == 0x03 and frame[3] == 0x00:
return frame[4] * 256 + frame[5]
@@ -299,7 +215,7 @@ def check_frame_modbus(frame, block=None):
if frame[2] == len(frame[3:-2]):
if type(block) is dict:
data_addr = block['data_addr']
return display_data(data_addr, list(frame[3:-2]))
return display_data(block['data_define'], data_addr, list(frame[3:-2]))
else:
return list(frame[3:-2])
else:
@@ -316,14 +232,14 @@ def check_frame_modbus(frame, block=None):
def check_frame_dlt645(frame, block=None):
""" 校验dlt645帧回复 """
if sum(frame[:-2]) % 0x100 != frame[-2]:
raise Exception("DLT645 Frame Verify error.")
raise("DLT645 Frame Verify error.")
elif len(frame[10:-2]) != frame[9]:
raise Exception("DLT645 Frame len error.")
raise("DLT645 Frame len error.")
code_func = frame[8]
if code_func == 0x9F:
block = block['data']
return check_frame_modbus(frame[10:-2], block)
else:
raise Exception("DLT645 Frame type error.")
raise("DLT645 Frame type error.")

107
source/utl_upgrade.py
View File

@@ -2,9 +2,9 @@
import hashlib
from pathlib import Path
from utl import conv_int_to_array
from utl import conv_int_to_array, trans_list_to_str
header_tag = {
Header_Tag = {
'file_type': [0x00, 2], # 0 - 文件类型; 2byte
'file_version': [0x01, 2], # 1 - 文件版本; 2byte
'file_length': [0x02, 4], # 2 - 文件长度; 4byte
@@ -20,7 +20,7 @@ header_tag = {
'hex_name': [0xFF, -1, 80], # 255 - Hex文件名; less than 80byte
}
def file_hex_to_bin(file_data, base_address, len_max=0xC000, conv_end=False):
def file_IntelHex_to_Bin(file_data, base_address, len_max=0xC000, conv_end=False):
"""
将Intel Hex格式文件转换为Bin格式;
"""
@@ -34,7 +34,7 @@ def file_hex_to_bin(file_data, base_address, len_max=0xC000, conv_end=False):
if line[0] == ':':
checksum = sum([int(x, 16) * (15 * (~i & 0x01) + 1) for i, x in enumerate(line[1:])])
if (checksum & 0x00FF) != 0:
raise Exception('Hex file checksum error!')
raise('Hex file checksum error!')
if line[7:9] == '00':
len = int(line[1:3], 16)
@@ -78,40 +78,40 @@ def file_encryption(buffer):
return result
def build_header(data_bin: bytearray, config: dict, len_max=512):
def build_header(config: dict, len_max=512):
"""
基于配置参数, 生成文件信息头;
"""
# 定义文件头
m_file_header = bytearray(len_max)
m_file_header[0:8] = b"TOPSCOMM"
header_len = 11
tag_num = 0
for tag, value in config.items():
if tag in header_tag.keys():
if header_tag[tag][1] == -1:
tag_len = min(len(value), header_tag[tag][2])
if tag in Header_Tag.keys():
if Header_Tag[tag][1] == -1:
tag_len = min(len(value), Header_Tag[tag][2])
else:
tag_len = header_tag[tag][1]
tag_date = [header_tag[tag][0], tag_len] + value[:tag_len]
tag_len = Header_Tag[tag][1]
tag_date = [Header_Tag[tag][0], tag_len] + value[:tag_len]
m_file_header[header_len: header_len + tag_len + 2] = bytearray(tag_date)
tag_num += 1
header_len += 2 + tag_len
m_file_header[0:8] = b"TOPSCOMM"
m_file_header[8] = ((header_len - 10) % 0x100)
m_file_header[9] = ((header_len - 10) // 0x100)
m_file_header[10] = tag_num
m_file_header[header_len] = sum(m_file_header[:header_len]) % 0x100
m_file_header[header_len+1] = sum(m_file_header[:header_len]) // 0x100
if header_len+2 > 64:
if header_len+2 > len_max:
return None
else:
return m_file_header
def build_header1(data_bin: bytearray, config: dict):
def build_header_lite(config: dict):
"""
基于配置参数, 生成文件信息头; 基于新版本升级方案;
基于配置参数, 生成文件信息头;
"""
# 定义文件头
m_file_header = bytearray(64)
@@ -122,13 +122,11 @@ def build_header1(data_bin: bytearray, config: dict):
return m_file_header
def make_datafile(file_path: str, config: dict, header_len=512):
""" 升级文件生成函数 """
""" 升级文件生成函数; 完整文件头, TI-hex转换 """
file_path = Path(file_path)
file_data = file_path.read_text()
data_bin = file_hex_to_bin(file_data, config['flash_addr'], len_max=config['flash_size'] * 2)
data_bin = file_IntelHex_to_Bin(file_data, config['flash_addr'], len_max=config['flash_size'] * 2)
md5_ctx = hashlib.md5()
md5_ctx.update(data_bin)
config['file_length'] = conv_int_to_array(len(data_bin))
@@ -136,17 +134,17 @@ def make_datafile(file_path: str, config: dict, header_len=512):
if header_len > 256:
config['hex_name'] = list(file_path.name.encode())[:80]
data_header = build_header(data_bin, config, header_len)
data_header = build_header(config, header_len)
if data_header is None:
raise ("header tag oversize! ")
data_encrypt = file_encryption(data_bin)
return data_header, data_encrypt, data_bin
return data_header, data_bin, data_encrypt
def make_datafile1(file_path: str, config: dict, header_len=512):
""" 升级文件生成函数; 基于新版本方案 """
""" 升级文件生成函数; 简易文件头, TI-Hex转换 """
file_path = Path(file_path)
file_data = file_path.read_text()
data_bin = file_hex_to_bin(file_data, config['flash_addr'], len_max=config['flash_size'] * 2)
data_bin = file_IntelHex_to_Bin(file_data, config['flash_addr'], len_max=config['flash_size'] * 2)
md5_ctx = hashlib.md5()
md5_ctx.update(data_bin)
@@ -155,16 +153,34 @@ def make_datafile1(file_path: str, config: dict, header_len=512):
if header_len > 256:
config['hex_name'] = list(file_path.name.encode())[:80]
data_header = build_header1(data_bin, config)
data_header = build_header_lite(config)
data_encrypt = file_encryption(data_bin)
return data_header, data_encrypt, data_bin
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()
md5_ctx.update(data_bin)
config['file_length'] = conv_int_to_array(len(data_bin))
config["md5"] = list(md5_ctx.digest())
if header_len > 256:
config['hex_name'] = list(file_path.name.encode())[:80]
data_header = build_header(config, header_len)
if data_header is None:
raise("Header tag oversize. ")
data_encrypt = file_encryption(data_bin)
return data_header, data_bin, data_encrypt
def test1(fp):
""" bin文件生成测试 """
file1 = Path(fp)
path1 = file1.parent / (file1.stem + ".bin")
data1 = file1.read_text()
data2 = file_hex_to_bin(data1, 0x3F0100)
data2 = file_IntelHex_to_Bin(data1, 0x3F0100)
path1.write_bytes(data2)
pass
@@ -183,36 +199,33 @@ def test2():
def test3():
""" 完整升级包生成测试 """
config = {
# 'file_type': [0x43, 0x00], # 叠光DSP升级文件
# 'file_version': [0x00, 0x00], # 文件版本-00
# 'file_length': [], # 文件长度
# 'md5': [], # 文件MD5
'file_type': [0x10, 0x01], # Xilinx-Demo 自机升级文件
'file_version': [0x00, 0x00], # 文件版本-00 用于兼容文件格式升级
# 'file_length': [], # 文件长度(自动生成)
# 'md5': [], # 文件MD5(自动生成)
'encrypt': [0x01], # 默认加密算法
'update_type': [0x01], # APP升级
# 'update_spec': [], # 升级特征字
'update_verison': [0x01, 0x00, 0x00, 0x00], # 升级版本号; 4byte
'update_date': [0x23, 0x12, 0x16], # 升级版本日期; 3byte
# 'area_code': [], # 省份特征; 4byte
# 'uptate_str': [], # 升级段描述; less than 64byte
# 'device_str': [], # 设备特征描述; less than 64byte
# 'hex_name': [], # Hex文件名; less than 80byte
'update_spec': [0x00, 0x00, 0x00, 0x00], # 升级特征字
'update_verison': [0x02, 0x00, 0x00, 0x01], # 升级版本号
'update_date': [0x19, 0x04, 0x24], # 升级版本日期
# 'area_code': [], # 省份特征
# 'uptate_str': [], # 升级段描述
# 'device_str': [], # 设备特征描述
# 'hex_name': [], # Hex文件名(自动读取)
# 文件结构信息
'flash_addr': 0x3E8020, # 程序文件起始地址
'flash_size': 0x005FC0, # 程序文件大小
}
file1 = Path("D:\WorkingProject\LightStackController\lamina_pvdc\Source\PVDC20\Debug\LAMINA_PVDC.hex")
func_gen = [make_datafile, make_datafile1]
func_para = [{'header_len':64}, {}]
path_bin = Path("F:\\Work\\FPGA\\Test\\Vivado\\test_update\\test_update.vitis\\upgrade_system\\Debug\\sd_card\\BOOT.BIN")
count = 0
for func, para in zip(func_gen, func_para):
header, data_e, data_b = func(file1, config, **para)
print(list(map(hex, header)))
print(list(map(hex, data_e)))
print(list(map(hex, data_b)))
file2 = file1.parent / (file1.stem + f'_v{count}.dat')
file2.write_bytes(header + data_b)
header, data_b, _ = make_datafile2(path_bin, config, header_len=128)
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.write_bytes(header + data_b)
if __name__ == "__main__":