459 lines
17 KiB
Python
459 lines
17 KiB
Python
import socket
|
|
import struct
|
|
import tkinter as tk
|
|
from tkinter import messagebox
|
|
import threading
|
|
import random
|
|
from udpmsg import *
|
|
from udpmsg import msg_CcuStateFbMsg
|
|
from datetime import datetime
|
|
from tkinter import font
|
|
from tkinter import Frame
|
|
from tkinter.ttk import *
|
|
|
|
class ccuSys(Frame):
|
|
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):
|
|
# 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", "fcuGZ_1L", "fcuGZ_1H",
|
|
"fcuGZ_2L", "fcuGZ_2H", "fcuGZ_3L", "fcuGZ_3H"
|
|
])
|
|
|
|
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)
|
|
|
|
# FCU GZ control
|
|
tk.Label(self.fcu_frame, text="GZ 1L:").pack(side=tk.LEFT)
|
|
self.fcu_gz1l_var = tk.IntVar(value=0)
|
|
self.fcu_gz1l_entry = tk.Entry(
|
|
self.fcu_frame, textvariable=self.fcu_gz1l_var, width=5
|
|
)
|
|
self.fcu_gz1l_entry.pack(side=tk.LEFT, padx=5)
|
|
self.fcu_gz1l_entry.bind("<Return>", lambda e: self.set_fcu_gz1l())
|
|
|
|
tk.Label(self.fcu_frame, text="GZ 1H:").pack(side=tk.LEFT)
|
|
self.fcu_gz1h_var = tk.IntVar(value=0)
|
|
self.fcu_gz1h_entry = tk.Entry(
|
|
self.fcu_frame, textvariable=self.fcu_gz1h_var, width=5
|
|
)
|
|
self.fcu_gz1h_entry.pack(side=tk.LEFT, padx=5)
|
|
self.fcu_gz1h_entry.bind("<Return>", lambda e: self.set_fcu_gz1h())
|
|
|
|
# Add more GZ controls
|
|
tk.Label(self.fcu_frame, text="GZ 2L:").pack(side=tk.LEFT)
|
|
self.fcu_gz2l_var = tk.IntVar(value=0)
|
|
self.fcu_gz2l_entry = tk.Entry(
|
|
self.fcu_frame, textvariable=self.fcu_gz2l_var, width=5
|
|
)
|
|
self.fcu_gz2l_entry.pack(side=tk.LEFT, padx=5)
|
|
self.fcu_gz2l_entry.bind("<Return>", lambda e: self.set_fcu_gz2l())
|
|
|
|
tk.Label(self.fcu_frame, text="GZ 2H:").pack(side=tk.LEFT)
|
|
self.fcu_gz2h_var = tk.IntVar(value=0)
|
|
self.fcu_gz2h_entry = tk.Entry(
|
|
self.fcu_frame, textvariable=self.fcu_gz2h_var, width=5
|
|
)
|
|
self.fcu_gz2h_entry.pack(side=tk.LEFT, padx=5)
|
|
self.fcu_gz2h_entry.bind("<Return>", lambda e: self.set_fcu_gz2h())
|
|
|
|
tk.Label(self.fcu_frame, text="GZ 3L:").pack(side=tk.LEFT)
|
|
self.fcu_gz3l_var = tk.IntVar(value=0)
|
|
self.fcu_gz3l_entry = tk.Entry(
|
|
self.fcu_frame, textvariable=self.fcu_gz3l_var, width=5
|
|
)
|
|
self.fcu_gz3l_entry.pack(side=tk.LEFT, padx=5)
|
|
self.fcu_gz3l_entry.bind("<Return>", lambda e: self.set_fcu_gz3l())
|
|
|
|
tk.Label(self.fcu_frame, text="GZ 3H:").pack(side=tk.LEFT)
|
|
self.fcu_gz3h_var = tk.IntVar(value=0)
|
|
self.fcu_gz3h_entry = tk.Entry(
|
|
self.fcu_frame, textvariable=self.fcu_gz3h_var, width=5
|
|
)
|
|
self.fcu_gz3h_entry.pack(side=tk.LEFT, padx=5)
|
|
self.fcu_gz3h_entry.bind("<Return>", lambda e: self.set_fcu_gz3h())
|
|
|
|
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)
|
|
|
|
self.ccuState.fcuGZ_1L=self.fcu_gz1l_var.get()
|
|
self.ccuState.fcuGZ_1H=self.fcu_gz1h_var.get()
|
|
self.ccuState.fcuGZ_2L=self.fcu_gz2l_var.get()
|
|
self.ccuState.fcuGZ_2H=self.fcu_gz2h_var.get()
|
|
self.ccuState.fcuGZ_3L=self.fcu_gz3l_var.get()
|
|
self.ccuState.fcuGZ_3H=self.fcu_gz3h_var.get()
|
|
|
|
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 set_fcu_gz1l(self, value=None):
|
|
"""Set FCU GZ 1L and update display"""
|
|
if value is None:
|
|
try:
|
|
value = self.fcu_gz1l_var.get()
|
|
except tk.TclError:
|
|
tk.messagebox.showwarning("Input Error", "Invalid GZ 1L value")
|
|
return
|
|
self.ccuState.fcuGZ_1L = value
|
|
self.update_display()
|
|
|
|
def set_fcu_gz1h(self, value=None):
|
|
"""Set FCU GZ 1H and update display"""
|
|
if value is None:
|
|
try:
|
|
value = self.fcu_gz1h_var.get()
|
|
except tk.TclError:
|
|
tk.messagebox.showwarning("Input Error", "Invalid GZ 1H value")
|
|
return
|
|
self.ccuState.fcuGZ_1H = value
|
|
self.update_display()
|
|
|
|
def set_fcu_gz2l(self, value=None):
|
|
"""Set FCU GZ 2L and update display"""
|
|
if value is None:
|
|
try:
|
|
value = self.fcu_gz2l_var.get()
|
|
except tk.TclError:
|
|
tk.messagebox.showwarning("Input Error", "Invalid GZ 2L value")
|
|
return
|
|
self.ccuState.fcuGZ_2L = value
|
|
self.update_display()
|
|
|
|
def set_fcu_gz2h(self, value=None):
|
|
"""Set FCU GZ 2H and update display"""
|
|
if value is None:
|
|
try:
|
|
value = self.fcu_gz2h_var.get()
|
|
except tk.TclError:
|
|
tk.messagebox.showwarning("Input Error", "Invalid GZ 2H value")
|
|
return
|
|
self.ccuState.fcuGZ_2H = value
|
|
self.update_display()
|
|
|
|
def set_fcu_gz3l(self, value=None):
|
|
"""Set FCU GZ 3L and update display"""
|
|
if value is None:
|
|
try:
|
|
value = self.fcu_gz3l_var.get()
|
|
except tk.TclError:
|
|
tk.messagebox.showwarning("Input Error", "Invalid GZ 3L value")
|
|
return
|
|
self.ccuState.fcuGZ_3L = value
|
|
self.update_display()
|
|
|
|
def set_fcu_gz3h(self, value=None):
|
|
"""Set FCU GZ 3H and update display"""
|
|
if value is None:
|
|
try:
|
|
value = self.fcu_gz3h_var.get()
|
|
except tk.TclError:
|
|
tk.messagebox.showwarning("Input Error", "Invalid GZ 3H value")
|
|
return
|
|
self.ccuState.fcuGZ_3H = value
|
|
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()
|