diff --git a/source/dev_EnergyRouter.py b/source/dev_EnergyRouter.py index 00ff4a9..4f157c0 100644 --- a/source/dev_EnergyRouter.py +++ b/source/dev_EnergyRouter.py @@ -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) + + + \ No newline at end of file diff --git a/source/dev_Lamina.py b/source/dev_Lamina.py index 9ff41b9..58a65ee 100644 --- a/source/dev_Lamina.py +++ b/source/dev_Lamina.py @@ -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() diff --git a/source/func_frame.py b/source/func_frame.py index 55f5a2c..d4b06c5 100644 --- a/source/func_frame.py +++ b/source/func_frame.py @@ -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,48 +143,49 @@ 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 - current_map = modbus_map[idx] - if current_map[1] == 1: - """ Hex字符表示 """ item = data[0] * 0x0100 + data[1] item = display_hex(item, 4) - elif current_map[1] == 2: - """ 16位数值表示 """ - item = data[0] * 0x0100 + data[1] - if item > 0x8000: - item -= 0x1_0000 - elif current_map[1] == 3: - """ 32位数值表示 """ - item = data[2] * 0x0100 + data[3] - item *= 0x10000 - item += data[0] * 0x0100 + data[1] - if data[2] > 0x80: - item -= 0x1_0000_0000 - len_data = 2 - elif current_map[1] == 4: - """ 字符串表示 """ - len_data = 16 - item = display_str(data, len_data) - item = item.decode() - elif current_map[1] == 5: - """ 载波地址表示 """ - len_data = 3 - item = display_str(data, len_data) - item = trans_list_to_str(item) + else: + current_map = modbus_map[idx] + data_label = current_map[0] + if current_map[1] == 1: + """ Hex字符表示 """ + item = data[0] * 0x0100 + data[1] + item = display_hex(item, 4) + elif current_map[1] == 2: + """ 16位数值表示 """ + item = data[0] * 0x0100 + data[1] + if item > 0x8000: + item -= 0x1_0000 + elif current_map[1] == 3: + """ 32位数值表示 """ + item = data[2] * 0x0100 + data[3] + item *= 0x10000 + item += data[0] * 0x0100 + data[1] + if data[2] > 0x80: + item -= 0x1_0000_0000 + data_len = 2 + elif current_map[1] == 4: + """ 字符串表示 """ + data_len = 16 + item = display_str(data, data_len) + item = item.decode() + elif current_map[1] == 5: + """ 载波地址表示 """ + 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.") diff --git a/source/utl_upgrade.py b/source/utl_upgrade.py index 32f3d1e..8367454 100644 --- a/source/utl_upgrade.py +++ b/source/utl_upgrade.py @@ -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__":