linghtiin/DebugTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

积累修改;

Browse Source
This commit is contained in:
何 泽隆
2024-12-12 22:02:09 +08:00
parent 38d6ff30a3
commit 810bac464f
12 changed files with 599 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
source/device/DeviceSerial.py
View File

@@ -1,6 +1,8 @@
import time
from pathlib import Path
from serial import Serial
from serial.tools import list_ports
from abc import ABC, abstractmethod
from .tools import ByteConv
from .function import protocols
@@ -119,6 +121,7 @@ class DeviceSerial:
'bytesize': 8,
'stopbits': 1,
}
kwargs = kwargs if kwargs else None
serial_close = lambda com: com.close() if com.isOpen() else None
serial_port_check = lambda port: port.upper() in (com.name for com in list_ports.comports())
@@ -126,9 +129,13 @@ class DeviceSerial:
match (self.__com, port, kwargs):
case (None, str() as port, None):
""" 使用默认参数打开串口 """
if not serial_port_check(port):
raise ValueError("无效串口端口: %s" % port)
self.__com = Serial(port, timeout=0, **com_config)
case (None, str() as port, dict() as kwargs):
""" 使用指定参数打开串口 """
if not serial_port_check(port):
raise ValueError("无效串口端口: %s" % port)
com_config['baudrate'] = kwargs['baudrate'] if 'baudrate' in kwargs.keys() else com_config['baudrate']
com_config['parity'] = kwargs['parity'] if 'parity' in kwargs.keys() else com_config['parity']
com_config['bytesize'] = kwargs['bytesize'] if 'bytesize' in kwargs.keys() else com_config['bytesize']
@@ -136,19 +143,19 @@ class DeviceSerial:
self.__com = Serial(port, timeout=0, **com_config)
case (Serial() as com, None, None):
""" 无参数重开串口 """
serial_close(self.__com)
serial_close(com)
com.open()
case (Serial() as com, port, None):
""" 重新指定端口号并打开串口 """
serial_close(self.__com)
if serial_port_check(port):
serial_close(com)
if not serial_port_check(port):
raise ValueError("无效串口端口: %s" % port)
com.port = port
com.open()
case (None, str() as port, dict() as kwargs):
case (Serial() as com, str() as port, dict() as kwargs):
""" 重新指定端口号与配置并打开串口 """
serial_close(self.__com)
if serial_port_check(port):
serial_close(com)
if not serial_port_check(port):
raise ValueError("无效串口端口: %s" % port)
com.port = port
com.baudrate = kwargs['baudrate'] if 'baudrate' in kwargs.keys() else com_config['baudrate']
@@ -162,6 +169,26 @@ class DeviceSerial:
return self.__com.is_open
@abstractmethod
def frame_read(self, daddr=0x60, dlen=0x50) -> bool:
pass
@abstractmethod
def frame_write_one(self, daddr=0x85, dval=-900) -> bool:
pass
@abstractmethod
def frame_write_dual(self, daddr=0x91, dval=600) -> bool:
pass
@abstractmethod
def frame_write_str(self, daddr=0x82, dval=[0x06, 0x05, 0x04, 0x03, 0x02, 0x01]) -> bool:
pass
@abstractmethod
def frame_update(self, path_file: Path, makefile: bool = False) -> bool:
pass

15
source/device/LaminaAdapter.py
View File

@@ -16,6 +16,7 @@ ParamMap_LaminaAdapter = {
# 4 - str
# 5 - addr
# 6 - float
# 8 - func_callback
0x00: ["硬件版本识别电压", 2, 1000],
0x01: ["输出电容电电压", 2, 10],
0x02: ["参考电压", 2, 10],
@@ -128,6 +129,13 @@ ParamMap_LaminaAdapter = {
0x170: ["SN", 4, 16],
0x180: ["MES", 4, 16],
0x190: ["Datetime", 4, 16],
0x1A0: ["事件记录1", 8, 16],
0x1B0: ["事件记录2", 8, 16],
0x1C0: ["事件记录3", 8, 16],
0x1D0: ["事件记录4", 8, 16],
0x1E0: ["事件记录5", 8, 16],
0x1F0: ["事件记录6", 8, 16],
0x200: ["事件记录-Log", 8, 16],
}
MemoryMap_SLCP001 = {
@@ -221,6 +229,11 @@ class LaminaAdapter(DeviceSerial):
self.block['data']['data_val'] = dval
return self._transfer_data()
def frame_log(self, step='next') -> bool:
self.block['data']['type'] = 'log'
self.block['data']['step'] = step
return self._transfer_data()
def frame_update(self, path_file: Path, makefile: bool = False) -> bool:
""" 程序升级
注意: 在使用单板升级测试时, 需要关闭低电压检测功能, 否则无法启动升级流程;
@@ -761,4 +774,4 @@ def GenerateImage_DLSY001_p460(path_boot: Path, path_main: Path, path_back: Path
offset_image = 0x056000
Image[offset_image: offset_image + len(back_header)] = back_header
return bytearray(Image), main_header, back_header
return bytearray(Image), main_header, back_header

191
source/device/LaminaController.py
View File

@@ -21,6 +21,10 @@ ParamMap_LaminaController = {
# 7 - numberList
0x00: ["绝缘检测正电阻", 6],
0x02: ["绝缘检测负电阻", 6],
0x04: ["负对地绝缘检测电压(动作)" , 2, 100],
0x05: ["负对地绝缘检测电压(未动作)" , 2, 100],
0x06: ["正对地绝缘检测电压(动作)" , 2, 100],
0x07: ["正对地绝缘检测电压(未动作)" , 2, 100],
0x0B: ["事件标志", 1],
0x0C: ["告警字1", 1],
0x0D: ["告警字2", 1],
@@ -117,16 +121,17 @@ ParamMap_LaminaController = {
MemoryMap_280039 = {
'image_size': 2*0x030000, # 镜像文件大小
'app_size': 2*0x004000, # 应用程序大小
'boot_size': 2*0x014000, # Boot程序大小
'boot_size': 2*0x004000, # Boot程序大小
'app_size': 2*0x014000, # 应用程序大小
'boot_addr': 0x000000, # Boot程序地址
'main_header': 0x006000, # main信息地址
'back_header': 0x007000, # back信息地址
'main_info': 0x088000, # main版本地址
'back_info': 0x007000, # back版本地址
'main_addr': 0x008000, # main程序地址
'back_addr': 0x01C000, # back程序地址
'main_info': 0x088000, # main版本地址
'back_info': 0x09C000, # back版本地址
}
class LaminaController:
@@ -553,44 +558,49 @@ class LaminaController_new(DeviceSerial):
self.flag_print = False
self.retry = 3
self.time_out = 1.5
self.block['file_block_size'] = 240
# 读取config
self.block['type'] = 'record_cfg'
self.block['step'] = 'start'
frame_master = function.protocols.make_frame_modbus(self.block)
try:
status = 'init' # 初始化
self.block['file_block_size'] = 240
ret, pbar = True, None
frame_data_cfg = []
while ret:
if ret := self.__transfer_data(frame_master):
frame_data_cfg.append(self.output['record'])
if self.output['record']['seq'] == 0:
pbar = tqdm(total=self.output['record']['total'] + 1, desc="Record Config Reading")
self.block['step'] = 'next'
frame_master = function.protocols.make_frame_modbus(self.block)
elif (self.output['record']['seq']) >= self.output['record']['total']:
ret = False
pbar and pbar.update()
pbar and pbar.close()
status = 'read_config' # 读取config
self.block['type'] = 'record_cfg'
self.block['step'] = 'start'
ret, pbar = True, None
frame_data_cfg = []
while ret:
if ret := self._transfer_data():
frame_data_cfg.append(self.output['record'])
if self.output['record']['seq'] == 0:
pbar = tqdm(total=self.output['record']['total'] + 1, desc="Record Config Reading")
self.block['step'] = 'next'
elif (self.output['record']['seq']) >= self.output['record']['total']:
ret = False
pbar and pbar.update()
pbar and pbar.close()
# 读取data
self.block['type'] = 'record_data'
self.block['step'] = 'start'
frame_master = function.protocols.make_frame_modbus(self.block)
ret, pbar = True, None
frame_data_record = []
while ret:
if ret := self.__transfer_data(frame_master):
frame_data_record.append(self.output['record'])
if self.output['record']['seq'] == 0:
pbar = tqdm(total=self.output['record']['total'] + 1, desc="Record Data Reading")
self.block['step'] = 'next'
frame_master = function.protocols.make_frame_modbus(self.block)
elif (self.output['record']['seq']) >= self.output['record']['total']:
ret = False
pbar and pbar.update()
pbar and pbar.close()
status = 'read_data' # 读取data
self.block['type'] = 'record_data'
self.block['step'] = 'start'
ret, pbar = True, None
frame_data_record = []
while ret:
if ret := self._transfer_data():
frame_data_record.append(self.output['record'])
if self.output['record']['seq'] == 0:
pbar = tqdm(total=self.output['record']['total'] + 1, desc="Record Data Reading")
self.block['step'] = 'next'
elif (self.output['record']['seq']) >= self.output['record']['total']:
ret = False
pbar and pbar.update()
pbar and pbar.close()
except Exception as ex:
""" 通用异常处理 """
self.flag_print, self.retry, self.time_out = param_saved
report = f'Record Fail: {status}'
report += f', Frame in {self.block["record"]["seq"]}' if status == 'read_config' or status == 'read_data' else ''
report += f'\n Error by {ex}' if not isinstance(ex, AssertionError) else ''
print(report)
return False
self.flag_print, self.retry, self.time_out = param_saved
if (len(frame_data_record) != 32) or (len(frame_data_cfg) != 3):
@@ -602,7 +612,7 @@ class LaminaController_new(DeviceSerial):
config_record = {'LinePos': []}
pos = 0
if data_cfg_bare[pos+1] != 0xF1 or data_cfg_bare[pos] != 0xF1:
Warning("config 配置文件格式异常")
raise Warning("config 配置文件格式异常")
pos += 2
config_record['LinePos'].append(pos)
len_faultword = (data_cfg_bare[3] * 0x100 + data_cfg_bare[2])
@@ -621,7 +631,7 @@ class LaminaController_new(DeviceSerial):
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 配置文件格式异常")
raise Warning("config 配置文件格式异常")
pos += 2
config_record['LinePos'].append(pos)
config_record['ChannelNum'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
@@ -631,7 +641,7 @@ class LaminaController_new(DeviceSerial):
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 配置文件格式异常")
raise Warning("config 配置文件格式异常")
pos += 2
config_record['LinePos'].append(pos)
config_record['ChannelDescription'] = []
@@ -648,7 +658,7 @@ class LaminaController_new(DeviceSerial):
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 配置文件格式异常")
raise Warning("config 配置文件格式异常")
pos += 2
config_record['LinePos'].append(pos)
config_record['SysFreq'] = data_cfg_bare[pos+1] * 0x100 + data_cfg_bare[pos]
@@ -664,7 +674,7 @@ class LaminaController_new(DeviceSerial):
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 配置文件格式异常")
raise Warning("config 配置文件格式异常")
pos += 2
config_record['LinePos'].append(pos)
time_stamp = {
@@ -726,56 +736,57 @@ class LaminaController_new(DeviceSerial):
return True
def frame_update(self, path_bin: Path, makefile: bool = False) -> bool:
def frame_update(self, path_file: Path, makefile: bool = False) -> bool:
""" 程序升级
注意: 在使用单板升级测试时, 需要关闭低电压检测功能, 否则无法启动升级流程;
"""
if makefile:
self.block['file'], file_bin = GeneratePackage_DLSP001_p280039(path_bin)
else:
self.block['file'] = path_bin.read_bytes()
self.block['header_offset'] = 184
self.block['type'] = 'update'
param_saved = self.flag_print, self.retry, self.time_out
self.retry = 5
self.retry = 3
self.time_out = 0.5
self.flag_print = False
try:
status = 'init' # 初始化
if not path_file.exists():
raise Exception("工程编译目标文件不存在.")
if makefile and self.make_package is not None:
self.block['file'] = self.make_package(path_file)
else:
self.block['file'] = path_file.read_bytes()
# 启动帧
self.block['step'] = 'start'
self.block['index'] = 0
frame_master = function.protocols.make_frame_modbus(self.block)
if not self.__transfer_data(frame_master):
self.block['type'] = 'update'
self.block['header_offset'] = 184
status = 'start' # 启动帧
self.block['step'] = 'start'
self.block['index'] = 0
assert self._transfer_data()
self.block['file_block_size'] = self.output['upgrade']['length']
# 避免接收到延迟返回报文
time.sleep(self.time_out)
status = 'trans' # 文件传输
self.retry = 3
self.time_out = 1.5
self.block['step'] = 'trans'
self.block['index'] = 0
frame_total = ceil((len(self.block['file']) - self.block['header_offset']) / self.block['file_block_size'])
for idx in tqdm(range(frame_total), desc="File Transmitting"):
self.block['index'] = idx
assert self._transfer_data()
status = 'end' # 结束升级
self.time_out = 1
self.block['step'] = 'end'
self.block['index'] += 1
assert self._transfer_data()
except Exception as ex:
""" 通用异常处理 """
self.flag_print, self.retry, self.time_out = param_saved
print('Upgrade Fail: start')
return False
self.block["data"]['file_block_size'] = self.output['upgrade']['length']
# 避免接收到延迟返回报文
time.sleep(self.time_out)
# 文件传输
self.retry = 3
self.time_out = 1.5
self.block["data"]['step'] = 'trans'
self.block['index'] = 0
frame_total = ceil((len(self.block["data"]['file']) - self.block['header_offset']) / self.block["data"]['file_block_size'])
for idx in tqdm(range(frame_total), desc="File Transmitting"):
self.block["data"]['index'] = idx
frame_master = function.protocols.make_frame_modbus(self.block)
if not self.__transfer_data(frame_master):
self.flag_print, self.retry, self.time_out = param_saved
print(f'Upgrade Fail: trans data in {idx}')
return False
# 结束升级
self.time_out = 1
self.block["data"]['step'] = 'end'
self.block["data"]['index'] += 1
frame_master = function.protocols.make_frame_modbus(self.block)
if not self.__transfer_data(frame_master):
self.flag_print, self.retry, self.time_out = param_saved
print(f'Upgrade Fail: end')
report = f'Upgrade Fail: {status}'
report += f', Frame in {self.block["index"]}' if status == 'trans' else ''
report += f'\n Error by {ex}' if not isinstance(ex, AssertionError) else ''
print(report)
return False
self.flag_print, self.retry, self.time_out = param_saved
@@ -814,7 +825,7 @@ def GeneratePackage(type_dev: str, path_hex: Path, **kwargs) -> bytearray:
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))}]")
print(f"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main) // 2)}]")
# 组装镜像
Image = [0xFF] * (len(main_header) + len(encrypt_main))
@@ -883,9 +894,9 @@ def GenerateImage(type_dev: str, path_boot: Path, path_main: Path, path_back: Pa
# Log打印
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"\t header_length={len(main_header)}, bin_length={len(bin_main)}[{hex(len(bin_main) // 2)}]")
print(f"Back File:")
print(f"\t header_length={len(back_header)}, bin_length={len(bin_back)}[{hex(len(bin_back))}]")
print(f"\t header_length={len(back_header)}, bin_length={len(bin_back)}[{hex(len(bin_back) // 2)}]")
# 额外文件
if 'output_header' in kwargs.keys() and kwargs['output_header']:

46
source/device/function/protocols.py
View File

@@ -29,6 +29,7 @@ frame_modbus = {
0x04: [5, 1, 2],
0x06: [8, 0],
0x07: [8, 3, 2, {0x01: 2, 0x02: 0, 0x03: 0}],
0x08: [5, 3, 2, {0x01: 1, 0x02: 0x22, 0x03: 1}],
0x10: [8, 0],
0x11: [11, 2, 7]
}
@@ -94,6 +95,19 @@ def make_frame_modbus(block:dict) -> bytearray:
frame.append(block['file_block_size'] // 256)
frame.append(block['file_block_size'] % 256)
elif block['type'] == 'log':
""" 事件记录系列报文 """
frame.append(0x08)
if block['step'] == 'start':
frame.append(0x01)
frame.append(block['data_len'] // 0x100 % 0x100)
frame.append(block['data_len'] % 0x100)
elif block['step'] == 'next':
frame.append(0x02)
elif block['step'] == 'end':
frame.append(0x03)
else:
raise Exception("Modbus Log Frame Step Error.")
else:
""" 数据读取系列报文 """
data_addr = block['data_addr']
@@ -217,6 +231,12 @@ def check_frame_modbus(frame:bytes, block:dict) -> dict:
else:
raise Exception("Frame read data length error.")
output['result'] = True
elif output['code_func'] == 0x08:
if frame[2] == 0x02:
output['Regs'] = display_data(0x200, frame[5:-2], block['data_define'])
else:
raise Exception(" Log SubFunc code error.")
output['result'] = True
elif output['code_func'] == 0x06:
""" 单个数据写入帧 """
output['result'] = True
@@ -374,6 +394,32 @@ def display_data(address: int, data: bytes, modbus_map: dict=modbus_map) -> dict
""" 正序数值表示 """
data_len = current_map[2]
item = list(map(lambda x: x, data[:2 * data_len]))
elif current_map[1] == 8:
""" 事件记录解析 """
data_len = current_map[2]
time_stamp, event_id, event_num = struct.unpack('<IBB', data[:6])
item = f"Event: {event_id}\n"
item += f"\ttime: {time_stamp}\n"
item += f"\tnumber: {event_num}\n"
if event_id == 2:
""" 故障记录事件 """
info_len, status, flag_fault, flag_alarm, volt_in, volt_out, curr_in, curr_induc, cbox_power_limit, flag_control = struct.unpack('<BBHHhhhhhH', data[6:6+18])
info = data[6+18:6+18+info_len].decode().strip('\x00')
item += f"\tData: {info}\n"
item += f"\t\tstatus: {status:x}\n"
item += f"\t\tflag_fault: {flag_fault:x}\n"
item += f"\t\tflag_alarm: {flag_alarm:x}\n"
item += f"\t\tvolt_in: {volt_in / 10}\n"
item += f"\t\tvolt_out: {volt_out / 10}\n"
item += f"\t\tcurr_in: {curr_in / 100}\n"
item += f"\t\tcurr_induc: {curr_induc / 100}\n"
item += f"\t\tcbox_power_limit: {cbox_power_limit / 10}\n"
item += f"\t\tflag_control: {flag_control:b}\n"
else:
info_len, = struct.unpack('<B', data[6:6+1])
info = data[7:7+info_len].decode().strip('\x00')
item += f"\tData: {info}\n"
output_data[idx] = data_label, item
idx += data_len

18
source/device/tools/ByteConv.py
View File

@@ -1,7 +1,7 @@
def trans_list_to_str(data: list) -> str:
def trans_list_to_str(data: list, word_len=2, prefix='') -> str:
""" 标准串口字符串表示 """
func_trans = lambda x: ('00' + hex(x % 256)[2:])[-2:].upper()
return " ".join(map(func_trans, data))
func_trans_word = lambda x: prefix + ('0' * word_len + hex(x % (0x10 ** word_len))[2:])[-word_len:].upper()
return " ".join(map(func_trans_word, data))
def trans_str_to_list(data: str) -> list:
@@ -26,4 +26,14 @@ def conv_int_to_array(num: int, big_end=False):
def display_hex(data:int, length:int=2) -> str:
""" Hex字符固定最小长度表示 """
return f"0x{data:0{length}X}"
return f"0x{data:0{length}X}"
def trans_list_to_str_by_word(data: list, big_end=True) -> str:
""" 将数据以word格式显示 """
s = [data[i+1] * 0x100 + data[i] for i in range(0, len(data), 2)]
r = " ".join(map(hex, s))
print(data)
print(trans_list_to_str(data))
print(r)
return r