375 lines
12 KiB
Python
375 lines
12 KiB
Python
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: ["抖动频率上限", 2],
|
|
0xA6: ["抖动频率下限", 2],
|
|
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)
|
|
elif block['type'] == "write_str":
|
|
frame.append(0x10)
|
|
data_len = len(block['data_val'])
|
|
data_val = block['data_val']
|
|
if data_len > 0x0100:
|
|
raise("modbus data len oversize.")
|
|
frame.append(data_addr // 256 % 256)
|
|
frame.append(data_addr % 256)
|
|
frame.append(0x00)
|
|
frame.append(data_len // 2)
|
|
frame.append(data_len)
|
|
for i in range(data_len//2):
|
|
frame.append(data_val[2*i + 1])
|
|
frame.append(data_val[2*i])
|
|
|
|
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, dev_addr=[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA]):
|
|
block_modbus = {
|
|
'addr_dev' : 0x01,
|
|
'data_addr' : daddr,
|
|
'data_len' : dlen,
|
|
'type' : 'read',
|
|
}
|
|
frame1 = make_frame_modbus(block_modbus)
|
|
block_dlt645 = {
|
|
'addr' : dev_addr,
|
|
'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, dev_addr=[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA]):
|
|
block_modbus = {
|
|
'addr_dev' : 0x01,
|
|
'data_addr' : daddr,
|
|
'data_val' : dval,
|
|
'type' : 'write_one',
|
|
}
|
|
frame1 = make_frame_modbus(block_modbus)
|
|
block_dlt645 = {
|
|
'addr' : dev_addr,
|
|
'data' : frame1,
|
|
}
|
|
frame2 = make_frame_dlt645(block_dlt645)
|
|
com.write(bytearray(frame2))
|
|
|
|
def frame_write_dual(daddr=0x91, dval=600, dev_addr=[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA]):
|
|
block_modbus = {
|
|
'addr_dev' : 0x01,
|
|
'data_addr' : daddr,
|
|
'data_val' : dval,
|
|
'type' : 'write_dual',
|
|
}
|
|
frame1 = make_frame_modbus(block_modbus)
|
|
block_dlt645 = {
|
|
'addr' : dev_addr,
|
|
'data' : frame1,
|
|
}
|
|
frame2 = make_frame_dlt645(block_dlt645)
|
|
com.write(bytearray(frame2))
|
|
|
|
def frame_write_str(daddr=0x82, dval=[0x06, 0x05, 0x04, 0x03, 0x02, 0x01], dev_addr=[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA]):
|
|
block_modbus = {
|
|
'addr_dev' : 0x01,
|
|
'data_addr' : daddr,
|
|
'data_val' : dval,
|
|
'type' : 'write_str',
|
|
}
|
|
frame1 = make_frame_modbus(block_modbus)
|
|
block_dlt645 = {
|
|
'addr' : dev_addr,
|
|
'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)
|
|
address = [0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA]
|
|
|
|
pass
|
|
|