783 lines
31 KiB
Python
783 lines
31 KiB
Python
import time
|
|
from pathlib import Path
|
|
from serial import Serial
|
|
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
|
|
# 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],
|
|
0x54: ["模式更改命令" , 2],
|
|
0x55: ["短时停机命令(未启用)" , 2],
|
|
0x56: ["手动录波命令" , 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: ["LLC输出电压给定值", 2],
|
|
0x7F: ["Boost输出电压给定值", 2],
|
|
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],
|
|
}
|
|
|
|
|
|
class LaminaController:
|
|
def __init__(self, com_name="COM16", addr_645=[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA], addr_modbus=0x01, **kwargs):
|
|
# 初始化串口通信
|
|
if com_name is not None:
|
|
com_config = {}
|
|
com_config['baudrate'] = kwargs['baudrate'] if 'baudrate' in kwargs.keys() else 115200
|
|
com_config['parity'] = kwargs['parity'] if 'parity' in kwargs.keys() else 'N'
|
|
com_config['bytesize'] = kwargs['bytesize'] if 'bytesize' in kwargs.keys() else 8
|
|
com_config['stopbits'] = kwargs['stopbits'] if 'stopbits' in kwargs.keys() else 1
|
|
self.__com = Serial(com_name, **com_config)
|
|
else:
|
|
self.__com =None
|
|
|
|
self.flag_print = 'frame_print' in kwargs.keys()
|
|
self.time_out = kwargs['time_out'] if 'time_out' in kwargs.keys() else 1
|
|
self.retry = kwargs['retry'] if 'retry' in kwargs.keys() else 1
|
|
self.retry_sub = kwargs['retry_sub'] if 'retry_sub' in kwargs.keys() else 1
|
|
|
|
self.block = {
|
|
'addr' : addr_645,
|
|
'type' : 'modbus',
|
|
'data' : {
|
|
'addr_dev' : addr_modbus,
|
|
'data_define': modbus_map,
|
|
},
|
|
}
|
|
|
|
def __transfer_data(self, frame):
|
|
""" 报文数据传输 """
|
|
|
|
if self.__com is None:
|
|
print(trans_list_to_str(frame))
|
|
return False
|
|
|
|
cnt = 0
|
|
while cnt < self.retry:
|
|
self.__com.read_all()
|
|
self.__com.write(bytearray(frame))
|
|
|
|
cnt_sub = 0
|
|
frame_recv = None
|
|
while not frame_recv:
|
|
time.sleep(self.time_out)
|
|
frame_recv = self.__com.read_all()
|
|
cnt_sub += 1
|
|
if cnt_sub >= self.retry_sub:
|
|
break
|
|
|
|
try:
|
|
output_text = check_frame_modbus(frame_recv, self.block['data'])
|
|
except Exception as e:
|
|
print(e)
|
|
cnt += 1
|
|
continue
|
|
print(output_text)
|
|
break
|
|
|
|
if self.flag_print:
|
|
print(trans_list_to_str(frame))
|
|
print(trans_list_to_str(frame_recv))
|
|
|
|
return cnt < self.retry
|
|
|
|
def frame_read(self, daddr=0x60, dlen=0x30):
|
|
self.block['data']['type'] = 'read'
|
|
self.block['data']['data_addr'] = daddr
|
|
self.block['data']['data_len'] = dlen
|
|
frame = make_frame_modbus(self.block['data'])
|
|
|
|
return self.__transfer_data(frame)
|
|
|
|
def frame_write_one(self, daddr=0x85, dval=-900):
|
|
self.block['data']['type'] = 'write_one'
|
|
self.block['data']['data_addr'] = daddr
|
|
self.block['data']['data_val'] = dval
|
|
frame = make_frame_modbus(self.block['data'])
|
|
|
|
return self.__transfer_data(frame)
|
|
|
|
def frame_write_dual(self, daddr=0x91, dval=600):
|
|
self.block['data']['type'] = 'write_dual'
|
|
self.block['data']['data_addr'] = daddr
|
|
self.block['data']['data_val'] = dval
|
|
frame = make_frame_modbus(self.block['data'])
|
|
|
|
return self.__transfer_data(frame)
|
|
|
|
def frame_write_str(self, daddr=0x82, dval=[0x06, 0x05, 0x04, 0x03, 0x02, 0x01]):
|
|
self.block['data']['type'] = 'write_str'
|
|
self.block['data']['data_addr'] = daddr
|
|
self.block['data']['data_val'] = dval
|
|
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 = 2 * (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):
|
|
""" 程序升级
|
|
注意: 在使用单板升级测试时, 需要关闭低电压检测功能, 否则无法启动升级流程;
|
|
|
|
"""
|
|
self.block['data']['type'] = 'update'
|
|
self.block['data']['step'] = 'start'
|
|
self.block['data']['index'] = 0
|
|
self.block['data']['file'] = Path(path_bin).read_bytes()
|
|
self.block['data']['header_offset'] = 184
|
|
# 启动帧
|
|
frame_master = bytearray(make_frame_modbus(self.block['data']))
|
|
|
|
# 等待擦除完成返回
|
|
try_times = 30
|
|
self.__com.read_all()
|
|
while try_times:
|
|
time.sleep(self.time_out)
|
|
self.__com.write(frame_master)
|
|
frame_slave = self.__com.read_all()
|
|
if not frame_slave:
|
|
try_times -= 1
|
|
continue
|
|
|
|
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.")
|
|
|
|
# 避免接收到延迟返回报文
|
|
time.sleep(self.time_out)
|
|
|
|
# 文件传输
|
|
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_modbus(self.block['data']))
|
|
|
|
cnt = 0
|
|
while cnt < self.retry:
|
|
self.__com.read_all()
|
|
self.__com.write(frame_master)
|
|
|
|
cnt_sub = 0
|
|
frame_slave = None
|
|
while not frame_slave:
|
|
time.sleep(self.time_out)
|
|
frame_slave = self.__com.read_all()
|
|
cnt_sub += 1
|
|
if cnt_sub >= self.retry_sub:
|
|
break
|
|
|
|
try:
|
|
ret = check_frame_modbus(frame_slave, self.block['data'])
|
|
except Exception as e:
|
|
print(e)
|
|
ret = False, 0, 0
|
|
if ret[0]:
|
|
break
|
|
else:
|
|
cnt += 1
|
|
|
|
if cnt >= self.retry:
|
|
raise Exception("Error, Retry failed.")
|
|
|
|
self.block["data"]['index'] += 1
|
|
data_remain -= self.block["data"]['file_block_size']
|
|
|
|
# 结束升级
|
|
self.block["data"]['step'] = 'end'
|
|
frame_master = bytearray(make_frame_modbus(self.block['data']))
|
|
|
|
cnt = 0
|
|
while cnt < self.retry:
|
|
self.__com.read_all()
|
|
self.__com.write(frame_master)
|
|
|
|
cnt_sub = 0
|
|
frame_slave = None
|
|
while not frame_slave:
|
|
time.sleep(self.time_out)
|
|
frame_slave = self.__com.read_all()
|
|
cnt_sub += 1
|
|
if cnt_sub >= self.retry_sub:
|
|
break
|
|
|
|
try:
|
|
ret = check_frame_modbus(frame_slave, self.block['data'])
|
|
except Exception as e:
|
|
print(e)
|
|
ret = False, 0, 0
|
|
if ret[0]:
|
|
break
|
|
else:
|
|
cnt += 1
|
|
|
|
if cnt >= self.retry:
|
|
raise Exception("Error, Retry failed.")
|
|
|
|
|
|
def test_communication(time_out=2):
|
|
""" 通信成功率测试 """
|
|
log_success = 0
|
|
log_failed = 0
|
|
log_failedseries = 0
|
|
cnt_failedseries = 0
|
|
time_start = time.time()
|
|
saveconfig_print = dev_lamina.flag_print
|
|
dev_lamina.flag_print = False
|
|
try:
|
|
while 1:
|
|
if dev_lamina.frame_read(0x0C, 0x20):
|
|
log_success += 1
|
|
cnt_failedseries = 0
|
|
else:
|
|
log_failed += 1
|
|
cnt_failedseries += 1
|
|
if log_failedseries <= cnt_failedseries:
|
|
log_failedseries = cnt_failedseries
|
|
print(f"Time Stamp: {time.ctime()}")
|
|
print(f"Success Frame: {log_success}")
|
|
print(f"Failed Frame: {log_failed}")
|
|
print(f"Max Series Failed Frame: {log_failedseries}")
|
|
time.sleep(time_out)
|
|
finally:
|
|
time_end = time.time()
|
|
print("Test Result: ")
|
|
print(f"Time Start: {time.strftime(r'%Y-%m-%d %H:%M:%S', time.localtime(time_start))}, \tTime End: {time.strftime(r'%Y-%m-%d %H:%M:%S', time.localtime(time_end))}")
|
|
print(f"Time Elapsed: {time_end - time_start}")
|
|
print(f"Success Rate: {log_success / (log_success + log_failed) * 100}%")
|
|
dev_lamina.flag_print = saveconfig_print
|
|
|
|
|
|
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)
|
|
|
|
|
|
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")
|
|
|
|
# 正常启动镜像
|
|
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])
|
|
dev_lamina.frame_write_str(0x82, [0x0A, 0x00, 0x00, 0x00, 0x00, 0x00])
|
|
dev_lamina.frame_write_str(0x82, [0xFF, 0x00, 0x00, 0x00, 0x00, 0x00])
|
|
dev_lamina.frame_write_str(0x82, [0xA1, 0x00, 0x00, 0x00, 0x00, 0x00])
|
|
dev_lamina.frame_write_str(0x82, [0x1A, 0x00, 0x00, 0x00, 0x00, 0x00])
|
|
dev_lamina.frame_write_str(0x82, [0x99, 0x99, 0x99, 0x99, 0x99, 0x99])
|
|
dev_lamina.frame_write_str(0x82, [0x00, 0x00, 0x00, 0x00, 0x00, 0x01])
|
|
if 0:
|
|
dev_lamina.frame_write_one(0x52, 0x01)
|
|
time.sleep(0.5)
|
|
dev_lamina.frame_write_one(0x50, 0x00)
|
|
if 0:
|
|
while 1:
|
|
code_mes = input("扫描数据")
|
|
temp = code_mes[5:7] + code_mes[-10:]
|
|
code_addr = [int(temp[i:i+2], 16) for i in range(0, len(temp), 2)]
|
|
print(f"扫描结果: {code_mes}")
|
|
print(f"载波地址: {trans_list_to_str(code_addr)}")
|
|
dev_lamina.frame_read(0x100, 0x20)
|
|
time.sleep(0.5)
|
|
dev_lamina.frame_write_str(0x0180, list(code_mes))
|
|
time.sleep(0.5)
|
|
dev_lamina.frame_read(0x180, 0x10)
|
|
time.sleep(0.5)
|
|
dev_lamina.frame_write_str(0x82, code_addr)
|
|
time.sleep(0.5)
|
|
dev_lamina.frame_read(0x80, 0x08)
|
|
time.sleep(0.5)
|
|
dev_lamina.frame_write_one(0xA3, 0x01)
|
|
time.sleep(0.5)
|
|
dev_lamina.frame_write_one(0x51, 0x01)
|
|
time.sleep(5)
|
|
dev_lamina.frame_read(0x0E, 0x13)
|
|
time.sleep(5)
|
|
dev_lamina.frame_read(0x0E, 0x13)
|
|
time.sleep(0.5)
|
|
dev_lamina.frame_write_one(0x50, 0x00)
|
|
|
|
if 0:
|
|
dev_lamina.frame_write_one(0x0054, 0x01)
|
|
dev_lamina.frame_read(0x000E, 0x02)
|
|
dev_lamina.frame_write_one(0x0054, 0x03)
|
|
dev_lamina.frame_read(0x000E, 0x02)
|
|
dev_lamina.frame_write_one(0x0054, 0x07)
|
|
dev_lamina.frame_read(0x000E, 0x02)
|
|
dev_lamina.frame_write_one(0x0054, 0x06)
|
|
dev_lamina.frame_read(0x000E, 0x02)
|
|
dev_lamina.frame_write_one(0x0054, 0x04)
|
|
dev_lamina.frame_read(0x000E, 0x02)
|
|
dev_lamina.frame_write_one(0x0054, 0x00)
|
|
dev_lamina.frame_read(0x000E, 0x02)
|
|
if 0:
|
|
dev_lamina.frame_read(0x0E, 0x14)
|
|
if 0:
|
|
dev_lamina.frame_read(0x60, 0x60)
|
|
if 0:
|
|
dev_lamina.frame_read(0x70, 0x02)
|
|
dev_lamina.frame_write_one(0x70, 2100)
|
|
dev_lamina.frame_write_one(0x71, 2200)
|
|
dev_lamina.frame_read(0x70, 0x02)
|
|
dev_lamina.frame_write_one(0x70, 2300)
|
|
dev_lamina.frame_write_one(0x71, 2100)
|
|
dev_lamina.frame_read(0x70, 0x02)
|
|
if 0:
|
|
dev_lamina.frame_write_str(0x0170, list("SN202405201117-1"))
|
|
dev_lamina.frame_write_str(0x0180, list("MES202405201117-2"))
|
|
dev_lamina.frame_write_str(0x0190, list("D202405211500-3"))
|
|
time.sleep(2)
|
|
dev_lamina.frame_read(0x0170, 0x30)
|
|
|
|
|
|
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)
|
|
|
|
|