275 lines
9.4 KiB
Python
275 lines
9.4 KiB
Python
import struct
|
|
from crc import Calculator, Crc16
|
|
from utl import display_hex, trans_list_to_str
|
|
|
|
modbus_map = {
|
|
# 1 - Hex
|
|
# 2 - Int16
|
|
# 3 - lnt32
|
|
# 4 - str
|
|
# 5 - addr
|
|
# 6 - float
|
|
0x01: ["Hex示例", 1],
|
|
0x02: ["Int示例", 2],
|
|
0x03: ["Int32示例", 3],
|
|
0x04: ["str示例", 4, 16],
|
|
0x10: ["addr示例", 5, 6],
|
|
0x20: ["Float示例", 6],
|
|
}
|
|
|
|
|
|
def make_frame_modbus(block:dict):
|
|
""" modbus 生成函数"""
|
|
frame = []
|
|
calculator = Calculator(Crc16.MODBUS)
|
|
|
|
frame.append(block['addr_dev'])
|
|
if block['type'] == 'update':
|
|
""" 升级系列报文 """
|
|
frame.append(0x07)
|
|
if len(block['file']) <= block["header_offset"]:
|
|
raise Exception("Modbus Update error, file too small.")
|
|
|
|
if block['step'] == 'start':
|
|
frame.append(0x01)
|
|
frame.append(0x00)
|
|
frame.append(0x00)
|
|
frame.append(0x00)
|
|
frame.append(block['header_offset'] // 256)
|
|
frame.append(block['header_offset'] % 256)
|
|
frame += list(block['file'][:block['header_offset']])
|
|
elif block['step'] == 'trans':
|
|
file_offset = block["header_offset"] + block['index'] * block['file_block_size']
|
|
file_block = block['file'][file_offset: file_offset + block['file_block_size']]
|
|
frame.append(0x02)
|
|
frame.append(0x00)
|
|
frame.append(block['index'] // 256)
|
|
frame.append(block['index'] % 256)
|
|
frame.append(len(file_block) // 256)
|
|
frame.append(len(file_block) % 256)
|
|
frame += list(file_block)
|
|
elif block['step'] == 'end':
|
|
frame.append(0x03)
|
|
frame.append(0x00)
|
|
frame.append(block['index'] // 256)
|
|
frame.append(block['index'] % 256)
|
|
frame.append(0x00)
|
|
frame.append(0x00)
|
|
else:
|
|
raise Exception("Modbus Update Frame Step Error.")
|
|
else:
|
|
""" 数据读取系列报文 """
|
|
data_addr = block['data_addr']
|
|
if block['type'] == "read":
|
|
frame.append(0x03)
|
|
data_len = block['data_len']
|
|
frame.append(data_addr // 256 % 256)
|
|
frame.append(data_addr % 256)
|
|
frame.append(data_len // 256 % 256)
|
|
frame.append(data_len % 256)
|
|
elif block['type'] == "write_one":
|
|
frame.append(0x06)
|
|
data_val = block['data_val']
|
|
if data_val < 0:
|
|
data_val += 0x1_0000
|
|
frame.append(data_addr // 256 % 256)
|
|
frame.append(data_addr % 256)
|
|
frame.append(data_val // 256 % 256)
|
|
frame.append(data_val % 256)
|
|
elif block['type'] == "write_dual":
|
|
frame.append(0x10)
|
|
data_val = block['data_val']
|
|
if data_val < 0:
|
|
data_val += 0x1_0000_0000
|
|
frame.append(data_addr // 256 % 256)
|
|
frame.append(data_addr % 256)
|
|
frame.append(0x00)
|
|
frame.append(0x02)
|
|
frame.append(0x04)
|
|
frame.append(data_val // 256 % 256)
|
|
frame.append(data_val % 256)
|
|
data_val //= 0x1_0000
|
|
frame.append(data_val // 256 % 256)
|
|
frame.append(data_val % 256)
|
|
elif block['type'] == "write_str":
|
|
frame.append(0x10)
|
|
data_len = len(block['data_val'])
|
|
item_len = 2 * block['data_define'][data_addr][2]
|
|
data_val = block['data_val']
|
|
if data_len > item_len:
|
|
raise Exception("Modbus data len oversize.")
|
|
elif data_len < item_len:
|
|
data_val += '\000' * (item_len - data_len)
|
|
data_len = len(data_val)
|
|
frame.append(data_addr // 256 % 256)
|
|
frame.append(data_addr % 256)
|
|
frame.append(0x00)
|
|
frame.append(data_len // 2)
|
|
frame.append(data_len)
|
|
|
|
if type(data_val[0]) == str:
|
|
data_val = list(map(lambda x: x.encode()[0], data_val))
|
|
for i in range(data_len//2):
|
|
frame.append(data_val[2*i + 1])
|
|
frame.append(data_val[2*i])
|
|
else:
|
|
raise Exception("Modbus Frame Type Error.")
|
|
|
|
|
|
crc = calculator.checksum(bytearray(frame))
|
|
frame.append(crc % 256)
|
|
frame.append(crc // 256)
|
|
|
|
return frame
|
|
|
|
def make_frame_dlt645(block:dict):
|
|
""" dlt645 生成函数"""
|
|
frame = []
|
|
|
|
if block['type'] == 'modbus':
|
|
""" Modbus 透传帧 """
|
|
ctrl_code = 0x1F # Modbus 透传帧功能码
|
|
data_frame = make_frame_modbus(block["data"])
|
|
len_data = len(data_frame)
|
|
else:
|
|
raise Exception("Unknown dlt645 frame type.")
|
|
|
|
frame.append(0x68)
|
|
frame += block['addr'][:6]
|
|
frame.append(0x68)
|
|
frame.append(ctrl_code)
|
|
frame.append(len_data)
|
|
frame += data_frame
|
|
frame.append(sum(frame) % 256)
|
|
frame.append(0x16)
|
|
|
|
return frame
|
|
|
|
def display_data(modbus_map: dict, address: int, data: list):
|
|
""" 格式化表示数据 """
|
|
def display_str(data, len_data):
|
|
item = bytearray(data[:2 * len_data])
|
|
for i in range(len_data):
|
|
t = item[2*i]
|
|
item[2*i] = item[2*i+1]
|
|
item[2*i+1] = t
|
|
return item
|
|
|
|
output = "Parse Result: \n"
|
|
idx = address
|
|
len_max = 0
|
|
while data:
|
|
data_len = 1
|
|
data_label = "未知数据"
|
|
if idx not in modbus_map.keys():
|
|
item = data[0] * 0x0100 + data[1]
|
|
item = display_hex(item, 4)
|
|
else:
|
|
current_map = modbus_map[idx]
|
|
data_label = current_map[0]
|
|
if current_map[1] == 1:
|
|
""" Hex字符表示 """
|
|
item = data[0] * 0x0100 + data[1]
|
|
item = display_hex(item, 4)
|
|
elif current_map[1] == 2:
|
|
""" 16位数值表示 """
|
|
item = data[0] * 0x0100 + data[1]
|
|
if item > 0x8000:
|
|
item -= 0x1_0000
|
|
elif current_map[1] == 3:
|
|
""" 32位数值表示 """
|
|
item = data[2] * 0x0100 + data[3]
|
|
item *= 0x10000
|
|
item += data[0] * 0x0100 + data[1]
|
|
if data[2] > 0x80:
|
|
item -= 0x1_0000_0000
|
|
data_len = 2
|
|
elif current_map[1] == 4:
|
|
""" 字符串表示 """
|
|
data_len = current_map[2]
|
|
item = display_str(data, data_len)
|
|
item = item.replace(b'\xff', b' 0xFF')
|
|
item = item.decode()
|
|
elif current_map[1] == 5:
|
|
""" 载波地址表示 """
|
|
data_len = current_map[2]
|
|
item = display_str(data, data_len)
|
|
item = trans_list_to_str(item)
|
|
elif current_map[1] == 6:
|
|
""" 浮点数值表示 """
|
|
temp = [data[2], data[3], data[0], data[1]]
|
|
item = struct.unpack('>f', bytes(temp))[0]
|
|
data_len = 2
|
|
|
|
if len_max < len(data_label):
|
|
len_max = len(data_label)
|
|
len_remain = len_max - len(data_label)
|
|
output_line = "\t".join([display_hex(idx, 4), data_label + " " * len_remain, str(item)])
|
|
|
|
idx += data_len
|
|
del data[0:data_len * 2]
|
|
output += output_line + "\n"
|
|
return output
|
|
|
|
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.")
|
|
|
|
code_func = frame[1]
|
|
code_subfunc = frame[2]
|
|
if code_func == 0x07:
|
|
""" 升级回复帧 """
|
|
if frame[3] == 0x00:
|
|
check_result = True
|
|
else:
|
|
check_result = False
|
|
update_index = frame[4] * 256 + frame[5]
|
|
check_hope = 0
|
|
if code_subfunc == 0x01:
|
|
""" 处理帧头 """
|
|
check_hope = frame[6] * 256 + frame[7]
|
|
elif code_subfunc == 0x02:
|
|
""" 解析帧序号及返回值, 不做序列号校验 """
|
|
elif code_subfunc == 0x03:
|
|
""" 升级结束 """
|
|
else:
|
|
raise Exception("Func code or Return code error.")
|
|
return check_result, update_index, check_hope
|
|
elif code_func == 0x03 or code_func == 0x04:
|
|
""" 数据读取帧 """
|
|
if frame[2] == len(frame[3:-2]):
|
|
if type(block) is dict:
|
|
data_addr = block['data_addr']
|
|
return display_data(block['data_define'], data_addr, list(frame[3:-2]))
|
|
else:
|
|
return list(frame[3:-2])
|
|
else:
|
|
raise Exception("Frame read error.")
|
|
elif code_func == 0x06:
|
|
""" 单个数据写入帧 """
|
|
return frame[2:-2]
|
|
elif code_func == 0x10:
|
|
""" 多个数据写入帧 """
|
|
return frame[2:-2]
|
|
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.")
|
|
|
|
code_func = frame[8]
|
|
if code_func == 0x9F:
|
|
block = block['data']
|
|
return check_frame_modbus(frame[10:-2], block)
|
|
else:
|
|
raise Exception("DLT645 Frame type error.")
|
|
|