增加了新的调试界面,优化了一些功能
This commit is contained in:
+319
-6
@@ -2,19 +2,332 @@ import socket
|
||||
import struct
|
||||
import tkinter as tk
|
||||
import threading
|
||||
import random
|
||||
from udpmsg import *
|
||||
from udpmsg import msg_CcuStateFbMsg
|
||||
from datetime import datetime
|
||||
from tkinter import font
|
||||
from guiEle import ElementDisplayApp
|
||||
from until import *
|
||||
from guiInput import InputGUI
|
||||
from tkinter import Frame
|
||||
|
||||
from tkinter.ttk import *
|
||||
|
||||
class ccuSys(Frame):
|
||||
def __init__(self,root):
|
||||
def __init__(self, root):
|
||||
super().__init__(root)
|
||||
# Initialize CCU state
|
||||
self.ccuState = msg_CcuStateFbMsg()
|
||||
self.col_frames = []
|
||||
self.state_entries = []
|
||||
|
||||
# Timer settings
|
||||
self.update_interval = 1000 # Default update interval in milliseconds
|
||||
|
||||
# Network settings
|
||||
self.udp_ip = "127.0.0.1"
|
||||
self.udp_port = 5005
|
||||
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
self.sock.settimeout(1.0)
|
||||
|
||||
# Initialize GUI
|
||||
self.master = root
|
||||
self.createWidgets()
|
||||
|
||||
# Start data update thread
|
||||
self.running = True
|
||||
self.update_thread = threading.Thread(target=self._update_thread)
|
||||
self.update_thread.daemon = True
|
||||
self.update_thread.start()
|
||||
|
||||
# Start timer for periodic display update
|
||||
self.update_timer = None
|
||||
self.start_update_timer()
|
||||
|
||||
def createWidgets(self):
|
||||
pass
|
||||
# Main window configuration
|
||||
# self.master.title("CCU Monitoring System")
|
||||
# self.master.geometry("1200x800")
|
||||
|
||||
# Create status display area
|
||||
self.status_frame = tk.Frame(self.master)
|
||||
self.status_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
|
||||
|
||||
# Create status columns
|
||||
self._create_status_column("FCU Status", [
|
||||
"fcuState", "fcuFaultLeve", "powerLimit",
|
||||
"totalPowerGeneration", "fcuVoltage", "fcuCurrent",
|
||||
"fcuOutPower", "fcuOutIns"
|
||||
])
|
||||
|
||||
self._create_status_column("Battery Status", [
|
||||
"insBatSoc", "powerBatSoc", "insBatLimit",
|
||||
"powerBatLimit", "insBatTotalElectricity",
|
||||
"powerBatTotalElectricity", "insCurrent",
|
||||
"powerCurrent"
|
||||
])
|
||||
|
||||
self._create_status_column("Sensor Status", [
|
||||
"h2Concentration1", "h2Concentration2",
|
||||
"o2Concentration1", "o2Concentration2",
|
||||
"ch4oConcentration1", "ch4oConcentration2",
|
||||
"backpressure1", "backpressure2"
|
||||
])
|
||||
|
||||
self._create_status_column("System Status", [
|
||||
"insPosRelayState", "insNegRelayState",
|
||||
"powerPosRelayState", "powerNegRelayState",
|
||||
"flameDetection", "exchangerTemperature",
|
||||
"insEmergencyStatusWord", "powerEmergencyStatusWord"
|
||||
])
|
||||
|
||||
# Create control panel
|
||||
self.control_frame = tk.Frame(self.master)
|
||||
self.control_frame.pack(fill=tk.X, padx=10, pady=5)
|
||||
|
||||
# Add control buttons
|
||||
self.refresh_btn = tk.Button(
|
||||
self.control_frame, text="Refresh", command=self.update_display
|
||||
)
|
||||
self.refresh_btn.pack(side=tk.LEFT, padx=5)
|
||||
|
||||
self.export_btn = tk.Button(
|
||||
self.control_frame, text="Export Data", command=self.export_data
|
||||
)
|
||||
self.export_btn.pack(side=tk.LEFT, padx=5)
|
||||
|
||||
# Timer control
|
||||
self.timer_frame = tk.Frame(self.control_frame)
|
||||
self.timer_frame.pack(side=tk.RIGHT, padx=10)
|
||||
|
||||
# Timer interval label
|
||||
tk.Label(self.timer_frame, text="Update Interval (ms):").pack(side=tk.LEFT)
|
||||
|
||||
# Timer interval entry
|
||||
self.interval_entry = tk.Entry(self.timer_frame, width=6)
|
||||
self.interval_entry.insert(0, str(self.update_interval))
|
||||
self.interval_entry.pack(side=tk.LEFT, padx=5)
|
||||
|
||||
# Set interval button
|
||||
self.set_interval_btn = tk.Button(
|
||||
self.timer_frame,
|
||||
text="Set",
|
||||
command=self.set_update_interval
|
||||
)
|
||||
self.set_interval_btn.pack(side=tk.LEFT)
|
||||
|
||||
# Status indicator
|
||||
self.status_indicator = tk.Label(
|
||||
self.timer_frame, text="●", fg="green", font=("Arial", 16)
|
||||
)
|
||||
self.status_indicator.pack(side=tk.RIGHT, padx=10)
|
||||
|
||||
# FCU control panel
|
||||
self.fcu_frame = tk.Frame(self.control_frame)
|
||||
self.fcu_frame.pack(side=tk.LEFT, padx=10)
|
||||
|
||||
# FCU state control
|
||||
tk.Label(self.fcu_frame, text="FCU State:").pack(side=tk.LEFT)
|
||||
self.fcu_state_var = tk.IntVar(value=0)
|
||||
self.fcu_state_menu = tk.OptionMenu(
|
||||
self.fcu_frame, self.fcu_state_var,
|
||||
*range(4), command=self.set_fcu_state
|
||||
)
|
||||
self.fcu_state_menu.pack(side=tk.LEFT, padx=5)
|
||||
|
||||
# FCU fault level control
|
||||
tk.Label(self.fcu_frame, text="Fault Level:").pack(side=tk.LEFT)
|
||||
self.fcu_fault_var = tk.IntVar(value=0)
|
||||
self.fcu_fault_menu = tk.OptionMenu(
|
||||
self.fcu_frame, self.fcu_fault_var,
|
||||
*range(4), command=self.set_fcu_fault_level
|
||||
)
|
||||
self.fcu_fault_menu.pack(side=tk.LEFT, padx=5)
|
||||
|
||||
def _create_status_column(self, title, fields):
|
||||
"""Create a status column with specified fields"""
|
||||
col_frame = tk.LabelFrame(
|
||||
self.status_frame, text=title, bd=2, relief=tk.GROOVE
|
||||
)
|
||||
col_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=5, pady=5)
|
||||
|
||||
for field in fields:
|
||||
field_frame = tk.Frame(col_frame)
|
||||
field_frame.pack(fill=tk.X, padx=5, pady=2)
|
||||
|
||||
# Field name
|
||||
name_label = tk.Label(
|
||||
field_frame, text=field, width=20, anchor=tk.W
|
||||
)
|
||||
name_label.pack(side=tk.LEFT)
|
||||
|
||||
# Value display
|
||||
value_entry = tk.Entry(
|
||||
field_frame, width=15, state='readonly'
|
||||
)
|
||||
value_entry.pack(side=tk.RIGHT, fill=tk.X, expand=True)
|
||||
|
||||
# Store reference for updates
|
||||
self.state_entries.append((field, value_entry))
|
||||
|
||||
def generate_simulated_data(self):
|
||||
"""Generate simulated CCU state data"""
|
||||
self.ccuState.fcuState = self.fcu_state_var.get()
|
||||
self.ccuState.fcuFaultLeve = self.fcu_fault_var.get()
|
||||
self.ccuState.powerLimit = random.randint(0, 100)
|
||||
self.ccuState.totalPowerGeneration = int(random.uniform(0, 1000))
|
||||
self.ccuState.fcuVoltage = int(random.uniform(200, 400))
|
||||
self.ccuState.fcuCurrent = int(random.uniform(0, 100))
|
||||
self.ccuState.fcuOutPower = random.randint(0, 1000)
|
||||
self.ccuState.fcuOutIns = random.randint(0, 100)
|
||||
|
||||
self.ccuState.insBatSoc = random.randint(0, 100)
|
||||
self.ccuState.powerBatSoc = random.randint(0, 100)
|
||||
self.ccuState.insBatLimit = random.randint(0, 100)
|
||||
self.ccuState.powerBatLimit = random.randint(0, 100)
|
||||
self.ccuState.insBatTotalElectricity = random.randint(0, 1000)
|
||||
self.ccuState.powerBatTotalElectricity = random.randint(0, 1000)
|
||||
self.ccuState.insCurrent = random.randint(0, 100)
|
||||
self.ccuState.powerCurrent = random.randint(0, 100)
|
||||
|
||||
self.ccuState.h2Concentration1 = random.randint(0, 100)
|
||||
self.ccuState.h2Concentration2 = random.randint(0, 100)
|
||||
self.ccuState.o2Concentration1 = random.randint(0, 100)
|
||||
self.ccuState.o2Concentration2 = random.randint(0, 100)
|
||||
self.ccuState.ch4oConcentration1 = random.randint(0, 100)
|
||||
self.ccuState.ch4oConcentration2 = random.randint(0, 100)
|
||||
self.ccuState.backpressure1 = random.randint(0, 10)
|
||||
self.ccuState.backpressure2 = random.randint(0, 10)
|
||||
|
||||
self.ccuState.insPosRelayState = random.randint(0, 1)
|
||||
self.ccuState.insNegRelayState = random.randint(0, 1)
|
||||
self.ccuState.powerPosRelayState = random.randint(0, 1)
|
||||
self.ccuState.powerNegRelayState = random.randint(0, 1)
|
||||
self.ccuState.flameDetection = random.randint(0, 1)
|
||||
self.ccuState.exchangerTemperature = random.randint(20, 100)
|
||||
self.ccuState.insEmergencyStatusWord = random.randint(0, 255)
|
||||
self.ccuState.powerEmergencyStatusWord = random.randint(0, 255)
|
||||
|
||||
def update_display(self):
|
||||
"""Update all displayed values"""
|
||||
# Generate simulated data if no real data available
|
||||
if not hasattr(self.ccuState, 'fcuState'):
|
||||
self.generate_simulated_data()
|
||||
|
||||
for field, entry in self.state_entries:
|
||||
value = getattr(self.ccuState, field, "N/A")
|
||||
entry.config(state='normal')
|
||||
entry.delete(0, tk.END)
|
||||
entry.insert(0, str(value))
|
||||
entry.config(state='readonly')
|
||||
|
||||
# Update status indicator
|
||||
self._update_status_indicator()
|
||||
|
||||
def _update_status_indicator(self):
|
||||
"""Update the status indicator based on system state"""
|
||||
if self.ccuState.fcuFaultLeve > 0:
|
||||
self.status_indicator.config(fg="red")
|
||||
else:
|
||||
self.status_indicator.config(fg="green")
|
||||
|
||||
def export_data(self):
|
||||
"""Export current status data to file"""
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
filename = f"ccu_status_{timestamp}.log"
|
||||
try:
|
||||
with open(filename, 'w') as f:
|
||||
for field, _ in self.state_entries:
|
||||
value = getattr(self.ccuState, field, "N/A")
|
||||
f.write(f"{field}: {value}\n")
|
||||
print(f"Data exported to {filename}")
|
||||
except Exception as e:
|
||||
print(f"Export failed: {str(e)}")
|
||||
|
||||
def _update_thread(self):
|
||||
"""Thread for updating CCU state"""
|
||||
while self.running:
|
||||
try:
|
||||
# Receive data from UDP
|
||||
data, _ = self.sock.recvfrom(1024)
|
||||
self.ccuState.unpack(data)
|
||||
|
||||
except socket.timeout:
|
||||
continue
|
||||
except Exception as e:
|
||||
print(f"Update error: {str(e)}")
|
||||
break
|
||||
|
||||
def set_update_interval(self):
|
||||
"""Set new update interval from GUI input"""
|
||||
try:
|
||||
new_interval = int(self.interval_entry.get())
|
||||
if new_interval < 100:
|
||||
raise ValueError("Interval must be >= 100ms")
|
||||
self.update_interval = new_interval
|
||||
self.stop_update_timer()
|
||||
self.start_update_timer()
|
||||
except ValueError as e:
|
||||
print(f"Invalid interval: {str(e)}")
|
||||
|
||||
def set_fcu_state(self, state):
|
||||
"""Set FCU state and update display"""
|
||||
self.ccuState.fcuState = state
|
||||
self.update_display()
|
||||
|
||||
def set_fcu_fault_level(self, level):
|
||||
"""Set FCU fault level and update display"""
|
||||
self.ccuState.fcuFaultLeve = level
|
||||
self.update_display()
|
||||
|
||||
def start_update_timer(self):
|
||||
"""Start the periodic display update timer"""
|
||||
if not hasattr(self, 'data_update_thread'):
|
||||
self.data_update_running = True
|
||||
self.data_update_thread = threading.Thread(
|
||||
target=self._data_update_loop
|
||||
)
|
||||
self.data_update_thread.daemon = True
|
||||
self.data_update_thread.start()
|
||||
|
||||
self._update_display_loop()
|
||||
|
||||
def _update_display_loop(self):
|
||||
"""Periodically update the display"""
|
||||
self.generate_simulated_data()
|
||||
self.update_display()
|
||||
self.update_timer = self.master.after(
|
||||
self.update_interval,
|
||||
self._update_display_loop
|
||||
)
|
||||
|
||||
def _data_update_loop(self):
|
||||
"""Periodically update data from network"""
|
||||
while self.data_update_running:
|
||||
try:
|
||||
# Receive data from UDP
|
||||
data, _ = self.sock.recvfrom(1024)
|
||||
self.ccuState.unpack(data)
|
||||
time.sleep(self.update_interval / 1000)
|
||||
except socket.timeout:
|
||||
continue
|
||||
except Exception as e:
|
||||
print(f"Data update error: {str(e)}")
|
||||
break
|
||||
|
||||
def stop_update_timer(self):
|
||||
"""Stop the periodic display update timer"""
|
||||
if self.update_timer:
|
||||
self.master.after_cancel(self.update_timer)
|
||||
self.update_timer = None
|
||||
|
||||
def __del__(self):
|
||||
"""Clean up resources"""
|
||||
self.running = False
|
||||
self.stop_update_timer()
|
||||
if hasattr(self, 'update_thread'):
|
||||
self.update_thread.join()
|
||||
self.sock.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
root = tk.Tk()
|
||||
app = ccuSys(root)
|
||||
app.pack(fill=tk.BOTH, expand=True)
|
||||
root.mainloop()
|
||||
|
||||
Reference in New Issue
Block a user