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("", 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("", 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("", 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("", 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("", 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("", 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()