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

允许合法帧子串通过校验;

Browse Source
This commit is contained in:
何 泽隆
2024-08-31 18:36:35 +08:00
parent 7bbe0a08ef
commit 518a325072
1 changed files with 178 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
source/func_frame.py
View File

@@ -1,6 +1,6 @@
import struct import struct
from crc import Calculator, Crc16 from crc import Calculator, Crc16
from tools.ByteConv import trans_list_to_str, display_hex from tools.ByteConv import trans_list_to_str, trans_str_to_list, display_hex
modbus_map = { modbus_map = {
# 1 - Hex # 1 - Hex
@@ -17,6 +17,19 @@ modbus_map = {
0x20: ["Float示例", 6], 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): def make_frame_modbus(block:dict):
""" modbus 生成函数""" """ modbus 生成函数"""
@@ -211,13 +224,62 @@ def display_data(modbus_map: dict, address: int, data: list):
output += output_line + "\n" output += output_line + "\n"
return output 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] not in frame_defines.keys():
continue
frame_define = frame_defines[buffer[i+1]]
j = frame_define[0]
if 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 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+8] != 0x68:
continue
if buffer[i+1] != bytearray(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): def check_frame_modbus(frame, block=None):
""" 校验modbus帧回复 """ """ 校验modbus帧回复 """
calculator = Calculator(Crc16.MODBUS)
if calculator.checksum(frame[:-2]) != (frame[-1] * 0x100 + frame[-2]): if len(frame:=find_frame_modbus(frame, block['addr_dev'])) == 0:
raise Exception("Frame Crc check failed.") raise Exception("No frame data")
elif frame[0] != 0x01:
raise Exception("Frame device address error.")
code_func = frame[1] code_func = frame[1]
code_subfunc = frame[2] code_subfunc = frame[2]
@@ -246,24 +308,22 @@ def check_frame_modbus(frame, block=None):
data_addr = block['data_addr'] data_addr = block['data_addr']
return display_data(block['data_define'], data_addr, list(frame[3:-2])) return display_data(block['data_define'], data_addr, list(frame[3:-2]))
else: else:
return list(frame[3:-2]) return trans_list_to_str(frame[3:-2])
else: else:
raise Exception("Frame read error.") raise Exception("Frame read error.")
elif code_func == 0x06: elif code_func == 0x06:
""" 单个数据写入帧 """ """ 单个数据写入帧 """
return frame[2:-2] return trans_list_to_str(frame[2:-2])
elif code_func == 0x10: elif code_func == 0x10:
""" 多个数据写入帧 """ """ 多个数据写入帧 """
return frame[2:-2] return trans_list_to_str(frame[2:-2])
else: else:
raise Exception(f"Frame Date error. func={code_func}, func_sub={code_subfunc}, len={len(frame)}") raise Exception(f"Frame Date error. func={code_func}, func_sub={code_subfunc}, len={len(frame)}")
def check_frame_dlt645(frame, block=None): def check_frame_dlt645(frame, block=None):
""" 校验dlt645帧回复 """ """ 校验dlt645帧回复 """
if sum(frame[:-2]) % 0x100 != frame[-2]: if len(frame:=find_frame_dlt645(frame, block['addr'])) == 0:
raise Exception("DLT645 Frame Verify error.") raise Exception("No frame data")
elif len(frame[10:-2]) != frame[9]:
raise Exception("DLT645 Frame len error.")
code_func = frame[8] code_func = frame[8]
if code_func == 0x9F: if code_func == 0x9F:
@@ -272,3 +332,108 @@ def check_frame_dlt645(frame, block=None):
else: else:
raise Exception("DLT645 Frame type error.") raise Exception("DLT645 Frame type error.")
if __name__ == "__main__":
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")
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_test2), block)