Files

1936 lines
95 KiB
Python

import tkinter as tk
from tkinter import ttk
import socket
import threading
import time
import struct
from db import DB
import os
from udpMsg import (
msg_disHighVolBusFbMsg,
msg_disHighAVolBusFbMsg,
msg_disLowMainBusFbMsg,
msg_disLowBusFbMsg,
msg_disHVBCmdMsg,
msg_disHVBACmdMsg,
msg_disLVMBCmdMsg,
msg_disLVBCmdMsg
)
from ccuMsg import (
msg_ccUFbMsg,
# msg_ccUCmdMsg
)
class PowerManagerApp:
def __init__(self, root):
self.root = root
# 解决字体模糊问题
style = ttk.Style()
style.configure('.', font=('Microsoft YaHei', 10))
self.simulation_HighVolBusFbMsg = msg_disHighVolBusFbMsg()
self.simulation_HighAVolBusFbMsg = msg_disHighAVolBusFbMsg()
self.simulation_LowMainBusFbMsg = msg_disLowMainBusFbMsg()
self.simulation_LowBusFbMsg = msg_disLowBusFbMsg()
self.simulation_HVBCmdMsg = msg_disHVBCmdMsg()
self.simulation_HVBACmdMsg = msg_disHVBACmdMsg()
self.simulation_LVMBCmdMsg = msg_disLVMBCmdMsg()
self.simulation_LVBCmdMsg = msg_disLVBCmdMsg()
self.simulation_ccUFbMsg = msg_ccUFbMsg()
# 菜单栏复选框状态变量
self.menu_checkbox_vars = {
'Start_Connection': tk.BooleanVar(value=False),
'Forward_Feedback': tk.BooleanVar(value=False),
'Transparent_Command': tk.BooleanVar(value=False),
'Simulation_Mode': tk.BooleanVar(value=False)
}
self.Start_Connection = self.menu_checkbox_vars['Start_Connection']
self.Start_Connection.trace_add('write', self.on_connection_changed)
self.Forward_Feedback = self.menu_checkbox_vars['Forward_Feedback']
self.Transparent_Command = self.menu_checkbox_vars['Transparent_Command']
self.Simulation_Mode = self.menu_checkbox_vars['Simulation_Mode']
# 添加模拟模式回调
self.Simulation_Mode.trace_add('write', self.on_simulation_mode_changed)
self.simulation_thread = None
self.simulation_running = False
self.root.title("电源管理系统")
self.root.tk.call('tk', 'scaling', 1.5) # 适配高DPI显示
# 移除固定窗口大小设置,实现自适应
self.root.minsize(800, 600) # 设置最小窗口大小
# 信息主机端口
self.udp_control_port = 5001
# self.udp_control_host = "192.168.0.223"
self.udp_control_host = "127.0.0.1"
# 本机端口
self.udp_ip = "127.0.0.1"
# self.udp_ip = "192.168.0.140"
self.udp_port = 7000
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.bind((self.udp_ip, self.udp_port))
# 高压配电板端口
self.disSysPortC1 = 6000
self.disSysHostC1 = '192.168.1.154'
# self.disSysHostC1 = '127.0.0.1'
# 仪表配电板端口
self.disSysPortC2 = 6000
self.disSysHostC2 = '192.168.1.158'
# self.disSysHostC2 = '127.0.0.1'
# 数据库配置
self.db = DB(os.path.join(os.path.dirname(__file__), 'test.db'))
self.db.connect()
# 状态数据 - 完全匹配udpMsg.py中的定义
self.status_data = {
"high_vol_bus": {
"breakers": {
"燃料电池断路器": 0,
"动力锂电池断路器": 0,
"推进电机断路器": 0,
"锂电池组仪表断路器": 0,
"DC/DC模块断路器": 0,
"艏高压配电盒断路器": 0,
"艉FT设备45断路器": 0,
"艉舵机开关1断路器": 0,
"艉舵机开关2断路器": 0,
"FB备用断路器": 0,
"TYZ备用断路器": 0,
"备用断路器": 0
},
"faults": {
"DC/DC故障1": 0,
"DC/DC故障2": 0
},
"insulation": {
"动力母线绝缘状态": 0,
"A母线绝缘状态": 0
},
"alarms": {
"仪表电源丢失信号": 0,
"应急电源丢失信号": 0
},
"voltage": {
"动力母线电压": 0.0,
"A母线电压": 0.0
},
"current": {
"动力母线电流": 0.0,
"DC/DC5模块电流": 0.0,
"推进电机控制箱电流": 0.0,
"A母线电流": 0.0,
"锂电池组仪表电流": 0.0,
"艏高压配电盒电流": 0.0,
"艉FT设备45电流": 0.0,
"TYZ备用开关电流": 0.0,
"艉舵机开关1电流": 0.0,
"备用电流1": 0.0,
"艉舵机开关2电流": 0.0,
"备用电流2": 0.0,
"FB备用开关电流": 0.0,
"备用电流3": 0.0
},
"cool_water_pressure": 0.0,
"dcdc_temp":0
},
"high_a_vol_bus": {
"breakers": {
"艏FT设备123断路器": 0,
"执行机构断路器": 0,
"桅舵机控制箱断路器": 0,
"XCZ断路器": 0,
"艏舵控制箱断路器": 0,
"开盖启动气缸断路器": 0
},
"alarms": {
"仪表电源故障": 0,
"应急电源故障": 0,
"进水报警": 0
},
"voltage": {
"B母线电压": 0.0
},
"current": {
"B母线电流": 0.0,
"艏FT设备123电流": 0.0,
"执行机构电流": 0.0,
"桅舵机控制箱电流": 0.0,
"XCZ电流": 0.0,
"艏舵控制箱电流": 0.0,
"开盖启动气缸电流": 0.0
}
},
"low_main_bus": {
"breakers": {
"锂电池组仪表断路器": 0,
"艏低压配电盒断路器": 0,
"单元4仪表48V断路器": 0,
"DC/C1直流配电板断路器": 0,
"备用断路器1": 0,
"备用断路器2": 0,
"应急锂电池组2断路器": 0
},
"alarms": {
"仪表电源故障": 0,
"应急电源故障": 0
},
"insulation": 0,
"voltage": {
"仪表母线电压": 0.0,
"应急2母线电压": 0.0
},
"current": {
"艏低压配电盒电流": 0.0,
"DC/C1仪表48V电流": 0.0,
"备用电流1": 0.0,
"应急锂电池组2电流": 0.0,
"锂电池组仪表电流": 0.0,
"单元4仪表48V电流": 0.0,
"备用电流2": 0.0,
"综合能量管理系统应急DC48V电流": 0.0,
"燃料电池安全系统应急DC48V电流": 0.0,
"PLC控制电源电流": 0.0
}
},
"low_bus": {
"breakers": {
"单元1断路器": 0,
"单元2断路器": 0,
"单元3断路器": 0,
"单元5断路器": 0,
"艏PZ设备断路器": 0,
"备用断路器1": 0,
"备用断路器2": 0,
"应急锂电池组1断路器": 0
},
"alarms": {
"仪表电源故障": 0,
"应急电源故障": 0,
"进水报警": 0
},
"voltage": {
"仪表母线电压": 0.0,
"应急1母线电压": 0.0
},
"current": {
"仪表母线电流": 0.0,
"单元1电流": 0.0,
"单元2电流": 0.0,
"单元3电流": 0.0,
"单元5电流": 0.0,
"艏PZ设备电流": 0.0,
"应急锂电池组1电流": 0.0
}
}
}
# 初始化界面
self.setup_ui()
# 初始化UDP接收线程(不自动启动)
self.running = False
self.recv_thread = threading.Thread(target=self.udp_receive_loop)
self.recv_thread.daemon = True
# 启动状态更新定时器
self.update_ui()
def setup_config_ui(self, parent):
"""创建通信配置窗口"""
config_window = tk.Toplevel(parent)
config_window.title("通信配置")
config_window.geometry("400x300")
# 本机配置
ttk.Label(config_window, text="本机配置").grid(row=0, column=0, padx=10, pady=5, sticky=tk.W)
ttk.Label(config_window, text="IP地址:").grid(row=1, column=0, padx=10, pady=5, sticky=tk.W)
self.local_ip_entry = ttk.Entry(config_window)
self.local_ip_entry.grid(row=1, column=1, padx=10, pady=5)
self.local_ip_entry.insert(0, self.udp_ip)
ttk.Label(config_window, text="端口:").grid(row=2, column=0, padx=10, pady=5, sticky=tk.W)
self.local_port_entry = ttk.Entry(config_window)
self.local_port_entry.grid(row=2, column=1, padx=10, pady=5)
self.local_port_entry.insert(0, str(self.udp_port))
# 高压配电板配置
ttk.Label(config_window, text="高压配电板配置").grid(row=3, column=0, padx=10, pady=5, sticky=tk.W)
ttk.Label(config_window, text="IP地址:").grid(row=4, column=0, padx=10, pady=5, sticky=tk.W)
self.high_vol_ip_entry = ttk.Entry(config_window)
self.high_vol_ip_entry.grid(row=4, column=1, padx=10, pady=5)
self.high_vol_ip_entry.insert(0, self.disSysHostC1)
ttk.Label(config_window, text="端口:").grid(row=5, column=0, padx=10, pady=5, sticky=tk.W)
self.high_vol_port_entry = ttk.Entry(config_window)
self.high_vol_port_entry.grid(row=5, column=1, padx=10, pady=5)
self.high_vol_port_entry.insert(0, str(self.disSysPortC1))
# 仪表配电板配置
ttk.Label(config_window, text="仪表配电板配置").grid(row=6, column=0, padx=10, pady=5, sticky=tk.W)
ttk.Label(config_window, text="IP地址:").grid(row=7, column=0, padx=10, pady=5, sticky=tk.W)
self.instrument_ip_entry = ttk.Entry(config_window)
self.instrument_ip_entry.grid(row=7, column=1, padx=10, pady=5)
self.instrument_ip_entry.insert(0, self.disSysHostC2)
ttk.Label(config_window, text="端口:").grid(row=8, column=0, padx=10, pady=5, sticky=tk.W)
self.instrument_port_entry = ttk.Entry(config_window)
self.instrument_port_entry.grid(row=8, column=1, padx=10, pady=5)
self.instrument_port_entry.insert(0, str(self.disSysPortC2))
# 信息主机配置
ttk.Label(config_window, text="信息主机配置").grid(row=9, column=0, padx=10, pady=5, sticky=tk.W)
ttk.Label(config_window, text="IP地址:").grid(row=10, column=0, padx=10, pady=5, sticky=tk.W)
self.control_host_entry = ttk.Entry(config_window)
self.control_host_entry.grid(row=10, column=1, padx=10, pady=5)
self.control_host_entry.insert(0, self.udp_control_host)
ttk.Label(config_window, text="端口:").grid(row=11, column=0, padx=10, pady=5, sticky=tk.W)
self.control_port_entry = ttk.Entry(config_window)
self.control_port_entry.grid(row=11, column=1, padx=10, pady=5)
self.control_port_entry.insert(0, str(self.udp_control_port))
# 保存按钮
save_btn = ttk.Button(config_window, text="保存配置", command=self.save_config)
save_btn.grid(row=12, column=0, columnspan=2, pady=10)
def setup_db_config_ui(self, parent):
"""创建数据库配置窗口"""
db_window = tk.Toplevel(parent)
db_window.title("数据库配置")
db_window.geometry("400x200")
# 数据库路径配置
ttk.Label(db_window, text="数据库路径:").grid(row=0, column=0, padx=10, pady=5, sticky=tk.W)
self.db_path_entry = ttk.Entry(db_window)
self.db_path_entry.grid(row=0, column=1, padx=10, pady=5)
self.db_path_entry.insert(0, os.path.dirname(self.db.db_path))
# 数据库名称配置
ttk.Label(db_window, text="数据库名称:").grid(row=1, column=0, padx=10, pady=5, sticky=tk.W)
self.db_name_entry = ttk.Entry(db_window)
self.db_name_entry.grid(row=1, column=1, padx=10, pady=5)
self.db_name_entry.insert(0, os.path.basename(self.db.db_path))
# 保存按钮
save_btn = ttk.Button(db_window, text="保存配置", command=self.save_db_config)
save_btn.grid(row=2, column=0, columnspan=2, pady=10)
def save_db_config(self):
"""保存数据库配置"""
try:
db_path = os.path.join(self.db_path_entry.get(), self.db_name_entry.get())
self.db.disconnect()
self.db = DB(db_path)
self.db.connect()
self.log_message(f"数据库配置已更新: {db_path}")
except Exception as e:
self.log_message(f"数据库配置保存失败: {str(e)}")
# 恢复原配置
self.db = DB(os.path.join(os.path.dirname(__file__), 'test.db'))
self.db.connect()
self.db = DB(os.path.join(os.path.dirname(__file__), 'test.db'))
self.db.connect()
def save_config(self):
"""保存配置"""
# 保存当前配置作为回滚用
old_ip = self.udp_ip
old_port = self.udp_port
old_host1 = self.disSysHostC1
old_port1 = self.disSysPortC1
old_host2 = self.disSysHostC2
old_port2 = self.disSysPortC2
try:
# 更新本机配置
self.udp_ip = self.local_ip_entry.get()
self.udp_port = int(self.local_port_entry.get())
# 更新高压配电板配置
self.disSysHostC1 = self.high_vol_ip_entry.get()
self.disSysPortC1 = int(self.high_vol_port_entry.get())
# 更新仪表配电板配置
self.disSysHostC2 = self.instrument_ip_entry.get()
self.disSysPortC2 = int(self.instrument_port_entry.get())
# 创建新socket并设置SO_REUSEADDR选项
new_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
new_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
new_sock.settimeout(0.1) # 保持一致的超时设置
try:
new_sock.bind((self.udp_ip, self.udp_port))
except OSError as e:
self.log_message(f"绑定socket失败: {str(e)} (IP: {self.udp_ip}, 端口: {self.udp_port})")
new_sock.close()
raise
# 成功后再替换旧socket
old_sock = self.sock
self.sock = new_sock
old_sock.close()
self.log_message(f"成功绑定新socket (IP: {self.udp_ip}, 端口: {self.udp_port})")
self.log_message("通信配置已更新")
return True
except Exception as e:
# 恢复旧配置
self.udp_ip = old_ip
self.udp_port = old_port
self.disSysHostC1 = old_host1
self.disSysPortC1 = old_port1
self.disSysHostC2 = old_host2
self.disSysPortC2 = old_port2
self.log_message(f"配置保存失败: {str(e)}")
# 确保socket仍然有效
if not hasattr(self, 'sock') or not isinstance(self.sock, socket.socket):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.settimeout(0.1)
self.sock.bind((self.udp_ip, self.udp_port))
return False
def setup_ui(self):
"""初始化用户界面"""
# 创建菜单栏
menubar = tk.Menu(self.root)
config_menu = tk.Menu(menubar, tearoff=0)
config_menu.add_checkbutton(label="启动连接", variable=self.menu_checkbox_vars['Start_Connection'])
config_menu.add_command(label="通信配置", command=lambda: self.setup_config_ui(self.root))
config_menu.add_command(label="数据库配置", command=lambda: self.setup_db_config_ui(self.root))
menubar.add_cascade(label="配置", menu=config_menu)
# 添加选项菜单
options_menu = tk.Menu(menubar, tearoff=0)
options_menu.add_checkbutton(label="转发反馈", variable=self.menu_checkbox_vars['Forward_Feedback'])
options_menu.add_checkbutton(label="透传指令", variable=self.menu_checkbox_vars['Transparent_Command'])
options_menu.add_checkbutton(label="模拟模式", variable=self.menu_checkbox_vars['Simulation_Mode'])
menubar.add_cascade(label="通信选项", menu=options_menu)
dis_options_menu = tk.Menu(menubar, tearoff=0)
dis_options_menu.add_command(label="动力母线操作", command=lambda: self.show_popup_window("动力母线操作"))
dis_options_menu.add_command(label="艏部高压操作", command=lambda: self.show_popup_window("艏部高压操作"))
dis_options_menu.add_command(label="仪表母线操作", command=lambda: self.show_popup_window("仪表母线操作"))
dis_options_menu.add_command(label="艏部低压操作", command=lambda: self.show_popup_window("艏部低压操作"))
menubar.add_cascade(label="配电操作", menu=dis_options_menu)
menubar.add_command(label="DCDC状态", command=lambda: self.show_popup_window("DCDC状态"))
menubar.add_command(label="复合管控面板", command=lambda: self.show_compound_control_window("复合管控面板"))
menubar.add_command(label="软件说明", command=lambda: self.show_help_window("程序说明"))
# 配置菜单栏
self.root.config(menu=menubar)
# 主框架
main_frame = ttk.Frame(self.root, padding="10")
main_frame.pack(fill=tk.BOTH, expand=True)
self.root.grid_rowconfigure(0, weight=1)
self.root.grid_columnconfigure(0, weight=1)
# 状态显示区
status_frame = ttk.LabelFrame(main_frame, text="系统状态", padding="10")
status_frame.pack(fill=tk.BOTH, expand=True, pady=5)
# 创建Canvas和Scrollbar
self.notebook = ttk.Notebook(status_frame)
self.notebook.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
# 高压母线状态页
high_vol_frame = ttk.Frame(self.notebook)
self.notebook.add(high_vol_frame, text="动力母线状态")
self.setup_high_vol_ui(high_vol_frame)
# 高压A母线状态页
high_a_vol_frame = ttk.Frame(self.notebook)
self.notebook.add(high_a_vol_frame, text="艏部高压状态")
self.setup_high_a_vol_ui(high_a_vol_frame)
# 仪表主母线状态页
low_main_frame = ttk.Frame(self.notebook)
self.notebook.add(low_main_frame, text="仪表母线状态")
self.setup_low_main_ui(low_main_frame)
# 仪表母线状态页
low_bus_frame = ttk.Frame(self.notebook)
self.notebook.add(low_bus_frame, text="艏部低压状态")
self.setup_low_bus_ui(low_bus_frame)
# 日志区
log_frame = ttk.LabelFrame(main_frame, text="系统日志", padding="10")
log_frame.pack(fill=tk.BOTH, expand=True, pady=5)
# 使用grid布局
self.log_text = tk.Text(log_frame, height=10, state=tk.DISABLED)
scrollbar = ttk.Scrollbar(log_frame, orient=tk.VERTICAL, command=self.log_text.yview)
# 配置grid布局
self.log_text.grid(row=0, column=0, sticky="nsew")
scrollbar.grid(row=0, column=1, sticky="ns")
# 设置权重使文本框可扩展
log_frame.grid_rowconfigure(0, weight=1)
log_frame.grid_columnconfigure(0, weight=1)
self.log_text.config(yscrollcommand=scrollbar.set)
def setup_hvb_control(self, parent):
"""设置动力母线配电指令控制面板"""
frame = ttk.LabelFrame(parent, text="动力母线配电指令", padding="10")
frame.pack(fill=tk.X, pady=5)
# 断路器控制按钮
breaker_names = [
"燃料电池断路器", "动力锂电池断路器", "推进电机断路器",
"锂电池组仪表断路器", "DC/DC模块断路器", "艏高压配电盒断路器",
"艉FT设备45断路器", "艉舵机开关1断路器", "艉舵机开关2断路器",
"FB备用断路器", "TYZ备用断路器", "备用断路器", "DC/DC模块启动","冷却系统启动"
]
for i, name in enumerate(breaker_names):
row = i // 3
col = i % 3
btn_frame = ttk.Frame(frame)
btn_frame.grid(row=row, column=col, padx=10, pady=2, sticky=tk.W)
ttk.Label(btn_frame, text=name, width=20).pack(side=tk.LEFT)
on_btn = ttk.Button(btn_frame, text="合", width=3,
command=lambda n=name: self.send_hvb_cmd(n, 0x55))
on_btn.pack(side=tk.LEFT, padx=2)
off_btn = ttk.Button(btn_frame, text="分", width=3,
command=lambda n=name: self.send_hvb_cmd(n, 0xAA))
off_btn.pack(side=tk.LEFT, padx=2)
def setup_hvba_control(self, parent):
"""设置动力汇流排A配电指令控制面板"""
frame = ttk.LabelFrame(parent, text="动力汇流排A配电指令", padding="10")
frame.pack(fill=tk.X, pady=5)
# 断路器控制按钮
breaker_names = [
"艏FT设备123断路器", "执行机构断路器", "桅舵机控制箱断路器",
"XCZ断路器", "艏舵控制箱断路器", "开盖启动气缸断路器"
]
for i, name in enumerate(breaker_names):
row = i // 3
col = i % 3
btn_frame = ttk.Frame(frame)
btn_frame.grid(row=row, column=col, padx=10, pady=2, sticky=tk.W)
ttk.Label(btn_frame, text=name, width=20).pack(side=tk.LEFT)
on_btn = ttk.Button(btn_frame, text="合", width=3,
command=lambda n=name: self.send_hvba_cmd(n, 0x55))
on_btn.pack(side=tk.LEFT, padx=2)
off_btn = ttk.Button(btn_frame, text="分", width=3,
command=lambda n=name: self.send_hvba_cmd(n, 0xAA))
off_btn.pack(side=tk.LEFT, padx=2)
def setup_lvmb_control(self, parent):
"""设置仪表母线配电指令控制面板"""
frame = ttk.LabelFrame(parent, text="仪表母线配电指令", padding="10")
frame.pack(fill=tk.X, pady=5)
# 断路器控制按钮
breaker_names = [
"锂电池组仪表断路器", "艏低压配电盒断路器", "单元4仪表48V断路器",
"DC/C1直流配电板断路器", "备用断路器1", "备用断路器2",
"应急锂电池组2断路器"
]
for i, name in enumerate(breaker_names):
row = i // 3
col = i % 3
btn_frame = ttk.Frame(frame)
btn_frame.grid(row=row, column=col, padx=10, pady=2, sticky=tk.W)
ttk.Label(btn_frame, text=name, width=20).pack(side=tk.LEFT)
on_btn = ttk.Button(btn_frame, text="合", width=3,
command=lambda n=name: self.send_lvmb_cmd(n, 0x55))
on_btn.pack(side=tk.LEFT, padx=2)
off_btn = ttk.Button(btn_frame, text="分", width=3,
command=lambda n=name: self.send_lvmb_cmd(n, 0xaa))
off_btn.pack(side=tk.LEFT, padx=2)
def setup_lvb_control(self, parent):
"""设置仪表汇流排配电指令控制面板"""
frame = ttk.LabelFrame(parent, text="仪表汇流排配电指令", padding="10")
frame.pack(fill=tk.X, pady=5)
# 断路器控制按钮
breaker_names = [
"单元1断路器", "单元2断路器", "单元3断路器",
"单元5断路器", "艏PZ设备断路器", "备用断路器1",
"备用断路器2", "应急锂电池组1断路器"
]
for i, name in enumerate(breaker_names):
row = i // 3
col = i % 3
btn_frame = ttk.Frame(frame)
btn_frame.grid(row=row, column=col, padx=10, pady=2, sticky=tk.W)
ttk.Label(btn_frame, text=name, width=20).pack(side=tk.LEFT)
on_btn = ttk.Button(btn_frame, text="合", width=3,
command=lambda n=name: self.send_lvb_cmd(n, 0x55))
on_btn.pack(side=tk.LEFT, padx=2)
off_btn = ttk.Button(btn_frame, text="分", width=3,
command=lambda n=name: self.send_lvb_cmd(n, 0xaa))
off_btn.pack(side=tk.LEFT, padx=2)
def send_hvb_cmd(self, breaker_name, state):
"""发送动力母线配电指令"""
# print(breaker_name,state)
cmd = msg_disHVBCmdMsg()
if breaker_name == "燃料电池断路器":
cmd.fuelCellCircuitBreaker = state
elif breaker_name == "动力锂电池断路器":
cmd.powerLithiumBatteryCircuitBreaker = state
elif breaker_name == "推进电机断路器":
cmd.propulsionMotorCircuitBreaker = state
elif breaker_name == "锂电池组仪表断路器":
cmd.lithiumBatteryGroupInstrumentCircuitBreaker = state
elif breaker_name == "DC/DC模块断路器":
cmd.dcDc5ModuleCircuitBreaker = state
elif breaker_name == "艏高压配电盒断路器":
cmd.bowHighVoltageDistributionBoxCircuitBreaker = state
elif breaker_name == "艉FT设备45断路器":
cmd.sternFTDevice45CircuitBreaker = state
elif breaker_name == "艉舵机开关1断路器":
cmd.sternRudderSwitch1CircuitBreaker = state
elif breaker_name == "艉舵机开关2断路器":
cmd.sternRudderSwitch2CircuitBreaker = state
elif breaker_name == "FB备用断路器":
cmd.fbReservedCircuitBreaker = state
elif breaker_name == "TYZ备用断路器":
cmd.tyzReservedCircuitBreaker = state
elif breaker_name == "备用断路器":
cmd.reservedCircuitBreaker = state
elif breaker_name == "DC/DC模块启动":
cmd.dcdc5Module = state
elif breaker_name == "冷却系统启动":
cmd.coolingSystem = state
if(self.Simulation_Mode):
if breaker_name == "燃料电池断路器":
self.simulation_HVBCmdMsg.fuelCellCircuitBreaker = state
elif breaker_name == "动力锂电池断路器":
self.simulation_HVBCmdMsg.powerLithiumBatteryCircuitBreaker = state
elif breaker_name == "推进电机断路器":
self.simulation_HVBCmdMsg.propulsionMotorCircuitBreaker = state
elif breaker_name == "锂电池组仪表断路器":
self.simulation_HVBCmdMsg.lithiumBatteryGroupInstrumentCircuitBreaker = state
elif breaker_name == "DC/DC模块断路器":
self.simulation_HVBCmdMsg.dcDc5ModuleCircuitBreaker = state
elif breaker_name == "艏高压配电盒断路器":
self.simulation_HVBCmdMsg.bowHighVoltageDistributionBoxCircuitBreaker = state
elif breaker_name == "艉FT设备45断路器":
self.simulation_HVBCmdMsg.sternFTDevice45CircuitBreaker = state
elif breaker_name == "艉舵机开关1断路器":
self.simulation_HVBCmdMsg.sternRudderSwitch1CircuitBreaker = state
elif breaker_name == "艉舵机开关2断路器":
self.simulation_HVBCmdMsg.sternRudderSwitch2CircuitBreaker = state
elif breaker_name == "FB备用断路器":
self.simulation_HVBCmdMsg.fbReservedCircuitBreaker = state
elif breaker_name == "TYZ备用断路器":
self.simulation_HVBCmdMsg.tyzReservedCircuitBreaker = state
elif breaker_name == "备用断路器":
self.simulation_HVBCmdMsg.reservedCircuitBreaker = state
elif breaker_name == "DC/DC模块启动":
self.simulation_HVBCmdMsg.dcdc5Module = state
elif breaker_name == "冷却系统启动":
self.simulation_HVBCmdMsg.coolingSystem = state
self.send_udp_command(cmd.pack(), self.disSysHostC1, self.disSysPortC1)
self.db.insertHighVolBusCmd(cmd)
if state==0x55:
buntu=True
else:
buntu=False
self.log_message(f"发送动力母线配电指令: {breaker_name} {'合' if buntu else '分'}")
def send_hvba_cmd(self, breaker_name, state):
"""发送动力汇流排A配电指令"""
cmd = msg_disHVBACmdMsg()
if breaker_name == "艏FT设备123断路器":
cmd.bowFTDevice123CircuitBreaker = state
elif breaker_name == "执行机构断路器":
cmd.actuatorCircuitBreaker = state
elif breaker_name == "桅舵机控制箱断路器":
cmd.mastSteeringGearControlBoxCircuitBreaker = state
elif breaker_name == "XCZ断路器":
cmd.xczCircuitBreaker = state
elif breaker_name == "艏舵控制箱断路器":
cmd.bowRudderControlBoxCircuitBreaker = state
elif breaker_name == "开盖启动气缸断路器":
cmd.openWaterCoverStartCylinderCircuitBreaker = state
if state==0x55:
buntu=True
else:
buntu=False
if(self.Simulation_Mode):
if breaker_name == "艏FT设备123断路器":
self.simulation_HVBACmdMsg.bowFTDevice123CircuitBreaker = state
elif breaker_name == "执行机构断路器":
self.simulation_HVBACmdMsg.actuatorCircuitBreaker = state
elif breaker_name == "桅舵机控制箱断路器":
self.simulation_HVBACmdMsg.mastSteeringGearControlBoxCircuitBreaker = state
elif breaker_name == "XCZ断路器":
self.simulation_HVBACmdMsg.xczCircuitBreaker = state
elif breaker_name == "艏舵控制箱断路器":
self.simulation_HVBACmdMsg.bowRudderControlBoxCircuitBreaker = state
elif breaker_name == "开盖启动气缸断路器":
self.simulation_HVBACmdMsg.openWaterCoverStartCylinderCircuitBreaker = state
self.log_message(f"发送动力汇流排A配电指令: {breaker_name} {'合' if buntu else '分'}")
# self.send_udp_command(cmd.pack())
self.send_udp_command(cmd.pack(), self.disSysHostC1, self.disSysPortC1)
self.db.insertHighAVolBusCmd(cmd)
def send_lvmb_cmd(self, breaker_name, state):
"""发送仪表母线配电指令"""
cmd = msg_disLVMBCmdMsg()
if breaker_name == "锂电池组仪表断路器":
cmd.lithiumBatteryGroupInstrumentCircuitBreaker = state
elif breaker_name == "艏低压配电盒断路器":
cmd.bowLowVoltageDistributionBoxCircuitBreaker = state
elif breaker_name == "单元4仪表48V断路器":
cmd.unit4InstrumentDC48VCircuitBreaker = state
elif breaker_name == "DC/C1直流配电板断路器":
cmd.dcC1DCDistributionPanelCircuitBreaker = state
elif breaker_name == "备用断路器1":
cmd.reservedCircuitBreaker1 = state
elif breaker_name == "备用断路器2":
cmd.reservedCircuitBreaker2 = state
elif breaker_name == "应急锂电池组2断路器":
cmd.emergencyLithiumBatteryGroup2CircuitBreaker = state
if(self.Simulation_Mode):
if breaker_name == "锂电池组仪表断路器":
self.simulation_LVMBCmdMsg.lithiumBatteryGroupInstrumentCircuitBreaker = state
elif breaker_name == "艏低压配电盒断路器":
self.simulation_LVMBCmdMsg.bowLowVoltageDistributionBoxCircuitBreaker = state
elif breaker_name == "单元4仪表48V断路器":
self.simulation_LVMBCmdMsg.unit4InstrumentDC48VCircuitBreaker = state
elif breaker_name == "DC/C1直流配电板断路器":
self.simulation_LVMBCmdMsg.dcC1DCDistributionPanelCircuitBreaker = state
elif breaker_name == "备用断路器1":
self.simulation_LVMBCmdMsg.reservedCircuitBreaker1 = state
elif breaker_name == "备用断路器2":
self.simulation_LVMBCmdMsg.reservedCircuitBreaker2 = state
# self.send_udp_command(cmd.pack())
print(breaker_name,state)
print(cmd)
# print(cmd.unpack())
self.send_udp_command(cmd.pack(), self.disSysHostC2, self.disSysPortC2)
self.db.insertLowMainBusCmd(cmd)
if state==0x55:
buntu=True
else:
buntu=False
self.log_message(f"发送仪表母线配电指令: {breaker_name} {'合' if buntu else '分'}")
def send_lvb_cmd(self, breaker_name, state):
"""发送仪表汇流排配电指令"""
cmd = msg_disLVBCmdMsg()
if breaker_name == "单元1断路器":
cmd.unit1CircuitBreaker = state
elif breaker_name == "单元2断路器":
cmd.unit2CircuitBreaker = state
elif breaker_name == "单元3断路器":
cmd.unit3CircuitBreaker = state
elif breaker_name == "单元5断路器":
cmd.unit5CircuitBreaker = state
elif breaker_name == "艏PZ设备断路器":
cmd.bowPZDeviceCircuitBreaker = state
elif breaker_name == "备用断路器1":
cmd.reservedCircuitBreaker1 = state
elif breaker_name == "备用断路器2":
cmd.reservedCircuitBreaker2 = state
elif breaker_name == "应急锂电池组1断路器":
cmd.emergencyLithiumBatteryGroup1CircuitBreaker = state
if(self.Simulation_Mode):
if breaker_name == "单元1断路器":
self.simulation_LVBCmdMsg.unit1CircuitBreaker = state
elif breaker_name == "单元2断路器":
self.simulation_LVBCmdMsg.unit2CircuitBreaker = state
elif breaker_name == "单元3断路器":
self.simulation_LVBCmdMsg.unit3CircuitBreaker = state
elif breaker_name == "单元5断路器":
self.simulation_LVBCmdMsg.unit5CircuitBreaker = state
elif breaker_name == "艏PZ设备断路器":
self.simulation_LVBCmdMsg.bowPZDeviceCircuitBreaker = state
elif breaker_name == "备用断路器1":
self.simulation_LVBCmdMsg.reservedCircuitBreaker1 = state
elif breaker_name == "备用断路器2":
self.simulation_LVBCmdMsg.reservedCircuitBreaker2 = state
elif breaker_name == "应急锂电池组1断路器":
self.simulation_LVBCmdMsg.emergencyLithiumBatteryGroup1CircuitBreaker = state
# self.send_udp_command(cmd.pack())
self.send_udp_command(cmd.pack(), self.disSysHostC2, self.disSysPortC2)
self.db.insertLowBusCmd(cmd)
if state==0x55:
buntu=True
else:
buntu=False
self.log_message(f"发送仪表汇流排配电指令: {breaker_name} {'合' if buntu else '分'}")
def udp_receive_loop(self):
"""UDP接收消息循环"""
while self.running:
if not self.Start_Connection.get():
time.sleep(0.1) # 连接未启动时短暂休眠
continue
try:
data, addr = self.sock.recvfrom(1024)
# 转发消息到上位机
if self.Forward_Feedback.get():
self.send_udp_command(data, self.udp_control_host, self.udp_control_port)
self.process_udp_message(data)
if self.Transparent_Command.get():
msg_id = int.from_bytes(data[2:4], byteorder='little')
if msg_id == 0x0001:
self.send_udp_command(data, self.disSysHostC1, self.disSysPortC1)
self.simulation_HVBCmdMsg.unpack(data)
elif msg_id == 0x0002:
self.send_udp_command(data, self.disSysHostC1, self.disSysPortC1)
self.simulation_HVBACmdMsg.unpack(data)
elif msg_id == 0x0003:
self.send_udp_command(data, self.disSysHostC2, self.disSysPortC2)
self.simulation_LVMBCmdMsg.unpack(data)
elif msg_id == 0x0004:
self.send_udp_command(data, self.disSysHostC2, self.disSysPortC2)
self.simulation_LVBCmdMsg.unpack(data)
except socket.timeout:
continue # 忽略超时错误
except Exception as e:
self.log_message(f"接收错误: {str(e)}")
time.sleep(0.1) # 发生错误时短暂休眠
def setup_dcdc_status(self, parent):
frame = parent
# 故障状态 - 扩展显示所有故障位
fault_labels1 = ["过温关机", "输出过流", "输出过压", "输出欠压", "输入过压", "输入欠压", "输出短路保护", "内部故障"]
fault_labels2 = ["通信故障", "工作状态", "冷却泵1故障", "冷却泵2故障", "冷却水泄漏", "冷却泵1运行状态", "冷却泵2运行状态", "保留"]
# DC/DC故障显示
fault_frame = ttk.Frame(frame)
fault_frame.grid(row=10, column=0, columnspan=4, sticky=tk.W, pady=2)
# 定义样式
style = ttk.Style()
style.configure("FaultTitle.TLabel", font=("Microsoft YaHei", 10, "bold"), background="#f0f0f0", padding=5)
style.configure("FaultLabel.TLabel", font=("Microsoft YaHei", 9))
ttk.Label(fault_frame, text="DC/DC状态字1:", style="FaultTitle.TLabel").grid(row=0, column=0, columnspan=4, sticky=tk.W, pady=5)
# 只在第一次初始化时创建列表
if not hasattr(self, 'high_vol_dcdc_fault1_bits'):
self.high_vol_dcdc_fault1_bits = []
self.high_vol_dcdc_fault2_bits = []
# 清空现有内容
self.high_vol_dcdc_fault1_bits.clear()
self.high_vol_dcdc_fault2_bits.clear()
for i, label in enumerate(fault_labels1):
pair_frame = tk.Frame(fault_frame, borderwidth=1, relief="solid", padx=4, pady=2)
pair_frame.grid(row=1+i//4, column=i%4, padx=5, pady=5, sticky=tk.W)
ttk.Label(pair_frame, text=label, style="FaultLabel.TLabel").pack(side=tk.LEFT, padx=(0, 5))
canvas = tk.Canvas(pair_frame, width=20, height=20, bg="white", highlightthickness=0)
canvas.create_oval(2, 2, 18, 18, fill="white", outline="gray")
canvas.pack(side=tk.RIGHT)
self.high_vol_dcdc_fault1_bits.append(canvas)
ttk.Label(fault_frame, text="DC/DC状态字2:", style="FaultTitle.TLabel").grid(row=3, column=0, columnspan=4, sticky=tk.W, pady=5)
for i, label in enumerate(fault_labels2):
pair_frame = tk.Frame(fault_frame, borderwidth=1, relief="solid", padx=4, pady=2)
pair_frame.grid(row=4+i//4, column=i%4, padx=5, pady=5, sticky=tk.W)
ttk.Label(pair_frame, text=label, style="FaultLabel.TLabel").pack(side=tk.LEFT, padx=(0, 5))
canvas = tk.Canvas(pair_frame, width=20, height=20, bg="white", highlightthickness=0)
canvas.create_oval(2, 2, 18, 18, fill="white", outline="gray")
canvas.pack(side=tk.RIGHT)
self.high_vol_dcdc_fault2_bits.append(canvas)
def setup_high_vol_ui(self, parent):
"""设置高压母线状态界面"""
# 初始化故障位列表
self.high_vol_dcdc_fault1_bits = []
self.high_vol_dcdc_fault2_bits = []
frame = ttk.LabelFrame(parent, text="高压母线状态")
frame.pack(fill=tk.BOTH, expand=True, padx=2, pady=2)
# 使用网格布局管理器
frame.grid_columnconfigure(0, weight=1)
frame.grid_columnconfigure(1, weight=1)
frame.grid_columnconfigure(2, weight=1)
frame.grid_columnconfigure(3, weight=1)
# 电压与断路器统一显示区域
status_frame = ttk.LabelFrame(frame, text="电压与断路器状态", padding=8)
status_frame.grid(row=0, column=0, columnspan=6, sticky=tk.W + tk.E, padx=8, pady=8)
# 电压显示(横向布局)
voltage_items = [
("动力母线电压(V):", "power_voltage"),
("艏部高压电压(V):", "a_voltage")
]
for i, (label_text, varname) in enumerate(voltage_items):
sub_frame = ttk.Frame(status_frame)
sub_frame.grid(row=0, column=i, padx=10, pady=4, sticky=tk.W)
ttk.Label(sub_frame, text=label_text, width=20, anchor="w").pack(side=tk.LEFT)
label_var = ttk.Label(sub_frame, text="0.0V", width=10, anchor="e", relief="groove")
label_var.pack(side=tk.RIGHT)
setattr(self, f"high_vol_{varname}", label_var)
# 断路器状态标签
ttk.Label(status_frame, text="断路器状态:", font=("Arial", 10, "bold")).grid(row=1, column=0, columnspan=6, sticky=tk.W, pady=(10, 4))
# 断路器状态显示区域
self.high_vol_breakers = {}
breaker_names = list(self.status_data["high_vol_bus"]["breakers"].keys())
columns_per_row = 4 # 每行 4 个断路器显示块
for i, name in enumerate(breaker_names):
row = 2 + i // columns_per_row
col = i % columns_per_row
# 每个断路器显示块:使用 Frame 带边框
breaker_frame = tk.Frame(status_frame, relief="solid", borderwidth=1, padx=6, pady=2)
breaker_frame.grid(row=row, column=col, padx=6, pady=4, sticky=tk.W)
# 名称标签
ttk.Label(breaker_frame, text=name, width=20, anchor="w").pack(side=tk.LEFT)
# 圆形指示灯
canvas = tk.Canvas(breaker_frame, width=18, height=18, bg="white", highlightthickness=0)
canvas.create_oval(2, 2, 16, 16, fill="white", outline="gray")
canvas.pack(side=tk.RIGHT)
self.high_vol_breakers[name] = canvas
# self.setup_dcdc_status(frame)
# # 绝缘状态 - 详细显示
# ttk.Label(frame, text="动力母线绝缘:").grid(row=16, column=0, sticky=tk.W)
# self.high_vol_power_insulation = ttk.Label(frame, text="正常")
# self.high_vol_power_insulation.grid(row=16, column=1, sticky=tk.W)
# ttk.Label(frame, text="A母线绝缘:").grid(row=16, column=2, sticky=tk.W)
# self.high_vol_a_insulation = ttk.Label(frame, text="正常")
# self.high_vol_a_insulation.grid(row=16, column=3, sticky=tk.W)
# 电流值显示分组框
current_group = ttk.LabelFrame(frame, text="电流值显示", padding=10)
current_group.grid(row=16, column=0, columnspan=6, sticky=tk.W+tk.E, padx=10, pady=10)
# 创建统一的显示区域(LabelFrame)
status_group = ttk.LabelFrame(frame, text="高压电系统状态", padding=10)
status_group.grid(row=16, column=0, columnspan=6, sticky=tk.W + tk.E, padx=10, pady=10)
# 所有参数标签及属性名
status_items = [
("动力母线电流(A):", "powerBusCurrent"),
("A汇流排电流(A):", "busbarACurrent"),
("DC/DC5模块电流(A):", "dcDc5ModuleCurrent"),
("推进电机控制箱电流(A):", "propulsionMotorControlBoxCurrent"),
("锂电池组(仪表)电流(A):", "lithiumBatteryGroupMeterCurrent"),
("艏部高压配电箱电流(A):", "bowHighVoltageDistributionBoxCurrent"),
("艉部FT装置4/5电流(A):", "sternFTDevice45Current"),
("TYZ预留开关电流(A):", "tyzReservedSwitchCurrent"),
("艉舵开关1电流(A):", "sternRudderSwitch1Current"),
("艉舵开关2电流(A):", "sternRudderSwitch2Current"),
("FB预留开关电流(A):", "fbReservedSwitchCurrent"),
("备用电流1(A):", "reservedCurrent1"),
("备用电流2(A):", "reservedCurrent2"),
("备用电流3(A):", "reservedCurrent3"),
("冷却水压力(MPa):", "cool_water_pressure"),
("DC/DC温度(℃):", "dcdc_temp"),
("仪表失电信号:", "losspower_insulation"),
("应急失电信号:", "loss_emerpower_insulation"),
("动力母线绝缘:", "power_insulation"),
("艏高压绝缘:", "a_insulation")
]
# 每行显示几列(可根据实际界面宽度调整)
columns_per_row = 3
# 布局显示
for i, (label_text, varname) in enumerate(status_items):
row = i // columns_per_row
col = i % columns_per_row
# 每个参数一个子 Frame(横向排列)
item_frame = ttk.Frame(status_group)
item_frame.grid(row=row, column=col, sticky=tk.W, padx=8, pady=3)
ttk.Label(item_frame, text=label_text, width=26, anchor="w").pack(side=tk.LEFT)
label_var = ttk.Label(item_frame, text="0.0", width=10, anchor="e", relief="groove")
label_var.pack(side=tk.RIGHT)
# 保存变量引用
setattr(self, f"high_vol_{varname}", label_var)
def setup_high_a_vol_ui(self, parent):
"""设置高压A母线状态界面"""
frame = ttk.LabelFrame(parent, text="高压A母线状态")
frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
# 使用网格布局管理器 - 调整为3列
frame.grid_columnconfigure(0, weight=1)
frame.grid_columnconfigure(1, weight=1)
frame.grid_columnconfigure(2, weight=1)
# 电压与断路器统一显示区域
status_frame = ttk.LabelFrame(frame, text="电压与断路器状态", padding=5)
status_frame.grid(row=0, column=0, columnspan=3, sticky=tk.W + tk.E, padx=5, pady=5)
# 电压显示(横向布局)
voltage_frame = ttk.Frame(status_frame)
voltage_frame.grid(row=0, column=0, padx=10, pady=4, sticky=tk.W)
ttk.Label(voltage_frame, text="汇流排B电压(V):", width=20, anchor="w").pack(side=tk.LEFT)
self.high_a_vol_voltage = ttk.Label(voltage_frame, text="0.0V", width=10, anchor="e", relief="groove")
self.high_a_vol_voltage.pack(side=tk.RIGHT)
# 断路器状态标签
ttk.Label(status_frame, text="断路器状态:", font=("Arial", 10, "bold")).grid(row=1, column=0, columnspan=6, sticky=tk.W, pady=(10, 4))
# 断路器状态显示区域
self.high_a_vol_breakers = {}
breaker_names = list(self.status_data["high_a_vol_bus"]["breakers"].keys())
columns_per_row = 3 # 每行3个断路器显示块
for i, name in enumerate(breaker_names):
row = 2 + i // columns_per_row
col = i % columns_per_row
# 每个断路器显示块:使用Frame带边框
breaker_frame = tk.Frame(status_frame, relief="solid", borderwidth=1, padx=6, pady=2)
breaker_frame.grid(row=row, column=col, padx=6, pady=4, sticky=tk.W)
# 名称标签
ttk.Label(breaker_frame, text=name, width=20, anchor="w").pack(side=tk.LEFT)
# 圆形指示灯
canvas = tk.Canvas(breaker_frame, width=18, height=18, bg="white", highlightthickness=0)
canvas.create_oval(2, 2, 16, 16, fill="white", outline="gray")
canvas.pack(side=tk.RIGHT)
self.high_a_vol_breakers[name] = canvas
# 创建统一的显示区域(LabelFrame)
status_group = ttk.LabelFrame(frame, text="艏部高压状态", padding=5)
status_group.grid(row=2, column=0, columnspan=3, sticky=tk.W + tk.E, padx=5, pady=5)
# 所有参数标签及属性名
status_items = [
("汇流排B电流(A):", "current"),
("艏部FT设备123电流(A):", "bow_ft_current"),
("启闭机构电流(A):", "actuator_current"),
("桅杆舵机控制箱电流(A):", "mast_steering_current"),
("XCZ电流(A):", "xcz_current"),
("艏舵控制箱电流(A):", "bow_rudder_current"),
("敞水盖启动电缸电流(A):", "open_cover_current"),
("仪表电源失电信号:", "instrument_power_fault"),
("应急电源失电信号:", "emergency_power_fault"),
("进水报警:", "water_alarm")
]
# 每行显示3列
columns_per_row = 3
# 布局显示
for i, (label_text, varname) in enumerate(status_items):
row = i // columns_per_row
col = i % columns_per_row
# 每个参数一个子Frame(横向排列)
item_frame = ttk.Frame(status_group)
item_frame.grid(row=row, column=col, sticky=tk.W, padx=8, pady=3)
ttk.Label(item_frame, text=label_text, width=26, anchor="w").pack(side=tk.LEFT)
label_var = ttk.Label(item_frame, text="0.0", width=10, anchor="e", relief="groove")
label_var.pack(side=tk.RIGHT)
# 保存变量引用
setattr(self, f"high_a_vol_{varname}", label_var)
def process_udp_message(self, data):
"""处理接收到的UDP消息"""
try:
msg_id = int.from_bytes(data[2:4], byteorder='little')
# print('----------------------------',hex(msg_id))
if msg_id == 0x0005: # 高压母线状态
msg = msg_disHighVolBusFbMsg()
if msg.unpack(data):
self.db.insertHighVolBusData(msg)
bus = self.status_data["high_vol_bus"]
bus["voltage"]["动力母线电压"] = msg.powerBusVoltage / 10.0
bus["voltage"]["A母线电压"] = msg.busbarAVoltage / 10.0
bus["current"]["动力母线电流"] = msg.powerBusCurrent / 10.0
bus["current"]["DC/DC5模块电流"] = msg.dcDc5ModuleCurrent / 10.0
bus["current"]["推进电机控制箱电流"] = msg.propulsionMotorControlBoxCurrent / 10.0
bus["current"]["A母线电流"] = msg.busbarACurrent / 10.0
bus["current"]["锂电池组仪表电流"] = msg.lithiumBatteryGroupMeterCurrent / 10.0
bus["current"]["艏高压配电盒电流"] = msg.bowHighVoltageDistributionBoxCurrent / 10.0
bus["current"]["艉FT设备45电流"] = msg.sternFTDevice45Current / 10.0
bus["current"]["TYZ备用开关电流"] = msg.tyzReservedSwitchCurrent / 10.0
bus["current"]["艉舵机开关1电流"] = msg.sternRudderSwitch1Current / 10.0
bus["current"]["备用电流1"] = msg.reservedCurrent1 / 10.0
bus["current"]["艉舵机开关2电流"] = msg.sternRudderSwitch2Current / 10.0
bus["current"]["备用电流2"] = msg.reservedCurrent2 / 10.0
bus["current"]["FB备用开关电流"] = msg.fbReservedSwitchCurrent / 10.0
bus["current"]["备用电流3"] = msg.reservedCurrent3 / 10.0
bus["breakers"]["燃料电池断路器"] = msg.fuelCellCircuitBreaker
bus["breakers"]["动力锂电池断路器"] = msg.powerLithiumBatteryCircuitBreaker
bus["breakers"]["推进电机断路器"] = msg.propulsionMotorCircuitBreaker
bus["breakers"]["锂电池组仪表断路器"] = msg.lithiumBatteryGroupInstrumentCircuitBreaker
bus["breakers"]["DC/DC模块断路器"] = msg.dcDc5ModuleCircuitBreaker
bus["breakers"]["艏高压配电盒断路器"] = msg.bowHighVoltageDistributionBoxCircuitBreaker
bus["breakers"]["艉FT设备45断路器"] = msg.sternFTDevice45CircuitBreaker
bus["breakers"]["艉舵机开关1断路器"] = msg.sternRudderSwitch1CircuitBreaker
bus["breakers"]["艉舵机开关2断路器"] = msg.sternRudderSwitch2CircuitBreaker
bus["breakers"]["FB备用断路器"] = msg.fbReservedCircuitBreaker
bus["breakers"]["TYZ备用断路器"] = msg.tyzReservedCircuitBreaker
bus["breakers"]["备用断路器"] = msg.reservedCircuitBreaker
bus["faults"]["DC/DC故障1"] = msg.dcDcFaultWord1
bus["faults"]["DC/DC故障2"] = msg.dcDcFaultWord2
if msg.powerBusInsulationStatus == 0x55:
bus["insulation"]["动力母线绝缘状态"] = 0
else:
bus["insulation"]["动力母线绝缘状态"] = 1
if msg.aBusInsulationStatus == 0x55:
bus["insulation"]["A母线绝缘状态"] = 0
else:
bus["insulation"]["A母线绝缘状态"] = 1
bus["alarms"]["应急电源丢失信号"] = msg.emergencyPowerLossSignal
bus["alarms"]["仪表电源丢失信号"] = msg.meterPowerLossSignal
bus["cool_water_pressure"] = msg.coolWaterPressure / 10.0
bus["dcdc_temp"] = msg.dcdcTemperature-40
self.status_data["high_vol_bus"] = bus
self.log_message("收到高压母线状态更新")
elif msg_id == 0x0006: # 高压A母线状态
msg = msg_disHighAVolBusFbMsg()
if msg.unpack(data):
self.db.insertHighAVolBusData(msg)
bus = self.status_data["high_a_vol_bus"]
bus["voltage"]["B母线电压"] = msg.busbarBVoltage / 10.0
bus["current"]["B母线电流"] = msg.busbarBCurrent / 10.0
bus["current"]["艏FT设备123电流"] = msg.bowFTDevice123Current / 10.0
bus["current"]["执行机构电流"] = msg.actuatorCurrent / 10.0
bus["current"]["桅舵机控制箱电流"] = msg.mastSteeringGearControlBoxCurrent / 10.0
bus["current"]["XCZ电流"] = msg.xczCurrent / 10.0
bus["current"]["艏舵控制箱电流"] = msg.bowRudderControlBoxCurrent / 10.0
bus["current"]["开盖启动气缸电流"] = msg.openWaterCoverStartCylinderCurrent / 10.0
bus["breakers"]["艏FT设备123断路器"] = msg.bowFTDevice123CircuitBreaker
bus["breakers"]["执行机构断路器"] = msg.actuatorCircuitBreaker
bus["breakers"]["桅舵机控制箱断路器"] = msg.mastSteeringGearControlBoxCircuitBreaker
bus["breakers"]["XCZ断路器"] = msg.xczCircuitBreaker
bus["breakers"]["艏舵控制箱断路器"] = msg.bowRudderControlBoxCircuitBreaker
bus["breakers"]["开盖启动气缸断路器"] = msg.openWaterCoverStartCylinderCircuitBreaker
bus["alarms"]["仪表电源故障"] = msg.instrumentPowerFailureSignal
bus["alarms"]["应急电源故障"] = msg.emergencyPowerFailureSignal
bus["alarms"]["进水报警"] = msg.waterIngressionAlarm
self.log_message("收到高压A母线状态更新")
elif msg_id == 0x0007: # 仪表主母线状态
msg = msg_disLowMainBusFbMsg()
if msg.unpack(data):
self.db.insertLowMainBusData(msg)
bus = self.status_data["low_main_bus"]
bus["voltage"]["仪表母线电压"] = msg.instrumentBusbarVoltage / 10.0
bus["voltage"]["应急2母线电压"] = msg.emergency2BusbarVoltage / 10.0
bus["current"]["艏低压配电盒电流"] = msg.bowLowVoltageDistributionBoxCurrent / 10.0
bus["current"]["DC/C1仪表48V电流"] = msg.dcC1InstrumentDC48VCurrent / 10.0
bus["current"]["备用电流1"] = msg.reservedCurrent1 / 10.0
bus["current"]["应急锂电池组2电流"] = msg.emergencyLithiumBatteryGroup2Current / 10.0
bus["current"]["锂电池组仪表电流"] = msg.lithiumBatteryGroupInstrumentCurrent / 10.0
bus["current"]["单元4仪表48V电流"] = msg.unit4InstrumentDC48VCurrent / 10.0
bus["current"]["备用电流2"] = msg.reservedCurrent2 / 10.0
bus["current"]["综合能量管理系统应急DC48V电流"] = msg.compositeEnergyManagementSystemEmergencyDC48VCurrent / 10.0
bus["current"]["燃料电池安全系统应急DC48V电流"] = msg.fuelCellSecuritySystemEmergencyDC48VCurrent / 10.0
bus["current"]["PLC控制电源电流"] = msg.plcControlPowerCurrent / 10.0
bus["breakers"]["锂电池组仪表断路器"] = msg.lithiumBatteryGroupInstrumentCircuitBreaker
bus["breakers"]["艏低压配电盒断路器"] = msg.bowLowVoltageDistributionBoxCircuitBreaker
bus["breakers"]["单元4仪表48V断路器"] = msg.unit4InstrumentDC48VCircuitBreaker
bus["breakers"]["DC/C1直流配电板断路器"] = msg.dcC1DCDistributionPanelCircuitBreaker
bus["breakers"]["备用断路器1"] = msg.reservedCircuitBreaker1
bus["breakers"]["备用断路器2"] = msg.reservedCircuitBreaker2
bus["breakers"]["应急锂电池组2断路器"] = msg.emergencyLithiumBatteryGroup2CircuitBreaker
# 0x55 正常
# 0xaa 故障
bus["alarms"]["仪表电源故障"] = msg.instrumentPowerFailureSignal
bus["alarms"]["应急电源故障"] = msg.emergencyPowerFailureSignal
bus["insulation"] = msg.instrumentBusbarInsulationLow
self.log_message("收到仪表主母线状态更新")
else:
print("仪表主母线状态","解析失败")
elif msg_id == 0x0008: # 仪表母线状态
msg = msg_disLowBusFbMsg()
if msg.unpack(data):
self.db.insertLowBusData(msg)
bus = self.status_data["low_bus"]
bus["voltage"]["仪表母线电压"] = msg.instrumentBusbarVoltage / 10.0
bus["voltage"]["应急1母线电压"] = msg.emergency1BusbarVoltage / 10.0
bus["current"]["仪表母线电流"] = msg.instrumentBusbarCurrent / 10.0
bus["current"]["单元1电流"] = msg.unit1Current / 10.0
bus["current"]["单元2电流"] = msg.unit2Current / 10.0
bus["current"]["单元3电流"] = msg.unit3Current / 10.0
bus["current"]["单元5电流"] = msg.unit5Current / 10.0
bus["current"]["艏PZ设备电流"] = msg.bowPZDeviceCurrent / 10.0
bus["current"]["应急锂电池组1电流"] = msg.emergencyLithiumBatteryGroup1Current / 10.0
bus["breakers"]["单元1断路器"] = msg.unit1CircuitBreaker
bus["breakers"]["单元2断路器"] = msg.unit2CircuitBreaker
bus["breakers"]["单元3断路器"] = msg.unit3CircuitBreaker
bus["breakers"]["单元5断路器"] = msg.unit5CircuitBreaker
bus["breakers"]["艏PZ设备断路器"] = msg.bowPZDeviceCircuitBreaker
bus["breakers"]["备用断路器1"] = msg.reservedCircuitBreaker1
bus["breakers"]["备用断路器2"] = msg.reservedCircuitBreaker2
bus["breakers"]["应急锂电池组1断路器"] = msg.emergencyLithiumBatteryGroup1CircuitBreaker
bus["alarms"]["仪表电源故障"] = msg.instrumentPowerFailureSignal
bus["alarms"]["应急电源故障"] = msg.emergencyPowerFailureSignal
bus["alarms"]["进水报警"] = msg.waterIngressionAlarm
self.log_message("收到仪表母线状态更新")
else:
print("仪表母线状态","解析失败")
else:
self.log_message("消息解析失败")
print('消息处理完成')
except Exception as e:
self.log_message(f"消息处理错误: {str(e)}")
def setup_low_main_ui(self, parent):
"""设置仪表主母线状态界面"""
frame = ttk.LabelFrame(parent, text="仪表主母线状态")
frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
# 使用网格布局管理器 - 调整为3列
frame.grid_columnconfigure(0, weight=1)
frame.grid_columnconfigure(1, weight=1)
frame.grid_columnconfigure(2, weight=1)
# 电压与断路器统一显示区域
status_frame = ttk.LabelFrame(frame, text="电压与断路器状态", padding=5)
status_frame.grid(row=0, column=0, columnspan=3, sticky=tk.W + tk.E, padx=5, pady=5)
# 电压显示(横向布局)
voltage_frame = ttk.Frame(status_frame)
voltage_frame.grid(row=0, column=0, padx=10, pady=4, sticky=tk.W)
ttk.Label(voltage_frame, text="仪表母线电压(V):", width=20, anchor="w").pack(side=tk.LEFT)
self.low_main_instrument_voltage = ttk.Label(voltage_frame, text="0.0V", width=10, anchor="e", relief="groove")
self.low_main_instrument_voltage.pack(side=tk.RIGHT)
voltage_frame2 = ttk.Frame(status_frame)
voltage_frame2.grid(row=0, column=1, padx=10, pady=4, sticky=tk.W)
ttk.Label(voltage_frame2, text="应急2汇流排电压(V):", width=20, anchor="w").pack(side=tk.LEFT)
self.low_main_emergency2_voltage = ttk.Label(voltage_frame2, text="0.0V", width=10, anchor="e", relief="groove")
self.low_main_emergency2_voltage.pack(side=tk.RIGHT)
# 断路器状态标签
ttk.Label(status_frame, text="断路器状态:", font=("Arial", 10, "bold")).grid(row=1, column=0, columnspan=3, sticky=tk.W, pady=(10, 4))
# 断路器状态显示区域
self.low_main_breakers = {}
breaker_names = list(self.status_data["low_main_bus"]["breakers"].keys())
columns_per_row = 3 # 每行3个断路器显示块
for i, name in enumerate(breaker_names):
row = 2 + i // columns_per_row
col = i % columns_per_row
# 每个断路器显示块:使用Frame带边框
breaker_frame = tk.Frame(status_frame, relief="solid", borderwidth=1, padx=6, pady=2)
breaker_frame.grid(row=row, column=col, padx=6, pady=4, sticky=tk.W)
# 名称标签
ttk.Label(breaker_frame, text=name, width=20, anchor="w").pack(side=tk.LEFT)
# 圆形指示灯
canvas = tk.Canvas(breaker_frame, width=18, height=18, bg="white", highlightthickness=0)
canvas.create_oval(2, 2, 16, 16, fill="white", outline="gray")
canvas.pack(side=tk.RIGHT)
self.low_main_breakers[name] = canvas
# 创建统一的显示区域(LabelFrame)
status_group = ttk.LabelFrame(frame, text="仪表主母线系统状态", padding=5)
status_group.grid(row=1, column=0, columnspan=3, sticky=tk.W + tk.E, padx=5, pady=5)
# 所有参数标签及属性名
status_items = [
("艏部低压配电箱电流(A):", "bow_current"),
("锂电池组(仪表)电流(A):", "battery_current"),
("DC-C1仪表DC48V电流(A):", "dcc1_current"),
("应急锂电池组2电流(A):", "emergency2_current"),
("单元4仪表48V电流(A):", "unit4_current"),
("复合能源管控系统应急DC48V电流(A):", "energy_current"),
("燃料电池安保系统应急DC48V电流(A):", "fuelcell_current"),
("PLC控制电源电流(A):", "plc_current"),
("备用电流1(A):", "reserved1"),
("备用电流2(A):", "reserved2"),
("仪表电源失电信号:", "instrument_fault"),
("应急电源失电信号:", "emergency_fault"),
("仪表汇流排绝缘低:", "insulation")
]
# 每行显示3列
columns_per_row = 3
# 布局显示
for i, (label_text, varname) in enumerate(status_items):
row = i // columns_per_row
col = i % columns_per_row
# 每个参数一个子Frame(横向排列)
item_frame = ttk.Frame(status_group)
item_frame.grid(row=row, column=col, sticky=tk.W, padx=8, pady=3)
ttk.Label(item_frame, text=label_text, width=26, anchor="w").pack(side=tk.LEFT)
label_var = ttk.Label(item_frame, text="0.0", width=10, anchor="e", relief="groove")
label_var.pack(side=tk.RIGHT)
# 保存变量引用
setattr(self, f"low_main_{varname}", label_var)
return frame
def setup_low_bus_ui(self, parent):
"""设置仪表母线状态界面"""
frame = ttk.LabelFrame(parent, text="仪表母线状态")
frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
# 使用网格布局管理器 - 调整为3列
frame.grid_columnconfigure(0, weight=1)
frame.grid_columnconfigure(1, weight=1)
frame.grid_columnconfigure(2, weight=1)
# 电压与断路器统一显示区域
status_frame = ttk.LabelFrame(frame, text="电压与断路器状态", padding=5)
status_frame.grid(row=0, column=0, columnspan=3, sticky=tk.W + tk.E, padx=5, pady=5)
# 电压显示(横向布局)
voltage_frame = ttk.Frame(status_frame)
voltage_frame.grid(row=0, column=0, padx=10, pady=4, sticky=tk.W)
ttk.Label(voltage_frame, text="仪表母排电压(V):", width=20, anchor="w").pack(side=tk.LEFT)
self.low_bus_instrument_voltage = ttk.Label(voltage_frame, text="0.0V", width=10, anchor="e", relief="groove")
self.low_bus_instrument_voltage.pack(side=tk.RIGHT)
voltage_frame2 = ttk.Frame(status_frame)
voltage_frame2.grid(row=0, column=1, padx=10, pady=4, sticky=tk.W)
ttk.Label(voltage_frame2, text="应急1母线电压(V):", width=20, anchor="w").pack(side=tk.LEFT)
self.low_bus_emergency1_voltage = ttk.Label(voltage_frame2, text="0.0V", width=10, anchor="e", relief="groove")
self.low_bus_emergency1_voltage.pack(side=tk.RIGHT)
# 断路器状态标签
ttk.Label(status_frame, text="断路器状态:", font=("Arial", 10, "bold")).grid(row=1, column=0, columnspan=3, sticky=tk.W, pady=(10, 4))
# 断路器状态显示区域
self.low_bus_breakers = {}
breaker_names = list(self.status_data["low_bus"]["breakers"].keys())
columns_per_row = 3 # 每行3个断路器显示块
for i, name in enumerate(breaker_names):
row = 2 + i // columns_per_row
col = i % columns_per_row
# 每个断路器显示块:使用Frame带边框
breaker_frame = tk.Frame(status_frame, relief="solid", borderwidth=1, padx=6, pady=2)
breaker_frame.grid(row=row, column=col, padx=6, pady=4, sticky=tk.W)
# 名称标签
ttk.Label(breaker_frame, text=name, width=20, anchor="w").pack(side=tk.LEFT)
# 圆形指示灯
canvas = tk.Canvas(breaker_frame, width=18, height=18, bg="white", highlightthickness=0)
canvas.create_oval(2, 2, 16, 16, fill="white", outline="gray")
canvas.pack(side=tk.RIGHT)
self.low_bus_breakers[name] = canvas
# 创建统一的显示区域(LabelFrame)
status_group = ttk.LabelFrame(frame, text="仪表母线系统状态", padding=5)
status_group.grid(row=1, column=0, columnspan=3, sticky=tk.W + tk.E, padx=5, pady=5)
# 所有参数标签及属性名
status_items = [
("仪表母排电流(A):", "main_current"),
("单元1电流(A):", "unit1_current"),
("单元2电流(A):", "unit2_current"),
("单元3电流(A):", "unit3_current"),
("单元5电流(A):", "unit5_current"),
("艏PZ设备电流(A):", "bow_pz_current"),
("应急锂电池组1电流(A):", "emergency1_current"),
("仪表电失电信号:", "power_fault"),
("应急电失电信号:", "emgpower_fault"),
("浸水报警:", "water_alarm")
]
# 每行显示3列
columns_per_row = 3
# 布局显示
for i, (label_text, varname) in enumerate(status_items):
row = i // columns_per_row
col = i % columns_per_row
# 每个参数一个子Frame(横向排列)
item_frame = ttk.Frame(status_group)
item_frame.grid(row=row, column=col, sticky=tk.W, padx=8, pady=3)
ttk.Label(item_frame, text=label_text, width=26, anchor="w").pack(side=tk.LEFT)
label_var = ttk.Label(item_frame, text="0.0", width=10, anchor="e", relief="groove")
label_var.pack(side=tk.RIGHT)
# 保存变量引用
setattr(self, f"low_bus_{varname}", label_var)
return frame
def send_udp_command(self, data, ip, port):
"""发送UDP命令"""
if not self.Start_Connection.get():
return
try:
# self.sock.sendto(data, (self.udp_ip, self.udp_port + 1)
self.sock.sendto(data, (ip, port))
print(f"发送命令到 {ip}:{port} ", data.hex())
except Exception as e:
self.log_message(f"发送命令失败: {str(e)}")
def updata_dcdc_status(self,status):
# 更新DC/DC状态 (按位检查)
dcdc_fault1 = status['faults']['DC/DC故障1']
dcdc_fault2 = status['faults']['DC/DC故障2']
for i in range(8):
# 故障位为1时显示红色,否则绿色
color1 = 'red' if (dcdc_fault1 & (1 << i)) else 'green'
color2 = 'red' if (dcdc_fault2 & (1 << i)) else 'green'
self.high_vol_dcdc_fault1_bits[i].config(bg=color1)
self.high_vol_dcdc_fault2_bits[i].config(bg=color2)
def update_high_vol_ui(self, status):
"""更新高压母线UI显示"""
# 更新电压显示
self.high_vol_power_voltage.config(text=f"{status['voltage']['动力母线电压']:.1f} V")
self.high_vol_a_voltage.config(text=f"{status['voltage']['A母线电压']:.1f} V")
# 更新所有电流值显示
current_vars = [
('powerBusCurrent', '动力母线电流'),
('busbarACurrent', 'A母线电流'),
('dcDc5ModuleCurrent', 'DC/DC5模块电流'),
('propulsionMotorControlBoxCurrent', '推进电机控制箱电流'),
('lithiumBatteryGroupMeterCurrent', '锂电池组仪表电流'),
('bowHighVoltageDistributionBoxCurrent', '艏高压配电盒电流'),
('sternFTDevice45Current', '艉FT设备45电流'),
('tyzReservedSwitchCurrent', 'TYZ备用开关电流'),
('sternRudderSwitch1Current', '艉舵机开关1电流'),
('sternRudderSwitch2Current', '艉舵机开关2电流'),
('fbReservedSwitchCurrent', 'FB备用开关电流'),
('reservedCurrent1', '备用电流1'),
('reservedCurrent2', '备用电流2'),
('reservedCurrent3', '备用电流3')
]
for var_name, data_key in current_vars:
label = getattr(self, f"high_vol_{var_name}")
label.config(text=f"{status['current'][data_key]:.1f} A")
# 更新断路器状态 (0x55:合闸-绿色, 0xAA:分闸-黑色, 0x5A:故障-红色)
for name, state in status['breakers'].items():
if name in self.high_vol_breakers:
canvas = self.high_vol_breakers[name]
if state == 0x55:
color = "green" # 合闸
elif state == 0xAA:
color = "black" # 分闸
elif state == 0x5A:
color = "red" # 故障
else:
color = "white" # 未知状态
canvas.delete("all")
canvas.create_oval(2, 2, 18, 18, fill=color, outline="black")
# 更新绝缘状态 (0x55:正常-绿色, 其他:故障-红色)
power_insul_ok = status['insulation']['动力母线绝缘状态'] == 0
a_insul_ok = status['insulation']['A母线绝缘状态'] == 0
self.high_vol_power_insulation.config(
text="正常" if power_insul_ok else "故障",
foreground="green" if power_insul_ok else "red"
)
self.high_vol_a_insulation.config(
text="正常" if a_insul_ok else "故障",
foreground="green" if a_insul_ok else "red"
)
# 更新冷却水压力 (显示为Kpa)
self.high_vol_cool_water_pressure.config(
text=f"{status['cool_water_pressure']:.1f} Kpa"
)
# 更新DC/DC温度 (显示为摄氏度)
self.high_vol_dcdc_temp.config(
text=f"{status['dcdc_temp']} ℃"
)
# 更新失电故障显示 (0x55:正常, 其他:故障)
self.high_vol_losspower_insulation.config(
text="正常" if status['alarms']['仪表电源丢失信号'] == 0x55 else "故障"
)
self.high_vol_loss_emerpower_insulation.config(
text="正常" if status['alarms']['应急电源丢失信号'] == 0x55 else "故障"
)
def update_ui(self):
"""更新界面状态"""
# 更新高压母线状态
high_vol = self.status_data["high_vol_bus"]
self.update_high_vol_ui(high_vol)
# 更新高压A母线状态
high_a_vol = self.status_data["high_a_vol_bus"]
self.high_a_vol_voltage.config(text=f"{high_a_vol['voltage']['B母线电压']:.1f} V")
self.high_a_vol_current.config(text=f"{high_a_vol['current']['B母线电流']:.1f} A")
self.high_a_vol_bow_ft_current.config(text=f"{high_a_vol['current']['艏FT设备123电流']:.1f} A")
self.high_a_vol_actuator_current.config(text=f"{high_a_vol['current']['执行机构电流']:.1f} A")
# 更新新增的电流显示项
self.high_a_vol_mast_steering_current.config(text=f"{high_a_vol['current']['桅舵机控制箱电流']:.1f} A")
self.high_a_vol_xcz_current.config(text=f"{high_a_vol['current']['XCZ电流']:.1f} A")
self.high_a_vol_bow_rudder_current.config(text=f"{high_a_vol['current']['艏舵控制箱电流']:.1f} A")
self.high_a_vol_open_cover_current.config(text=f"{high_a_vol['current']['开盖启动气缸电流']:.1f} A")
for name, canvas in self.high_a_vol_breakers.items():
state = high_a_vol["breakers"].get(name, 0)
if state == 0x55:
color = "green"
elif state == 0xAA:
color = "black"
elif state == 0x5A:
color = "red"
else:
color = "white"
# color = "green" if state else "red"
canvas.create_oval(2, 2, 18, 18, fill=color, outline="black")
# 更新仪表主母线状态
low_main = self.status_data["low_main_bus"]
for name, canvas in self.low_main_breakers.items():
state = low_main["breakers"].get(name, 0)
# color = "green" if state else "red"
if state == 0x55:
color = "green"
elif state == 0xAA:
color = "black"
elif state == 0x5A:
color = "red"
else:
color = "white"
canvas.create_oval(2, 2, 18, 18, fill=color, outline="black")
# 更新仪表母线状态
low_bus = self.status_data["low_bus"]
for name, canvas in self.low_bus_breakers.items():
state = low_bus["breakers"].get(name, 0)
# color = "green" if state else "red"
if state == 0x55:
color = "green"
elif state == 0xAA:
color = "black"
elif state == 0x5A:
color = "red"
else:
color = "white"
canvas.create_oval(2, 2, 18, 18, fill=color, outline="black")
self.high_a_vol_instrument_power_fault.config(
text="正常" if high_a_vol["alarms"]["仪表电源故障"]==0x55 else "故障"
)
self.high_a_vol_emergency_power_fault.config(
text="正常" if high_a_vol["alarms"]["应急电源故障"]==0x55 else "故障"
)
self.high_a_vol_water_alarm.config(
text="正常" if high_a_vol["alarms"]["进水报警"]==0x55 else "故障"
)
# 更新仪表主母线状态
low_main = self.status_data["low_main_bus"]
self.low_main_instrument_voltage.config(text=f"{low_main['voltage']['仪表母线电压']:.1f} V")
self.low_main_emergency2_voltage.config(text=f"{low_main['voltage']['应急2母线电压']:.1f} V")
self.low_main_bow_current.config(text=f"{low_main['current']['艏低压配电盒电流']:.1f} A")
self.low_main_battery_current.config(text=f"{low_main['current']['锂电池组仪表电流']:.1f} A")
# 更新新增的电流显示项
self.low_main_dcc1_current.config(text=f"{low_main['current']['DC/C1仪表48V电流']:.1f} A")
self.low_main_emergency2_current.config(text=f"{low_main['current']['应急锂电池组2电流']:.1f} A")
self.low_main_unit4_current.config(text=f"{low_main['current']['单元4仪表48V电流']:.1f} A")
self.low_main_energy_current.config(text=f"{low_main['current']['综合能量管理系统应急DC48V电流']:.1f} A")
self.low_main_fuelcell_current.config(text=f"{low_main['current']['燃料电池安全系统应急DC48V电流']:.1f} A")
self.low_main_plc_current.config(text=f"{low_main['current']['PLC控制电源电流']:.1f} A")
# 新增加备用电流
self.low_main_reserved1.config(text=f"{low_main['current']['备用电流1']:.1f} A")
self.low_main_reserved2.config(text=f"{low_main['current']['备用电流2']:.1f} A")
for name, canvas in self.low_main_breakers.items():
state = low_main["breakers"].get(name, 0)
# color = "green" if state else "red"
if state == 0x55:
color = "green"
elif state == 0xAA:
color = "black"
elif state == 0x5A:
color = "red"
else:
color = "white"
canvas.create_oval(2, 2, 18, 18, fill=color, outline="black")
self.low_main_instrument_fault.config(
text="正常" if low_main["alarms"]["仪表电源故障"]==0x55 else "故障"
)
self.low_main_emergency_fault.config(
text="正常" if low_main["alarms"]["应急电源故障"]==0x55 else "故障"
)
self.low_main_insulation.config(
text="正常" if low_main["insulation"]==0x55 else "故障"
)
# 更新仪表母线状态
low_bus = self.status_data["low_bus"]
self.low_bus_instrument_voltage.config(text=f"{low_bus['voltage']['仪表母线电压']:.1f} V")
self.low_bus_emergency1_voltage.config(text=f"{low_bus['voltage']['应急1母线电压']:.1f} V")
self.low_bus_unit1_current.config(text=f"{low_bus['current']['单元1电流']:.1f} A")
self.low_bus_unit2_current.config(text=f"{low_bus['current']['单元2电流']:.1f} A")
# 更新新增的电流显示项
self.low_bus_main_current.config(text=f"{low_bus['current']['仪表母线电流']:.1f} A")
self.low_bus_unit3_current.config(text=f"{low_bus['current']['单元3电流']:.1f} A")
self.low_bus_unit5_current.config(text=f"{low_bus['current']['单元5电流']:.1f} A")
self.low_bus_bow_pz_current.config(text=f"{low_bus['current']['艏PZ设备电流']:.1f} A")
self.low_bus_emergency1_current.config(text=f"{low_bus['current']['应急锂电池组1电流']:.1f} A")
for name, canvas in self.low_bus_breakers.items():
state = low_bus["breakers"].get(name, 0)
# color = "green" if state else "red"
if state == 0x55:
color = "green"
elif state == 0xAA:
color = "black"
elif state == 0x5A:
color = "red"
else:
color = "white"
canvas.create_oval(2, 2, 18, 18, fill=color, outline="black")
self.low_bus_power_fault.config(
text="正常" if low_bus["alarms"]["仪表电源故障"]==0x55 else "故障"
)
self.low_bus_emgpower_fault.config(
text="正常" if low_bus["alarms"]["应急电源故障"]==0x55 else "故障"
)
self.low_bus_water_alarm.config(
text="正常" if low_bus["alarms"]["进水报警"]==0x55 else "故障"
)
# print('update_ui')
# 每500ms更新一次
self.root.after(1000, self.update_ui)
def log_message(self, message):
"""记录日志消息"""
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
self.log_text.config(state=tk.NORMAL)
self.log_text.insert(tk.END, f"[{timestamp}] {message}\n")
# 获取当前所有行数
line_count = int(self.log_text.index('end-1c').split('.')[0])
# 如果超过50行,删除最旧的行
if line_count > 50:
self.log_text.delete(1.0, 2.0)
self.log_text.config(state=tk.DISABLED)
self.log_text.see(tk.END)
def on_simulation_mode_changed(self, *args):
"""模拟模式复选框回调"""
if self.Simulation_Mode.get():
self.start_simulation()
else:
self.stop_simulation()
def start_simulation(self):
"""启动模拟线程"""
if not self.simulation_running:
self.simulation_running = True
self.simulation_thread = threading.Thread(target=self.simulation_loop)
self.simulation_thread.daemon = True
self.simulation_thread.start()
self.log_message("模拟模式已启动")
def stop_simulation(self):
"""停止模拟线程"""
if self.simulation_running:
self.simulation_running = False
if self.simulation_thread:
self.simulation_thread.join()
self.log_message("模拟模式已停止")
def simulation_loop(self):
"""模拟线程主循环"""
self.simulation_HighVolBusFbMsg.dcDcFaultWord1 = 0x01
self.simulation_HighVolBusFbMsg.dcDcFaultWord2 = 0x01
while self.simulation_running:
# 发送模拟数据
self.simulation_HighVolBusFbMsg.fuelCellCircuitBreaker = self.simulation_HVBCmdMsg.fuelCellCircuitBreaker
self.simulation_HighVolBusFbMsg.powerLithiumBatteryCircuitBreaker = self.simulation_HVBCmdMsg.powerLithiumBatteryCircuitBreaker
self.simulation_HighVolBusFbMsg.propulsionMotorCircuitBreaker = self.simulation_HVBCmdMsg.propulsionMotorCircuitBreaker
self.simulation_HighVolBusFbMsg.lithiumBatteryGroupInstrumentCircuitBreaker = self.simulation_HVBCmdMsg.lithiumBatteryGroupInstrumentCircuitBreaker
self.simulation_HighVolBusFbMsg.dcDc5ModuleCircuitBreaker = self.simulation_HVBCmdMsg.dcDc5ModuleCircuitBreaker
self.simulation_HighVolBusFbMsg.bowHighVoltageDistributionBoxCircuitBreaker = self.simulation_HVBCmdMsg.bowHighVoltageDistributionBoxCircuitBreaker
self.simulation_HighVolBusFbMsg.sternFTDevice45CircuitBreaker = self.simulation_HVBCmdMsg.sternFTDevice45CircuitBreaker
self.simulation_HighVolBusFbMsg.sternRudderSwitch1CircuitBreaker = self.simulation_HVBCmdMsg.sternRudderSwitch1CircuitBreaker
self.simulation_HighVolBusFbMsg.sternRudderSwitch2CircuitBreaker = self.simulation_HVBCmdMsg.sternRudderSwitch2CircuitBreaker
self.simulation_HighVolBusFbMsg.fbReservedCircuitBreaker = self.simulation_HVBCmdMsg.fbReservedCircuitBreaker
self.simulation_HighVolBusFbMsg.tyzReservedCircuitBreaker = self.simulation_HVBCmdMsg.tyzReservedCircuitBreaker
self.simulation_HighVolBusFbMsg.reservedCircuitBreaker = self.simulation_HVBCmdMsg.reservedCircuitBreaker
# dcdc状态字按位进行循环偏移
self.simulation_HighVolBusFbMsg.dcDcFaultWord1 = ((self.simulation_HighVolBusFbMsg.dcDcFaultWord1 << 1) | (self.simulation_HighVolBusFbMsg.dcDcFaultWord1 >> 7)) & 0xFF
self.simulation_HighVolBusFbMsg.dcDcFaultWord2 = ((self.simulation_HighVolBusFbMsg.dcDcFaultWord2 >> 1) | (self.simulation_HighVolBusFbMsg.dcDcFaultWord2 << 7)) & 0xFF
# 故障往复切换
self.simulation_HighAVolBusFbMsg.actuatorCircuitBreaker = self.simulation_HVBACmdMsg.actuatorCircuitBreaker
self.simulation_HighAVolBusFbMsg.bowFTDevice123CircuitBreaker = self.simulation_HVBACmdMsg.bowFTDevice123CircuitBreaker
self.simulation_HighAVolBusFbMsg.mastSteeringGearControlBoxCircuitBreaker = self.simulation_HVBACmdMsg.mastSteeringGearControlBoxCircuitBreaker
self.simulation_HighAVolBusFbMsg.xczCircuitBreaker = self.simulation_HVBACmdMsg.xczCircuitBreaker
self.simulation_HighAVolBusFbMsg.bowRudderControlBoxCircuitBreaker = self.simulation_HVBACmdMsg.bowRudderControlBoxCircuitBreaker
self.simulation_HighAVolBusFbMsg.openWaterCoverStartCylinderCircuitBreaker = self.simulation_HVBACmdMsg.openWaterCoverStartCylinderCircuitBreaker
self.simulation_LowMainBusFbMsg.lithiumBatteryGroupInstrumentCircuitBreaker = self.simulation_LVMBCmdMsg.lithiumBatteryGroupInstrumentCircuitBreaker
self.simulation_LowMainBusFbMsg.bowLowVoltageDistributionBoxCircuitBreaker = self.simulation_LVMBCmdMsg.bowLowVoltageDistributionBoxCircuitBreaker
self.simulation_LowMainBusFbMsg.unit4InstrumentDC48VCircuitBreaker = self.simulation_LVMBCmdMsg.unit4InstrumentDC48VCircuitBreaker
self.simulation_LowMainBusFbMsg.dcC1DCDistributionPanelCircuitBreaker = self.simulation_LVMBCmdMsg.dcC1DCDistributionPanelCircuitBreaker
self.simulation_LowMainBusFbMsg.reservedCircuitBreaker1 = self.simulation_LVMBCmdMsg.reservedCircuitBreaker1
self.simulation_LowMainBusFbMsg.reservedCircuitBreaker2 = self.simulation_LVMBCmdMsg.reservedCircuitBreaker2
self.simulation_LowMainBusFbMsg.emergencyLithiumBatteryGroup2CircuitBreaker = self.simulation_LVMBCmdMsg.emergencyLithiumBatteryGroup2CircuitBreaker
self.simulation_LowBusFbMsg.unit1CircuitBreaker = self.simulation_LVBCmdMsg.unit1CircuitBreaker
self.simulation_LowBusFbMsg.unit2CircuitBreaker = self.simulation_LVBCmdMsg.unit2CircuitBreaker
self.simulation_LowBusFbMsg.unit3CircuitBreaker = self.simulation_LVBCmdMsg.unit3CircuitBreaker
self.simulation_LowBusFbMsg.unit5CircuitBreaker = self.simulation_LVBCmdMsg.unit5CircuitBreaker
self.simulation_LowBusFbMsg.bowPZDeviceCircuitBreaker = self.simulation_LVBCmdMsg.bowPZDeviceCircuitBreaker
self.simulation_LowBusFbMsg.reservedCircuitBreaker1 = self.simulation_LVBCmdMsg.reservedCircuitBreaker1
self.simulation_LowBusFbMsg.reservedCircuitBreaker2 = self.simulation_LVBCmdMsg.reservedCircuitBreaker2
self.simulation_LowBusFbMsg.emergencyLithiumBatteryGroup1CircuitBreaker = self.simulation_LVBCmdMsg.emergencyLithiumBatteryGroup1CircuitBreaker
# self.send_udp_command(self.simulation_HighVolBusFbMsg.pack(), self.udp_ip, self.udp_control_port)
# time.sleep(0.1)
# self.send_udp_command(self.simulation_HighAVolBusFbMsg.pack(), self.udp_ip, self.udp_control_port)
# time.sleep(0.1)
# self.send_udp_command(self.simulation_LowBusFbMsg.pack(), self.udp_ip, self.udp_control_port)
# time.sleep(0.1)
# self.send_udp_command(self.simulation_LowMainBusFbMsg.pack(), self.udp_ip, self.udp_control_port)
# time.sleep(0.1) # 1秒间隔
# 处理高压母线数据
packed_data = self.simulation_HighVolBusFbMsg.pack()
# error_packed_data = packed_data[:-1] + b'\x00' # 修改校验位为0x00
self.send_udp_command(packed_data, self.udp_ip, self.udp_control_port)
time.sleep(0.01)
# 处理高压A母线数据
packed_data = self.simulation_HighAVolBusFbMsg.pack()
# error_packed_data = packed_data[:-1] + b'\x00' # 修改校验位为0x00
self.send_udp_command(packed_data, self.udp_ip, self.udp_control_port)
time.sleep(0.01)
# 处理低压母线数据
packed_data = self.simulation_LowBusFbMsg.pack()
# error_packed_data = packed_data[:-1] + b'\x00' # 修改校验位为0x00
self.send_udp_command(packed_data, self.udp_ip, self.udp_control_port)
time.sleep(0.01)
# 处理仪表主母线数据
packed_data = self.simulation_LowMainBusFbMsg.pack()
# error_packed_data = packed_data[:-1] + b'\x00' # 修改校验位为0x00
self.send_udp_command(packed_data, self.udp_ip, self.udp_control_port)
time.sleep(0.01)
packed_data = self.simulation_ccUFbMsg.pack()
self.send_udp_command(packed_data, self.udp_ip, self.udp_control_port)
time.sleep(0.01)
time.sleep(0.7) # 循环间隔
def on_connection_changed(self, *args):
"""连接状态改变回调"""
if self.Start_Connection.get():
self.running = True
if not self.recv_thread.is_alive():
self.recv_thread.start()
self.log_message("通信连接已启动")
else:
self.running = False
self.log_message("通信连接已停止")
def on_closing(self):
"""窗口关闭事件处理"""
self.running = False
self.stop_simulation()
self.sock.close()
self.root.destroy()
def show_popup_window(self, title):
"""创建并显示弹出窗口"""
popup = tk.Toplevel(self.root)
popup.title(title)
# 根据不同的标题创建对应的操作面板
if title == "动力母线操作":
frame = ttk.Frame(popup)
frame.pack(fill="both", expand=True, padx=5, pady=5)
self.setup_hvb_control(frame)
elif title == "艏部高压操作":
frame = ttk.Frame(popup)
frame.pack(fill="both", expand=True, padx=5, pady=5)
self.setup_hvba_control(frame)
elif title == "仪表母线操作":
frame = ttk.Frame(popup)
frame.pack(fill="both", expand=True, padx=5, pady=5)
self.setup_lvmb_control(frame)
elif title == "艏部低压操作":
frame = ttk.Frame(popup)
frame.pack(fill="both", expand=True, padx=5, pady=5)
self.setup_lvb_control(frame)
elif title == "DCDC状态":
frame = ttk.Frame(popup)
frame.pack(fill="both", expand=True, padx=5, pady=5)
self.setup_dcdc_status(frame)
# 弹窗显示时立即更新一次状态
self.updata_dcdc_status(self.status_data["high_vol_bus"])
# 设置定时器每1秒更新一次状态
def update_dcdc():
if popup.winfo_exists(): # 检查窗口是否还存在
self.updata_dcdc_status(self.status_data["high_vol_bus"])
popup.after(1000, update_dcdc)
popup.after(1000, update_dcdc)
close_button = ttk.Button(popup, text="关闭", command=popup.destroy)
close_button.pack(pady=10)
def show_compound_control_window(self, title):
"""创建并显示弹出窗口"""
popup = tk.Toplevel(self.root)
popup.title(title)
frame = ttk.Frame(popup)
popup.geometry("600x400")
popup.resizable(True, True)
frame.pack(fill="both", expand=True, padx=5, pady=5)
def show_help_window(self, title):
"""显示程序说明弹窗"""
popup = tk.Toplevel(self.root)
popup.title(title)
popup.geometry("600x400")
popup.resizable(True, True)
# 创建文本框和滚动条
frame = ttk.Frame(popup)
frame.pack(fill="both", expand=True, padx=10, pady=10)
text_widget = tk.Text(frame, wrap="word", font=("Microsoft YaHei", 10))
scrollbar = ttk.Scrollbar(frame, orient="vertical", command=text_widget.yview)
text_widget.configure(yscrollcommand=scrollbar.set)
text_widget.pack(side="left", fill="both", expand=True)
scrollbar.pack(side="right", fill="y")
# 程序说明内容
help_text = """H100动力管理系统程序说明
版本:v1.0
功能概述:
1. 高压母线状态监控 - 实时显示动力母线电压、电流及断路器状态
2. 高压A母线状态监控 - 监控汇流排B电压电流及断路器状态
3. 仪表主母线状态监控 - 显示仪表母线电压电流及配电状态
4. 仪表母线状态监控 - 监控应急电源系统状态
5. 配电指令控制 - 支持远程断路器合分操作
6. 数据记录 - 所有状态数据自动保存到SQLite数据库
7. 模拟模式 - 支持离线模拟测试
通信协议:
- 采用UDP协议与分布式系统通信
- 消息ID范围:0x0001-0x0008
- 支持实时状态反馈和指令下发
操作说明:
1. 主界面显示各母线实时状态
2. 通过菜单栏可打开各配电控制面板
3. 断路器状态:黄色=合闸,黑色=分闸,红色=故障
4. 启用模拟模式可进行离线测试
注意事项:
- 操作前请确认系统连接正常
- 重要操作需双重确认
- 定期检查数据库存储状态"""
text_widget.insert("1.0", help_text)
text_widget.config(state="disabled") # 设置为只读
# 关闭按钮
close_btn = ttk.Button(popup, text="关闭", command=popup.destroy)
close_btn.pack(pady=10)
if __name__ == "__main__":
root = tk.Tk()
app = PowerManagerApp(root)
root.protocol("WM_DELETE_WINDOW", app.on_closing)
print('start')
root.mainloop()
print('end')