初始化工程;
This commit is contained in:
345
source/main.py
Normal file
345
source/main.py
Normal file
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user