# target_simulator/gui/sync_tool_window.py import tkinter as tk from tkinter import ttk, messagebox import time import random from typing import Dict, Optional, List from collections import deque from target_simulator.core.sfp_communicator import SFPCommunicator from target_simulator.gui.payload_router import DebugPayloadRouter # Tentativo di importare matplotlib per il grafico try: import matplotlib matplotlib.use("TkAgg") from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg from matplotlib.figure import Figure MATPLOTLIB_AVAILABLE = True except ImportError: MATPLOTLIB_AVAILABLE = False class SyncToolWindow(tk.Toplevel): """Finestra per il test di latenza tramite messaggi SYNC con grafico in tempo reale.""" def __init__( self, master, communicator: SFPCommunicator, router: DebugPayloadRouter ): super().__init__(master) self.communicator = communicator self.router = router self.title("Latency Sync Tool") self.geometry("900x700") self.transient(master) self.grab_set() self.pending_requests: Dict[int, float] = {} # {cookie: send_timestamp} # Stato per invio periodico self.periodic_active = False self.periodic_interval_ms = 500 # Default 500ms self.periodic_after_id = None # Storico delle latenze per il grafico (max 200 campioni) self.latency_history: deque = deque(maxlen=200) self.time_history: deque = deque(maxlen=200) # Statistiche self.latency_values: List[float] = [] self._create_widgets() self.protocol("WM_DELETE_WINDOW", self._on_close) self.after(100, self._process_sync_queue) def _create_widgets(self): main_frame = ttk.Frame(self, padding=10) main_frame.pack(fill=tk.BOTH, expand=True) # === CONTROLLI SUPERIORI === controls_frame = ttk.LabelFrame(main_frame, text="Controls", padding=5) controls_frame.pack(fill=tk.X, pady=(0, 10)) # Riga 1: Invio singolo row1 = ttk.Frame(controls_frame) row1.pack(fill=tk.X, pady=2) self.send_button = ttk.Button( row1, text="Send Single SYNC", command=self._send_sync, width=18 ) self.send_button.pack(side=tk.LEFT, padx=(0, 10)) self.status_var = tk.StringVar(value="Idle") ttk.Label(row1, textvariable=self.status_var, foreground="blue").pack( side=tk.LEFT ) # Riga 2: Invio periodico row2 = ttk.Frame(controls_frame) row2.pack(fill=tk.X, pady=2) self.start_stop_button = ttk.Button( row2, text="Start Periodic", command=self._toggle_periodic, width=18 ) self.start_stop_button.pack(side=tk.LEFT, padx=(0, 10)) ttk.Label(row2, text="Interval (ms):").pack(side=tk.LEFT, padx=(0, 5)) self.interval_var = tk.StringVar(value="500") interval_entry = ttk.Entry(row2, textvariable=self.interval_var, width=8) interval_entry.pack(side=tk.LEFT, padx=(0, 10)) ttk.Button(row2, text="Clear History", command=self._clear_history).pack( side=tk.LEFT ) # === STATISTICHE === stats_frame = ttk.LabelFrame(main_frame, text="Statistics", padding=5) stats_frame.pack(fill=tk.X, pady=(0, 10)) stats_grid = ttk.Frame(stats_frame) stats_grid.pack(fill=tk.X) ttk.Label(stats_grid, text="Samples:").grid( row=0, column=0, sticky=tk.W, padx=5 ) self.samples_var = tk.StringVar(value="0") ttk.Label( stats_grid, textvariable=self.samples_var, font=("TkDefaultFont", 10, "bold"), ).grid(row=0, column=1, sticky=tk.W, padx=5) ttk.Label(stats_grid, text="Last:").grid(row=0, column=2, sticky=tk.W, padx=5) self.last_var = tk.StringVar(value="-- ms") ttk.Label( stats_grid, textvariable=self.last_var, font=("TkDefaultFont", 10, "bold") ).grid(row=0, column=3, sticky=tk.W, padx=5) ttk.Label(stats_grid, text="Min:").grid(row=0, column=4, sticky=tk.W, padx=5) self.min_var = tk.StringVar(value="-- ms") ttk.Label( stats_grid, textvariable=self.min_var, font=("TkDefaultFont", 10, "bold") ).grid(row=0, column=5, sticky=tk.W, padx=5) ttk.Label(stats_grid, text="Max:").grid(row=0, column=6, sticky=tk.W, padx=5) self.max_var = tk.StringVar(value="-- ms") ttk.Label( stats_grid, textvariable=self.max_var, font=("TkDefaultFont", 10, "bold") ).grid(row=0, column=7, sticky=tk.W, padx=5) ttk.Label(stats_grid, text="Avg:").grid(row=0, column=8, sticky=tk.W, padx=5) self.avg_var = tk.StringVar(value="-- ms") ttk.Label( stats_grid, textvariable=self.avg_var, font=("TkDefaultFont", 10, "bold") ).grid(row=0, column=9, sticky=tk.W, padx=5) # === GRAFICO === if MATPLOTLIB_AVAILABLE: graph_frame = ttk.LabelFrame(main_frame, text="Latency Graph", padding=5) graph_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10)) self.figure = Figure(figsize=(8, 3), dpi=100) self.ax = self.figure.add_subplot(111) self.ax.set_xlabel("Time (s)") self.ax.set_ylabel("Latency (ms)") self.ax.set_title("One-Way Latency Over Time") self.ax.grid(True, alpha=0.3) self.canvas = FigureCanvasTkAgg(self.figure, graph_frame) self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) (self.line,) = self.ax.plot([], [], "b-", linewidth=1.5) else: graph_frame = ttk.LabelFrame(main_frame, text="Latency Graph", padding=5) graph_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10)) ttk.Label( graph_frame, text="Matplotlib not available. Install with: pip install matplotlib", foreground="red", ).pack(pady=20) # === TABELLA RISULTATI === tree_frame = ttk.LabelFrame(main_frame, text="Recent Results", padding=5) tree_frame.pack(fill=tk.BOTH, expand=True) columns = ("time", "rtt", "latency") self.tree = ttk.Treeview(tree_frame, columns=columns, show="headings", height=8) self.tree.heading("time", text="Timestamp") self.tree.heading("rtt", text="Round Trip (ms)") self.tree.heading("latency", text="One-Way Latency (ms)") self.tree.column("time", width=150) self.tree.column("rtt", width=150, anchor=tk.E) self.tree.column("latency", width=150, anchor=tk.E) scrollbar = ttk.Scrollbar( tree_frame, orient=tk.VERTICAL, command=self.tree.yview ) self.tree.configure(yscrollcommand=scrollbar.set) self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) def _send_sync(self): """Invia una singola richiesta SYNC.""" if not self.communicator or not self.communicator.is_open: messagebox.showerror( "Error", "SFP Communicator is not connected.", parent=self ) return cookie = random.randint(0, 2**32 - 1) send_time = time.monotonic() if self.communicator.send_sync_request(cookie): self.pending_requests[cookie] = send_time if not self.periodic_active: self.status_var.set( f"Sent request with cookie {cookie}. Waiting for reply..." ) else: self.status_var.set("Failed to send SYNC request.") def _toggle_periodic(self): """Attiva/disattiva l'invio periodico.""" if self.periodic_active: # Stop self.periodic_active = False if self.periodic_after_id: self.after_cancel(self.periodic_after_id) self.periodic_after_id = None self.start_stop_button.config(text="Start Periodic") self.status_var.set("Periodic sending stopped.") else: # Start try: interval = int(self.interval_var.get()) if interval < 50: messagebox.showwarning( "Warning", "Interval too short. Minimum is 50ms.", parent=self ) return self.periodic_interval_ms = interval except ValueError: messagebox.showerror("Error", "Invalid interval value.", parent=self) return self.periodic_active = True self.start_stop_button.config(text="Stop Periodic") self.status_var.set( f"Periodic sending active (every {self.periodic_interval_ms} ms)" ) self._periodic_send() def _periodic_send(self): """Invia periodicamente richieste SYNC.""" if not self.periodic_active: return self._send_sync() self.periodic_after_id = self.after( self.periodic_interval_ms, self._periodic_send ) def _clear_history(self): """Pulisce lo storico e le statistiche.""" self.latency_history.clear() self.time_history.clear() self.latency_values.clear() self._update_statistics() self._update_graph() # Pulisce anche la tabella for item in self.tree.get_children(): self.tree.delete(item) def _update_statistics(self): """Aggiorna le statistiche visualizzate.""" if not self.latency_values: self.samples_var.set("0") self.last_var.set("-- ms") self.min_var.set("-- ms") self.max_var.set("-- ms") self.avg_var.set("-- ms") else: self.samples_var.set(str(len(self.latency_values))) self.last_var.set(f"{self.latency_values[-1]:.2f} ms") self.min_var.set(f"{min(self.latency_values):.2f} ms") self.max_var.set(f"{max(self.latency_values):.2f} ms") self.avg_var.set( f"{sum(self.latency_values)/len(self.latency_values):.2f} ms" ) def _update_graph(self): """Aggiorna il grafico della latenza.""" if not MATPLOTLIB_AVAILABLE: return if len(self.time_history) > 0: # Converti i tempi assoluti in relativi (secondi dall'inizio) times = list(self.time_history) latencies = list(self.latency_history) start_time = times[0] relative_times = [(t - start_time) for t in times] self.line.set_data(relative_times, latencies) self.ax.relim() self.ax.autoscale_view() try: self.canvas.draw() except: pass # Ignora errori durante il disegno def _process_sync_queue(self): """Estrae i risultati dalla coda del router e aggiorna la UI.""" result = self.router.get_sync_result() if result: cookie = result.get("cookie") reception_time = result.get("reception_timestamp") if cookie in self.pending_requests: send_time = self.pending_requests.pop(cookie) rtt_s = reception_time - send_time rtt_ms = rtt_s * 1000 latency_ms = rtt_ms / 2.0 # Aggiorna storico self.latency_history.append(latency_ms) self.time_history.append(reception_time) self.latency_values.append(latency_ms) # Limita latency_values per non far crescere troppo la lista if len(self.latency_values) > 200: self.latency_values.pop(0) # Aggiorna UI timestamp_str = time.strftime("%H:%M:%S") self.tree.insert( "", 0, values=(timestamp_str, f"{rtt_ms:.3f}", f"{latency_ms:.3f}") ) # Limita il numero di righe nella tabella items = self.tree.get_children() if len(items) > 50: self.tree.delete(items[-1]) if not self.periodic_active: self.status_var.set( f"Received reply for cookie {cookie}. Latency: {latency_ms:.2f} ms" ) self._update_statistics() self._update_graph() # Continua il polling self.after(100, self._process_sync_queue) def _on_close(self): """Gestisce la chiusura della finestra.""" if self.periodic_active: self.periodic_active = False if self.periodic_after_id: self.after_cancel(self.periodic_after_id) self.destroy()