import tkinter as tk from tkinter import ttk, scrolledtext, messagebox import serial import threading import time import math class MaselliSimulatorApp: def __init__(self, root_window): self.root = root_window self.root.title("Simulador Protocolo Maselli") self.serial_port = None # Para simulación continua self.simulating = False self.simulation_thread = None self.simulation_step = 0 # --- Configuration Frame --- config_frame = ttk.LabelFrame(self.root, text="Configuración") config_frame.grid(row=0, column=0, padx=10, pady=10, sticky="ew", columnspan=2) ttk.Label(config_frame, text="Puerto COM:").grid(row=0, column=0, padx=5, pady=5, sticky="w") self.com_port_var = tk.StringVar(value="COM3") self.com_port_entry = ttk.Entry(config_frame, textvariable=self.com_port_var, width=10) self.com_port_entry.grid(row=0, column=1, padx=5, pady=5, sticky="ew") ttk.Label(config_frame, text="Baud Rate:").grid(row=0, column=2, padx=5, pady=5, sticky="w") self.baud_rate_var = tk.StringVar(value="115200") self.baud_rate_entry = ttk.Entry(config_frame, textvariable=self.baud_rate_var, width=10) self.baud_rate_entry.grid(row=0, column=3, padx=5, pady=5, sticky="ew") ttk.Label(config_frame, text="ADAM Address (2c):").grid(row=1, column=0, padx=5, pady=5, sticky="w") self.adam_address_var = tk.StringVar(value="01") self.adam_address_entry = ttk.Entry(config_frame, textvariable=self.adam_address_var, width=5) self.adam_address_entry.grid(row=1, column=1, padx=5, pady=5, sticky="ew") ttk.Label(config_frame, text="Función:").grid(row=1, column=2, padx=5, pady=5, sticky="w") self.function_type_var = tk.StringVar(value="Lineal") self.function_type_combo = ttk.Combobox(config_frame, textvariable=self.function_type_var, values=["Lineal", "Sinusoidal", "Manual"], state="readonly", width=10) self.function_type_combo.grid(row=1, column=3, padx=5, pady=5, sticky="ew") self.function_type_combo.bind("<>", self.on_function_type_change) # Parámetros para mapeo 4-20mA (y generación en Lineal/Sinusoidal) ttk.Label(config_frame, text="Valor Mínimo (Brix) [p/ 4mA]:").grid(row=2, column=0, padx=5, pady=5, sticky="w") self.min_brix_map_var = tk.StringVar(value="0") self.min_brix_map_entry = ttk.Entry(config_frame, textvariable=self.min_brix_map_var, width=10) self.min_brix_map_entry.grid(row=2, column=1, padx=5, pady=5, sticky="ew") ttk.Label(config_frame, text="Valor Máximo (Brix) [p/ 20mA]:").grid(row=2, column=2, padx=5, pady=5, sticky="w") self.max_brix_map_var = tk.StringVar(value="80") self.max_brix_map_entry = ttk.Entry(config_frame, textvariable=self.max_brix_map_var, width=10) self.max_brix_map_entry.grid(row=2, column=3, padx=5, pady=5, sticky="ew") # Parámetros específicos para simulación continua ttk.Label(config_frame, text="Periodo Sim. (s):").grid(row=3, column=0, padx=5, pady=5, sticky="w") self.period_var = tk.StringVar(value="1.0") self.period_entry = ttk.Entry(config_frame, textvariable=self.period_var, width=5) self.period_entry.grid(row=3, column=1, padx=5, pady=5, sticky="ew") # Parámetros específicos para modo Manual ttk.Label(config_frame, text="Valor Brix Manual:").grid(row=4, column=0, padx=5, pady=5, sticky="w") self.manual_brix_var = tk.StringVar(value="10.0") self.manual_brix_entry = ttk.Entry(config_frame, textvariable=self.manual_brix_var, width=10, state=tk.DISABLED) self.manual_brix_entry.grid(row=4, column=1, padx=5, pady=5, sticky="ew") self.manual_send_button = ttk.Button(config_frame, text="Enviar Manual", command=self.send_manual_value, state=tk.DISABLED) self.manual_send_button.grid(row=4, column=2, columnspan=2, padx=5, pady=5, sticky="ew") # --- Controls Frame (para simulación continua) --- controls_frame = ttk.LabelFrame(self.root, text="Control Simulación Continua") controls_frame.grid(row=1, column=0, padx=10, pady=5, sticky="ew") self.start_button = ttk.Button(controls_frame, text="Iniciar Simulación", command=self.start_simulation) self.start_button.pack(side=tk.LEFT, padx=5) self.stop_button = ttk.Button(controls_frame, text="Detener Simulación", command=self.stop_simulation, state=tk.DISABLED) self.stop_button.pack(side=tk.LEFT, padx=5) # --- Display Frame --- display_frame = ttk.LabelFrame(self.root, text="Visualización") display_frame.grid(row=1, column=1, rowspan=2, padx=10, pady=10, sticky="nsew") ttk.Label(display_frame, text="Valor Brix Actual:").grid(row=0, column=0, padx=5, pady=5, sticky="w") self.current_brix_display_var = tk.StringVar(value="---") self.current_brix_label = ttk.Label(display_frame, textvariable=self.current_brix_display_var, font=("Courier", 10)) self.current_brix_label.grid(row=0, column=1, padx=5, pady=5, sticky="w") ttk.Label(display_frame, text="Valor mA Correspondiente:").grid(row=1, column=0, padx=5, pady=5, sticky="w") self.current_ma_display_var = tk.StringVar(value="--.-- mA") self.current_ma_label = ttk.Label(display_frame, textvariable=self.current_ma_display_var, font=("Courier", 10)) self.current_ma_label.grid(row=1, column=1, padx=5, pady=5, sticky="w") # --- Log Frame --- log_frame = ttk.LabelFrame(self.root, text="Log de Comunicación") log_frame.grid(row=2, column=0, padx=10, pady=10, sticky="nsew", columnspan=2) self.com_log_text = scrolledtext.ScrolledText(log_frame, height=12, width=70, wrap=tk.WORD, state=tk.DISABLED) self.com_log_text.pack(padx=5,pady=5,fill=tk.BOTH, expand=True) self.root.columnconfigure(1, weight=1) # Allow display_frame to expand log_frame.columnconfigure(0, weight=1) log_frame.rowconfigure(0, weight=1) self.root.protocol("WM_DELETE_WINDOW", self.on_closing) self.on_function_type_change() # Set initial state of widgets def on_function_type_change(self, event=None): func_type = self.function_type_var.get() if func_type == "Manual": if self.simulating: self.stop_simulation() # Detiene simulación continua y cierra puerto si estaba abierto por ella self.manual_brix_entry.config(state=tk.NORMAL) self.manual_send_button.config(state=tk.NORMAL) self.period_entry.config(state=tk.DISABLED) self.start_button.config(state=tk.DISABLED) self.stop_button.config(state=tk.DISABLED) else: # Lineal o Sinusoidal self.manual_brix_entry.config(state=tk.DISABLED) self.manual_send_button.config(state=tk.DISABLED) self.period_entry.config(state=tk.NORMAL) if not self.simulating: self.start_button.config(state=tk.NORMAL) self.stop_button.config(state=tk.DISABLED) else: self.start_button.config(state=tk.DISABLED) self.stop_button.config(state=tk.NORMAL) def _log_message(self, message): self.com_log_text.configure(state=tk.NORMAL) self.com_log_text.insert(tk.END, message + "\n") self.com_log_text.see(tk.END) self.com_log_text.configure(state=tk.DISABLED) def calculate_checksum(self, message_part): s = sum(ord(c) for c in message_part) checksum_byte = s % 256 return f"{checksum_byte:02X}" def scale_to_mA(self, brix_value, min_brix_map, max_brix_map): if max_brix_map == min_brix_map: return 4.0 percentage = (brix_value - min_brix_map) / (max_brix_map - min_brix_map) percentage = max(0.0, min(1.0, percentage)) mA_value = 4.0 + percentage * 16.0 return mA_value def format_mA_value(self, mA_val): return f"{mA_val:06.3f}" def _get_common_params(self): try: com_port = self.com_port_var.get() baud_rate = int(self.baud_rate_var.get()) adam_address = self.adam_address_var.get() if len(adam_address) != 2: messagebox.showerror("Error", "La dirección ADAM debe tener 2 caracteres.") return None min_brix_map = float(self.min_brix_map_var.get()) max_brix_map = float(self.max_brix_map_var.get()) if min_brix_map > max_brix_map: # Allow min_brix_map == max_brix_map messagebox.showerror("Error", "Valor Mínimo (Brix) para mapeo no puede ser mayor que Valor Máximo (Brix).") return None return com_port, baud_rate, adam_address, min_brix_map, max_brix_map except ValueError as e: messagebox.showerror("Error de Entrada", f"Valor inválido en la configuración común: {e}") return None def send_manual_value(self): common_params = self._get_common_params() if not common_params: return com_port, baud_rate, adam_address, min_brix_map, max_brix_map = common_params try: manual_brix = float(self.manual_brix_var.get()) except ValueError: messagebox.showerror("Error de Entrada", "Valor Brix Manual inválido.") return mA_val = self.scale_to_mA(manual_brix, min_brix_map, max_brix_map) mA_str = self.format_mA_value(mA_val) self.current_brix_display_var.set(f"{manual_brix:.3f} Brix") self.current_ma_display_var.set(f"{mA_str} mA") message_part = f"#{adam_address}{mA_str}" checksum = self.calculate_checksum(message_part) full_string_to_send = f"{message_part}{checksum}\r" log_display_string = full_string_to_send.replace('\r', '') temp_serial_port = None try: temp_serial_port = serial.Serial(com_port, baud_rate, timeout=1) self._log_message(f"Puerto {com_port} abierto temporalmente para envío manual.") self._log_message(f"Enviando Manual: {log_display_string}") temp_serial_port.write(full_string_to_send.encode('ascii')) except serial.SerialException as e: self._log_message(f"Error al enviar manualmente por puerto COM: {e}") messagebox.showerror("Error de Puerto COM", f"No se pudo abrir/escribir en {com_port}: {e}") finally: if temp_serial_port and temp_serial_port.is_open: temp_serial_port.close() self._log_message(f"Puerto {com_port} cerrado tras envío manual.") def start_simulation(self): if self.simulating: messagebox.showwarning("Advertencia", "La simulación ya está en curso.") return common_params = self._get_common_params() if not common_params: return com_port, baud_rate, self.adam_address, self.min_brix_map, self.max_brix_map = common_params try: self.simulation_period = float(self.period_var.get()) if self.simulation_period <= 0: messagebox.showerror("Error", "El periodo debe ser un número positivo.") return self.function_type = self.function_type_var.get() if self.function_type == "Manual": # Should not happen if GUI logic is correct messagebox.showinfo("Info", "Seleccione modo Lineal o Sinusoidal para simulación continua.") return except ValueError as e: messagebox.showerror("Error de Entrada", f"Valor inválido en la configuración de simulación: {e}") return try: self.serial_port = serial.Serial(com_port, baud_rate, timeout=1) self._log_message(f"Puerto {com_port} abierto a {baud_rate} baud para simulación continua.") except serial.SerialException as e: messagebox.showerror("Error de Puerto COM", f"No se pudo abrir el puerto {com_port}: {e}") self.serial_port = None return self.simulating = True self.simulation_step = 0 self.start_button.config(state=tk.DISABLED) self.stop_button.config(state=tk.NORMAL) self._set_config_entries_state(tk.DISABLED) self.simulation_thread = threading.Thread(target=self.run_simulation, daemon=True) self.simulation_thread.start() self._log_message("Simulación continua iniciada.") def _set_config_entries_state(self, state): self.com_port_entry.config(state=state) self.baud_rate_entry.config(state=state) self.adam_address_entry.config(state=state) self.function_type_combo.config(state=state) self.min_brix_map_entry.config(state=state) self.max_brix_map_entry.config(state=state) # Specific to sim type current_func_type = self.function_type_var.get() if current_func_type != "Manual": self.period_entry.config(state=state) self.manual_brix_entry.config(state=tk.DISABLED) self.manual_send_button.config(state=tk.DISABLED) else: # Manual self.period_entry.config(state=tk.DISABLED) # For manual, these are handled by on_function_type_change based on main state # If state is tk.DISABLED, ensure manual ones are also disabled if state == tk.DISABLED: self.manual_brix_entry.config(state=tk.DISABLED) self.manual_send_button.config(state=tk.DISABLED) else: # tk.NORMAL self.manual_brix_entry.config(state=tk.NORMAL) self.manual_send_button.config(state=tk.NORMAL) def stop_simulation(self): if not self.simulating: # Could be called when switching to Manual mode, even if not simulating if self.function_type_var.get() != "Manual": self._log_message("Simulación continua ya estaba detenida.") # Ensure GUI elements are in a consistent state for non-manual modes if self.function_type_var.get() != "Manual": self.start_button.config(state=tk.NORMAL) self.stop_button.config(state=tk.DISABLED) self._set_config_entries_state(tk.NORMAL) return self.simulating = False if self.simulation_thread and self.simulation_thread.is_alive(): try: self.simulation_thread.join(timeout=max(0.1, self.simulation_period * 1.5 if hasattr(self, 'simulation_period') else 2.0)) except Exception as e: self._log_message(f"Error al esperar el hilo de simulación: {e}") if self.serial_port and self.serial_port.is_open: self.serial_port.close() self._log_message(f"Puerto {self.serial_port.name} cerrado (simulación continua).") self.serial_port = None self.start_button.config(state=tk.NORMAL) self.stop_button.config(state=tk.DISABLED) self._set_config_entries_state(tk.NORMAL) # Call on_function_type_change to ensure manual fields are correctly set # if the current mode is manual after stopping. self.on_function_type_change() self._log_message("Simulación continua detenida.") self.current_brix_display_var.set("---") self.current_ma_display_var.set("--.-- mA") def run_simulation(self): steps_for_full_cycle = 100 while self.simulating: current_brix_val = 0.0 if self.function_type == "Lineal": progress = (self.simulation_step % (2 * steps_for_full_cycle)) / steps_for_full_cycle if progress > 1.0: progress = 2.0 - progress current_brix_val = self.min_brix_map + (self.max_brix_map - self.min_brix_map) * progress elif self.function_type == "Sinusoidal": phase = (self.simulation_step / steps_for_full_cycle) * 2 * math.pi sin_val = (math.sin(phase) + 1) / 2 current_brix_val = self.min_brix_map + (self.max_brix_map - self.min_brix_map) * sin_val mA_val = self.scale_to_mA(current_brix_val, self.min_brix_map, self.max_brix_map) mA_str = self.format_mA_value(mA_val) self.current_brix_display_var.set(f"{current_brix_val:.3f} Brix") self.current_ma_display_var.set(f"{mA_str} mA") message_part = f"#{self.adam_address}{mA_str}" checksum = self.calculate_checksum(message_part) full_string_to_send = f"{message_part}{checksum}\r" log_display_string = full_string_to_send.replace('\r', '') self._log_message(f"Enviando Sim: {log_display_string}") if self.serial_port and self.serial_port.is_open: try: self.serial_port.write(full_string_to_send.encode('ascii')) except serial.SerialException as e: self._log_message(f"Error al escribir en puerto COM (sim): {e}") self.root.after(0, self.stop_simulation_from_thread_error) # Schedule stop from main thread break self.simulation_step += 1 time.sleep(self.simulation_period) # Ensure GUI updates if loop exits due to self.simulating = False if not self.simulating and self.root.winfo_exists(): self.root.after(0, self.ensure_gui_stopped_state_sim) def stop_simulation_from_thread_error(self): """Called from main thread if serial error occurs in sim thread.""" if self.simulating: # Check if it wasn't already stopped messagebox.showerror("Error de Simulación", "Error de puerto COM durante la simulación. Simulación detenida.") self.stop_simulation() def ensure_gui_stopped_state_sim(self): """Asegura que la GUI refleje el estado detenido si el hilo de simulación continua termina.""" if not self.simulating: # If stop_simulation wasn't called or completed fully self.start_button.config(state=tk.NORMAL) self.stop_button.config(state=tk.DISABLED) self._set_config_entries_state(tk.NORMAL) # self.on_function_type_change() # Reset based on current function type if self.serial_port and self.serial_port.is_open: self.serial_port.close() self._log_message(f"Puerto {self.serial_port.name} cerrado (auto, sim_end).") self.serial_port = None self._log_message("Simulación continua terminada (hilo finalizado).") def on_closing(self): if self.simulating: self.stop_simulation() # Esto ya maneja el thread y el puerto elif self.serial_port and self.serial_port.is_open: # Si el puerto quedó abierto por otra razón (improbable con lógica actual) self.serial_port.close() self.root.destroy() if __name__ == "__main__": main_root = tk.Tk() app = MaselliSimulatorApp(main_root) main_root.mainloop()