使用状态机增加了初始状态和故障状态,编写了gui调试程序
This commit is contained in:
@@ -0,0 +1,42 @@
|
||||
import socket
|
||||
import struct
|
||||
import tkinter as tk
|
||||
import threading
|
||||
from udpmsg import *
|
||||
from datetime import datetime
|
||||
from tkinter import font
|
||||
from guiEle import ElementDisplayApp
|
||||
from until import *
|
||||
from guiInput import InputGUI
|
||||
|
||||
# 创建UDP套接字
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
server_address = ('localhost', 5001)
|
||||
sock.bind(server_address)
|
||||
|
||||
# 创建主窗口
|
||||
root = tk.Tk()
|
||||
root.title("UDP消息显示")
|
||||
eleplay = ElementDisplayApp(root)
|
||||
inp = InputGUI(root)
|
||||
|
||||
# 创建一个文本框用于显示消息
|
||||
custom_font = font.Font(family="Arial", size=12, weight="normal")
|
||||
text_box = tk.Text(root, height=10, width=50, font=custom_font)
|
||||
# .pack(side='left', fill='y', expand=True)
|
||||
text_box.grid(row=0, column=4, sticky='ne')
|
||||
|
||||
# 创建并启动接收和发送消息的线程
|
||||
thread = threading.Thread(target=receive_messages, args=(sock, text_box,eleplay))
|
||||
thread.daemon = True
|
||||
thread.start()
|
||||
|
||||
thread2 = threading.Thread(target=send_messages, args=(sock, eleplay))
|
||||
thread2.daemon = True
|
||||
thread2.start()
|
||||
|
||||
# 运行主循环
|
||||
root.mainloop()
|
||||
|
||||
# 关闭套接字
|
||||
sock.close()
|
||||
@@ -0,0 +1,255 @@
|
||||
import tkinter as tk
|
||||
from tkinter import font
|
||||
from udpmsg import *
|
||||
|
||||
class ElementDisplayApp:
|
||||
def __init__(self, root):
|
||||
self.root = root
|
||||
|
||||
self.disHVB = msg_disHighVolBusFbMsg()
|
||||
self.disHVBA = msg_disHighAVolBusFbMsg()
|
||||
self.disHVBB = msg_disHighBVolBusFbMsg()
|
||||
self.disLVB = msg_disLowBusFbMsg()
|
||||
|
||||
# 定义每组的颜色
|
||||
self.colors = ["#FFC0CB", "#ADD8E6", "#90EE90", "#FFD700"]
|
||||
|
||||
# 创建自定义字体
|
||||
self.custom_font = font.Font(family="Arial", size=12, weight='bold')
|
||||
|
||||
# 创建4个Frame来分组显示元素的值
|
||||
self.frames = []
|
||||
for i in range(4):
|
||||
frame = tk.Frame(self.root, bg=self.colors[i], padx=10, pady=10)
|
||||
# frame.pack(padx=10, pady=10, expand=True, fill='both')
|
||||
frame.grid(row=0, column=i, sticky='nsew') # 使用 grid 布局管理器
|
||||
self.frames.append(frame)
|
||||
|
||||
# 定义四个字典来存储元素的值
|
||||
# self.dict1 = {f"元素{i+1}": f"Value {i+1}" for i in range(16)}
|
||||
self.dict1={'燃料电池断路器':'闭合','动力锂电池断路器':'闭合','推进电机断路器':'闭合','仪表锂电池断路器':'闭合','DC/DC5模块断路器':'闭合',
|
||||
'艏部高压配电断路器':'闭合','艏部FT4/5断路器':'闭合','艉舵1/2断路器':'闭合','艉舵3/4断路器':'闭合',
|
||||
'FB预留断路器':'闭合','TYZ预留断路器':'闭合','预留断路器':'闭合'}
|
||||
# self.dict2 = {f"元素{i+17}": f"Value {i+17}" for i in range(16)}
|
||||
self.dict2 = {'艏部FT1-3断路器':'闭合','启闭机构断路器':'闭合','桅杆舵机断路器':'闭合','XCZ断路器':'闭合',
|
||||
'艏舵断路器':'闭合','敞开电缸断路器':'闭合'}
|
||||
# self.dict3 = {f"元素{i+33}": f"Value {i+33}" for i in range(16)}
|
||||
self.dict3 = {'仪表锂电池断路器':'闭合','艏部低压断路器':'闭合','UNIT4断路器':'闭合','DC/DC1断路器':'闭合',
|
||||
'预留断路器1':'闭合', '预留断路器2':'闭合','艉部应急电池开关':'闭合'}
|
||||
# self.dict4 = {f"元素{i+49}": f"Value {i+49}" for i in range(16)}
|
||||
self.dict4 = {'UNIT1断路器':'闭合','UNIT2断路器':'闭合','UNIT3断路器':'闭合','UNIT5断路器':'闭合','艏部抛载断路器':'闭合',
|
||||
'预留断路器1':'闭合','预留断路器2':'断开','艏部应急电池开关':'闭合'}
|
||||
|
||||
# 创建Label来显示每个字典中的元素值，并应用自定义字体
|
||||
self.labels = []
|
||||
for i, dictionary in enumerate([self.dict1, self.dict2, self.dict3, self.dict4]):
|
||||
for j, (key, value) in enumerate(dictionary.items()):
|
||||
# 根据元素的值选择颜色
|
||||
if value == "闭合":
|
||||
color = "green"
|
||||
else:
|
||||
color = "red"
|
||||
label = tk.Label(self.frames[i], text=f"{key}: {value}", fg=color, font=self.custom_font, anchor="w")
|
||||
label.grid(row=j//2, column=j%2, padx=5, pady=5)
|
||||
# self.labels.append(label)
|
||||
|
||||
def update_disp(self):
|
||||
# 更新Label的文本
|
||||
for i, dictionary in enumerate([self.dict1, self.dict2, self.dict3, self.dict4]):
|
||||
for j, (key, value) in enumerate(dictionary.items()):
|
||||
# 根据元素的值选择颜色
|
||||
if value == "闭合":
|
||||
color = "green"
|
||||
else:
|
||||
color = "red"
|
||||
# self.labels[i + j].config(text=f"{key}: {value}", fg=color, font=self.custom_font, anchor="w")
|
||||
label = tk.Label(self.frames[i], text=f"{key}: {value}", fg=color, font=self.custom_font, anchor="w")
|
||||
label.grid(row=j//2, column=j%2, padx=5, pady=5)
|
||||
|
||||
def updata_data(self):
|
||||
if (self.dict1['燃料电池断路器']=='闭合'):
|
||||
self.disHVB.DCB1 = 0x55
|
||||
else:
|
||||
self.disHVB.DCB1 = 0xAA
|
||||
|
||||
if (self.dict1['动力锂电池断路器']=='闭合'):
|
||||
self.disHVB.DCB2 = 0x55
|
||||
else:
|
||||
self.disHVB.DCB2 = 0xAA
|
||||
|
||||
if (self.dict1['推进电机断路器']=='闭合'):
|
||||
self.disHVB.ZK1 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK1 = 0xAA
|
||||
|
||||
if (self.dict1['仪表锂电池断路器']=='闭合'):
|
||||
self.disHVB.ZK2 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK2 = 0xAA
|
||||
|
||||
if (self.dict1['DC/DC5模块断路器']=='闭合'):
|
||||
self.disHVB.ZK3 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK3 = 0xAA
|
||||
|
||||
if (self.dict1['艏部高压配电断路器']=='闭合'):
|
||||
self.disHVB.ZK4 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK4 = 0xAA
|
||||
|
||||
if (self.dict1['艏部FT4/5断路器']=='闭合'):
|
||||
self.disHVB.ZK5 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK5 = 0xAA
|
||||
|
||||
if (self.dict1['艉舵1/2断路器']=='闭合'):
|
||||
self.disHVB.ZK6 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK6 = 0xAA
|
||||
|
||||
if (self.dict1['艉舵3/4断路器']=='闭合'):
|
||||
self.disHVB.ZK7 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK7 = 0xAA
|
||||
|
||||
if (self.dict1['FB预留断路器']=='闭合'):
|
||||
self.disHVB.ZK8 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK8 = 0xAA
|
||||
|
||||
if (self.dict1['TYZ预留断路器']=='闭合'):
|
||||
self.disHVB.ZK9 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK9 = 0xAA
|
||||
|
||||
if (self.dict1['预留断路器']=='闭合'):
|
||||
self.disHVB.ZK10 = 0x55
|
||||
else:
|
||||
self.disHVB.ZK10 = 0xAA
|
||||
######################################################
|
||||
if (self.dict2['艏部FT1-3断路器']=='闭合'):
|
||||
self.disHVBA.bowFTDevice123CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBA.bowFTDevice123CircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict2['启闭机构断路器']=='闭合'):
|
||||
self.disHVBA.actuatorCircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBA.actuatorCircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict2['桅杆舵机断路器']=='闭合'):
|
||||
self.disHVBA.mastSteeringGearControlBoxCircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBA.mastSteeringGearControlBoxCircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict2['XCZ断路器']=='闭合'):
|
||||
self.disHVBA.xczCircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBA.xczCircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict2['艏舵断路器']=='闭合'):
|
||||
self.disHVBA.bowRudderControlBoxCircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBA.bowRudderControlBoxCircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict2['敞开电缸断路器']=='闭合'):
|
||||
self.disHVBA.openWaterCoverStartCylinderCircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBA.openWaterCoverStartCylinderCircuitBreaker = 0xAA
|
||||
|
||||
####################################################################
|
||||
if (self.dict3['仪表锂电池断路器']=='闭合'):
|
||||
self.disHVBB.lithiumBatteryGroupInstrumentCircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBB.lithiumBatteryGroupInstrumentCircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict3['艏部低压断路器']=='闭合'):
|
||||
self.disHVBB.bowLowVoltageDistributionCircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBB.bowLowVoltageDistributionCircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict3['UNIT4断路器']=='闭合'):
|
||||
self.disHVBB.unit4CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBB.unit4CircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict3['DC/DC1断路器']=='闭合'):
|
||||
self.disHVBB.dcdc1CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBB.dcdc1CircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict3['预留断路器1']=='闭合'):
|
||||
self.disHVBB.reservedCircuitBreaker1 = 0x55
|
||||
else:
|
||||
self.disHVBB.reservedCircuitBreaker1 = 0xAA
|
||||
|
||||
if (self.dict3['预留断路器2']=='闭合'):
|
||||
self.disHVBB.reservedCircuitBreaker2 = 0x55
|
||||
else:
|
||||
self.disHVBB.reservedCircuitBreaker2 = 0xAA
|
||||
|
||||
if (self.dict3['艉部应急电池开关']=='闭合'):
|
||||
self.disHVBB.emergencyLithiumBatteryGroup2CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disHVBB.emergencyLithiumBatteryGroup2CircuitBreaker = 0xAA
|
||||
|
||||
######################################################################
|
||||
if (self.dict4['UNIT1断路器']=='闭合'):
|
||||
self.disLVB.unit1CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disLVB.unit1CircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict4['UNIT2断路器']=='闭合'):
|
||||
self.disLVB.unit2CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disLVB.unit2CircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict4['UNIT3断路器']=='闭合'):
|
||||
self.disLVB.unit3CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disLVB.unit3CircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict4['UNIT5断路器']=='闭合'):
|
||||
self.disLVB.unit5CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disLVB.unit5CircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict4['预留断路器1']=='闭合'):
|
||||
self.disLVB.reservedCircuitBreaker1 = 0x55
|
||||
else:
|
||||
self.disLVB.reservedCircuitBreaker1 = 0xAA
|
||||
|
||||
if (self.dict4['预留断路器2']=='闭合'):
|
||||
self.disLVB.reservedCircuitBreaker2 = 0x55
|
||||
else:
|
||||
self.disLVB.reservedCircuitBreaker2 = 0xAA
|
||||
|
||||
if (self.dict4['艏部抛载断路器']=='闭合'):
|
||||
self.disLVB.bowPZDeviceCircuitBreaker = 0x55
|
||||
else:
|
||||
self.disLVB.bowPZDeviceCircuitBreaker = 0xAA
|
||||
|
||||
if (self.dict4['艏部应急电池开关']=='闭合'):
|
||||
self.disLVB.emergencyLithiumBatteryGroup1CircuitBreaker = 0x55
|
||||
else:
|
||||
self.disLVB.emergencyLithiumBatteryGroup1CircuitBreaker = 0xAA
|
||||
|
||||
a = self.disHVB.pack()
|
||||
b = self.disHVBA.pack()
|
||||
c = self.disHVBB.pack()
|
||||
d = self.disLVB.pack()
|
||||
return a,b,c,d
|
||||
|
||||
if __name__ == "__main__":
|
||||
root = tk.Tk()
|
||||
app = ElementDisplayApp(root)
|
||||
root.mainloop()
|
||||
|
||||
# # 创建主窗口
|
||||
# root = tk.Tk()
|
||||
|
||||
# # 创建应用实例并在root中显示
|
||||
# app = ElementDisplayApp(root)
|
||||
|
||||
# # 运行主循环
|
||||
# root.mainloop()
|
||||
@@ -0,0 +1,121 @@
|
||||
import tkinter as tk
|
||||
from tkinter import messagebox
|
||||
from udpmsg import msg_CcuStateFbMsg
|
||||
|
||||
class msg_CcuStateFbMsg:
|
||||
def __init__(self):
|
||||
self.start1 = 0x40
|
||||
self.start2 = 0x40
|
||||
self.id = 0x0001
|
||||
self.length = 0x0000
|
||||
self.flag = 0x00
|
||||
self.failvreCode = 0x00
|
||||
|
||||
def pack(self):
|
||||
# 初始化数据
|
||||
pass # function body is omitted
|
||||
|
||||
class InputGUI:
|
||||
def __init__(self, master):
|
||||
self.master = master
|
||||
self.printing = False
|
||||
self.msg_instance = msg_CcuStateFbMsg()
|
||||
|
||||
# 创建一个Frame来包含所有的输入框和标签
|
||||
row = 0
|
||||
column = 0
|
||||
|
||||
self.input_frame = tk.Frame(master, bg='lightgray')
|
||||
self.input_frame.grid(row=1, column=0, sticky='nsew')
|
||||
|
||||
self.title_label = tk.Label(self.input_frame, text="复合管控消息交互", font=("Arial", 16, "bold"))
|
||||
self.title_label.grid(row=0, column=0, columnspan=2, pady=(10, 20), sticky="nsew") # 调整标题的上下边距并居中显示
|
||||
|
||||
# 创建输入框和标签
|
||||
self.start1_label = tk.Label(self.input_frame, text="Start1:")
|
||||
self.start1_label.grid(row=1, column=0)
|
||||
self.start1_entry = tk.Entry(self.input_frame)
|
||||
self.start1_entry.grid(row=1, column=1)
|
||||
|
||||
self.start2_label = tk.Label(self.input_frame, text="Start2:")
|
||||
self.start2_label.grid(row=2, column=0)
|
||||
self.start2_entry = tk.Entry(self.input_frame)
|
||||
self.start2_entry.grid(row=2, column=1)
|
||||
|
||||
self.id_label = tk.Label(self.input_frame, text="ID:")
|
||||
self.id_label.grid(row=3, column=0)
|
||||
self.id_entry = tk.Entry(self.input_frame)
|
||||
self.id_entry.grid(row=3, column=1)
|
||||
|
||||
self.length_label = tk.Label(self.input_frame, text="Length:")
|
||||
self.length_label.grid(row=4, column=0)
|
||||
self.length_entry = tk.Entry(self.input_frame)
|
||||
self.length_entry.grid(row=4, column=1)
|
||||
|
||||
self.flag_label = tk.Label(self.input_frame, text="Flag:")
|
||||
self.flag_label.grid(row=5, column=0)
|
||||
self.flag_entry = tk.Entry(self.input_frame)
|
||||
self.flag_entry.grid(row=5, column=1)
|
||||
|
||||
self.failvreCode_label = tk.Label(self.input_frame, text="Failvre Code:")
|
||||
self.failvreCode_label.grid(row=6, column=0)
|
||||
self.failvreCode_entry = tk.Entry(self.input_frame)
|
||||
self.failvreCode_entry.grid(row=6, column=1)
|
||||
|
||||
# 创建更新按钮
|
||||
self.update_button = tk.Button(self.input_frame, text="Update", command=self.update_msg)
|
||||
self.update_button.grid(row=7, column=0, columnspan=2)
|
||||
|
||||
# 创建打印按钮
|
||||
self.print_button = tk.Button(self.input_frame, text="Print", command=self.start_printing)
|
||||
self.print_button.grid(row=8, column=0)
|
||||
|
||||
# 创建停止打印按钮
|
||||
self.stop_button = tk.Button(self.input_frame, text="Stop", command=self.stop_printing, state=tk.DISABLED)
|
||||
self.stop_button.grid(row=8, column=1)
|
||||
|
||||
# 创建显示打印消息数量的标签
|
||||
self.print_count_label = tk.Label(self.input_frame, text="Print Count: 0", font=("Arial", 12))
|
||||
self.print_count_label.grid(row=9, column=0, columnspan=2)
|
||||
# 创建更新按钮
|
||||
self.update_button = tk.Button(self.input_frame, text="Update", command=self.update_msg)
|
||||
self.update_button.grid(row=7, column=0, columnspan=2)
|
||||
|
||||
def update_msg(self):
|
||||
try:
|
||||
self.msg_instance.start1 = int(self.start1_entry.get(), 16)
|
||||
self.msg_instance.start2 = int(self.start2_entry.get(), 16)
|
||||
self.msg_instance.id = int(self.id_entry.get(), 16)
|
||||
self.msg_instance.length = int(self.length_entry.get(), 16)
|
||||
self.msg_instance.flag = int(self.flag_entry.get(), 16)
|
||||
self.msg_instance.failvreCode = int(self.failvreCode_entry.get(), 16)
|
||||
messagebox.showinfo("Success", "msg_CcuStateFbMsg updated successfully!")
|
||||
except ValueError:
|
||||
messagebox.showerror("Error", "Invalid input. Please enter valid hexadecimal values.")
|
||||
|
||||
def start_printing(self):
|
||||
self.printing = True
|
||||
self.print_count = 0
|
||||
self.print_count_label.config(text="Print Count: 0")
|
||||
self.print_button.config(state=tk.DISABLED)
|
||||
self.stop_button.config(state=tk.NORMAL)
|
||||
self.print_message()
|
||||
|
||||
def stop_printing(self):
|
||||
self.printing = False
|
||||
self.print_button.config(state=tk.NORMAL)
|
||||
self.stop_button.config(state=tk.DISABLED)
|
||||
|
||||
def print_message(self, callback=None):
|
||||
if self.printing:
|
||||
print("Printing message...")
|
||||
self.print_count += 1
|
||||
self.print_count_label.config(text=f"Print Count: {self.print_count}")
|
||||
if callback:
|
||||
callback()
|
||||
self.master.after(1000, self.print_message, callback) # 每秒打印一次
|
||||
|
||||
if __name__ == "__main__":
|
||||
root = tk.Tk()
|
||||
gui = InputGUI(root)
|
||||
root.mainloop()
|
||||
@@ -20,6 +20,8 @@ print(valid_msg)
|
||||
|
||||
# 创建UDP套接字
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
server_address = ('localhost', 5001)
|
||||
sock.bind(server_address)
|
||||
|
||||
# 要发送的信息
|
||||
# message = "Hello, this is a test message."
|
||||
|
||||
+43
-1
@@ -81,7 +81,7 @@ class msg_CcuStateFbMsg:
|
||||
self.placeholder3 = 0x00
|
||||
self.placeholder4 = 0x00
|
||||
self.placeholder5 = 0x00
|
||||
self.checkCode = 0x00
|
||||
self.checkCode = 0x00
|
||||
|
||||
def pack(self):
|
||||
# 初始化数据
|
||||
@@ -481,7 +481,49 @@ class msg_disLowBusFbMsg:
|
||||
checkCode_L = self.checkCode & 0xFF
|
||||
data += struct.pack('<B', checkCode_L)
|
||||
return data
|
||||
|
||||
class msg_disHVBCmd:
|
||||
def __init__(self):
|
||||
self.start1 = None
|
||||
self.start2 = None
|
||||
self.id = None
|
||||
self.length = None
|
||||
self.fuelCellCircuitBreaker = None
|
||||
self.powerLithiumBatteryCircuitBreaker = None
|
||||
self.propulsionMotorCircuitBreaker = None
|
||||
self.lithiumBatteryGroupInstrumentCircuitBreaker = None
|
||||
self.dcDc5ModuleCircuitBreaker = None
|
||||
self.bowHighVoltageDistributionBoxCircuitBreaker = None
|
||||
self.sternFTDevice45CircuitBreaker = None
|
||||
self.sternRudderSwitch1CircuitBreaker = None
|
||||
self.sternRudderSwitch2CircuitBreaker = None
|
||||
self.fbReservedCircuitBreaker = None
|
||||
self.tyzReservedCircuitBreaker = None
|
||||
self.reservedCircuitBreaker = None
|
||||
self.checkCode = None
|
||||
|
||||
def unpack(self,data):
|
||||
try:
|
||||
unpacked_data = struct.unpack('<BBHHBBBBBBBBBBBBB',data)
|
||||
self.start1 = unpacked_data[0]
|
||||
self.start2 = unpacked_data[1]
|
||||
self.id = unpacked_data[2]
|
||||
self.length = unpacked_data[3]
|
||||
self.fuelCellCircuitBreaker = unpacked_data[4]
|
||||
self.powerLithiumBatteryCircuitBreaker = unpacked_data[5]
|
||||
self.propulsionMotorCircuitBreaker = unpacked_data[6]
|
||||
self.lithiumBatteryGroupInstrumentCircuitBreaker = unpacked_data[7]
|
||||
self.dcDc5ModuleCircuitBreaker = unpacked_data[8]
|
||||
self.bowHighVoltageDistributionBoxCircuitBreaker = unpacked_data[9]
|
||||
self.sternFTDevice45CircuitBreaker = unpacked_data[10]
|
||||
self.sternRudderSwitch1CircuitBreaker = unpacked_data[11]
|
||||
self.sternRudderSwitch2CircuitBreaker = unpacked_data[12]
|
||||
self.fbReservedCircuitBreaker = unpacked_data[13]
|
||||
self.tyzReservedCircuitBreaker = unpacked_data[14]
|
||||
self.reservedCircuitBreaker = unpacked_data[15]
|
||||
self.checkCode = unpacked_data[16]
|
||||
except struct.error as e:
|
||||
print(f"Error unpacking data: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
msg = msg_CcuStateFbMsg()
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
import socket
|
||||
import struct
|
||||
import tkinter as tk
|
||||
import threading
|
||||
from udpmsg import *
|
||||
from datetime import datetime
|
||||
from tkinter import font
|
||||
from guiEle import ElementDisplayApp
|
||||
import threading
|
||||
|
||||
target_ip = '127.0.0.1'
|
||||
target_port = 5000
|
||||
|
||||
def receive_messages(sock, text_box,eleplay:ElementDisplayApp):
|
||||
while True:
|
||||
current_time = datetime.now()
|
||||
formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
data, address = sock.recvfrom(4096)
|
||||
headBuf = data[:6]
|
||||
header = struct.unpack('<BBHH',headBuf)
|
||||
start1 = data[0]
|
||||
start2 = data[1]
|
||||
msg_id = data[2]
|
||||
msg_len = data[3]
|
||||
if(start1==0x40 and start2==0x40):
|
||||
if(msg_id==0x0001):
|
||||
msg = msg_disHVBCmd()
|
||||
msg.unpack(data)
|
||||
text_box.insert(tk.END, f"{formatted_time}->接收到来自 {address} | 高压母线配电指令:\n")
|
||||
if(msg.fuelCellCircuitBreaker==0x55):
|
||||
eleplay.dict1['燃料电池断路器']='闭合'
|
||||
elif(msg.fuelCellCircuitBreaker==0xAA):
|
||||
eleplay.dict1['燃料电池断路器']='断开'
|
||||
|
||||
if(msg.powerLithiumBatteryCircuitBreaker==0x55):
|
||||
eleplay.dict1['动力锂电池断路器']='闭合'
|
||||
elif(msg.powerLithiumBatteryCircuitBreaker==0xAA):
|
||||
eleplay.dict1['动力锂电池断路器']='断开'
|
||||
|
||||
if(msg.propulsionMotorCircuitBreaker==0x55):
|
||||
eleplay.dict1['推进电机断路器']='闭合'
|
||||
elif(msg.propulsionMotorCircuitBreaker==0xAA):
|
||||
eleplay.dict1['推进电机断路器']='断开'
|
||||
|
||||
if(msg.lithiumBatteryGroupInstrumentCircuitBreaker==0x55):
|
||||
eleplay.dict1['仪表锂电池断路器']='闭合'
|
||||
elif(msg.lithiumBatteryGroupInstrumentCircuitBreaker==0xAA):
|
||||
eleplay.dict1['仪表锂电池断路器']='断开'
|
||||
|
||||
if(msg.dcDc5ModuleCircuitBreaker==0x55):
|
||||
eleplay.dict1['DC/DC5模块断路器']='闭合'
|
||||
elif(msg.dcDc5ModuleCircuitBreaker==0xAA):
|
||||
eleplay.dict1['DC/DC5模块断路器']='断开'
|
||||
|
||||
if(msg.bowHighVoltageDistributionBoxCircuitBreaker==0x55):
|
||||
eleplay.dict1['艏部高压配电断路器']='闭合'
|
||||
elif(msg.bowHighVoltageDistributionBoxCircuitBreaker==0xAA):
|
||||
eleplay.dict1['艏部高压配电断路器']='断开'
|
||||
|
||||
if(msg.sternFTDevice45CircuitBreaker==0x55):
|
||||
eleplay.dict1['艏部FT4/5断路器']='闭合'
|
||||
elif(msg.sternFTDevice45CircuitBreaker==0xAA):
|
||||
eleplay.dict1['艏部FT4/5断路器']='断开'
|
||||
|
||||
if(msg.sternRudderSwitch1CircuitBreaker==0x55):
|
||||
eleplay.dict1['艉舵1/2断路器']='闭合'
|
||||
elif(msg.sternRudderSwitch1CircuitBreaker==0xAA):
|
||||
eleplay.dict1['艉舵1/2断路器']='断开'
|
||||
|
||||
if(msg.sternRudderSwitch2CircuitBreaker==0x55):
|
||||
eleplay.dict1['艉舵3/4断路器']='闭合'
|
||||
elif(msg.sternRudderSwitch2CircuitBreaker==0xAA):
|
||||
eleplay.dict1['艉舵3/4断路器']='断开'
|
||||
|
||||
if(msg.fbReservedCircuitBreaker==0x55):
|
||||
eleplay.dict1['FB预留断路器']='闭合'
|
||||
elif(msg.fbReservedCircuitBreaker==0xAA):
|
||||
eleplay.dict1['FB预留断路器']='断开'
|
||||
|
||||
if(msg.tyzReservedCircuitBreaker==0x55):
|
||||
eleplay.dict1['TYZ预留断路器']='闭合'
|
||||
elif(msg.tyzReservedCircuitBreaker==0xAA):
|
||||
eleplay.dict1['TYZ预留断路器']='断开'
|
||||
|
||||
if(msg.reservedCircuitBreaker==0x55):
|
||||
eleplay.dict1['预留断路器']='闭合'
|
||||
elif(msg.reservedCircuitBreaker==0xAA):
|
||||
eleplay.dict1['预留断路器']='断开'
|
||||
eleplay.update_disp()
|
||||
text_box.see(tk.END)
|
||||
|
||||
|
||||
def send_messages(sock, eleplay:ElementDisplayApp):
|
||||
while True:
|
||||
# 配电消息反馈
|
||||
a,b,c,d = eleplay.updata_data()
|
||||
sock.sendto(a, (target_ip, target_port))
|
||||
sock.sendto(b, (target_ip, target_port))
|
||||
sock.sendto(c, (target_ip, target_port))
|
||||
sock.sendto(d, (target_ip, target_port))
|
||||
|
||||
#TODO复合管控消息反馈
|
||||
|
||||
|
||||
threading.Event().wait(1) # 每隔1秒发送一次消息
|
||||
Reference in New Issue
Block a user