调整设备文件结构;

简化适配器升级流程;

source/dev_LaminaAdapter.py

@@ -1,8 +1,9 @@
 import time
 from pathlib import Path
 from serial import Serial
-from utl import trans_list_to_str
+from tools.ByteConv import trans_list_to_str
 from func_frame import make_frame_dlt645, check_frame_dlt645
+from func_upgrade import GeneratePackage_SLCP101_p460
 
 modbus_map = {
     # 1 - Hex
@@ -91,8 +92,8 @@ modbus_map = {
     0xA5: ["抖动频率上限",                2],
     0xA6: ["抖动频率下限",                2],
     0xA7: ["电池电压判断限值",            2],
-    0xA8: ["保留",                        1],
-    0xA9: ["保留",                        1],
+    0xA8: ["MPPT追踪模式",                1],
+    0xA9: ["ADC参考电压",                 2],
     0xAA: ["保留",                        1],
     0xAB: ["保留",                        1],
     0xAC: ["保留",                        1],
@@ -139,6 +140,7 @@ class LaminaAdapter:
                 'data_define': modbus_map,    
             },
         }
+        self.output = {}
 
     def __transfer_data(self, frame):
         """ 报文数据传输 """
@@ -168,6 +170,13 @@ class LaminaAdapter:
                 cnt += 1
                 continue
             print(output_text)
+            for line in output_text.split('\n'):
+                line = line.strip()
+                line_info = line.split('\t')
+                if line_info[0] == '0x0100':
+                    self.output['version'] = line_info[2].rstrip('\x00')
+                elif line_info[0] == '0x0082':
+                    self.output['address'] = line_info[2].rstrip('\x00')
             break
         
         if self.flag_print:
@@ -341,6 +350,20 @@ def test_communication(time_out=2):
         print(f"Success Rate: {log_success / (log_success + log_failed) * 100}%")
     dev_lamina.flag_print = saveconfig_print
 
+def make_Pakeage(fp: Path):
+    """ 生成升级包 """
+
+    hex_update = fp
+    file_package = fp.parent / f'{hex_update.stem}.dat'
+    file_update_bin = fp.parent / f'{hex_update.stem}.bin'
+
+    data_package, data_update_bin = GeneratePackage_SLCP101_p460(hex_update)
+
+    file_package.write_bytes(data_package)
+    file_update_bin.write_bytes(data_update_bin)
+
+    return file_package
+
 if __name__=='__main__':
     mode_config = {
         "Log": {'com_name': None, 
@@ -350,17 +373,32 @@ if __name__=='__main__':
                 #   'addr_645': [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], 
                   'frame_print': True, 
                   'time_out': 0.1, 'retry': 1, 'retry_sub': 10},
-        "HPLC":  {'com_name': 'COM4', 'baudrate': 9600, 'parity': 'E', 'bytesize': 8, 'stopbits': 1,
+        "HPLC":  {'com_name': 'COM10', 'baudrate': 9600, 'parity': 'E', 'bytesize': 8, 'stopbits': 1,
                 #   'addr_645': [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], 
                   'frame_print': True, 
                   'time_out': 0.5, 'retry': 3, 'retry_sub': 10},
     }
 
-
     dev_lamina = LaminaAdapter(**mode_config['Debug'])
 
     dev_lamina.frame_read(0x0100, 0x20)
 
+    if 0:
+        dev_lamina.frame_read(0xA9, 1)      # 读ADC参考电压
+        dev_lamina.frame_write_one(0xA9, 2000)      # 写ADC参考电压:2.0V
+        dev_lamina.frame_write_one(0xA9, 3300)      # 写ADC参考电压:3.3V
+        dev_lamina.frame_read(0x13, 1)      # 读输入电压
+        dev_lamina.frame_read(0x16, 1)      # 读输出电压
+        dev_lamina.frame_read(0x97, 4)      # 读校准参数: Vin_a, Vin_b, Vout_a, Vout_b
+        dev_lamina.frame_write_one(0x97, 1000)      # 写校准参数Vin_a: 1.000
+        dev_lamina.frame_write_one(0x97, 1500)      # 写校准参数Vin_a: 1.500
+        dev_lamina.frame_write_one(0x98, 100)       # 写校准参数Vin_b: 1.00
+        dev_lamina.frame_write_one(0x98, 150)       # 写校准参数Vin_b: 1.50
+        dev_lamina.frame_write_one(0x99, 1000)      # 写校准参数Vout_a: 1.000
+        dev_lamina.frame_write_one(0x99, 1500)      # 写校准参数Vout_a: 1.500
+        dev_lamina.frame_write_one(0x9A, 1000)      # 写校准参数Vout_a: 1.00
+        dev_lamina.frame_write_one(0x9A, 1500)      # 写校准参数Vout_a: 1.50
+
     if 0:
         dev_lamina.frame_write_str(0x82, [0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA])
         dev_lamina.frame_write_str(0x82, [0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
@@ -434,18 +472,42 @@ if __name__=='__main__':
         dev_lamina.frame_read(0x0170, 0x30)
 
     if not hasattr(__builtins__,"__IPYTHON__"):
-        path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240604_1200_V1.01a.bin")
+        # file_package = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p460_o1\result\lamina_optimizer_t1.dat")
         # path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240520_0000_T1.11.bin")
-        # 生产镜像版本
-        # path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240525_1800_V1.12.bin")
+        # 工程-即时转换
+        file_hex = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\lamina_adapter\Debug\lamina_adapter.hex")
+        if not file_hex.exists():
+            raise Exception("工程编译目标文件不存在.")
+        file_package = make_Pakeage(file_hex)
+        
         # 江苏发货产品灌装版本
         # path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240603_2100_V1.18.bin")
-        dev_lamina.frame_update(path_bin)
         
-        time.sleep(6)
+        version = "DLSY001_240911_1600_V1.01"
+        addr = [0x24, 0x09, 0x12, 0x00, 0x00, 0x00]
 
-        dev_lamina.frame_read(0x0100, 0x20)
+        while True:
+            """ 自动检测升级流程 """
+            ret = False
+            while not ret or ('version' not in dev_lamina.output.keys()) or (version == dev_lamina.output['version']):
+                dev_lamina.frame_read(0x82, 3)
+                ret = dev_lamina.frame_read(0x0100, 0x20)
+                time.sleep(1)
 
-        dev_lamina.frame_write_one(0x52, 0x01)
+            dev_lamina.frame_update(file_package)
+
+            time.sleep(6)
+
+            ret = dev_lamina.frame_read(0x0100, 0x20)
+
+            if ret and (version == dev_lamina.output['version']):
+                dev_lamina.frame_write_one(0x52, 0x01)
+
+                print(f"address: {' '.join(map(lambda x: ('000' + hex(x)[2:])[-2:], addr))}")
+                dev_lamina.frame_write_str(0x82, addr)
+                dev_lamina.frame_read(0x82, 3)
+                addr[5] += 1
+                if addr[5] & 0x0F >= 10:
+                    addr[5] += 0x10 - 10
 
 

source/dev_LaminaController.py

@@ -1,8 +1,11 @@
 import time
 from pathlib import Path
 from serial import Serial
-from utl import trans_list_to_str
-from func_frame import make_frame_dlt645, check_frame_dlt645
+from datetime import datetime
+from tools.ByteConv import trans_list_to_str
+from tools.IntelHex import file_Bin_to_IntelHex
+from func_frame import make_frame_modbus, check_frame_modbus
+from func_upgrade import GenerateImage_DLSP001_p280039, GeneratePackage_DLSP001_p280039
 
 modbus_map = {
     # 1 - Hex
@@ -11,6 +14,7 @@ modbus_map = {
     # 4 - str
     # 5 - addr
     # 6 - float
+    0x0B: ["事件标志", 1],
     0x0C: ["告警字1", 1],
     0x0D: ["告警字2", 1],
     0x0E: ["故障字1", 1],
@@ -42,6 +46,10 @@ modbus_map = {
     0x51: ["故障清除命令" ,              2],
     0x52: ["参数还原命令" ,              2],
     0x53: ["设备复位命令" ,              2],
+    0x54: ["模式更改命令" ,              2],
+    0x55: ["短时停机命令(未启用)" ,      2],
+    0x56: ["手动录波命令" ,              2],
+    0x57: ["时间配置命令" ,              7, 3],
 
     0x60: ["整机运行使能",               1],
     0x61: ["最小启动允许输入电压",       2],
@@ -70,8 +78,8 @@ modbus_map = {
     0x7B: ["过温告警值",                 2],
     0x7C: ["过温恢复值",                 2],
     0x7D: ["输出继电器故障判断差值",     2],
-    0x7E: ["保留数据项",                1],
-    0x7F: ["保留数据项",                1],
+    0x7E: ["LLC输出电压给定值",          2],
+    0x7F: ["Boost输出电压给定值",        2],
     0x80: ["三点法中间阈值",             2],
     0x81: ["浮充电压",                  2],
     0x82: ["恒压充电电压",               2],
@@ -150,7 +158,7 @@ class LaminaController:
                     break
 
             try:
-                output_text = check_frame_dlt645(frame_recv, block=self.block)
+                output_text = check_frame_modbus(frame_recv, self.block['data'])
             except Exception as e:
                 print(e)
                 cnt += 1
@@ -168,7 +176,7 @@ class LaminaController:
         self.block['data']['type'] = 'read'
         self.block['data']['data_addr'] = daddr
         self.block['data']['data_len'] = dlen
-        frame = make_frame_dlt645(self.block)
+        frame = make_frame_modbus(self.block['data'])
 
         return self.__transfer_data(frame)
 
@@ -176,7 +184,7 @@ class LaminaController:
         self.block['data']['type'] = 'write_one'
         self.block['data']['data_addr'] = daddr
         self.block['data']['data_val'] = dval
-        frame = make_frame_dlt645(self.block)
+        frame = make_frame_modbus(self.block['data'])
 
         return self.__transfer_data(frame)
 
@@ -184,7 +192,7 @@ class LaminaController:
         self.block['data']['type'] = 'write_dual'
         self.block['data']['data_addr'] = daddr
         self.block['data']['data_val'] = dval
-        frame = make_frame_dlt645(self.block)
+        frame = make_frame_modbus(self.block['data'])
 
         return self.__transfer_data(frame)
 
@@ -192,10 +200,216 @@ class LaminaController:
         self.block['data']['type'] = 'write_str'
         self.block['data']['data_addr'] = daddr
         self.block['data']['data_val'] = dval
-        frame = make_frame_dlt645(self.block)
+        frame = make_frame_modbus(self.block['data'])
 
         return self.__transfer_data(frame)
 
+    def frame_record(self):
+        """ 读取录波数据 
+        """
+        self.block['data']['file_block_size'] = 240
+        # 读取config
+        self.block['data']['type'] = 'record_cfg'
+        self.block['data']['step'] = 'start'
+        frame_master = make_frame_modbus(self.block['data'])
+
+        ret, cnt, cnt_sub = True, 0, 0
+        frame_data_cfg = []
+        while ret:
+            self.__com.read_all()
+            self.__com.write(frame_master)
+            frame_slave = None
+            while not frame_slave:
+                time.sleep(0.1)
+                frame_slave = self.__com.read_all()
+                cnt_sub += 1
+                if cnt_sub >= self.retry_sub:
+                    break
+
+            try:
+                ret, *block_cfg = check_frame_modbus(frame_slave, self.block['data'])
+                print(f"{trans_list_to_str(frame_slave[:9])}")
+            except Exception as ex:
+                print(ex)
+                cnt += 1
+                ret = False
+
+            if ret:
+                cnt = 0
+                frame_data_cfg.append(block_cfg[0])
+                if block_cfg[0]['seq'] == 0:
+                    self.block['data']['step'] = 'next'
+                    frame_master = make_frame_modbus(self.block['data'])
+                elif (block_cfg[0]['seq']) >= block_cfg[0]['total']:
+                    ret = False
+            elif cnt < self.retry:
+                ret = True
+
+        # 读取data
+        self.block['data']['type'] = 'record_data'
+        self.block['data']['step'] = 'start'
+        frame_master = make_frame_modbus(self.block['data'])
+
+        ret, cnt, cnt_sub = True, 0, 0
+        frame_data_record = []
+        while ret:
+            self.__com.read_all()
+            self.__com.write(frame_master)
+            time.sleep(0.3)
+            frame_slave = None
+            while not frame_slave:
+                time.sleep(0.1)
+                frame_slave = self.__com.read_all()
+                cnt_sub += 1
+                if cnt_sub >= self.retry_sub:
+                    break
+
+            try:
+                ret, *block_record = check_frame_modbus(frame_slave, self.block['data'])
+                print(f"{trans_list_to_str(frame_slave[:9])}")
+            except Exception as ex:
+                print(ex)
+                cnt += 1
+                ret = False
+
+            if ret:
+                cnt = 0
+                frame_data_record.append(block_record[0])
+                if block_record[0]['seq'] == 0:
+                    self.block['data']['step'] = 'next'
+                    frame_master = make_frame_modbus(self.block['data'])
+                elif (block_record[0]['seq']) >= block_record[0]['total']:
+                    ret = False
+            elif cnt < self.retry:
+                ret = True
+        
+        if len(frame_data_record) == 0:
+            raise Exception("未取得录波数据.")
+
+        # 处理配置信息
+        data_cfg_bare = b"".join([x['data'] for x in frame_data_cfg])
+        config_record = {'LinePos': []}
+        pos = 0
+        if data_cfg_bare[pos+1] != 0xF1 or data_cfg_bare[pos] != 0xF1:
+            Warning("config 配置文件格式异常")
+        pos += 2
+        config_record['LinePos'].append(pos)
+        len_faultword = (data_cfg_bare[3] * 0x100 + data_cfg_bare[2])
+        pos += 2
+        config_record['FaultWord'] = []
+        for i in range(0, len_faultword, 2):
+            faultword = data_cfg_bare[pos+i+1] * 0x100 + data_cfg_bare[pos+i]
+            config_record['FaultWord'].append(faultword)
+        pos += len_faultword
+        config_record['SysStatus'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['FaultRecordPosition'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['FaultNum'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['Standard'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        if data_cfg_bare[pos+1] != 0xF2 or data_cfg_bare[pos] != 0xF2:
+            Warning("config 配置文件格式异常")
+        pos += 2
+        config_record['LinePos'].append(pos)
+        config_record['ChannelNum'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['ChNum_Alg'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['ChNum_Dgt'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        if data_cfg_bare[pos+1] != 0xF3 or data_cfg_bare[pos] != 0xF3:
+            Warning("config 配置文件格式异常")
+        pos += 2
+        config_record['LinePos'].append(pos)
+        config_record['ChannelDescription'] = []
+        config_record['ChannelCoefficient'] = []
+        for i in range(config_record['ChannelNum']):
+            len_str = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+            pos += 2
+            ch_desc = []
+            for j in range(0, len_str, 2):
+                ch_desc.append(data_cfg_bare[pos + j])
+            pos += len_str
+            ch_coe = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+            pos += 2
+            config_record['ChannelDescription'].append(bytearray(ch_desc).decode())
+            config_record['ChannelCoefficient'].append(ch_coe)
+        if data_cfg_bare[pos+1] != 0xF4 or data_cfg_bare[pos] != 0xF4:
+            Warning("config 配置文件格式异常")
+        pos += 2
+        config_record['LinePos'].append(pos)
+        config_record['SysFreq'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['FreqNum'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['SampleFreq'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['SamplePoint'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['DataType'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        config_record['TimeFactor'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
+        pos += 2
+        if data_cfg_bare[pos+1] != 0xF5 or data_cfg_bare[pos] != 0xF5:
+            Warning("config 配置文件格式异常")
+        pos += 2
+        config_record['LinePos'].append(pos)
+        time_stamp = {
+            'year':     data_cfg_bare[pos+1]    * 0x100 + data_cfg_bare[pos],
+            'month':    data_cfg_bare[pos+3]    * 0x100 + data_cfg_bare[pos+2],
+            'day':      data_cfg_bare[pos+5]    * 0x100 + data_cfg_bare[pos+4],
+            'hour':     data_cfg_bare[pos+7]    * 0x100 + data_cfg_bare[pos+6],
+            'minute':   data_cfg_bare[pos+9]    * 0x100 + data_cfg_bare[pos+8],
+            'second':   data_cfg_bare[pos+11]   * 0x100 + data_cfg_bare[pos+10] +
+                (data_cfg_bare[pos+13]   * 0x100 + data_cfg_bare[pos+12]) * 0.001,
+        }
+        config_record['StartTime'] = time_stamp
+        pos += 14
+        time_stamp = {
+            'year':     data_cfg_bare[pos+1]    * 0x100 + data_cfg_bare[pos],
+            'month':    data_cfg_bare[pos+3]    * 0x100 + data_cfg_bare[pos+2],
+            'day':      data_cfg_bare[pos+5]    * 0x100 + data_cfg_bare[pos+4],
+            'hour':     data_cfg_bare[pos+7]    * 0x100 + data_cfg_bare[pos+6],
+            'minute':   data_cfg_bare[pos+9]    * 0x100 + data_cfg_bare[pos+8],
+            'second':   data_cfg_bare[pos+11]   * 0x100 + data_cfg_bare[pos+10] +
+                (data_cfg_bare[pos+13]   * 0x100 + data_cfg_bare[pos+12]) * 0.001,
+        }
+        config_record['TriggerTime'] = time_stamp
+        # 处理录波数据
+        data_record_bare =  b"".join([x['data'] for x in frame_data_record])
+        data_record = []
+        pos = 0
+        while pos < len(data_record_bare):
+            record_point = {
+                'index': data_record_bare[pos+3] * 0x1000 + 
+                    data_record_bare[pos+2] * 0x100 + 
+                    data_record_bare[pos+1] * 0x10 + 
+                    data_record_bare[pos],
+                'timestamp': data_record_bare[pos+7] * 0x1000 + 
+                    data_record_bare[pos+6] * 0x100 + 
+                    data_record_bare[pos+5] * 0x10 + 
+                    data_record_bare[pos + 4],
+                'ChAlg': [],
+                'ChDgt': [],
+            }
+            pos += 8
+            for i in range(config_record['ChNum_Alg']):
+                point_data_alg = data_record_bare[pos+1] * 0x100 + data_record_bare[pos]
+                if data_record_bare[pos+1] & 0x80:
+                    point_data_alg -= 0x10000
+                record_point['ChAlg'].append(point_data_alg)
+                pos += 2
+            for i in range(config_record['ChNum_Dgt']):
+                point_data_dgt = (data_record_bare[pos+(i // 8)] & (0x01 << (i % 8))) == (0x01 << (i % 8))
+                record_point['ChDgt'].append(point_data_dgt)
+                pos += 2
+            data_record.append(record_point)
+
+        return config_record, data_record
+
+
     def frame_update(self, path_bin):
         """ 程序升级
             注意: 在使用单板升级测试时, 需要关闭低电压检测功能, 否则无法启动升级流程;
@@ -207,7 +421,7 @@ class LaminaController:
         self.block['data']['file'] = Path(path_bin).read_bytes()
         self.block['data']['header_offset'] = 184
         # 启动帧
-        frame_master = bytearray(make_frame_dlt645(self.block))
+        frame_master = bytearray(make_frame_modbus(self.block['data']))
 
         # 等待擦除完成返回
         try_times = 30
@@ -220,8 +434,11 @@ class LaminaController:
                 try_times -= 1
                 continue
             
-            _, _, self.block["data"]['file_block_size'] = check_frame_dlt645(frame_slave, self.block)
-            break
+            try:
+                ret, _, self.block["data"]['file_block_size'] = check_frame_modbus(frame_slave, self.block['data'])
+                break
+            except Exception as e:
+                print(e)
 
         if self.block["data"]['file_block_size'] == 0:
             raise Exception("Error slave response.")
@@ -233,7 +450,7 @@ class LaminaController:
         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(self.block))
+            frame_master = bytearray(make_frame_modbus(self.block['data']))
 
             cnt = 0
             while cnt < self.retry:
@@ -250,7 +467,7 @@ class LaminaController:
                         break
 
                 try:
-                    ret = check_frame_dlt645(frame_slave, self.block)
+                    ret = check_frame_modbus(frame_slave, self.block['data'])
                 except Exception as e:
                     print(e)
                     ret = False, 0, 0
@@ -267,7 +484,7 @@ class LaminaController:
 
         # 结束升级
         self.block["data"]['step'] = 'end'
-        frame_master = bytearray(make_frame_dlt645(self.block))
+        frame_master = bytearray(make_frame_modbus(self.block['data']))
 
         cnt = 0
         while cnt < self.retry:
@@ -284,7 +501,7 @@ class LaminaController:
                     break
 
             try:
-                ret = check_frame_dlt645(frame_slave, self.block)
+                ret = check_frame_modbus(frame_slave, self.block['data'])
             except Exception as e:
                 print(e)
                 ret = False, 0, 0
@@ -308,7 +525,7 @@ def test_communication(time_out=2):
     dev_lamina.flag_print = False
     try:
         while 1:
-            if dev_lamina.frame_read(0x0E, 0x13):
+            if dev_lamina.frame_read(0x0C, 0x20):
                 log_success += 1
                 cnt_failedseries = 0
             else:
@@ -329,22 +546,122 @@ def test_communication(time_out=2):
         print(f"Success Rate: {log_success / (log_success + log_failed) * 100}%")
     dev_lamina.flag_print = saveconfig_print
 
-if __name__=='__main__':
-    mode_config = {
-        "Log": {'com_name': None, 
-                # 'addr_645': [0x01, 0x00, 0x00, 0x00, 0x00, 0x40], 
-                },
-        "Debug": {'com_name': 'COM8', 'baudrate': 250000, 'parity': 'E', 'bytesize': 8, 'stopbits': 1,
-                  'addr_645': [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], 
-                  'frame_print': True, 
-                  'time_out': 0.2, 'retry': 1, 'retry_sub': 10},
-    }
+
+def test_record(path_CFG=None, path_data=None):
+    """ 执行录波数据读取流程 """
+    if path_CFG is None: path_CFG = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p280039\record") / f"record_{'-'.join(time.ctime().replace(':', '').split(' '))}.cfg"
+    if path_data is None: path_data = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p280039\record") / f"record_{'-'.join(time.ctime().replace(':', '').split(' '))}.dat"
+
+    config, data = dev_lamina.frame_record()
+
+        # Header File
+        
+        # Config File
+    text_cfg = ""
+    text_cfg += f"LaminaController,1,{config['Standard']}\r\n"
+    text_cfg += f"{config['ChannelNum']},{config['ChNum_Alg']}A,{config['ChNum_Dgt']}D\r\n"
+    for i in range(config['ChNum_Alg']):
+        if config['ChannelDescription'][i][-1] == 'V':
+            text_cfg += f"{config['ChannelDescription'][i]},{1/config['ChannelCoefficient'][i]},0,0,-32767,32767,0.001,1,S\r\n"
+        elif config['ChannelDescription'][i][-1] == 'A':
+            text_cfg += f"{config['ChannelDescription'][i]},{1/config['ChannelCoefficient'][i]},0,0,-32767,32767,1,1,S\r\n"
+        else:
+            raise Exception("Unknow Channel Units.")
+    for i in range(config['ChNum_Dgt']):
+        text_cfg += f"{config['ChannelDescription'][i + config['ChNum_Alg']]}\r\n"
+    text_cfg += f"{config['SysFreq']}\r\n"
+    text_cfg += f"{config['FreqNum']}\r\n"
+    for i in range(config['FreqNum']):
+        text_cfg += f"{config['SampleFreq']},{config['SamplePoint']}\r\n"
+        # text_cfg += f"{datetime(**config['StartTime']).strftime('%y/%m/%d,%H:%M:%S')}\r\n"
+        # text_cfg += f"{datetime(**config['TriggerTime']).strftime('%y/%m/%d,%H:%M:%S')}\r\n"
+    text_cfg += f"{datetime.now().strftime('%y/%m/%d,%H:%M:%S')}\r\n"
+    text_cfg += f"{datetime.now().strftime('%y/%m/%d,%H:%M:%S')}\r\n"
+    text_cfg += f"ASCII\r\n"
+    text_cfg += f"1\r\n"
+    path_CFG.write_text(text_cfg)
+
+        # Data File
+    text_record = ""
+    for point in data:
+            # line_record = f"{point['index']},{point['timestamp'] * 1000},"
+        line_record = f"{point['index']},{30},"
+        for data_alg in point['ChAlg']:
+            line_record += f"{data_alg},"
+        for data_dgt in point['ChDgt']:
+            line_record += f"{data_dgt},"
+        text_record += line_record + "\r\n"
+    path_data.write_text(text_record)
 
 
-    dev_lamina = LaminaController(**mode_config['Debug'])
+def make_Image():
+    """ 叠光控制器DSP镜像与升级包生成流程 """
+    root = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p280039")
+    result = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p280039\result")
 
-    dev_lamina.frame_read(0x0100, 0x20)
+    # 正常启动镜像
+    hex_boot = Path(r"D:\WorkingProject\LightStackOptimizer\software\lamina_controller_dsp\bootloader\Debug\bootloader.hex")
+    hex_main = Path(r"D:\WorkingProject\LightStackOptimizer\software\lamina_controller_dsp\lamina_controller_dsp\DEBUG\lamina_controller_dsp.hex")
+    hex_back = root / r"DLSP001_240828_0900_B1.01.hex"
 
+    file_image = result / f'{hex_main.stem}_ROM.hex'
+    file_main_header = result / 'header_main.bin'
+    file_back_header = result / 'header_back.bin'
+
+    data_bins = GenerateImage_DLSP001_p280039(hex_boot, hex_main, hex_back)
+
+    file_main_header.write_bytes(data_bins[1])
+    file_back_header.write_bytes(data_bins[2])
+
+    data_hex = file_Bin_to_IntelHex(data_bins[0], 0x80000, memory_width=2)
+    file_image.write_text(data_hex)
+
+    # 异常镜像-主分区md5错误
+    file_image1 = result / f'{file_image.stem}_b1.hex'
+    data_image = data_bins[0].copy()
+    data_image[2 * 0x006018: 2 * 0x00601A] = [0x00, 0x01, 0x00, 0x02]
+    data_hex = file_Bin_to_IntelHex(data_image, 0x80000, memory_width=2)
+    file_image1.write_text(data_hex)
+
+    # 异常镜像-备份分区md5错误
+    file_image2 = result / f'{file_image.stem}_b2.hex'
+    data_image = data_bins[0].copy()
+    data_image[2 * 0x007018: 2 * 0x00701A] = [0x00, 0x01, 0x00, 0x02]
+    data_hex = file_Bin_to_IntelHex(data_image, 0x80000, memory_width=2)
+    file_image2.write_text(data_hex)
+
+    # 异常镜像-双分区md5错误
+    file_image3 = result / f'{file_image.stem}_b3.hex'
+    data_image = data_bins[0].copy()
+    data_image[2 * 0x006018: 2 * 0x00601A] = [0x00, 0x01, 0x00, 0x02]
+    data_image[2 * 0x007018: 2 * 0x00701A] = [0x00, 0x01, 0x00, 0x02]
+    data_hex = file_Bin_to_IntelHex(data_image, 0x80000, memory_width=2)
+    file_image3.write_text(data_hex)
+
+
+def make_Pakeage(fp: Path):
+    """ 生成升级包 """
+
+    hex_update = fp
+    file_package = fp.parent / f'{hex_update.stem}.dat'
+    file_update_bin = fp.parent / f'{hex_update.stem}.bin'
+    file_package_buffer = fp.parent / f'{hex_update.stem}.datbuffer'
+
+    data_package, data_update_bin = GeneratePackage_DLSP001_p280039(hex_update)
+    data_buffer = bytearray(2 * len(data_package))
+
+    for i in range(len(data_package)):
+        data_buffer[2*i] = data_package[i]
+
+
+    file_package.write_bytes(data_package)
+    file_update_bin.write_bytes(data_update_bin)
+    file_package_buffer.write_bytes(data_buffer)
+
+    return file_package
+
+def test():
+    """ 测试用代码 """
     if 0:
         dev_lamina.frame_write_str(0x82, [0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA])
         dev_lamina.frame_write_str(0x82, [0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
@@ -417,19 +734,51 @@ if __name__=='__main__':
         time.sleep(2)
         dev_lamina.frame_read(0x0170, 0x30)
 
-    if not hasattr(__builtins__,"__IPYTHON__"):
-        path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240625_2030_V1.04.bin")
-        # path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240520_0000_T1.11.bin")
-        # 生产镜像版本
-        # path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240525_1800_V1.12.bin")
-        # 江苏发货产品灌装版本
-        # path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240603_2100_V1.18.bin")
-        dev_lamina.frame_update(path_bin)
+
+if __name__=='__main__':
+    mode_config = {
+        "Log": {'com_name': None, 
+                # 'addr_645': [0x01, 0x00, 0x00, 0x00, 0x00, 0x40], 
+                },
+        "Debug": {'com_name': 'COM3', 'baudrate': 115200, 'parity': 'N', 'bytesize': 8, 'stopbits': 1,
+                #   'addr_645': [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], 
+                  'frame_print': True, 
+                  'time_out': 0.2, 'retry': 3, 'retry_sub': 10},
+    }
+
+    dev_lamina = LaminaController(**mode_config['Debug'])
+
+    dev_lamina.frame_read(0x0100, 0x20)
+
+
+    if not hasattr(__builtins__,"__IPYTHON__") and 0:   # 
+        """ 读取故障录波数据 """
+        path_CFG = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p280039\result\record4.cfg")
+        path_data = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p280039\result\record4.dat")
+        # 手动触发录波
+        dev_lamina.frame_write_one(0x56, 1)
+
+        test_record(path_CFG, path_data)
+
+    if not hasattr(__builtins__,"__IPYTHON__"):   #  and 0
+        """ 升级流程 """
+        path_project = Path("D:\WorkingProject\LightStackOptimizer\software\lamina_controller_dsp\lamina_controller_dsp")
+        file_hex = path_project / "DEBUG\lamina_controller_dsp.hex"
+        # file_hex = Path(r"C:\Users\wrqal\Documents\WXWork\1688856624403708\Cache\File\2024-09\lamina_controller_dsp(1).hex")
+        if not file_hex.exists():
+            raise Exception("工程编译目标文件不存在.")
+        file_package = make_Pakeage(file_hex)
+
+        dev_lamina.frame_write_one(0x60, 0)
+
+        time.sleep(1)
+
+        dev_lamina.frame_update(file_package)
 
         time.sleep(6)
 
         dev_lamina.frame_read(0x0100, 0x20)
 
-        dev_lamina.frame_write_one(0x52, 0x01)
+        # dev_lamina.frame_write_one(0x52, 0x01)
 
 

source/device/EnergyRouter.py

@@ -2,7 +2,7 @@ import time
 import socket
 import random
 from pathlib import Path
-from utl import trans_list_to_str
+from tools.ByteConv import trans_list_to_str
 from func_frame import make_frame_modbus, check_frame_modbus
 
 modbus_map = {

source/func_frame.py

@@ -1,6 +1,6 @@
 import struct
 from crc import Calculator, Crc16
-from utl import display_hex, trans_list_to_str
+from tools.ByteConv import trans_list_to_str, trans_str_to_list, display_hex
 
 modbus_map = {
     # 1 - Hex
@@ -17,6 +17,19 @@ modbus_map = {
     0x20: ["Float示例", 6],
 }
 
+frame_modbus = {
+    # func: len_base, type, idx
+        # type = 0, fixed length
+        # type = 1, add uint8 length by idx
+        # type = 2, add uint16 length by idx
+        # type = 3, check sub func code by idx, choose fixed length
+    0x03: [5, 1, 2],
+    0x04: [5, 1, 2],
+    0x06: [8, 0],
+    0x07: [8, 3, 2, {0x01: 2, 0x02: 0, 0x03: 0}],
+    0x10: [8, 0],
+    0x11: [11, 2, 7]
+}
 
 def make_frame_modbus(block:dict):
     """ modbus 生成函数"""
@@ -57,6 +70,29 @@ def make_frame_modbus(block:dict):
             frame.append(0x00)
         else:
             raise Exception("Modbus Update Frame Step Error.")
+    elif block['type'][:6] == 'record':
+        """ 录波系列报文 """
+        frame.append(0x11)
+        frame_types = block['type'].split('_')
+
+        if block['type'][7:10] == 'cfg':
+            frame.append(0x01)
+        elif block['type'][7:11] == 'data':
+            frame.append(0x02)
+        else:
+            raise Exception("Modbus Record Frame Type Error.")
+
+        if block['step'] == 'start':
+            frame.append(0x00)
+        elif block['step'] == 'next':
+            frame.append(0x01)
+        elif block['step'] == 'repeat':
+            frame.append(0x02)
+        else:
+            raise Exception("Modbus Record Frame Step Error.")
+        
+        frame.append(block['file_block_size'] // 256)
+        frame.append(block['file_block_size'] % 256)
     else:
         """ 数据读取系列报文 """
         data_addr = block['data_addr']
@@ -200,6 +236,10 @@ def display_data(modbus_map: dict, address: int, data: list):
                 temp = [data[2], data[3], data[0], data[1]]
                 item = struct.unpack('>f', bytes(temp))[0]
                 data_len = 2
+            elif current_map[1] == 7:
+                """ 正序数值表示 """
+                data_len = current_map[2]
+                item = " ".join(map(lambda x: str(x), data[:2 * data_len]))
 
         if len_max < len(data_label):
             len_max = len(data_label)
@@ -211,13 +251,64 @@ def display_data(modbus_map: dict, address: int, data: list):
         output += output_line + "\n"
     return output
 
+def find_frame_modbus(buffer, address, frame_defines: dict=frame_modbus):
+    """ 搜索合法modbus帧子串 """
+    len_buffer = len(buffer)
+
+    pos_frame, len_frame = 0, 0
+    calculator = Calculator(Crc16.MODBUS)
+    for i in range(len_buffer):
+        if buffer[i] != address:
+            continue
+        if (buffer[i+1] & 0x7F) not in frame_defines.keys():
+            continue
+        
+        frame_define = frame_defines[buffer[i+1] & 0x7F]
+        j = frame_define[0]
+        if buffer[i+1] & 0x80:
+            j = 5
+        elif frame_define[1] == 0:
+            pass
+        elif frame_define[1] == 1:
+            j += buffer[i+frame_define[2]]
+        elif frame_define[1] == 2:
+            j += buffer[i+frame_define[2]] * 0x100 + buffer[i+frame_define[2]+1]
+        elif frame_define[1] == 3:
+            if buffer[i+frame_define[2]] not in frame_define[3].keys():
+                continue
+            j += frame_define[3][buffer[i+frame_define[2]]]
+        else:
+            raise Exception("Unknow Modbus Define type.")
+
+        if ((i+j) <= len_buffer) and calculator.checksum(buffer[i:i+j-2]) == (buffer[i+j-1] * 0x100 + buffer[i+j-2]):
+            pos_frame, len_frame = i, j
+            break
+    
+    return buffer[pos_frame: pos_frame+len_frame]
+
+def find_frame_dlt645(buffer, address: list):
+    """ 搜索合法645帧子串 """
+    len_buffer = len(buffer)
+
+    pos_frame, len_frame = 0, 0
+    for i in range(len_buffer):
+        if buffer[i] != 0x68 or buffer[i+7] != 0x68:
+            continue
+        if address[0] != 0xAA and buffer[i+1:i+7] == bytes(address):
+            continue
+
+        j = buffer[i+9] + 12
+        if sum(buffer[i:i+j-2]) % 0x100 == buffer[i+j-2]:
+            pos_frame, len_frame = i, j
+            break
+    
+    return buffer[pos_frame: pos_frame+len_frame]
+
 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.")
-    elif frame[0] != 0x01:
-        raise Exception("Frame device address error.")
+
+    if len(frame:=find_frame_modbus(frame, block['addr_dev'])) == 0:
+        raise Exception("No frame data")
     
     code_func = frame[1]
     code_subfunc = frame[2]
@@ -246,24 +337,41 @@ def check_frame_modbus(frame, block=None):
                 data_addr = block['data_addr']
                 return display_data(block['data_define'], data_addr, list(frame[3:-2]))
             else:
-                return list(frame[3:-2])
+                return trans_list_to_str(frame[3:-2])
         else:
             raise Exception("Frame read error.")
     elif code_func == 0x06:
         """ 单个数据写入帧 """
-        return frame[2:-2]
+        return trans_list_to_str(frame[2:-2])
     elif code_func == 0x10:
         """ 多个数据写入帧 """
-        return frame[2:-2]
+        return trans_list_to_str(frame[2:-2])
+    elif code_func == 0x11:
+        """ 录波功能帧 """
+        frame_data = {
+            'seq': frame[5] * 0x100 + frame[6],
+            'total': frame[3] * 0x100 + frame[4],
+            'data': frame[9:-2]
+        }
+        if frame[2] == 0x01:
+            frame_data['type'] = 'config'
+        elif frame[2] == 0x02:
+            frame_data["type"] = 'data'
+        else:
+            raise Exception("Unknow data type")
+        return True, frame_data
+    elif code_func & 0x80:
+        """ 错误返回帧 """
+        if code_func & 0x7F == 0x07:
+            return False, frame[3], code_func, code_subfunc
+        return False, frame[2], code_func
     else:
         raise Exception(f"Frame Date error. func={code_func}, func_sub={code_subfunc}, len={len(frame)}")
 
 def check_frame_dlt645(frame, block=None):
     """ 校验dlt645帧回复 """
-    if sum(frame[:-2]) % 0x100 != frame[-2]:
-        raise Exception("DLT645 Frame Verify error.")
-    elif len(frame[10:-2]) != frame[9]:
-        raise Exception("DLT645 Frame len error.")
+    if len(frame:=find_frame_dlt645(frame, block['addr'])) == 0:
+        raise Exception("No frame data")
     
     code_func = frame[8]
     if code_func == 0x9F:
@@ -272,3 +380,110 @@ def check_frame_dlt645(frame, block=None):
     else:
         raise Exception("DLT645 Frame type error.")
 
+
+if __name__ == "__main__":
+    frame_slave = '01 07 02 00 00 A7 00 F0 A1 A4 1F DC 0B 1D 27 08 0F CB A7 49 03 99 09 C6 87 10 FB 08 86 71 9F 2A 2F BB 8F 69 4D 5C 9F 90 51 8A 8B D3 E0 85 9F 2B C1 9A 7E A3 22 9B 29 EE 6B 70 28 D3 31 F6 EF 78 43 8A DF FC F3 8C 13 02 31 F4 65 B5 EE 46 80 F2 E9 D4 E9 C8 F2 84 13 3A DF 50 1D 45 53 52 1D 89 6F F8 CB 7F 56 28 DF A2 11 D4 47 93 04 04 FF AB 35 1F C3 BA D2 F0 65 F2 F6 A3 AC A5 A2 AF AF 3E 88 65 EC 7B 35 62 DA 4A CF A4 69 A5 9E C7 70 E6 DC DD BE 49 DD DA 09 CE 42 18 5C 57 86 12 E0 A0 74 5F 5C F7 35 B3 FE 7C 51 94 5C 57 28 1A 86 1E 9F DE F6 B2 4B A9 29 E6 30 EA F2 BB E6 72 81 E9 80 3A DE FC DC C2 F8 8E 30 F4 66 B3 25 12 30 FA 90 09 3C 8C 1D FD 49 8E 4A 58 1C 17 48 54 CF 6A DE B4 05 7F 3D DD 68 F2 7E C2 CE 01 11 D6 DE 5E 7C 27 10 FE 28 28 3E 06 4D C8 01 07 02 00 00 A7 F4 08'
+    buffer_test = trans_str_to_list("62 01 03 25 1C 00 01 03 3A 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 10 01 F3 00 06 00 12 01 DB 00 19 00 BA 00 00 24 62 00 00 25 1C 00 00 00 0F 00 01 00 07 00 2B 02 4D FD 6C FD 6C 01 20 01 20 73 08 4D FD 6C FD")
+    buffer_test1 = trans_str_to_list("01 03 3A 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 11 01 F4 00 06 00 13 01 DB 00 1A 00 D5 00 00 25 AE 00 00 26 83 00 00 00 0C FF FF 00 56 00 72 02 1D FD 6C FD 6C 01 26 01 26 94 E3")
+    buffer_test2 = trans_str_to_list("01 06 00 60 00 00 89 D4 94 E3")
+    buffer_test3 = trans_str_to_list(frame_slave)
+    modbus_map = {
+        # 1 - Hex
+        # 2 - Int16
+        # 3 - lnt32
+        # 4 - str
+        # 5 - addr
+        # 6 - float
+        0x0C: ["告警字1", 1],
+        0x0D: ["告警字2", 1],
+        0x0E: ["故障字1", 1],
+        0x0F: ["故障字2", 1],
+        0x10: ["系统工作状态" ,              1],
+        0x11: ["Boost1工作状态" ,            1],
+        0x12: ["Boost2工作状态" ,            1],
+        0x13: ["开关机状态" ,                1],
+        0x14: ["光伏组串1输入电压" ,         2],
+        0x15: ["光伏组串2输入电压" ,         2],
+        0x16: ["Boost1电感电流" ,            2],
+        0x17: ["Boost2电感电流" ,            2],
+        0x18: ["Boost输出电压" ,             2],
+        0x19: ["Boost输出总电流" ,           2],
+        0x1A: ["Boost1功率" ,                3],
+        0x1C: ["Boost2功率" ,                3],
+        0x1E: ["输入总功率" ,                3],
+        0x20: ["LLC输出电压" ,               2],
+        0x21: ["端口输出电压" ,              2],
+        0x22: ["LLC输出电流均值" ,           2],
+        0x23: ["LLC输出电流峰值" ,           2],
+        0x24: ["绝缘检测电压" ,              2],
+        0x25: ["散热片温度" ,                2],
+        0x26: ["腔体1温度" ,                 2],
+        0x27: ["腔体2温度" ,                 2],
+        0x28: ["设备温度" ,                  2],
+
+        0x50: ["启停控制命令" ,              2],
+        0x51: ["故障清除命令" ,              2],
+        0x52: ["参数还原命令" ,              2],
+        0x53: ["设备复位命令" ,              2],
+
+        0x60: ["整机运行使能",               1],
+        0x61: ["最小启动允许输入电压",       2],
+        0x62: ["最大启动允许输入电压",       2],
+        0x63: ["最小停机输入电压",           2],
+        0x64: ["最大停机输入电压",           2],
+        0x65: ["最小启动允许输出电压",       2],
+        0x66: ["最大启动允许输出电压",       2],
+        0x67: ["最小停止允许输出电压",       2],
+        0x68: ["最大停止允许输出电压",       2],
+        0x69: ["最小MPPT电流限值",           2],
+        0x6A: ["最大MPPT电流限值",           2],
+        0x6B: ["保留数据项",                 2],
+        0x6C: ["最大功率限值",               3],
+        0x6E: ["最大功率限值存储值",         3],
+        0x70: ["Boost输入过压保护值",        2],
+        0x71: ["Boost输出过压保护值",        2],
+        0x72: ["LLC输出过压保护值",          2],
+        0x73: ["LLC输出欠压保护值",          2],
+        0x74: ["Boost电感过流保护值",        2],
+        0x75: ["LLC输出电流均值保护值",      2],
+        0x76: ["LLC输出电流峰值保护值",      2],
+        0x77: ["保留数据项",                 2],
+        0x78: ["过载保护值",                 3],
+        0x7A: ["过温故障值",                 2],
+        0x7B: ["过温告警值",                 2],
+        0x7C: ["过温恢复值",                 2],
+        0x7D: ["输出继电器故障判断差值",     2],
+        0x7E: ["保留数据项",                1],
+        0x7F: ["保留数据项",                1],
+        0x80: ["三点法中间阈值",             2],
+        0x81: ["浮充电压",                  2],
+        0x82: ["恒压充电电压",               2],
+        0x83: ["llc软起开始电压",            2],
+        0x84: ["boost开始运行电压",          2],
+        0x85: ["boost停止运行电压",          2],
+        0x86: ["绝缘检测正阻抗限值",          3],
+        0x88: ["绝缘检测负阻抗限值",          3],
+        0x8A: ["保留地址项",                 2],
+        0x8B: ["保留地址项",                 2],
+        0x8C: ["保留地址项",                 2],
+        0x8D: ["保留地址项",                 2],
+        0x8E: ["保留地址项",                 2],
+        0x8F: ["保留地址项",                 2],
+
+        0x100: ["程序版本字符串",         4, 16],
+        0x110: ["设备型号字符串",         4, 16],
+        0x120: ["保留地址项",             4, 16],
+        0x130: ["生产厂家字符串",         4,  8],
+        0x138: ["保留地址项",             4,  8],
+        0x140: ["保留地址项",             4, 16],
+        0x150: ["保留地址项",             4, 16],
+        0x160: ["硬件版本字符串",         4, 16],
+        0x170: ["设备序列号",             4, 16],
+        0x180: ["设备MES码",              4, 16],
+        0x190: ["出厂日期批次",           4, 16],
+    }
+    block = {
+        'addr_dev' : 0x01,
+        'data_define': modbus_map,    
+    }
+    check_frame_modbus(bytearray(buffer_test3), block)

source/func_upgrade.py

@@ -0,0 +1,704 @@
+# 升级包生成脚本
+import hashlib
+from pathlib import Path
+from datetime import datetime
+from crc import Calculator, Crc16
+
+from tools.ByteConv import trans_list_to_str, conv_int_to_array
+from tools.IntelHex import file_IntelHex_to_Bin, file_Bin_to_IntelHex
+
+
+Header_Tag = {
+    'file_type':      [0x00,  2],               # 0 - 文件类型; 2byte
+    'file_version':   [0x01,  2],               # 1 - 文件版本; 2byte
+    'file_length':    [0x02,  4],               # 2 - 文件长度; 4byte
+    'md5':            [0x03, 16],               # 3 - 文件MD5; 16byte
+    'encrypt':        [0x04,  1],               # 4 - 加密算法; 1byte
+    'update_type':    [0x05,  1],               # 5 - 升级文件类别; 1byte
+    'update_spec':    [0x06,  4],               # 6 - 升级特征字; 4byte
+    'update_verison': [0x07,  4],               # 7 - 升级版本号; 4byte
+    'update_date':    [0x08,  3],               # 8 - 升级版本日期; 3byte
+    'area_code':      [0x09,  4],               # 9 - 省份特征; 4byte
+    'uptate_str':     [0x0A, -1, 64],           # 10 - 升级段描述; less than 64byte
+    'device_str':     [0x0D, -1, 64],           # 13 - 设备特征描述; less than 64byte
+    'hex_name':       [0xFF, -1, 80],           # 255 - Hex文件名; less than 80byte
+}
+
+
+def file_encryption(buffer):
+    """ 文件加密算法 """
+    pwd_idx = 0
+    pwd = b'moc.mmocspot.www'
+    pwd = list(map(lambda x: (x - 0x30 + 0x100) % 0x100, pwd))
+    result = bytearray(len(buffer))
+    for i in range(len(buffer)):
+        k = i
+        k |= i >> 8
+        k |= i >> 16
+        k |= i >> 24
+        result[i] = buffer[i] ^ pwd[pwd_idx] ^ (k & 0xFF)
+        pwd_idx = (pwd_idx + 1) % len(pwd)
+    return result
+
+
+def build_header(config: dict, len_max=512):
+    """
+        基于配置参数, 生成文件信息头;
+        V1版本, 依据字典生成tag标签组;
+    """
+    # 定义文件头
+    m_file_header = bytearray(len_max)
+
+    header_len = 11
+    tag_num = 0
+    for tag, value in config.items():
+        if tag in Header_Tag.keys():
+            if tag == 'hex_name' and len_max < 256:
+                """ 当文件头长度不足时, 跳过文件名标签 """
+                continue
+            elif 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]
+            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 > len_max:
+        return None
+    else:
+        return m_file_header
+
+
+def build_header_lite(config: dict):
+    """
+        基于配置参数, 生成文件信息头;
+        V2版本, 仅提供必要信息;
+    """
+    # 定义文件头
+    m_file_header = bytearray(64)
+    m_file_header[0:4] = bytearray(config["update_verison"])
+    m_file_header[4:8] = bytearray(config["file_length"])
+    m_file_header[8:24] = bytearray(config["md5"])
+
+    return m_file_header
+
+
+def build_header_new(config: dict):
+    """
+        基于配置参数, 生成新版文件信息头;
+        V3版本, 依据新版格式填充数据;
+    """
+    # 定义文件头
+    m_file_header = [0xFF] * 184
+
+    m_file_header[0:8] = list(b"TOPSCOMM")
+    m_file_header[8:10] = config['prod_type']
+    m_file_header[10:22] = config['prog_id'] + [0] * (12 - len(config['prog_id']))
+    if config['method_compress'] == True:
+        m_file_header[23] = 0x01
+    else:
+        m_file_header[23] = 0x00
+
+    if 'crc32' in config.keys():
+        m_file_header[22] = 0x00
+        m_file_header[24: 40] = config['crc32'] + [0x00] * 12
+    elif 'md5' in config.keys():
+        m_file_header[22] = 0x01
+        m_file_header[24: 40] = config['md5']
+    else:
+        raise Exception("Error, Unknown method verify.")
+    
+    # 时间戳生成
+    time_now = datetime.now()
+    time_stamp = list(map(lambda x: int(x), 
+                                     time_now.strftime("%Y-%m-%d-%H-%M").split('-')))
+    time_stamp.insert(1, time_stamp[0] // 0x100)
+    time_stamp[0] = time_stamp[0] % 0x100
+    m_file_header[40: 46] = time_stamp
+
+    m_file_header[46: 50] = config['file_length']
+    # Cpu1
+    m_file_header[50: 54] = [0x00] * 4
+    m_file_header[54: 70] = [0x00] * 16
+    # Cpu2
+    m_file_header[70: 74] = [0x00] * 4
+    m_file_header[74: 90] = [0x00] * 16
+
+    if config['prog_type'] == 'app':
+        m_file_header[90: 92] = [0x00, 0x00]
+    elif config['prog_type'] == 'boot':
+        m_file_header[90: 92] = [0x01, 0x00]
+    elif config['prog_type'] == 'diff':
+        m_file_header[90: 92] = [0x02, 0x00]
+    elif config['prog_type'] == 'font':
+        m_file_header[90: 92] = [0x03, 0x00]
+    elif config['prog_type'] == 'config':
+        m_file_header[90: 92] = [0x04, 0x00]
+    elif config['prog_type'] == 'data':
+        m_file_header[90: 92] = [0x05, 0x00]
+    elif config['prog_type'] == 'test':
+        m_file_header[90: 92] = [0x06, 0x00]
+    else:
+        raise Exception("Error, Unknown Program Type.")
+    
+    m_file_header[92: 94] = config['area_code']
+    m_file_header[94: 158] = config['hex_name']
+    if 'upgrade_type' in config.keys():
+        m_file_header[158: 160] = config['upgrade_type']
+        m_file_header[160: 182] = [0x00] * 22
+    else:
+        m_file_header[158: 182] = [0x00] * 24
+
+    m_file_header = bytearray(m_file_header)
+
+    calculator = Calculator(Crc16.MODBUS)
+    code_crc16 = calculator.checksum(m_file_header[:-2])
+    m_file_header[182: 184] = [code_crc16 // 0x100, code_crc16 % 0x100]
+
+    return m_file_header
+
+
+def test1(fp: Path):
+    """ bin文件生成测试 """
+    file1 = Path(fp)
+    path1 = file1.parent / (file1.stem + ".bin")
+    data1 = file1.read_text()
+    data2 = file_IntelHex_to_Bin(data1, 0x3F0100)
+    path1.write_bytes(data2)
+
+    # 测试Bin到IntelHex
+    bin = Path(fp).read_bytes()
+    result = file_Bin_to_IntelHex(bin, 0x80000, memory_width=2)
+    (fp.parent / (fp.stem + ".hex")).write_text(result)
+
+
+def test2():
+    """ 校验, 加密测试 """
+    # Header - Crc16
+    bin_offcial = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p460\result\lamina_adapter_t1.dat")
+    data_offcial = bin_offcial.read_bytes()
+    byte_data = data_offcial[:182]
+    crc = data_offcial[182:184]
+    # data = "54 4F 50 53 43 4F 4D 4D 45 00 53 4C 43 50 30 30 31 00 00 00 00 00 01 00 B6 61 A8 73 BF 82 9E A7 4C 79 F6 BB 94 E2 A5 18 E8 07 05 18 0B 17 18 4C 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 6C 61 6D 69 6E 61 5F 61 64 61 70 74 65 72 5F 6D 61 69 6E 02 00 00 00 00 20 10 53 06 00 00 00 00 80 A5 8F 02 00 00 00 00 EC 1F 40 00 00 00 00 00 00 00 00 00 00 00 00 00 7A CA 61 0A FD 7F 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00"
+    # byte_data = list(map(lambda x: int(x, 16), data.split(' ')))
+    # crc = "99 CB"
+    calculator = Calculator(Crc16.MODBUS)
+    code_crc16 = calculator.checksum(bytearray(byte_data))
+    print(f"Result = {hex(code_crc16)}, Offcial Result = {crc}")
+
+    # File - md5
+    data_bin = '123456 123456 '.encode()
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(data_bin)
+    hash = md5_ctx.hexdigest()
+    # File - encrypt
+    buffer1 = data_bin[:]
+    buffer1_en = file_encryption(buffer1)
+    buffer2 = buffer1_en[:6] + buffer1[6:]
+    buffer2_de = file_encryption(buffer2)
+    pass
+
+
+def task5():
+    """ 文件缓冲区对比测试 """
+    file_dat = Path(r"test\p280039\result\lamina_controller_dsp_t1.dat")
+    file_dat_buffer = Path(r"test\p280039\result\lamina_controller_dsp_buffer.bin")
+    file_dat_buffer.exists(), file_dat.exists()
+    data_dat = file_dat.read_bytes()
+    data_dat_buffer = file_dat_buffer.read_bytes()
+    for i in range(len(data_dat)):
+        if data_dat[i] != data_dat_buffer[2*i]:
+            print(f"Diff in {hex(i)}, Data: {data_dat[i]}!={data_dat_buffer[2*i]}")
+
+
+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'] = conv_int_to_array(len(data_bin))
+    config['hex_name'] = list(path_bin.name.encode())[:80]
+
+    if (header:= build_header(config, 128)) is None:
+        raise Exception("Header tag oversize. ")
+    if (header_512:= build_header(config, 512)) is None:
+        raise Exception("Header tag oversize. ")
+    
+    data_encrypt = 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: {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)
+
+
+def GenerateImage_SLCP001_p4a0(path_boot: Path, path_main: Path, path_back: Path):
+    """ 叠光适配器-4A0平台版本 镜像生成 """
+    config = {
+        'prod_type':        [0x45, 0x00],       # 产品类型
+        'method_compress':  False,              # 文件压缩
+        'prog_id':          list(b"SLCP001"),   # 程序识别号
+        'prog_type':        'app',              # 程序类型
+        'area_code':        [0x00, 0x00],       # 地区
+    }
+    bin_boot = file_IntelHex_to_Bin(path_boot.read_text(), len_max=0x010000)
+    bin_main = file_IntelHex_to_Bin(path_main.read_text(), len_max=0x0CC000)
+    bin_back = file_IntelHex_to_Bin(path_back.read_text(), len_max=0x040000)
+    encrypt_main = file_encryption(bin_main)
+    encrypt_back = file_encryption(bin_back)
+
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_main)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_main))
+    config['hex_name'] = list(path_main.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (main_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+    
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_back)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_back))
+    config['hex_name'] = list(path_back.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (back_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+
+    print("Merge Image generated successfully.")
+    print(f"Main File:")
+    print(f"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main))}]")
+    print(f"Back File:")
+    print(f"\t header_length={len(back_header)}, bin_length={len(bin_back)}[{hex(len(bin_back))}]")
+
+    # 组装镜像
+    Image = [0xFF] * 0x100000
+    offset_image = 0
+    Image[offset_image: offset_image + len(bin_boot)] = bin_boot
+    offset_image = 0x010000
+    Image[offset_image: offset_image + len(bin_main)] = bin_main
+    offset_image = 0x0BC000
+    Image[offset_image: offset_image + len(bin_back)] = bin_back
+    offset_image = 0x0FC000
+    Image[offset_image: offset_image + len(main_header)] = main_header
+    offset_image = 0x0FE000
+    Image[offset_image: offset_image + len(back_header)] = back_header
+
+    return bytearray(Image), main_header, back_header
+
+
+def GenerateImage_SLCP101_p460(path_boot: Path, path_main: Path, path_back: Path):
+    """ 叠光适配器-460平台版本 镜像生成 """
+    config = {
+        'prod_type':        [0x45, 0x00],       # 产品类型
+        'method_compress':  False,              # 文件压缩
+        'prog_id':          list(b"SLCP101"),   # 程序识别号
+        'prog_type':        'app',              # 程序类型
+        'area_code':        [0x00, 0x00],       # 地区
+    }
+    bin_boot = file_IntelHex_to_Bin(path_boot.read_text(), len_max=0x00C000)
+    bin_main = file_IntelHex_to_Bin(path_main.read_text(), len_max=0x024000)
+    bin_back = file_IntelHex_to_Bin(path_back.read_text(), len_max=0x024000)
+    encrypt_main = file_encryption(bin_main)
+    encrypt_back = file_encryption(bin_back)
+
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_main)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_main))
+    config['hex_name'] = list(path_main.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (main_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+    
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_back)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_back))
+    config['hex_name'] = list(path_back.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (back_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+
+    print("Merge Image generated successfully.")
+    print(f"Main File:")
+    print(f"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main))}]")
+    print(f"Back File:")
+    print(f"\t header_length={len(back_header)}, bin_length={len(bin_back)}[{hex(len(bin_back))}]")
+
+    # 组装镜像
+    Image = [0xFF] * 0x058000
+    offset_image = 0
+    Image[offset_image: offset_image + len(bin_boot)] = bin_boot
+    offset_image = 0x00C000
+    Image[offset_image: offset_image + len(bin_main)] = bin_main
+    offset_image = 0x030000
+    Image[offset_image: offset_image + len(bin_back)] = bin_back
+    offset_image = 0x054000
+    Image[offset_image: offset_image + len(main_header)] = main_header
+    offset_image = 0x056000
+    Image[offset_image: offset_image + len(back_header)] = back_header
+
+    return bytearray(Image), main_header, back_header
+
+
+def GeneratePackage_SLCP101_p460(path_hex: Path):
+    """ 叠光适配器-460平台版本 生成升级包 """
+    config = {
+        'prod_type':        [0x45, 0x00],       # 产品类型
+        'method_compress':  False,              # 文件压缩
+        'prog_id':          list(b"SLCP101"),   # 程序识别号
+        'prog_type':        'app',              # 程序类型
+        'area_code':        [0x00, 0x00],       # 地区
+    }
+    bin_main = file_IntelHex_to_Bin(path_hex.read_text(), len_max=0x024000)
+    encrypt_main = file_encryption(bin_main)
+
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_main)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_main))
+    config['hex_name'] = list(path_hex.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (main_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+
+    print("Package generated successfully.")
+    print(f"File name: {path_hex.name}")
+    print(f"File Info:")
+    print(f"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main))}]")
+
+    # 组装镜像
+    Image = [0xFF] * (len(main_header) + len(encrypt_main))
+    offset_image = 0
+    Image[offset_image: offset_image + len(main_header)] = main_header
+    offset_image += len(main_header)
+    Image[offset_image: offset_image + len(encrypt_main)] = encrypt_main
+
+    return bytearray(Image), bin_main
+
+
+def GenerateImage_DLSY001_p460(path_boot: Path, path_main: Path, path_back: Path):
+    """ 叠光优化器-460平台版本 镜像生成 """
+    config = {
+        'prod_type':        [0x45, 0x00],       # 产品类型
+        'method_compress':  False,              # 文件压缩
+        'prog_id':          list(b"DLSY001"),   # 程序识别号
+        'prog_type':        'app',              # 程序类型
+        'area_code':        [0x00, 0x00],       # 地区
+    }
+    bin_boot = file_IntelHex_to_Bin(path_boot.read_text(), len_max=0x00C000)
+    bin_main = file_IntelHex_to_Bin(path_main.read_text(), len_max=0x024000)
+    bin_back = file_IntelHex_to_Bin(path_back.read_text(), len_max=0x024000)
+    encrypt_main = file_encryption(bin_main)
+    encrypt_back = file_encryption(bin_back)
+
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_main)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_main))
+    config['hex_name'] = list(path_main.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (main_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+    
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_back)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_back))
+    config['hex_name'] = list(path_back.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (back_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+
+    print("Merge Image generated successfully.")
+    print(f"Main File:")
+    print(f"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main))}]")
+    print(f"Back File:")
+    print(f"\t header_length={len(back_header)}, bin_length={len(bin_back)}[{hex(len(bin_back))}]")
+
+    # 组装镜像
+    Image = [0xFF] * 0x058000
+    offset_image = 0
+    Image[offset_image: offset_image + len(bin_boot)] = bin_boot
+    offset_image = 0x00C000
+    Image[offset_image: offset_image + len(bin_main)] = bin_main
+    offset_image = 0x030000
+    Image[offset_image: offset_image + len(bin_back)] = bin_back
+    offset_image = 0x054000
+    Image[offset_image: offset_image + len(main_header)] = main_header
+    offset_image = 0x056000
+    Image[offset_image: offset_image + len(back_header)] = back_header
+
+    return bytearray(Image), main_header, back_header
+
+
+def GeneratePackage_DLSY001_p460(path_hex: Path):
+    """ 叠光优化器-460平台版本 生成升级包 """
+    config = {
+        'prod_type':        [0x45, 0x00],       # 产品类型
+        'method_compress':  False,              # 文件压缩
+        'prog_id':          list(b"DLSY001"),   # 程序识别号
+        'prog_type':        'app',              # 程序类型
+        'area_code':        [0x00, 0x00],       # 地区
+    }
+    bin_main = file_IntelHex_to_Bin(path_hex.read_text(), len_max=0x024000)
+    encrypt_main = file_encryption(bin_main)
+
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_main)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_main))
+    config['hex_name'] = list(path_hex.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (main_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+
+    print("Package generated successfully.")
+    print(f"File name: {path_hex.name}")
+    print(f"File Info:")
+    print(f"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main))}]")
+
+    # 组装镜像
+    Image = [0xFF] * (len(main_header) + len(encrypt_main))
+    offset_image = 0
+    Image[offset_image: offset_image + len(main_header)] = main_header
+    offset_image += len(main_header)
+    Image[offset_image: offset_image + len(encrypt_main)] = encrypt_main
+
+    return bytearray(Image), bin_main
+
+
+def GenerateImage_DLSP001_p280039(path_boot: Path, path_main: Path, path_back: Path):
+    """ 叠光适配器-460平台版本 镜像生成 """
+    config = {
+        'prod_type':        [0x46, 0x00],       # 产品类型
+        'method_compress':  False,              # 文件压缩
+        'prog_id':          list(b"DLSP001"),   # 程序识别号
+        'prog_type':        'app',              # 程序类型
+        'area_code':        [0x00, 0x00],       # 地区
+        'upgrade_type':     [0x00, 0x00],       # 升级方式(0-片外缓冲, 1-片内缓冲, 2-升级备份)
+    }
+
+    bin_boot = file_IntelHex_to_Bin(path_boot.read_text(), len_max=2 * 0x004000, conv_end=False)
+    bin_main = file_IntelHex_to_Bin(path_main.read_text(), len_max=2 * 0x014000, conv_end=False)
+    bin_back = file_IntelHex_to_Bin(path_back.read_text(), len_max=2 * 0x014000, conv_end=False)
+
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_main)
+    config["md5"] = list(md5_ctx.digest())
+    config['upgrade_type'] = [0x00, 0x00]   # 主程序-片外缓冲
+    config['file_length'] = conv_int_to_array(len(bin_main))
+    config['hex_name'] = list(path_main.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (main_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_back)
+    config["md5"] = list(md5_ctx.digest())
+    config['upgrade_type'] = [0x02, 0x00]   # 备份程序
+    config['file_length'] = conv_int_to_array(len(bin_back))
+    config['hex_name'] = list(path_back.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (back_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+    
+    main_header_buffer = bytearray()
+    for byte_org in main_header:
+        main_header_buffer.extend(bytearray([0x00, byte_org]))
+    back_header_buffer = bytearray()
+    for byte_org in back_header:
+        back_header_buffer.extend(bytearray([0x00, byte_org]))
+    main_encrypt = file_encryption(bin_main)
+    back_encrypt = file_encryption(bin_back)
+
+    print("Merge Image generated successfully.")
+    print(f"Main File:")
+    print(f"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main))}]")
+    print(f"Back File:")
+    print(f"\t header_length={len(back_header)}, bin_length={len(bin_back)}[{hex(len(bin_back))}]")
+
+    # 组装镜像
+    Image = [0xFF] * 2 * 0x030000
+    offset_image = 0
+    Image[offset_image: offset_image + len(bin_boot)] = bin_boot
+    offset_image = 2 * 0x006000
+    Image[offset_image: offset_image + len(main_header_buffer)] = main_header_buffer
+    offset_image = 2 * 0x007000
+    Image[offset_image: offset_image + len(back_header_buffer)] = back_header_buffer
+    offset_image = 2 * 0x008000
+    Image[offset_image: offset_image + len(bin_main)] = bin_main
+    offset_image = 2 * 0x01C000
+    Image[offset_image: offset_image + len(bin_back)] = bin_back
+
+    return bytearray(Image), main_header, back_header
+
+
+def GeneratePackage_DLSP001_p280039(path_hex: Path):
+    """ 叠光控制器DSP-280039平台版本 生成升级包 """
+    config = {
+        'prod_type':        [0x46, 0x00],       # 产品类型
+        'method_compress':  False,              # 文件压缩
+        'prog_id':          list(b"DLSP001"),   # 程序识别号
+        'prog_type':        'app',              # 程序类型
+        'area_code':        [0x00, 0x00],       # 地区
+        'upgrade_type':     [0x00, 0x00],       # 升级方式(0-片外缓冲, 1-片内缓冲, 2-升级备份)
+    }
+
+    bin_main = file_IntelHex_to_Bin(path_hex.read_text(), len_max=2 * 0x014000, conv_end=False)
+    encrypt_main = file_encryption(bin_main)
+
+    md5_ctx = hashlib.md5()
+    md5_ctx.update(bin_main)
+    config["md5"] = list(md5_ctx.digest())
+    config['file_length'] = conv_int_to_array(len(bin_main))
+    config['hex_name'] = list(path_hex.name.encode())[:64]
+    config['hex_name'] += [0] * (64 - len(config['hex_name']))
+    if (main_header:=build_header_new(config)) is None:
+        raise Exception("Header tag oversize. ")
+
+    print("Package generated successfully.")
+    print(f"File name: {path_hex.name}")
+    print(f"File Info:")
+    print(f"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main))}]")
+
+    # 组装镜像
+    Image = [0xFF] * (len(main_header) + len(encrypt_main))
+    offset_image = 0
+    Image[offset_image: offset_image + len(main_header)] = main_header
+    offset_image += len(main_header)
+    Image[offset_image: offset_image + len(encrypt_main)] = encrypt_main
+
+    return bytearray(Image), bin_main
+
+
+def Process1():
+    """ 镜像生成流程 """
+    root = Path(r"test\p460")
+    result = Path(r"test\p460\result")
+    # 正常启动镜像
+    hex_boot = root / r"bootloader.hex"
+    hex_main = root / r"lamina_adapter.hex"
+    hex_back = root / r"lamina_adapter_back.hex"
+    hex_update = root / r"lamina_adapter_t1.hex"
+
+    file_image = result / f'{hex_main.stem[:-6]}_ROM.bin'
+    file_main_header = result / 'SLCP101_header_main.bin'
+    file_back_header = result / 'SLCP101_header_back.bin'
+    file_package = result / f'{hex_update.stem}.dat'
+    file_bin = result / f'{hex_update.stem}.bin'
+
+    data_bins = GenerateImage_SLCP101_p460(hex_boot, hex_main, hex_back)
+    data_package, data_bins = GeneratePackage_SLCP101_p460(hex_update)
+
+    file_image.write_bytes(data_bins[0].copy())
+    file_main_header.write_bytes(data_bins[1].copy())
+    file_back_header.write_bytes(data_bins[2].copy())
+    file_package.write_bytes(data_package)
+    file_bin.write_bytes(data_bins)
+
+    # 异常镜像-主分区md5错误
+    file_image1 = result / f'{file_image.stem}_b1.bin'
+    data_image = data_bins[0].copy()
+    data_image[0x054018: 0x05401A] = [0x00, 0x01]
+    file_image1.write_bytes(data_image)
+
+    # 异常镜像-备份分区md5错误
+    file_image2 = result / f'{file_image.stem}_b2.bin'
+    data_image = data_bins[0].copy()
+    data_image[0x056018: 0x05601A] = [0x00, 0x01]
+    file_image2.write_bytes(data_image)
+
+    # 异常镜像-双分区md5错误
+    file_image3 = result / f'{file_image.stem}_b3.bin'
+    data_image = data_bins[0].copy()
+    data_image[0x054018: 0x05401A] = [0x00, 0x01]
+    data_image[0x056018: 0x05601A] = [0x00, 0x01]
+    file_image3.write_bytes(data_image)
+
+
+def Process2():
+    """ 镜像生成流程 """
+    root = Path(r"D:\WorkingProject\LightStackOptimizer\software\lamina_optimizer\lamina_optimizer\Debug")
+    # result = Path(r"test\p460_o1\result")
+    result = root
+    # 正常启动镜像
+    hex_boot = Path(r"test\p460_o1\bootloader.hex")
+    hex_main = root / r"lamina_optimizer.hex"
+    hex_back = root / r"lamina_optimizer.hex"
+    hex_update = root / r"lamina_optimizer.hex"
+
+    file_image = result / f'{hex_main.stem}_ROM.bin'
+    file_main_header = result / 'DLSY001_header_main.bin'
+    file_back_header = result / 'DLSY001_header_back.bin'
+    file_package = result / f'{hex_update.stem}.dat'
+    file_bin = result / f'{hex_update.stem}.bin'
+
+    data_bins = GenerateImage_DLSY001_p460(hex_boot, hex_main, hex_back)
+    data_package, data_bin = GeneratePackage_DLSY001_p460(hex_update)
+
+    file_image.write_bytes(data_bins[0].copy())
+    file_main_header.write_bytes(data_bins[1].copy())
+    file_back_header.write_bytes(data_bins[2].copy())
+    file_package.write_bytes(data_package)
+    file_bin.write_bytes(data_bin)
+
+    # 异常镜像-主分区md5错误
+    file_image1 = result / f'{file_image.stem}_b1.bin'
+    data_image = data_bins[0].copy()
+    data_image[0x054018: 0x05401A] = [0x00, 0x01]
+    file_image1.write_bytes(data_image)
+
+    # 异常镜像-备份分区md5错误
+    file_image2 = result / f'{file_image.stem}_b2.bin'
+    data_image = data_bins[0].copy()
+    data_image[0x056018: 0x05601A] = [0x00, 0x01]
+    file_image2.write_bytes(data_image)
+
+    # 异常镜像-双分区md5错误
+    file_image3 = result / f'{file_image.stem}_b3.bin'
+    data_image = data_bins[0].copy()
+    data_image[0x054018: 0x05401A] = [0x00, 0x01]
+    data_image[0x056018: 0x05601A] = [0x00, 0x01]
+    file_image3.write_bytes(data_image)
+
+
+if __name__ == "__main__":
+    # path_bin = Path("F:\\Work\\FPGA\\Test\\Vivado\\test_update\\test_update.vitis\\upgrade_system\\Debug\\sd_card\\BOOT.BIN")
+    # GeneratePackage_Demo_Xilinx(path_bin)
+
+    Process2()
+    pass

source/main.py

@@ -3,7 +3,7 @@ from webui import webui
 from pathlib import Path
 from func_frame import check_frame_dlt645
 from source.dev_LaminaAdapter import LaminaAdapter
-from dev_EnergyRouter import EnergyRouter
+from source.device.EnergyRouter import EnergyRouter
 
 
 def my_function(e : webui.event):

source/tools/ByteConv.py

@@ -1,14 +1,15 @@
-
 def trans_list_to_str(data: list) -> str:
     """ 标准串口字符串表示 """
     func_trans = lambda x: ('00' + hex(x % 256)[2:])[-2:].upper()
     return " ".join(map(func_trans, data))
 
+
 def trans_str_to_list(data: str) -> list:
     """ 标准串口字符串转换列表 """
     func_trans = lambda x: int(x, 16)
     return list(map(func_trans, data.strip().split(" ")))
 
+
 def conv_int_to_array(num: int, big_end=False):
     """ 数值转字节数组 """
     result = [0, 0, 0, 0]
@@ -22,9 +23,9 @@ def conv_int_to_array(num: int, big_end=False):
         num //= 0x100
     return result
 
+
 def display_hex(data, len) -> str:
     """ Hex字符表示 """
     data  %= 2 ** (4 * len)
     result = "0" * len + hex(data)[2:]
     return "0x" + result[-len:].upper()
-

source/tools/IntelHex.py

@@ -0,0 +1,142 @@
+def calculate_checksum(data):
+    """ 计算校验域 """
+    checksum = 0
+    for i in range(0, len(data), 2):
+        checksum += int(data[i: i+2], 16)
+    return (0x100 - checksum) & 0xFF
+
+
+def file_IntelHex_to_Bin(file_data, base_address=0, len_max=1, **kwargs):
+    """
+        将Intel Hex格式文件转换为Bin格式;
+    """
+    if base_address == 0:
+        if file_data[8] == '2':
+            base_address = int(file_data[9: 13], 16) * 0x100
+            offset_begin = 16
+        elif file_data[8] == '4':
+            base_address = int(file_data[9: 13], 16) * 0x10000
+            offset_begin = 16
+        else:
+            base_address = 0
+            offset_begin = 0
+        base_address += int(file_data[offset_begin + 3: offset_begin + 7], 16)
+        base_address &= ~0x0FFF
+
+    len_space = len_max
+    result = bytearray(len_space).replace(b'\x00', b'\xff')
+    lines = file_data.split('\n')
+    offset = 0
+    max_address = 0
+    for line in lines:
+        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!')
+
+            if line[7:9] == '00':
+                len_line = int(line[1:3], 16)
+                address = offset + int(line[3:7], 16) - base_address
+                data = bytearray.fromhex(line[9:-2])
+                if 'conv_end' in kwargs.keys():
+                    address *= 2
+                    if kwargs['conv_end']:
+                        for i in range(0, len_line, 2):
+                            t = data[i]
+                            data[i] = data[i+1]
+                            data[i+1] = t
+
+                if (address + len_line) > max_address:
+                    max_address = address + len_line
+                if max_address >= len_space:
+                    if len_max == 1:
+                        result += bytearray(max_address - len_space).replace(b'\x00', b'\xff')
+                        len_space += max_address - len_space
+                    else:
+                        raise Exception("Bin file Oversize.")
+                result[address:address+len_line] = data
+
+            elif line[7:9] == '01':
+                break
+            elif line[7:9] == '02':
+                offset = int(line[9:-2], 16) * 16
+            elif line[7:9] == '03':
+                pass
+            elif line[7:9] == '04':   # 拓展地址偏移
+                offset = int(line[9:-2], 16) * 2**16
+            elif line[7:9] == '05':
+                pass
+    return result[:max_address]
+
+
+def file_Bin_to_IntelHex(file_data: bytearray, base_address=0, **kwargs):
+    """
+        将Bin格式文件转换为Intel Hex格式;
+    """
+    chunk_size = 32
+    rom_width = 8
+    memory_width = 8
+    if 'chunk_size' in kwargs.keys():
+        """ 数据分块大小 """
+        chunk_size = kwargs['chunk_size']
+    if 'memory_width' in kwargs.keys():
+        """ 地址数据位宽 """
+        memory_width = kwargs['memory_width'] * 8
+    if 'rom_width' in kwargs.keys():
+        """ rom数据位宽 """
+        rom_width = kwargs['rom_width'] * 8
+
+    def record_extern_address(address, record_type):
+        """ 添加地址记录 """
+        if record_type == 4:
+            record_data = (address & 0xFFFF0000) >> 16
+        elif record_type == 2:
+            record_data = (address & 0xFFFF0) >> 4
+        else:
+            raise Exception("Unknow record tye.")
+
+        hex_record = f':020000{record_type:02X}{record_data:04X}'
+        hex_record += f'{calculate_checksum(hex_record[1:]):02X}\n'
+        return hex_record
+
+    result = str()
+
+    # 添加开始记录
+    extern_address = 0
+    if base_address >= 2 ** 32:
+        """ 超过最大可表示范围, 抛出异常 """
+        raise Exception("base_address oversize")
+    elif base_address >= 2 ** 16:
+        result += record_extern_address(base_address, 4)
+        extern_address = base_address & 0xFFFF0000
+    elif base_address >= 2 ** 4:
+        result += record_extern_address(base_address, 2)
+        extern_address = base_address & 0xFFFF0
+
+    record_type = 0
+    binary_offset = 0
+    while binary_offset < len(file_data):
+        """ 循环分块处理数据 """
+        data_chunk = file_data[binary_offset: binary_offset + chunk_size]
+
+        record_len = len(data_chunk)
+        record_data = data_chunk
+        record_addr = (base_address - extern_address + (binary_offset // (memory_width // 8)))
+        if record_addr >= 0x10000:
+            result += record_extern_address((base_address + (binary_offset // (memory_width // 8))), 4)
+            extern_address = (base_address + (binary_offset // (memory_width // 8))) & 0xFFFF0000
+            continue
+
+        # 构造记录
+        hex_record = f':{record_len:02X}{record_addr:04X}{record_type:02X}'
+        hex_record += ''.join(f'{byte:02X}' for byte in record_data)
+        hex_record += f'{calculate_checksum(hex_record[1:]):02X}\n'
+        result += hex_record
+
+        binary_offset += chunk_size
+
+    # 添加结束记录
+    hex_record = f':00000001FF\n'
+    result += hex_record
+
+    return result

source/utl_upgrade.py

@@ -1,517 +0,0 @@
-# 升级包生成脚本
-import hashlib
-from pathlib import Path
-from datetime import datetime
-from crc import Calculator, Crc16
-
-from utl import conv_int_to_array, trans_list_to_str
-
-Header_Tag = {
-    'file_type':      [0x00,  2],               # 0 - 文件类型; 2byte
-    'file_version':   [0x01,  2],               # 1 - 文件版本; 2byte
-    'file_length':    [0x02,  4],               # 2 - 文件长度; 4byte
-    'md5':            [0x03, 16],               # 3 - 文件MD5; 16byte
-    'encrypt':        [0x04,  1],               # 4 - 加密算法; 1byte
-    'update_type':    [0x05,  1],               # 5 - 升级文件类别; 1byte
-    'update_spec':    [0x06,  4],               # 6 - 升级特征字; 4byte
-    'update_verison': [0x07,  4],               # 7 - 升级版本号; 4byte
-    'update_date':    [0x08,  3],               # 8 - 升级版本日期; 3byte
-    'area_code':      [0x09,  4],               # 9 - 省份特征; 4byte
-    'uptate_str':     [0x0A, -1, 64],           # 10 - 升级段描述; less than 64byte
-    'device_str':     [0x0D, -1, 64],           # 13 - 设备特征描述; less than 64byte
-    'hex_name':       [0xFF, -1, 80],           # 255 - Hex文件名; less than 80byte
-}
-
-def file_IntelHex_to_Bin(file_data, base_address=-1, len_max=-1, **kwargs):
-    """
-        将Intel Hex格式文件转换为Bin格式;
-    """
-    if base_address == -1:
-        if file_data[8] == '2':
-            base_address = int(file_data[9: 13], 16) * 16
-            offset_begin = 16
-        elif file_data[8] == '4':
-            base_address = int(file_data[9: 13], 16) * 2**16
-            offset_begin = 16
-        else:
-            base_address = 0
-            offset_begin = 0
-        base_address += int(file_data[offset_begin + 3: offset_begin + 7], 16)
-        base_address &= ~0x0FFF
-    if len_max == -1:
-        len_space = 0x10
-    else:
-        len_space = len_max
-    result = bytearray(len_space).replace(b'\x00', b'\xff')
-    lines = file_data.split('\n')
-    offset = 0
-    max_address = 0
-    for line in lines:
-        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!')
-            
-            if line[7:9] == '00':
-                len_line = int(line[1:3], 16)
-                address = offset + int(line[3:7], 16) - base_address
-                data = bytearray.fromhex(line[9:-2])
-                if 'conv_end' in kwargs.keys():
-                    address *= 2
-                    if kwargs['conv_end']:
-                        for i in range(0, len_line, 2):
-                            t = data[i]
-                            data[i] = data[i+1]
-                            data[i+1] = t
-
-                if max_address >= len_space:
-                    if len_max == -1:
-                        result += bytearray((len_line & ~0x0F) + 0x10).replace(b'\x00', b'\xff')
-                        len_space += (len_line & ~0x0F) + 0x10
-                    else:
-                        raise Exception("Bin file Oversize.")
-                result[address:address+len_line] = data
-
-                if (address + len_line) > max_address:
-                    max_address = address + len_line
-
-            elif line[7:9] == '01':
-                break
-            elif line[7:9] == '02':
-                offset = int(line[9:-2], 16) * 16
-            elif line[7:9] == '03':
-                pass
-            elif line[7:9] == '04':   # 拓展地址偏移
-                offset = int(line[9:-2], 16) * 2**16
-            elif line[7:9] == '05':
-                pass
-    return result[:max_address]
-
-
-def file_encryption(buffer):
-    """ 文件加密算法 """
-    pwd_idx = 0
-    pwd = b'moc.mmocspot.www'
-    pwd = list(map(lambda x: (x - 0x30 + 0x100) % 0x100, pwd))
-    result = bytearray(len(buffer))
-    for i in range(len(buffer)):
-        k = i
-        k |= i >> 8
-        k |= i >> 16
-        k |= i >> 24
-        result[i] = buffer[i] ^ pwd[pwd_idx] ^ (k & 0xFF)
-        pwd_idx = (pwd_idx + 1) % len(pwd)
-    return result
-
-
-def build_header(config: dict, len_max=512):
-    """
-        基于配置参数, 生成文件信息头;
-    """
-    # 定义文件头
-    m_file_header = bytearray(len_max)
-
-    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])
-            else:
-                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 > len_max:
-        return None
-    else:
-        return m_file_header
-
-def build_header_lite(config: dict):
-    """
-        基于配置参数, 生成文件信息头;
-    """
-    # 定义文件头
-    m_file_header = bytearray(64)
-    m_file_header[0:4] = bytearray(config["update_verison"])
-    m_file_header[4:8] = bytearray(config["file_length"])
-    m_file_header[8:24] = bytearray(config["md5"])
-
-    return m_file_header
-
-def build_header_new(config: dict):
-    """
-        基于配置参数, 生成新版文件信息头;
-    """
-    # 定义文件头
-    m_file_header = [0xFF] * 184
-
-    m_file_header[0:8] = list(b"TOPSCOMM")
-    m_file_header[8:10] = config['prod_type']
-    m_file_header[10:22] = config['prog_id'] + [0] * (12 - len(config['prog_id']))
-    if config['method_compress'] == True:
-        m_file_header[22] = 0x01
-    else:
-        m_file_header[22] = 0x00
-
-    if 'crc32' in config.keys():
-        m_file_header[23] = 0x00
-        m_file_header[24: 40] = config['crc32'] + [0x00] * 12
-    elif 'md5' in config.keys():
-        m_file_header[23] = 0x01
-        m_file_header[24: 40] = config['md5']
-    else:
-        raise Exception("Error, Unknown method verify.")
-    
-    # 时间戳生成
-    time_now = datetime.now()
-    time_stamp = list(map(lambda x: int(x), 
-                                     time_now.strftime("%Y-%m-%d-%H-%M").split('-')))
-    time_stamp.insert(1, time_stamp[0] // 0x100)
-    time_stamp[0] = time_stamp[0] % 0x100
-    m_file_header[40: 46] = time_stamp
-
-    m_file_header[46: 50] = config['file_length']
-    # Cpu1
-    m_file_header[50: 54] = [0x00] * 4
-    m_file_header[54: 70] = [0x00] * 16
-    # Cpu2
-    m_file_header[70: 74] = [0x00] * 4
-    m_file_header[74: 90] = [0x00] * 16
-
-    if config['prog_type'] == 'app':
-        m_file_header[90: 92] = [0x00, 0x00]
-    elif config['prog_type'] == 'boot':
-        m_file_header[90: 92] = [0x01, 0x00]
-    elif config['prog_type'] == 'diff':
-        m_file_header[90: 92] = [0x02, 0x00]
-    elif config['prog_type'] == 'font':
-        m_file_header[90: 92] = [0x03, 0x00]
-    elif config['prog_type'] == 'config':
-        m_file_header[90: 92] = [0x04, 0x00]
-    elif config['prog_type'] == 'data':
-        m_file_header[90: 92] = [0x05, 0x00]
-    elif config['prog_type'] == 'test':
-        m_file_header[90: 92] = [0x06, 0x00]
-    else:
-        raise Exception("Error, Unknown Program Type.")
-    
-    m_file_header[92: 94] = config['area_code']
-    m_file_header[94: 158] = config['hex_name']
-    m_file_header[158: 182] = [0x00] * 24
-
-    m_file_header = bytearray(m_file_header)
-
-    calculator = Calculator(Crc16.MODBUS)
-    code_crc16 = calculator.checksum(m_file_header[:-2])
-    m_file_header[182: 184] = [code_crc16 % 0x100, code_crc16 // 0x100]
-
-    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_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))
-    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 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_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), True)
-    config["md5"] = list(md5_ctx.digest())
-    if header_len > 256:
-        config['hex_name'] = list(file_path.name.encode())[:80]
-
-    data_header = build_header_lite(config)
-    data_encrypt = file_encryption(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 Exception("Header tag oversize. ")
-    data_encrypt = file_encryption(data_bin)
-    return data_header, data_bin, data_encrypt
-
-def make_datafile3(file_path: str, config: dict, header_len=184):
-    """ 升级文件生成函数; 完整文件头 """
-    file_path = Path(file_path)
-    file_data = file_path.read_text()
-    data_bin = file_IntelHex_to_Bin(file_data)
-    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())
-    config['hex_name'] = list(file_path.name.encode())[:64]
-    config['hex_name'] += [0] * (64 - len(config['hex_name']))
-
-    data_header = build_header_new(config)
-    if data_header is None:
-        raise Exception("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_IntelHex_to_Bin(data1, 0x3F0100)
-    path1.write_bytes(data2)
-    pass
-
-def test2():
-    """ md5校验, 加密测试 """
-    bin_offcial = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\uart_tool\江苏发货版本\SLCP001_240517_1800_V1.11.bin")
-    data_offcial = bin_offcial.read_bytes()
-    byte_data = data_offcial[:182]
-    crc = data_offcial[182:184]
-    # data = "54 4F 50 53 43 4F 4D 4D 45 00 53 4C 43 50 30 30 31 00 00 00 00 00 01 00 B6 61 A8 73 BF 82 9E A7 4C 79 F6 BB 94 E2 A5 18 E8 07 05 18 0B 17 18 4C 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 6C 61 6D 69 6E 61 5F 61 64 61 70 74 65 72 5F 6D 61 69 6E 02 00 00 00 00 20 10 53 06 00 00 00 00 80 A5 8F 02 00 00 00 00 EC 1F 40 00 00 00 00 00 00 00 00 00 00 00 00 00 7A CA 61 0A FD 7F 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00"
-    # byte_data = list(map(lambda x: int(x, 16), data.split(' ')))
-    # crc = "99 CB"
-    calculator = Calculator(Crc16.MODBUS)
-    code_crc16 = calculator.checksum(bytearray(byte_data))
-    print(f"Result = {hex(code_crc16)}, Offcial Result = {crc}")
-
-    data_bin = '123456 123456 '.encode()
-    md5_ctx = hashlib.md5()
-    md5_ctx.update(data_bin)
-    hash = md5_ctx.hexdigest()
-    buffer1 = data_bin[:]
-    buffer1_en = file_encryption(buffer1)
-    buffer2 = buffer1_en[:6] + buffer1[6:]
-    buffer2_en = file_encryption(buffer2)
-    pass
-
-def test3(path_bin):
-    """ 完整升级包生成测试 """
-    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,               # 程序空间大小
-    }
-
-    header, data_b, _ = make_datafile2(path_bin, config, header_len=128)
-    header_512, data_b_512, _ = 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.write_bytes(header + data_b)
-    file2 = path_bin.parent / (path_bin.stem + '_h512.dat')
-    file2.write_bytes(header_512 + data_b)
-
-def test4(path_boot, path_main, path_back):
-    """ 叠光适配器新版升级方案测试 """
-    config = {
-        'prod_type':        [0x45, 0x00],       # 产品类型
-        'method_compress':  False,              # 文件压缩
-        'prog_id':          list(b"SLCP001"),   # 程序识别号
-        'prog_type':        'app',              # 程序类型
-        'area_code':        [0x00, 0x00],       # 地区
-    }
-    file_path = Path(path_boot)
-    file_data = file_path.read_text()
-    boot_bin = file_IntelHex_to_Bin(file_data, len_max=0x010000)
-    main_header, main_data_b, main_data_e = make_datafile3(path_main, config)
-    back_header, back_data_b, back_data_e = make_datafile3(path_back, config)
-
-    print("Merge Image generated successfully.")
-    print(f"Main File:")
-    print(f"\t header_length={len(main_header)}, bin_length={len(main_data_b)}[{hex(len(main_data_b))}]")
-    print(f"Back File:")
-    print(f"\t header_length={len(back_header)}, bin_length={len(back_data_b)}[{hex(len(back_data_b))}]")
-
-    # 组装镜像
-    Image = [0xFF] * 0x100000
-    offset_image = 0
-    Image[offset_image: offset_image + len(boot_bin)] = boot_bin
-    offset_image = 0x010000
-    Image[offset_image: offset_image + len(main_data_b)] = main_data_b
-    offset_image = 0x0BC000
-    Image[offset_image: offset_image + len(back_data_b)] = back_data_b
-    offset_image = 0x0FC000
-    Image[offset_image: offset_image + len(main_header)] = main_header
-    offset_image = 0x0FE000
-    Image[offset_image: offset_image + len(back_header)] = back_header
-
-    return bytearray(Image), main_header, main_data_b, main_data_e, back_header, back_data_b, back_data_e
-
-
-def GenerateImage_SLCP001_p460(path_boot, path_main, path_back):
-    """ 叠光适配器-460平台版本 镜像生成 """
-    config = {
-        'prod_type':        [0x45, 0x00],       # 产品类型
-        'method_compress':  False,              # 文件压缩
-        'prog_id':          list(b"SLCP001"),   # 程序识别号
-        'prog_type':        'app',              # 程序类型
-        'area_code':        [0x00, 0x00],       # 地区
-    }
-    file_path = Path(path_boot)
-    file_data = file_path.read_text()
-    boot_bin = file_IntelHex_to_Bin(file_data, len_max=0x00C000)
-    main_header, main_data_b, main_data_e = make_datafile3(path_main, config)
-    back_header, back_data_b, back_data_e = make_datafile3(path_back, config)
-
-    print("Merge Image generated successfully.")
-    print(f"Main File:")
-    print(f"\t header_length={len(main_header)}, bin_length={len(main_data_b)}[{hex(len(main_data_b))}]")
-    print(f"Back File:")
-    print(f"\t header_length={len(back_header)}, bin_length={len(back_data_b)}[{hex(len(back_data_b))}]")
-
-    # 组装镜像
-    Image = [0xFF] * 0x058000
-    offset_image = 0
-    Image[offset_image: offset_image + len(boot_bin)] = boot_bin
-    offset_image = 0x00C000
-    Image[offset_image: offset_image + len(main_data_b)] = main_data_b
-    offset_image = 0x030000
-    Image[offset_image: offset_image + len(back_data_b)] = back_data_b
-    offset_image = 0x054000
-    Image[offset_image: offset_image + len(main_header)] = main_header
-    offset_image = 0x056000
-    Image[offset_image: offset_image + len(back_header)] = back_header
-
-    return bytearray(Image), main_header, main_data_b, main_data_e, back_header, back_data_b, back_data_e
-
-def GeneratePackage_SLCP001_p460(path_hex):
-    """ 叠光适配器-460平台版本 生成升级包 """
-    config = {
-        'prod_type':        [0x45, 0x00],       # 产品类型
-        'method_compress':  False,              # 文件压缩
-        'prog_id':          list(b"SLCP001"),   # 程序识别号
-        'prog_type':        'app',              # 程序类型
-        'area_code':        [0x00, 0x00],       # 地区
-    }
-    main_header, main_data_b, main_data_e = make_datafile3(path_hex, config)
-
-    print("Merge Image generated successfully.")
-    print(f"File Info:")
-    print(f"\t header_length={len(main_header)}, bin_length={len(main_data_b)}[{hex(len(main_data_b))}]")
-
-    # 组装镜像
-    Image = [0xFF] * (len(main_header) + len(main_data_e))
-    offset_image = 0
-    Image[offset_image: offset_image + len(main_header)] = main_header
-    offset_image += len(main_header)
-    Image[offset_image: offset_image + len(main_data_e)] = main_data_e
-
-    return bytearray(Image), main_data_b
-
-def Process1():
-    """ 镜像生成流程 """
-    root = Path(r"test\p460")
-    result = Path(r"test\p460\result")
-    # 正常启动镜像
-    hex_boot = root / r"bootloader.hex"
-    hex_main = root / r"lamina_adapter.hex"
-    hex_back = root / r"lamina_adapter_back.hex"
-
-    file_image = result / f'{hex_main.stem[:-6]}_ROM.bin'
-    file_main_header = result / 'SLCP001_header_main.bin'
-    file_back_header = result / 'SLCP001_header_back.bin'
-
-    data_bins = GenerateImage_SLCP001_p460(hex_boot, hex_main, hex_back)
-    file_image.write_bytes(data_bins[0].copy())
-    file_main_header.write_bytes(data_bins[1].copy())
-    file_back_header.write_bytes(data_bins[4].copy())
-
-    # 异常镜像-主分区md5错误
-    file_image1 = result / f'{file_image.stem}_b1.bin'
-    data_image = data_bins[0].copy()
-    data_image[0x054018: 0x05401A] = [0x00, 0x01]
-    file_image1.write_bytes(data_image)
-
-    # 异常镜像-备份分区md5错误
-    file_image2 = result / f'{file_image.stem}_b2.bin'
-    data_image = data_bins[0].copy()
-    data_image[0x056018: 0x05601A] = [0x00, 0x01]
-    file_image2.write_bytes(data_image)
-
-    # 异常镜像-双分区md5错误
-    file_image3 = result / f'{file_image.stem}_b3.bin'
-    data_image = data_bins[0].copy()
-    data_image[0x054018: 0x05401A] = [0x00, 0x01]
-    data_image[0x056018: 0x05601A] = [0x00, 0x01]
-    file_image3.write_bytes(data_image)
-
-def Process2():
-    """ 升级包生成流程 """
-    root = Path(r"test\p460")
-    result = Path(r"test\p460\result")
-    
-    hex_main = root / r"lamina_adapter_t1.hex"
-    file_package = result / f'{hex_main.stem}.dat'
-    file_bin = result / f'{hex_main.stem}.bin'
-
-    data_package, data_bins = GeneratePackage_SLCP001_p460(hex_main)
-    file_package.write_bytes(data_package)
-    file_bin.write_bytes(data_bins)
-
-
-if __name__ == "__main__":
-    # path_bin = Path("F:\\Work\\FPGA\\Test\\Vivado\\test_update\\test_update.vitis\\upgrade_system\\Debug\\sd_card\\BOOT.BIN")
-    # test3(path_bin)
-
-    # test2()
-
-    if 0:
-        bin_back = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\lamina_adapter_back.bin")
-        bin_main = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\lamina_adapter_main.bin")
-        main_header1 = bin_main.read_bytes()[:184]
-        main_data_e1 = bin_main.read_bytes()[184:]
-        back_header1 = bin_back.read_bytes()[:184]
-        back_data_e1 = bin_back.read_bytes()[184:]
-
-    Process1()