import struct from crc import Calculator, Crc16 from tools.ByteConv import trans_list_to_str, trans_str_to_list, display_hex 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], } 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 生成函数""" 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.") 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'] 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 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): """ 校验modbus帧回复 """ if len(frame:=find_frame_modbus(frame, block['addr_dev'])) == 0: raise Exception("No frame data") 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 trans_list_to_str(frame[3:-2]) else: raise Exception("Frame read error.") elif code_func == 0x06: """ 单个数据写入帧 """ return trans_list_to_str(frame[2:-2]) elif code_func == 0x10: """ 多个数据写入帧 """ 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 len(frame:=find_frame_dlt645(frame, block['addr'])) == 0: raise Exception("No frame data") 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.") 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)