import time import socket import random import hashlib from pathlib import Path from . import tools from . import function ParamMap_EnergyRouter = { 0x00: ['编译日期', 4, 6], 0x06: ['编译时间', 4, 5], } class EnergyRouter: """ 能量路由器远程升级测试(未完成) """ def __init__(self, ip="192.168.100.10", port=7, adddr_modbus=0x01, **kwargs): self._ip = ip self._port = port self.flag_print = 'frame_print' in kwargs.keys() self.time_out = kwargs['time_out'] if 'time_out' in kwargs.keys() else 1 self.retry = kwargs['retry'] if 'retry' in kwargs.keys() else 1 self.tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.tcp_socket.connect((self._ip, self._port)) self.block = { 'addr_dev' : adddr_modbus, 'data_define': ParamMap_EnergyRouter, } self.output = { 'result': False, 'code_func': 0x00, } self.log = { 'send': 0, 'read': 0, 'keep-fail': 0, 'record': { 'config': None, 'data': None, }, } def __transfer_data(self, frame: bytes) -> bool: """ 数据传输处理函数 """ if self.tcp_socket is None: print(tools.ByteConv.trans_list_to_str(frame)) return False try: self.tcp_socket.send(frame) time.sleep(self.time_out) frame_recv = self.tcp_socket.recv(128) self.output = function.protocols.check_frame_modbus(frame_recv, self.block) if self.flag_print: print("Read Frame: ", tools.ByteConv.trans_list_to_str(frame_recv)) if 'Regs' in self.output.keys(): function.protocols.print_display(self.output['Regs']) except Exception as ex: print("Error Info: ", ex) if self.flag_print and frame_recv: print("Fail Data: " , tools.ByteConv.trans_list_to_str(frame_recv)) self.output['result'] = False return self.output['result'] def frame_read(self, daddr=0x00, dlen=0x10): self.block['type'] = 'read' self.block['data_addr'] = daddr self.block['data_len'] = dlen frame = function.protocols.make_frame_modbus(self.block) return self.__transfer_data(frame) def frame_update(self, path_bin): """ 程序升级 """ param_saved = self.flag_print, self.retry, self.time_out self.block['type'] = 'update' self.block['step'] = 'start' self.block['file'] = Path(path_bin).read_bytes() self.block['header_offset'] = 128 # 启动帧 frame_master = function.protocols.make_frame_modbus(self.block) if not self.__transfer_data(frame_master): self.flag_print, self.retry, self.time_out = param_saved print('Upgrade Fail: start') return False self.block['file_block_size'] = self.output['upgrade']['length'] # 避免接收到延迟返回报文 time.sleep(0.4) # 文件传输 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: # 更新并发送帧 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] = function.protocols.make_frame_modbus(self.block) self.tcp_socket.send(seq_frame_master[i]) # 接收帧回复 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 Exception("TCP closed.") self.output = function.protocols.check_frame_modbus(seq_frame_slave[i], None) seq_current, seq_hope = self.output['upgrade']['index'], self.output['upgrade']['hope'] if seq_current < seq_offset: raise Exception("Error.") elif self.output['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] == 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 # 结束升级 self.block['step'] = 'end' frame_master = function.protocols.make_frame_modbus(self.block) while self.output['result'] is False: self.tcp_socket.send(frame_master) frame_slave = self.tcp_socket.recv(8) if frame_slave == '': raise Exception("TCP closed.") self.output = function.protocols.check_frame_modbus(frame_slave[:18], self.block) def GeneratePackage_Demo_Xilinx(path_bin: Path): """ 完整升级包生成测试 """ config = { 'file_type': [0x10, 0x01], # Xilinx-Demo 自机升级文件 'file_version': [0x00, 0x00], # 文件版本-00 用于兼容文件格式升级 # 'file_length': [], # 文件长度(自动生成) # 'md5': [], # 文件MD5(自动生成) 'encrypt': [0x01], # 默认加密算法 'update_type': [0x01], # APP升级 'update_spec': [0x00, 0x00, 0x00, 0x00], # 升级特征字 'update_verison': [0x02, 0x00, 0x00, 0x01], # 升级版本号 'update_date': [0x22, 0x04, 0x24], # 升级版本日期 # 'area_code': [], # 省份特征 # 'uptate_str': [], # 升级段描述 # 'device_str': [], # 设备特征描述 # 'hex_name': [], # Hex文件名(自动读取) # 文件Hex结构信息 # 'flash_addr': 0x3E8020, # 程序起始地址 # 'flash_size': 0x005FC0, # 程序空间大小 } data_bin = path_bin.read_bytes() md5_ctx = hashlib.md5() md5_ctx.update(data_bin) config["md5"] = list(md5_ctx.digest()) config['file_length'] = tools.ByteConv.conv_int_to_array(len(data_bin)) config['hex_name'] = list(path_bin.name.encode())[:80] if (header:= function.file_upgrade.build_header(config, 128)) is None: raise Exception("Header tag oversize. ") if (header_512:= function.file_upgrade.build_header(config, 512)) is None: raise Exception("Header tag oversize. ") data_encrypt = function.file_upgrade.file_encryption(data_bin) print("Upgrade file generated successfully.") print(f"\t header_length={len(header)}, bin_length={len(data_bin)}[{hex(len(data_bin))}]") print(f"\t file md5: {tools.ByteConv.trans_list_to_str(config['md5'])}") file1 = path_bin.parent / (path_bin.stem + '.dat') file1.write_bytes(header + data_bin) file2 = path_bin.parent / (path_bin.stem + '_h512.dat') file2.write_bytes(header_512 + data_bin) 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) pass