Frame Generator

commit c97ec0835ddc1bb84f7313bf2ecde9583a5cb101 Author: 何泽隆 Date: Sat Apr 13 00:44:44 2024 +0800 初始化工程; diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f3eb25a --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +# IDE配置 +.vscode + +# 缓存文件 +**/__pycache__/*.pyc diff --git a/resource/main.html b/resource/main.html new file mode 100644 index 0000000..adc1055 --- /dev/null +++ b/resource/main.html @@ -0,0 +1,64 @@ + + + + + + + + Frame Generator + + + + +

+ + + + + + +

+ + + +

+ +

+ + + \ No newline at end of file diff --git a/source/main.py b/source/main.py new file mode 100644 index 0000000..c2d1712 --- /dev/null +++ b/source/main.py @@ -0,0 +1,345 @@ +import time +from webui import webui +from pathlib import Path +from serial import Serial +from crc import Calculator, Crc16 +from utl import trans_list_to_str, trans_str_to_list, display_hex + +root = Path(".") +com_name = "COM16" + +modbus_map = { + # 1 - Hex + # 2 - Int16 + # 3 - lnt32 + # 4 - str + # 5 - addr + 0x0E: ["故障字1", 1], + 0x0F: ["故障字2", 1], + 0x10: ["MPPT工作状态", 1], + 0x11: ["系统工作状态", 1], + 0x12: ["系统工作模式", 1], + 0x13: ["输入电压", 2], + 0x14: ["电感电流", 2], + 0x15: ["12V电压", 2], + 0x16: ["输出电压", 2], + 0x17: ["输入电流", 2], + 0x18: ["温度1", 2], + 0x19: ["温度2", 2], + 0x1A: ["输入功率", 3], + 0x1C: ["设备温度", 2], + 0x1D: ["开关机状态", 1], + 0x1E: ["保留", 1], + 0x1F: ["保留", 1], + + 0x60: ["光伏通道使能", 1], + 0x61: ["最小启动输入电压", 2], + 0x62: ["最大启动输入电压", 2], + 0x63: ["最小停止输入电压", 2], + 0x64: ["最大停止输入电压", 2], + 0x65: ["最小MPPT电压", 2], + 0x66: ["最大MPPT电压", 2], + 0x67: ["最小启动输出电压", 2], + 0x68: ["最大启动输出电压", 2], + 0x69: ["最小停止输出电压", 2], + 0x6A: ["最大停止输出电压", 2], + 0x6B: ["输入过压保护值", 2], + 0x6C: ["输出过压保护值", 2], + 0x6D: ["输出欠压保护值", 2], + 0x6E: ["电感过流保护值", 2], + 0x6F: ["输入过流保护值", 2], + 0x70: ["最小电感电流限值", 2], + 0x71: ["最大电感电流限值", 2], + 0x72: ["浮充电压阈值", 2], + 0x73: ["三点法中间阈值", 2], + 0x74: ["恒压充电电压", 2], + 0x75: ["过温故障值", 2], + 0x76: ["过温告警值", 2], + 0x77: ["温度恢复值", 2], + 0x78: ["最低满载电压", 2], + 0x79: ["最高满载电压", 2], + 0x7A: ["输入过载保护值", 3], + 0x7C: ["最小功率限值", 3], + 0x7E: ["最大功率限值", 3], + 0x80: ["最大功率限值存储值", 3], + 0x82: ["载波通信地址", 5], + 0x85: ["电压环out_max", 2], + 0x86: ["电压环out_min", 2], + 0x87: ["电流环out_max", 2], + 0x88: ["电流环out_min", 2], + 0x89: ["MPPT扰动系数k_d_vin", 2], + 0x8A: ["dmin", 2], + 0x8B: ["dmax", 2], + 0x8C: ["扫描电压偏移scanvolt_offset", 2], + 0x8D: ["电压环Kp", 3], + 0x8F: ["电压环Ki", 3], + 0x91: ["电流环Kp", 3], + 0x93: ["电流环Ki", 3], + 0x95: ["日志级别", 1], + 0x96: ["日志输出方式", 1], + 0x97: ["采样校准volt_in_a", 2], + 0x98: ["采样校准volt_in_b", 2], + 0x99: ["采样校准volt_out_a", 2], + 0x9A: ["采样校准volt_out_b", 2], + 0x9B: ["采样校准curr_in_a", 2], + 0x9C: ["采样校准curr_in_b", 2], + 0x9D: ["采样校准curr_induc_a", 2], + 0x9E: ["采样校准curr_induc_b", 2], + 0x9F: ["采样校准volt_12V_a", 2], + 0xA0: ["采样校准volt_12V_b", 2], + 0xA1: ["温度补偿temp1_b", 2], + 0xA2: ["温度补偿temp2_b", 2], + 0xA3: ["系统工作模式", 1], + 0xA4: ["电感电流给定值curr_set", 2], + 0xA5: ["保留", 1], + 0xA6: ["保留", 1], + 0xA7: ["保留", 1], + 0xA8: ["保留", 1], + 0xA9: ["保留", 1], + 0xAA: ["保留", 1], + 0xAB: ["保留", 1], + 0xAC: ["保留", 1], + 0xAD: ["保留", 1], + 0xAE: ["保留", 1], + 0xAF: ["保留", 1], +} + +def make_frame_modbus(block:dict): + """ modbus 生成函数""" + frame = [] + calculator = Calculator(Crc16.MODBUS) + + func_code = 0x1F # Modbus 透传帧 + + frame.append(block['addr_dev']) + 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) + + crc = calculator.checksum(bytearray(frame)) + frame.append(crc % 256) + frame.append(crc // 256) + + return frame + +def make_frame_dlt645(block:dict): + """ dlt645 生成函数""" + frame = [] + len_data = len(block['data']) + + ctrl_code = 0x1F # Modbus 透传帧 + + frame.append(0x68) + frame += block['addr'][:6] + frame.append(0x68) + frame.append(ctrl_code) + frame.append(len_data) + frame += block['data'] + frame.append(sum(frame) % 256) + frame.append(0x16) + + return frame + +def display_data(address: int, data: list): + """ 格式化表示数据 """ + output = "Phase Result: \n" + idx = address + len_max = 0 + while data: + if idx not in modbus_map.keys(): + output_line = "\t Error, doesn`t match item.\n" + output += output_line + break + + len_data = 1 + current_map = modbus_map[idx] + 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 + len_data = 2 + elif current_map[1] == 4: + """ 字符串表示 """ + item = bytearray(data[:16]) + item = item[4:6] + item[2:4] + item[0:2] + len_data = 16 + elif current_map[1] == 5: + """ 载波地址表示 """ + item = trans_list_to_str(data[:6]) + len_data = 3 + + + if len_max < len(current_map[0]): + len_max = len(current_map[0]) + len_remain = len_max - len(current_map[0]) + output_line = "\t".join([display_hex(idx, 4), current_map[0] + " " * len_remain, str(item)]) + + idx += len_data + del data[0:len_data * 2] + output += output_line + "\n" + return output + +def test1(): + """ 测试1 """ + frame = "68 01 02 03 04 05 06 68 9F 29 01 03 24 20 30 00 00 00 00 00 04 00 00 00 9A 00 00 00 00 00 00 00 00 01 02 01 10 00 01 00 00 01 10 00 00 FF FF FF FF 5D 3B 7D 16 " + frame1 = trans_str_to_list(frame) + print(frame1) + frame2 = frame1[13:-4] + print(frame2) + output_text = display_data(0x0e, frame2) + print(output_text) + +def test2(): + """ 测试2 """ + block_modbus = { + 'addr_dev' : 0x01, + 'data_addr' : 0x0E, + 'data_len' : 0x12, + 'type' : 'read_one', + } + + frame1 = make_frame_modbus(block_modbus) + + block_dlt645 = { + 'addr' : [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], + 'data' : frame1, + } + + frame2 = make_frame_dlt645(block_dlt645) + + frame_string = trans_list_to_str(frame2) + + print(frame_string) + +def frame_read(daddr=0x60, dlen=0x50): + block_modbus = { + 'addr_dev' : 0x01, + 'data_addr' : daddr, + 'data_len' : dlen, + 'type' : 'read', + } + frame1 = make_frame_modbus(block_modbus) + block_dlt645 = { + 'addr' : [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], + 'data' : frame1, + } + frame2 = make_frame_dlt645(block_dlt645) + if com is None: + print(trans_list_to_str(frame2)) + return + com.read_all() + com.write(bytearray(frame2)) + time.sleep(0.5) + frame3 = com.read_all() + frame4 = list(frame3[13:-4]) + output_text = display_data(block_modbus['data_addr'], frame4) + print(output_text) + +def frame_write_one(daddr=0x85, dval=-900): + block_modbus = { + 'addr_dev' : 0x01, + 'data_addr' : daddr, + 'data_val' : dval, + 'type' : 'write_one', + } + frame1 = make_frame_modbus(block_modbus) + block_dlt645 = { + 'addr' : [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], + 'data' : frame1, + } + frame2 = make_frame_dlt645(block_dlt645) + com.write(bytearray(frame2)) + +def frame_write_dual(daddr=0x91, dval=600): + block_modbus = { + 'addr_dev' : 0x01, + 'data_addr' : daddr, + 'data_val' : dval, + 'type' : 'write_dual', + } + frame1 = make_frame_modbus(block_modbus) + block_dlt645 = { + 'addr' : [0x01, 0x02, 0x03, 0x04, 0x05, 0x06], + 'data' : frame1, + } + frame2 = make_frame_dlt645(block_dlt645) + com.write(bytearray(frame2)) + +def my_function(e : webui.event): + """ WebUI回调函数 """ + global events + events.append(e) + print(len(events)) + + print("Data from JavaScript: " + e.window.get_str(e, 0)) # Message from JS + frame = e.window.get_str(e, 0) + frame1 = trans_str_to_list(frame) + frame2 = frame1[13:-4] + output_text = display_data(0x0e, frame2) + + return output_text + +events = [] +def main_webui(): + myWindow = webui.window() + + file_main = root / 'resource\main.html' + + myWindow.bind("myID1", my_function) + myWindow.bind("my_function", my_function) + myWindow.show(str(file_main)) + + while 1: + """ 调试断点 """ + webui.wait() + print(events) + myWindow.show(str(file_main)) + +com = None +if __name__ == "__main__": + com = Serial("Com16", baudrate=115200, parity='N', timeout=2) + + + pass + diff --git a/source/utl.py b/source/utl.py new file mode 100644 index 0000000..a3c36e7 --- /dev/null +++ b/source/utl.py @@ -0,0 +1,17 @@ + +def trans_list_to_str(data: list) -> str: + """ 标准串口字符串表示 """ + func_trans = lambda x: ('00' + hex(x % 256)[2:])[-2:].upper() + return " ".join(map(func_trans, data)) + +def trans_str_to_list(data: str) -> list: + """ 标准串口字符串转换列表 """ + func_trans = lambda x: int(x, 16) + return list(map(func_trans, data.strip().split(" "))) + + +def display_hex(data, len) -> str: + """ Hex字符表示 """ + data %= 2 ** (4 * len) + result = "0" * len + hex(data)[2:] + return "0x" + result[-len:].upper() \ No newline at end of file