diff --git a/source/dev_LaminaAdapter.py b/source/dev_LaminaAdapter.py index db39f4f..1c20a40 100644 --- a/source/dev_LaminaAdapter.py +++ b/source/dev_LaminaAdapter.py @@ -1,8 +1,10 @@ import time +from math import ceil +from tqdm import tqdm from pathlib import Path from serial import Serial from tools.ByteConv import trans_list_to_str -from func_frame import make_frame_dlt645, check_frame_dlt645 +from func_frame import make_frame_dlt645, check_frame_dlt645, print_display from func_upgrade import GeneratePackage_SLCP101_p460 modbus_map = { @@ -115,6 +117,8 @@ modbus_map = { class LaminaAdapter: + """ 叠光适配器\优化器 + """ def __init__(self, com_name="COM16", addr_645=[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA], addr_modbus=0x01, **kwargs): # 初始化串口通信 if com_name is not None: @@ -129,6 +133,7 @@ class LaminaAdapter: self.flag_print = 'frame_print' in kwargs.keys() self.time_out = kwargs['time_out'] if 'time_out' in kwargs.keys() else 1 + self.time_gap = kwargs['time_gap'] if 'time_gap' in kwargs.keys() else 0.01 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 @@ -140,50 +145,74 @@ class LaminaAdapter: 'data_define': modbus_map, }, } - self.output = {} + self.output = { + 'result': False, + 'code_func': 0x00, + } + self.log = { + 'send': 0, + 'read': 0, + 'keep-fail': 0, + 'record': { + 'config': None, + 'data': None, + }, + } - def __transfer_data(self, frame): + def __read_frame(self) -> bool: + """ 使用帧字节超时策略读报文帧, 并进行解析数据, 打印异常 """ + frame_recv = b'' + time_start, time_current, flag_frame = time.time(), time.time(), False + while (time_current - time_start) < self.time_out: + time.sleep(self.time_gap) + bytes_read = self.__com.read_all() + time_current = time.time() + if flag_frame and len(bytes_read) == 0: + break + elif len(bytes_read): + flag_frame = True + frame_recv += bytes_read + try: + self.output = check_frame_dlt645(frame_recv, self.block['data']) + if self.flag_print: + print("Read Frame: ", trans_list_to_str(frame_recv)) + except Exception as ex: + print("Error Info: ", ex) + if self.flag_print and frame_recv: + print("Fail Data: " , trans_list_to_str(frame_recv)) + self.output['result'] = False + + return self.output['result'] + + def __transfer_data(self, frame: bytearray) -> bool: """ 报文数据传输 """ - 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)) + fail_count = 0 + while fail_count < self.retry: + frame_discard = self.__com.read_all() + self.__com.write(frame) + self.log['send'] += 1 - 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 + if self.flag_print and frame_discard: + print("Discard Data: " , frame_discard) + if self.flag_print: + print("Send Frame: ", trans_list_to_str(frame)) - try: - output_text = check_frame_dlt645(frame_recv, block=self.block) - except Exception as e: - print(e) - cnt += 1 - continue - print(output_text) - for line in output_text.split('\n'): - line = line.strip() - line_info = line.split('\t') - if line_info[0] == '0x0100': - self.output['version'] = line_info[2].rstrip('\x00') - elif line_info[0] == '0x0082': - self.output['address'] = line_info[2].rstrip('\x00') - break + if self.__read_frame(): + if 'Regs' in self.output.keys(): + print_display(self.output['Regs']) + self.log['read'] += 1 + break + else: + fail_count += 1 + if fail_count >= self.log['keep-fail']: + self.log['keep-fail'] = fail_count + time.sleep(2*self.time_out) - if self.flag_print: - print(trans_list_to_str(frame)) - print(trans_list_to_str(frame_recv)) - - return cnt < self.retry + return fail_count < self.retry def frame_read(self, daddr=0x60, dlen=0x50): self.block['data']['type'] = 'read' @@ -222,6 +251,8 @@ class LaminaAdapter: 注意: 在使用单板升级测试时, 需要关闭低电压检测功能, 否则无法启动升级流程; """ + param_saved = self.flag_print, self.retry, self.time_out + self.block['data']['type'] = 'update' self.block['data']['step'] = 'start' self.block['data']['index'] = 0 @@ -230,125 +261,65 @@ class LaminaAdapter: # 启动帧 frame_master = bytearray(make_frame_dlt645(self.block)) - # 等待擦除完成返回 - 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 - - _, _, self.block["data"]['file_block_size'] = check_frame_dlt645(frame_slave, self.block) - break - - if self.block["data"]['file_block_size'] == 0: - raise Exception("Error slave response.") + if not self.__transfer_data(frame_master): + 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' - data_remain = len(self.block["data"]['file']) - self.block['data']['header_offset'] - while data_remain > 0: - frame_master = bytearray(make_frame_dlt645(self.block)) - - 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_dlt645(frame_slave, self.block) - 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']['index'] = 0 + frame_total = ceil((len(self.block["data"]['file']) - self.block['data']['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 = make_frame_dlt645(self.block['data']) + 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' - frame_master = bytearray(make_frame_dlt645(self.block)) + self.block["data"]['index'] += 1 + frame_master = make_frame_dlt645(self.block['data']) + if not self.__transfer_data(frame_master): + self.flag_print, self.retry, self.time_out = param_saved + print(f'Upgrade Fail: end') + return False - 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_dlt645(frame_slave, self.block) - 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.flag_print, self.retry, self.time_out = param_saved + return True 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 + param_saved = dev_lamina.flag_print, dev_lamina.retry, dev_lamina.time_out dev_lamina.flag_print = False + dev_lamina.retry = 1 try: - while 1: - if dev_lamina.frame_read(0x0E, 0x13): - log_success += 1 - cnt_failedseries = 0 - else: - log_failed += 1 - cnt_failedseries += 1 - if log_failedseries <= cnt_failedseries: - log_failedseries = cnt_failedseries + while True: + dev_lamina.frame_read(0x0C, 0x20) 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}") + print(f"Success Frame: {dev_lamina.log['read']}") + print(f"Failed Frame: {dev_lamina.log['send'] - dev_lamina.log['read']}") + print(f"Max Series Failed Frame: {dev_lamina.log['keep-fail']}") 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 + print(f"Success Rate: {dev_lamina.log['read'] / dev_lamina.log['send'] * 100}%") + dev_lamina.flag_print, dev_lamina.retry, dev_lamina.time_out = param_saved + def make_Pakeage(fp: Path): """ 生成升级包 """ @@ -364,25 +335,8 @@ def make_Pakeage(fp: Path): return file_package -if __name__=='__main__': - mode_config = { - "Log": {'com_name': None, - # 'addr_645': [0x01, 0x00, 0x00, 0x00, 0x00, 0x40], - }, - "Debug": {'com_name': 'COM8', 'baudrate': 115200, 'parity': 'N', 'bytesize': 8, 'stopbits': 1, - # 'addr_645': [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], - 'frame_print': True, - 'time_out': 0.1, 'retry': 1, 'retry_sub': 10}, - "HPLC": {'com_name': 'COM10', 'baudrate': 9600, 'parity': 'E', 'bytesize': 8, 'stopbits': 1, - # 'addr_645': [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], - 'frame_print': True, - 'time_out': 0.5, 'retry': 3, 'retry_sub': 10}, - } - - dev_lamina = LaminaAdapter(**mode_config['Debug']) - - dev_lamina.frame_read(0x0100, 0x20) +def test(): if 0: dev_lamina.frame_read(0xA9, 1) # 读ADC参考电压 dev_lamina.frame_write_one(0xA9, 2000) # 写ADC参考电压:2.0V @@ -398,7 +352,6 @@ if __name__=='__main__': dev_lamina.frame_write_one(0x99, 1500) # 写校准参数Vout_a: 1.500 dev_lamina.frame_write_one(0x9A, 1000) # 写校准参数Vout_a: 1.00 dev_lamina.frame_write_one(0x9A, 1500) # 写校准参数Vout_a: 1.50 - 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]) @@ -438,7 +391,6 @@ if __name__=='__main__': 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) @@ -471,25 +423,41 @@ if __name__=='__main__': 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': 'COM8', 'baudrate': 115200, 'parity': 'N', 'bytesize': 8, 'stopbits': 1, + # 'addr_645': [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], + 'frame_print': True, + 'time_out': 0.1, 'retry': 1, 'retry_sub': 10}, + "HPLC": {'com_name': 'COM10', 'baudrate': 9600, 'parity': 'E', 'bytesize': 8, 'stopbits': 1, + # 'addr_645': [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], + 'frame_print': True, + 'time_out': 0.5, 'retry': 3, 'retry_sub': 10}, + } + + dev_lamina = LaminaAdapter(**mode_config['Debug']) + + dev_lamina.frame_read(0x0100, 0x20) + + if not hasattr(__builtins__,"__IPYTHON__"): - # file_package = Path(r"D:\WorkSpace\UserTool\SelfTool\FrameParser\test\p460_o1\result\lamina_optimizer_t1.dat") - # path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240520_0000_T1.11.bin") # 工程-即时转换 file_hex = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\lamina_adapter\Debug\lamina_adapter.hex") if not file_hex.exists(): raise Exception("工程编译目标文件不存在.") file_package = make_Pakeage(file_hex) - # 江苏发货产品灌装版本 - # path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240603_2100_V1.18.bin") - version = "DLSY001_240911_1600_V1.01" addr = [0x24, 0x09, 0x12, 0x00, 0x00, 0x00] while True: - """ 自动检测升级流程 """ + """ 自动检测升级流程(需要调试) """ ret = False - while not ret or ('version' not in dev_lamina.output.keys()) or (version == dev_lamina.output['version']): + while not ret or ('Reg' not in dev_lamina.output.keys()) or (version == dev_lamina.output['Reg'][0x0100][1]): dev_lamina.frame_read(0x82, 3) ret = dev_lamina.frame_read(0x0100, 0x20) time.sleep(1) @@ -500,7 +468,7 @@ if __name__=='__main__': ret = dev_lamina.frame_read(0x0100, 0x20) - if ret and (version == dev_lamina.output['version']): + if ret and (version == dev_lamina.output['Reg'][0x0100][1]): dev_lamina.frame_write_one(0x52, 0x01) print(f"address: {' '.join(map(lambda x: ('000' + hex(x)[2:])[-2:], addr))}") diff --git a/source/dev_LaminaController.py b/source/dev_LaminaController.py index 3cf128f..06a6125 100644 --- a/source/dev_LaminaController.py +++ b/source/dev_LaminaController.py @@ -7,7 +7,7 @@ from datetime import datetime from tenacity import retry, stop_after_attempt, wait_fixed from tools.ByteConv import trans_list_to_str from tools.IntelHex import file_Bin_to_IntelHex -from func_frame_re import make_frame_modbus, check_frame_modbus, print_display +from func_frame import make_frame_modbus, check_frame_modbus, print_display from func_upgrade import GenerateImage_DLSP001_p280039, GeneratePackage_DLSP001_p280039 modbus_map = { @@ -114,6 +114,8 @@ modbus_map = { 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: @@ -177,7 +179,7 @@ class LaminaController: self.output['result'] = False return self.output['result'] - def __transfer_data(self, frame: bytes) -> bool: + def __transfer_data(self, frame: bytearray) -> bool: """ 串口收发报文, 包含重试逻辑与数据打印 """ if self.__com is None: print(trans_list_to_str(frame)) diff --git a/source/device/EnergyRouter.py b/source/device/EnergyRouter.py index 8d94301..cbb0fa1 100644 --- a/source/device/EnergyRouter.py +++ b/source/device/EnergyRouter.py @@ -3,7 +3,7 @@ import socket import random from pathlib import Path from tools.ByteConv import trans_list_to_str -from func_frame import make_frame_modbus, check_frame_modbus +from source.func_frame import make_frame_modbus, check_frame_modbus, print_display modbus_map = { 0x00: ['编译日期', 4, 6], @@ -11,18 +11,59 @@ modbus_map = { } class EnergyRouter: - def __init__(self, ip="192.168.100.10", port=7, adddr_modbus=0x01): + """ 能量路由器远程升级测试(未完成) + """ + def __init__(self, ip="192.168.100.10", port=7, adddr_modbus=0x01, **kwargs): self._ip = ip self._port = port - self._addr_modbus =adddr_modbus + 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.tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.tcp_socket.connect((self._ip, self._port)) self.block = { - 'addr_dev' : self._addr_modbus, + 'addr_dev' : adddr_modbus, 'data_define': modbus_map, } + self.output = { + 'result': False, + 'code_func': 0x00, + } + self.log = { + 'send': 0, + 'read': 0, + 'keep-fail': 0, + 'record': { + 'config': None, + 'data': None, + }, + } + + def __transfer_data(self, frame: bytes) -> bool: + """ 数据传输处理函数 """ + if self.tcp_socket is None: + print(trans_list_to_str(frame)) + return False + + try: + self.tcp_socket.send(frame) + time.sleep(self.time_out) + frame_recv = self.tcp_socket.recv(128) + + self.output = check_frame_modbus(frame_recv, self.block) + if self.flag_print: + print("Read Frame: ", trans_list_to_str(frame_recv)) + if 'Regs' in self.output.keys(): + print_display(self.output['Regs']) + except Exception as ex: + print("Error Info: ", ex) + if self.flag_print and frame_recv: + print("Fail Data: " , trans_list_to_str(frame_recv)) + self.output['result'] = False + + return self.output['result'] def frame_read(self, daddr=0x00, dlen=0x10): self.block['type'] = 'read' @@ -30,40 +71,26 @@ class EnergyRouter: self.block['data_len'] = dlen frame = make_frame_modbus(self.block) - if self.tcp_socket is None: - print(trans_list_to_str(frame)) - return - - # self.tcp_socket.recv(8) - self.tcp_socket.send(bytearray(frame)) - time.sleep(0.5) - frame_recv = self.tcp_socket.recv(128) - output_text = check_frame_modbus(frame_recv, self.block) - print(output_text) - + return self.__transfer_data(frame) def frame_update(self, path_bin): """ 程序升级 """ + param_saved = self.flag_print, self.retry, self.time_out + self.block['type'] = 'update' self.block['step'] = 'start' self.block['file'] = Path(path_bin).read_bytes() self.block['header_offset'] = 128 # 启动帧 - frame_master = bytearray(make_frame_modbus(self.block)) + frame_master = make_frame_modbus(self.block) - # 等待擦除完成返回 - time.sleep(0.4) - self.tcp_socket.send(frame_master) - frame_slave = self.tcp_socket.recv(32) - if frame_slave == '': - raise Exception("TCP closed.") - - self.block['file_block_size'] = check_frame_modbus(frame_slave, self.block) - - if self.block['file_block_size'] == 0: - raise Exception("Error slave response.") + if not self.__transfer_data(frame_master): + self.flag_print, self.retry, self.time_out = param_saved + print('Upgrade Fail: start') + return False + self.block['file_block_size'] = self.output['upgrade']['length'] # 避免接收到延迟返回报文 time.sleep(0.4) @@ -87,7 +114,7 @@ class EnergyRouter: continue seq_window[i] = 1 self.block['index'] = seq_offset + i - seq_frame_master[i] = bytearray(make_frame_modbus(self.block)) + seq_frame_master[i] = make_frame_modbus(self.block) self.tcp_socket.send(seq_frame_master[i]) # 接收帧回复 tmp = list(zip(range(len(seq_window)), seq_window)) @@ -97,10 +124,11 @@ class EnergyRouter: # 接收到空数据, 对端已关闭连接 if seq_frame_slave[i] == '': raise Exception("TCP closed.") - result, seq_current, seq_hope = check_frame_modbus(seq_frame_slave[i], None) + self.output = check_frame_modbus(seq_frame_slave[i], None) + seq_current, seq_hope = self.output['upgrade']['index'], self.output['upgrade']['hope'] if seq_current < seq_offset: raise Exception("Error.") - elif result: + elif self.output['result']: seq_window[seq_current - seq_offset] = 2 data_remain -= self.block['file_block_size'] if seq_hope is not None and seq_hope < seq_offset: @@ -121,16 +149,15 @@ class EnergyRouter: seq_window[i] = 0 # 结束升级 - ret = 0 self.block['step'] = 'end' - frame_master = bytearray(make_frame_modbus(self.block)) + frame_master = make_frame_modbus(self.block) - while ret == 0: + while self.output['result'] is False: self.tcp_socket.send(frame_master) frame_slave = self.tcp_socket.recv(8) if frame_slave == '': raise Exception("TCP closed.") - ret = check_frame_modbus(frame_slave[:18], self.block) + self.output = check_frame_modbus(frame_slave[:18], self.block) if __name__ == "__main__": diff --git a/source/func_frame.py b/source/func_frame.py index 734a627..ac876b8 100644 --- a/source/func_frame.py +++ b/source/func_frame.py @@ -1,3 +1,4 @@ +import re import struct from crc import Calculator, Crc16 from tools.ByteConv import trans_list_to_str, trans_str_to_list, display_hex @@ -31,7 +32,8 @@ frame_modbus = { 0x11: [11, 2, 7] } -def make_frame_modbus(block:dict): + +def make_frame_modbus(block:dict) -> bytearray: """ modbus 生成函数""" frame = [] calculator = Calculator(Crc16.MODBUS) @@ -73,8 +75,6 @@ def make_frame_modbus(block:dict): 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': @@ -151,14 +151,14 @@ def make_frame_modbus(block:dict): else: raise Exception("Modbus Frame Type Error.") - crc = calculator.checksum(bytearray(frame)) frame.append(crc % 256) frame.append(crc // 256) - return frame + return bytearray(frame) -def make_frame_dlt645(block:dict): + +def make_frame_dlt645(block:dict) -> bytearray: """ dlt645 生成函数""" frame = [] @@ -179,79 +179,85 @@ def make_frame_dlt645(block:dict): frame.append(sum(frame) % 256) frame.append(0x16) - return frame + return bytearray(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) +def check_frame_modbus(frame:bytes, block:dict) -> dict: + """ 校验modbus帧回复 """ + if len(frame:=find_frame_modbus(frame, block['addr_dev'])) == 0: + raise Exception("No frame data") + + output = { + 'result': False, + 'code_func': frame[1], + 'code_sub': frame[2], + } + if output['code_func'] == 0x07: + """ 升级回复帧 """ + output['upgrade'] = {} + output['upgrade']['index'] = frame[4] * 256 + frame[5] + if output['code_sub'] == 0x01: + """ 升级开始, 处理文件头 """ + output['upgrade']['length'] = frame[6] * 256 + frame[7] + elif output['code_sub'] == 0x02: + """ 升级传输, 解析帧序号及返回值, 不做序列号校验 """ + pass + elif output['code_sub'] == 0x03: + """ 升级结束 """ + pass 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 - elif current_map[1] == 7: - """ 正序数值表示 """ - data_len = current_map[2] - item = " ".join(map(lambda x: str(x), data[:2 * data_len])) + raise Exception(" Upgrade SubFunc code error.") + output['result'] = True if frame[3] == 0x00 else False + output['code_error'] = frame[3] + elif output['code_func'] == 0x03 or output['code_func'] == 0x04: + """ 数据读取帧 """ + if frame[2] == len(frame[3:-2]): + output['Regs'] = display_data(block['data_addr'], frame[3:-2], block['data_define']) + else: + raise Exception("Frame read data length error.") + output['result'] = True + elif output['code_func'] == 0x06: + """ 单个数据写入帧 """ + output['result'] = True + elif output['code_func'] == 0x10: + """ 多个数据写入帧 """ + output['result'] = True + elif output['code_func'] == 0x11: + """ 录波功能帧 """ + data_record = {} + if frame[2] == 0x01: + data_record['type'] = 'config' + elif frame[2] == 0x02: + data_record["type"] = 'data' + else: + raise Exception("Unknow data type") + data_record['seq'] = frame[5] * 0x100 + frame[6] + data_record['total'] = frame[3] * 0x100 + frame[4] + data_record['data'] = frame[9:-2] + output['record'] = data_record + output['result'] = True + elif output['code_func'] & 0x80: + """ 错误返回帧 """ + output['code_error'] = frame[2] if output['code_func'] == 0x87 else frame[3] + else: + raise Exception(f"Frame Date error. func={output['code_func']}, func_sub={output['code_sub']}, len={len(frame)}") - 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): + +def check_frame_dlt645(frame:bytes, block:dict) -> dict: + """ 校验dlt645帧回复 """ + if len(frame:=find_frame_dlt645(frame, block['addr'])) == 0: + raise Exception("No frame data") + + code_func = frame[8] + if code_func == 0x9F: + return check_frame_modbus(frame[10:-2], block['data']) + else: + raise Exception("DLT645 Frame type error.") + + +def find_frame_modbus(buffer:bytes, address:int, frame_defines: dict=frame_modbus) -> bytes: """ 搜索合法modbus帧子串 """ len_buffer = len(buffer) @@ -286,7 +292,8 @@ def find_frame_modbus(buffer, address, frame_defines: dict=frame_modbus): return buffer[pos_frame: pos_frame+len_frame] -def find_frame_dlt645(buffer, address: list): + +def find_frame_dlt645(buffer:bytes, address: list) -> bytes: """ 搜索合法645帧子串 """ len_buffer = len(buffer) @@ -304,186 +311,81 @@ def find_frame_dlt645(buffer, address: list): 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 +def display_data(address: int, data: bytes, modbus_map: dict=modbus_map) -> dict: + """ 格式化表示数据, 得到显示数据字典 """ + def swapping_words(data:bytes, length:int) -> bytearray: + item = bytearray(data[:2 * length]) + for i in range(length): + item[2*i], item[2*i+1] = item[2*i+1], item[2*i] + return item + def convert_regs_to_int(data:bytes, length:int, signed: bool=True) -> int: + result = 0 + for idx_data, byte_data in enumerate(swapping_words(data, length)): + result += byte_data * 2 ** (8 * idx_data) + result -= 2 ** (16 * length) if signed and result >= (2 ** (16 * length - 1)) else 0 + return result + + output_data = {} + idx = address + while data: + data_label, data_len = "未知数据", 1 + if idx not in modbus_map.keys(): + item = convert_regs_to_int(data, 1, signed=False) + item = display_hex(item, 4) 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)}") + current_map = modbus_map[idx] + data_label = current_map[0] + if current_map[1] == 1: + """ Hex字符表示 """ + item = convert_regs_to_int(data, 1, signed=False) + item = display_hex(item, 4) + elif current_map[1] == 2: + """ 16位数值表示 """ + item = convert_regs_to_int(data, 1) + if len(current_map) > 2: + item /= current_map[2] + elif current_map[1] == 3: + """ 32位数值表示 """ + data_len = 2 + item = convert_regs_to_int(data, 2) + if len(current_map) > 2: + item /= current_map[2] + elif current_map[1] == 4: + """ 字符串表示 """ + data_len = current_map[2] + item = swapping_words(data, data_len) + item = item.replace(b'\xff', b' 0xFF').decode() + elif current_map[1] == 5: + """ 载波地址表示 """ + data_len = current_map[2] + item = swapping_words(data, data_len) + item = trans_list_to_str(item) + elif current_map[1] == 6: + """ 浮点数值表示 """ + data_len = 2 + item = struct.unpack('>f', bytes([data[2], data[3], data[0], data[1]]))[0] + elif current_map[1] == 7: + """ 正序数值表示 """ + data_len = current_map[2] + item = list(map(lambda x: x, data[:2 * data_len])) -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.") + output_data[idx] = data_label, item + idx += data_len + data = data[2*data_len:] + return output_data -if __name__ == "__main__": - frame_slave = '01 07 02 00 00 A7 00 F0 A1 A4 1F DC 0B 1D 27 08 0F CB A7 49 03 99 09 C6 87 10 FB 08 86 71 9F 2A 2F BB 8F 69 4D 5C 9F 90 51 8A 8B D3 E0 85 9F 2B C1 9A 7E A3 22 9B 29 EE 6B 70 28 D3 31 F6 EF 78 43 8A DF FC F3 8C 13 02 31 F4 65 B5 EE 46 80 F2 E9 D4 E9 C8 F2 84 13 3A DF 50 1D 45 53 52 1D 89 6F F8 CB 7F 56 28 DF A2 11 D4 47 93 04 04 FF AB 35 1F C3 BA D2 F0 65 F2 F6 A3 AC A5 A2 AF AF 3E 88 65 EC 7B 35 62 DA 4A CF A4 69 A5 9E C7 70 E6 DC DD BE 49 DD DA 09 CE 42 18 5C 57 86 12 E0 A0 74 5F 5C F7 35 B3 FE 7C 51 94 5C 57 28 1A 86 1E 9F DE F6 B2 4B A9 29 E6 30 EA F2 BB E6 72 81 E9 80 3A DE FC DC C2 F8 8E 30 F4 66 B3 25 12 30 FA 90 09 3C 8C 1D FD 49 8E 4A 58 1C 17 48 54 CF 6A DE B4 05 7F 3D DD 68 F2 7E C2 CE 01 11 D6 DE 5E 7C 27 10 FE 28 28 3E 06 4D C8 01 07 02 00 00 A7 F4 08' - 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") - buffer_test3 = trans_str_to_list(frame_slave) - 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], +def print_display(output_data: dict): + """ 格式化表示输出数据 """ + print("Parse Result:") + label_len_max = max(map(lambda x: len(x[0]), output_data.values())) + data_len_max = max(map(lambda x: len(str(x[1])), output_data.values())) + for key, value in output_data.items(): + label = value[0] + data = "-".join(map(str, value[1])) if type(value) == list else value[1] + print(f"{display_hex(key, 4)}: {data:<{data_len_max}} {label:<{label_len_max}}") + - 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_test3), block) diff --git a/source/func_frame_re.py b/source/func_frame_re.py deleted file mode 100644 index ac876b8..0000000 --- a/source/func_frame_re.py +++ /dev/null @@ -1,391 +0,0 @@ -import re -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) -> bytearray: - """ 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) - 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 bytearray(frame) - - -def make_frame_dlt645(block:dict) -> bytearray: - """ 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 bytearray(frame) - - -def check_frame_modbus(frame:bytes, block:dict) -> dict: - """ 校验modbus帧回复 """ - if len(frame:=find_frame_modbus(frame, block['addr_dev'])) == 0: - raise Exception("No frame data") - - output = { - 'result': False, - 'code_func': frame[1], - 'code_sub': frame[2], - } - if output['code_func'] == 0x07: - """ 升级回复帧 """ - output['upgrade'] = {} - output['upgrade']['index'] = frame[4] * 256 + frame[5] - if output['code_sub'] == 0x01: - """ 升级开始, 处理文件头 """ - output['upgrade']['length'] = frame[6] * 256 + frame[7] - elif output['code_sub'] == 0x02: - """ 升级传输, 解析帧序号及返回值, 不做序列号校验 """ - pass - elif output['code_sub'] == 0x03: - """ 升级结束 """ - pass - else: - raise Exception(" Upgrade SubFunc code error.") - output['result'] = True if frame[3] == 0x00 else False - output['code_error'] = frame[3] - elif output['code_func'] == 0x03 or output['code_func'] == 0x04: - """ 数据读取帧 """ - if frame[2] == len(frame[3:-2]): - output['Regs'] = display_data(block['data_addr'], frame[3:-2], block['data_define']) - else: - raise Exception("Frame read data length error.") - output['result'] = True - elif output['code_func'] == 0x06: - """ 单个数据写入帧 """ - output['result'] = True - elif output['code_func'] == 0x10: - """ 多个数据写入帧 """ - output['result'] = True - elif output['code_func'] == 0x11: - """ 录波功能帧 """ - data_record = {} - if frame[2] == 0x01: - data_record['type'] = 'config' - elif frame[2] == 0x02: - data_record["type"] = 'data' - else: - raise Exception("Unknow data type") - data_record['seq'] = frame[5] * 0x100 + frame[6] - data_record['total'] = frame[3] * 0x100 + frame[4] - data_record['data'] = frame[9:-2] - output['record'] = data_record - output['result'] = True - elif output['code_func'] & 0x80: - """ 错误返回帧 """ - output['code_error'] = frame[2] if output['code_func'] == 0x87 else frame[3] - else: - raise Exception(f"Frame Date error. func={output['code_func']}, func_sub={output['code_sub']}, len={len(frame)}") - - return output - - -def check_frame_dlt645(frame:bytes, block:dict) -> dict: - """ 校验dlt645帧回复 """ - if len(frame:=find_frame_dlt645(frame, block['addr'])) == 0: - raise Exception("No frame data") - - code_func = frame[8] - if code_func == 0x9F: - return check_frame_modbus(frame[10:-2], block['data']) - else: - raise Exception("DLT645 Frame type error.") - - -def find_frame_modbus(buffer:bytes, address:int, frame_defines: dict=frame_modbus) -> bytes: - """ 搜索合法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] & 0x7F) not in frame_defines.keys(): - continue - - frame_define = frame_defines[buffer[i+1] & 0x7F] - j = frame_define[0] - if buffer[i+1] & 0x80: - j = 5 - elif 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 ((i+j) <= len_buffer) and 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:bytes, address: list) -> bytes: - """ 搜索合法645帧子串 """ - len_buffer = len(buffer) - - pos_frame, len_frame = 0, 0 - for i in range(len_buffer): - if buffer[i] != 0x68 or buffer[i+7] != 0x68: - continue - if address[0] != 0xAA and buffer[i+1:i+7] == bytes(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 display_data(address: int, data: bytes, modbus_map: dict=modbus_map) -> dict: - """ 格式化表示数据, 得到显示数据字典 """ - def swapping_words(data:bytes, length:int) -> bytearray: - item = bytearray(data[:2 * length]) - for i in range(length): - item[2*i], item[2*i+1] = item[2*i+1], item[2*i] - return item - def convert_regs_to_int(data:bytes, length:int, signed: bool=True) -> int: - result = 0 - for idx_data, byte_data in enumerate(swapping_words(data, length)): - result += byte_data * 2 ** (8 * idx_data) - result -= 2 ** (16 * length) if signed and result >= (2 ** (16 * length - 1)) else 0 - return result - - output_data = {} - idx = address - while data: - data_label, data_len = "未知数据", 1 - if idx not in modbus_map.keys(): - item = convert_regs_to_int(data, 1, signed=False) - item = display_hex(item, 4) - else: - current_map = modbus_map[idx] - data_label = current_map[0] - if current_map[1] == 1: - """ Hex字符表示 """ - item = convert_regs_to_int(data, 1, signed=False) - item = display_hex(item, 4) - elif current_map[1] == 2: - """ 16位数值表示 """ - item = convert_regs_to_int(data, 1) - if len(current_map) > 2: - item /= current_map[2] - elif current_map[1] == 3: - """ 32位数值表示 """ - data_len = 2 - item = convert_regs_to_int(data, 2) - if len(current_map) > 2: - item /= current_map[2] - elif current_map[1] == 4: - """ 字符串表示 """ - data_len = current_map[2] - item = swapping_words(data, data_len) - item = item.replace(b'\xff', b' 0xFF').decode() - elif current_map[1] == 5: - """ 载波地址表示 """ - data_len = current_map[2] - item = swapping_words(data, data_len) - item = trans_list_to_str(item) - elif current_map[1] == 6: - """ 浮点数值表示 """ - data_len = 2 - item = struct.unpack('>f', bytes([data[2], data[3], data[0], data[1]]))[0] - elif current_map[1] == 7: - """ 正序数值表示 """ - data_len = current_map[2] - item = list(map(lambda x: x, data[:2 * data_len])) - - output_data[idx] = data_label, item - idx += data_len - data = data[2*data_len:] - return output_data - - -def print_display(output_data: dict): - """ 格式化表示输出数据 """ - print("Parse Result:") - label_len_max = max(map(lambda x: len(x[0]), output_data.values())) - data_len_max = max(map(lambda x: len(str(x[1])), output_data.values())) - for key, value in output_data.items(): - label = value[0] - data = "-".join(map(str, value[1])) if type(value) == list else value[1] - print(f"{display_hex(key, 4)}: {data:<{data_len_max}} {label:<{label_len_max}}") - - - - diff --git a/source/main.py b/source/main.py index e998d8e..8f4b094 100644 --- a/source/main.py +++ b/source/main.py @@ -1,7 +1,7 @@ import time from webui import webui from pathlib import Path -from func_frame import check_frame_dlt645 +from source.func_frame import check_frame_dlt645 from source.dev_LaminaAdapter import LaminaAdapter from source.device.EnergyRouter import EnergyRouter @@ -15,9 +15,9 @@ def my_function(e : webui.event): print("Data from JavaScript: " + e.window.get_str(e, 0)) # Message from JS frame = e.window.get_str(e, 0) block_dlt645 = e.window.get_str(e, 1) - output_text = check_frame_dlt645(frame, block=block_dlt645) + block_ouput = check_frame_dlt645(frame, block=block_dlt645) - return output_text + return block_ouput events = [] def main_webui():