from target_simulator.gui.ppi_display import PPIDisplay # target_simulator/gui/sfp_debug_window.py """ Provides a Toplevel window for debugging the SFP transport layer. This version uses a sampling approach to handle high-frequency data streams without overwhelming the GUI thread. """ import tkinter as tk from tkinter import ttk, scrolledtext, messagebox import logging import threading import collections import datetime import os import ctypes import time import math from typing import Dict, Callable, Optional, Any import socket # Third-party imports for image display try: from PIL import Image, ImageTk import numpy as np import cv2 _IMAGE_LIBS_AVAILABLE = True except ImportError: _IMAGE_LIBS_AVAILABLE = False # Imports from the project structure from target_simulator.core.sfp_transport import SfpTransport from target_simulator.core.sfp_structures import ( ImageLeaderData, SFPHeader, SfpRisStatusPayload, ) from target_simulator.gui.payload_router import DebugPayloadRouter from target_simulator.core.models import ( Target, Waypoint, ManeuverType, KNOTS_TO_FPS, ) from target_simulator.core import command_builder # default value for testing fdx protocolo DEF_TEST_ID = 1 DEF_TEST_RANGE = 30.0 DEF_TEST_AZIMUTH = 10.0 DEF_TEST_VELOCITY = 300.0 DEF_TEST_HEADING = 0.0 DEF_TEST_ALTITUDE = 10000.0 class SfpDebugWindow(tk.Toplevel): """Top-level window for SFP debugging and payload inspection.""" GUI_POLL_INTERVAL_MS = 250 def __init__(self, master=None): super().__init__(master) self.master = master try: self.geometry("1100x700") except Exception: pass try: self.protocol("WM_DELETE_WINDOW", self._on_close) except Exception: pass self.logger = logging.getLogger(__name__) self.payload_router = DebugPayloadRouter() self.shared_communicator = getattr(self.master, 'target_communicator', None) if self.shared_communicator: self.shared_communicator.add_connection_state_callback(self._update_toggle_state) handlers = self.payload_router.get_handlers() self.shared_communicator.add_payload_handlers(handlers) self.image_area_size = 150 self._ppi_visible = False # --- TK Variables --- self.ip_var = tk.StringVar(value="127.0.0.1") self.local_port_var = tk.StringVar(value="60002") self.server_port_var = tk.StringVar(value="60001") self.script_var = tk.StringVar(value="print('hello from client')") # Variables for the new target sender UI self.tgt_id_var = tk.IntVar(value=DEF_TEST_ID) self.tgt_range_var = tk.DoubleVar(value=DEF_TEST_RANGE) self.tgt_az_var = tk.DoubleVar(value=DEF_TEST_AZIMUTH) self.tgt_alt_var = tk.DoubleVar(value=DEF_TEST_ALTITUDE) self.tgt_vel_var = tk.DoubleVar(value=DEF_TEST_VELOCITY) self.tgt_hdg_var = tk.DoubleVar(value=DEF_TEST_HEADING) self.tgt_active_var = tk.BooleanVar(value=True) self.tgt_traceable_var = tk.BooleanVar(value=True) self.tgt_restart_var = tk.BooleanVar(value=False) self._master_mode_names = [ "idle_master_mode", "int_bit_master_mode", "gm_master_mode", "dbs_master_mode", "rws_master_mode", "vs_master_mode", "acm_master_mode", "tws_master_mode", "sea_low_master_mode", "sea_high_master_mode", "gmti_master_mode", "bcn_master_mode", "sam_master_mode", "ta_master_mode", "wa_master_mode", "stt_master_mode", "dtt_master_mode", "sstt_master_mode", "acq_master_mode", "ftt_master_mode", "agr_master_mode", "sar_master_mode", "invalid_master_mode_", "xtst_dummy_mode", "xtst_hw_validation_mode", "boot_master_mode", "master_mode_id_cardinality_", ] self._create_widgets() if self.shared_communicator: is_open = self.shared_communicator.is_open self.logger.info(f"SfpDebugWindow __init__: shared_communicator.is_open = {is_open}") self._update_toggle_state(is_open) self.after(self.GUI_POLL_INTERVAL_MS, self._process_latest_payloads) def _create_widgets(self): top_controls_frame = ttk.Frame(self) top_controls_frame.pack(side=tk.TOP, fill=tk.X, padx=5, pady=5) conn_frame = ttk.Frame(top_controls_frame) conn_frame.pack(side=tk.TOP, fill=tk.X, pady=(0, 5)) self._create_connection_widgets(conn_frame) target_sender_frame = ttk.LabelFrame(top_controls_frame, text="Simple Target Sender") target_sender_frame.pack(side=tk.TOP, fill=tk.X, pady=(5, 5)) self._create_target_sender_widgets(target_sender_frame) script_frame = ttk.LabelFrame(top_controls_frame, text="Script to send") script_frame.pack(side=tk.TOP, fill=tk.X, pady=(0, 5)) self._create_script_sender_widgets(script_frame) self.notebook = ttk.Notebook(self) self.notebook.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) self._create_notebook_tabs() def _toggle_ppi(self): try: if self._ppi_visible: self.ris_ppi_container.pack_forget() self.ris_table_container.pack(fill=tk.BOTH, expand=True) self._ppi_visible = False self.ppi_toggle_btn.config(text="Show PPI Map") else: self.ris_table_container.pack_forget() self.ris_ppi_container.pack(fill=tk.BOTH, expand=True) self._ppi_visible = True self.ppi_toggle_btn.config(text="Hide PPI Map") except Exception: self.logger.exception("Toggle PPI failed") def update_ppi_targets(self, targets): try: if self.ris_ppi_widget: self.ris_ppi_widget.update_targets(targets) except Exception: self.logger.exception("Failed to update RIS PPI targets") def on_ris_status_update(self, ris_status_payload): targets = [] for i in range(getattr(ris_status_payload.tgt, "tgt", []).__len__()): ris_tgt = ris_status_payload.tgt.tgt[i] if getattr(ris_tgt, "flags", 0) & 1: t = Target(target_id=i, trajectory=[], active=True, traceable=True) t.current_range_nm = (ris_tgt.x**2 + ris_tgt.y**2) ** 0.5 / 6076.12 try: t.current_azimuth_deg = math.degrees(math.atan2(ris_tgt.y, ris_tgt.x)) except Exception: t.current_azimuth_deg = 0.0 t.current_altitude_ft = ris_tgt.z t.current_heading_deg = getattr(ris_tgt, "heading", 0.0) t.active = True targets.append(t) self.update_ppi_targets(targets) def _create_connection_widgets(self, parent): ttk.Label(parent, text="IP:").pack(side=tk.LEFT, padx=(4, 2)) ttk.Entry(parent, textvariable=self.ip_var, width=18).pack(side=tk.LEFT, padx=(0, 6)) ttk.Label(parent, text="Local Port:").pack(side=tk.LEFT, padx=(0, 2)) ttk.Entry(parent, textvariable=self.local_port_var, width=8).pack(side=tk.LEFT, padx=(0, 6)) ttk.Label(parent, text="Server Port:").pack(side=tk.LEFT, padx=(0, 2)) ttk.Entry(parent, textvariable=self.server_port_var, width=8).pack(side=tk.LEFT, padx=(0, 6)) self.connect_toggle_btn = ttk.Button(parent, text="Connect", command=self._on_toggle_connect) self.connect_toggle_btn.pack(side=tk.LEFT, padx=(0, 6)) self.send_probe_btn = ttk.Button(parent, text="Send Probe", command=self._on_send_probe) self.send_probe_btn.pack(side=tk.LEFT, padx=(6, 4)) def _create_target_sender_widgets(self, parent): grid = ttk.Frame(parent, padding=5) grid.pack(fill=tk.X) grid.columnconfigure(1, pad=15) grid.columnconfigure(3, pad=15) grid.columnconfigure(5, pad=15) ttk.Label(grid, text="ID:").grid(row=0, column=0, sticky=tk.W) ttk.Spinbox(grid, from_=0, to=15, textvariable=self.tgt_id_var, width=8).grid(row=1, column=0, sticky=tk.W) ttk.Label(grid, text="Range (NM):").grid(row=0, column=1, sticky=tk.W) ttk.Spinbox(grid, from_=0, to=500, textvariable=self.tgt_range_var, width=10).grid(row=1, column=1, sticky=tk.W) ttk.Label(grid, text="Azimuth (°):").grid(row=0, column=2, sticky=tk.W) ttk.Spinbox(grid, from_=-180, to=180, textvariable=self.tgt_az_var, width=10).grid(row=1, column=2, sticky=tk.W) ttk.Label(grid, text="Velocity (kn):").grid(row=0, column=3, sticky=tk.W) ttk.Spinbox(grid, from_=0, to=2000, textvariable=self.tgt_vel_var, width=10).grid(row=1, column=3, sticky=tk.W) ttk.Label(grid, text="Heading (°):").grid(row=0, column=4, sticky=tk.W) ttk.Spinbox(grid, from_=0, to=360, textvariable=self.tgt_hdg_var, width=10).grid(row=1, column=4, sticky=tk.W) ttk.Label(grid, text="Altitude (ft):").grid(row=0, column=5, sticky=tk.W) ttk.Spinbox(grid, from_=0, to=80000, textvariable=self.tgt_alt_var, width=12).grid(row=1, column=5, sticky=tk.W) controls_frame = ttk.Frame(grid) controls_frame.grid(row=1, column=6, sticky="nsew", padx=(20, 0)) ttk.Checkbutton(controls_frame, text="Active", variable=self.tgt_active_var).pack(side=tk.LEFT, anchor="w") ttk.Checkbutton(controls_frame, text="Traceable", variable=self.tgt_traceable_var).pack(side=tk.LEFT, anchor="w", padx=5) ttk.Checkbutton(controls_frame, text="Restart", variable=self.tgt_restart_var).pack(side=tk.LEFT, anchor="w", padx=5) send_button = ttk.Button(controls_frame, text="Send Target", command=self._on_send_target) send_button.pack(side=tk.LEFT, padx=(10, 0)) ttk.Label(controls_frame, text="(debug: sends /s /t /r if set)", foreground="#555555", font=("Segoe UI", 8)).pack(side=tk.LEFT, padx=(6, 0)) quick_cmd_frame = ttk.Frame(parent) quick_cmd_frame.pack(fill=tk.X, pady=(6, 0)) ttk.Button(quick_cmd_frame, text="tgtreset", command=lambda: self._on_send_simple_command(command_builder.build_tgtreset())).pack(side=tk.LEFT, padx=4) ttk.Button(quick_cmd_frame, text="pause", command=lambda: self._on_send_simple_command(command_builder.build_pause())).pack(side=tk.LEFT, padx=4) ttk.Button(quick_cmd_frame, text="continue", command=lambda: self._on_send_simple_command(command_builder.build_continue())).pack(side=tk.LEFT, padx=4) ttk.Button(quick_cmd_frame, text="tgtset (cur)", command=lambda: self._on_send_tgtset()).pack(side=tk.LEFT, padx=8) self.ppi_toggle_btn = ttk.Button(quick_cmd_frame, text="Show PPI Map", command=self._toggle_ppi) self.ppi_toggle_btn.pack(side=tk.RIGHT, padx=4) def _create_script_sender_widgets(self, parent): ttk.Label(parent, text="Script to send:").pack(side=tk.LEFT, padx=(5, 2)) ttk.Entry(parent, textvariable=self.script_var, width=60).pack(side=tk.LEFT, expand=True, fill=tk.X, padx=(0, 5)) self.send_script_btn = ttk.Button(parent, text="Send script", command=self._on_send_script) self.send_script_btn.pack(side=tk.LEFT, padx=5) def _create_notebook_tabs(self): self.log_tab = scrolledtext.ScrolledText(self.notebook, state=tk.DISABLED, wrap=tk.WORD, font=("Consolas", 9)) self.notebook.add(self.log_tab, text="Raw Log") if _IMAGE_LIBS_AVAILABLE: self.mfd_tab = self._create_image_tab("MFD Image") self.notebook.add(self.mfd_tab["frame"], text="MFD Image") self.sar_tab = self._create_image_tab("SAR Image") self.notebook.add(self.sar_tab["frame"], text="SAR Image") ris_frame = ttk.Frame(self.notebook) paned = ttk.Panedwindow(ris_frame, orient=tk.HORIZONTAL) paned.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) left = ttk.Frame(paned) self.scenario_tree = ttk.Treeview(left, columns=("field", "value"), show="headings", height=12) self.scenario_tree.heading("field", text="Field") self.scenario_tree.heading("value", text="Value") self.scenario_tree.column("field", width=140, anchor="w") self.scenario_tree.column("value", width=160, anchor="w") self.scenario_tree.pack(fill=tk.BOTH, expand=True) paned.add(left, weight=1) right = ttk.Frame(paned) self.ris_table_container = ttk.Frame(right) self.ris_ppi_container = ttk.Frame(right) cols = ("idx", "flags", "heading", "x", "y", "z") self.ris_tree = ttk.Treeview(self.ris_table_container, columns=cols, show="headings", height=12) for c, txt in zip(cols, ("#", "flags", "heading", "x", "y", "z")): self.ris_tree.heading(c, text=txt) self.ris_tree.column(c, width=70, anchor="center") try: style = ttk.Style() small_font = ("Consolas", 8) style.configure("Small.Treeview", font=small_font) self.ris_tree.configure(style="Small.Treeview") self.scenario_tree.configure(style="Small.Treeview") except Exception: pass self.ris_tree.pack(fill=tk.BOTH, expand=True) self.ris_table_container.pack(fill=tk.BOTH, expand=True) paned.add(right, weight=2) try: gm = getattr(self.master, "config_manager", None) trail_len = None try: general = gm.get_general_settings() if gm else {} trail_len = general.get("ppi_trail_length") if isinstance(general, dict) else None except Exception: trail_len = None self.ris_ppi_widget = PPIDisplay(self.ris_ppi_container, max_range_nm=100, trail_length=trail_len) self.ris_ppi_widget.pack(fill=tk.BOTH, expand=True) except Exception: self.ris_ppi_widget = None btn_frame = ttk.Frame(ris_frame) btn_frame.pack(fill=tk.X, padx=5, pady=(0, 5)) self.scenario_view_mode = tk.StringVar(value="simplified") mode_frame = ttk.Frame(btn_frame) mode_frame.pack(side=tk.LEFT, padx=(4, 0)) ttk.Label(mode_frame, text="View:").pack(side=tk.LEFT, padx=(0, 6)) ttk.Radiobutton(mode_frame, text="Raw", value="raw", variable=self.scenario_view_mode).pack(side=tk.LEFT) ttk.Radiobutton(mode_frame, text="Simplified", value="simplified", variable=self.scenario_view_mode).pack(side=tk.LEFT) self.simplified_decimals = tk.IntVar(value=4) ttk.Label(mode_frame, text=" Decimals:").pack(side=tk.LEFT, padx=(8, 2)) try: sp = tk.Spinbox(mode_frame, from_=0, to=8, width=3, textvariable=self.simplified_decimals) sp.pack(side=tk.LEFT) except Exception: e = ttk.Entry(mode_frame, textvariable=self.simplified_decimals, width=3) e.pack(side=tk.LEFT) self.ris_save_csv_btn = ttk.Button(btn_frame, text="Save CSV", command=lambda: self._on_save_ris_csv()) self.ris_save_csv_btn.pack(side=tk.RIGHT) self.notebook.add(ris_frame, text="RIS Status") raw_frame = ttk.Frame(self.notebook) history_frame = ttk.Frame(raw_frame, width=380) history_frame.pack(side=tk.LEFT, fill=tk.Y, padx=(5, 2), pady=5) ttk.Label(history_frame, text="History (latest)").pack(anchor=tk.W, padx=4) try: history_font = ("Consolas", 8) except Exception: history_font = None list_container = ttk.Frame(history_frame) list_container.pack(fill=tk.BOTH, expand=True, padx=4, pady=(2, 4)) columns = ("ts", "flow", "tid", "size") self.history_tree = ttk.Treeview(list_container, columns=columns, show="headings", height=20) self.history_tree.heading("ts", text="Timestamp") self.history_tree.heading("flow", text="Flow") self.history_tree.heading("tid", text="TID") self.history_tree.heading("size", text="Size") self.history_tree.column("ts", width=100, anchor="w") self.history_tree.column("flow", width=50, anchor="w") self.history_tree.column("tid", width=40, anchor="center") self.history_tree.column("size", width=50, anchor="e") try: style = ttk.Style() style.configure("Small.Treeview", font=history_font) self.history_tree.configure(style="Small.Treeview") except Exception: pass self.history_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) self.history_vscroll = ttk.Scrollbar(list_container, orient=tk.VERTICAL, command=self.history_tree.yview) self.history_vscroll.pack(side=tk.RIGHT, fill=tk.Y) self.history_tree.config(yscrollcommand=self.history_vscroll.set) hb_frame = ttk.Frame(history_frame) hb_frame.pack(fill=tk.X, padx=4, pady=(4, 4)) self.history_settings_btn = ttk.Button(hb_frame, text="Settings", command=lambda: self._open_history_settings_dialog()) self.history_settings_btn.pack(side=tk.LEFT) self.history_clear_btn = ttk.Button(hb_frame, text="Clear", command=lambda: self._on_clear_history()) self.history_clear_btn.pack(side=tk.RIGHT) self.raw_tab_text = scrolledtext.ScrolledText(raw_frame, state=tk.DISABLED, wrap=tk.NONE, font=("Consolas", 9)) self.raw_tab_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(2, 5), pady=5) try: self.notebook.insert(1, raw_frame, text="SFP Raw") except Exception: self.notebook.add(raw_frame, text="SFP Raw") try: self.raw_tab_text.tag_config("flag_set", background="#d4ffd4", foreground="#006400") self.raw_tab_text.tag_config("flag_unset", background="#f0f0f0", foreground="#808080") self.raw_tab_text.tag_config("flag_error", background="#ffd4d4", foreground="#800000") self.raw_tab_text.tag_config("hdr_field", foreground="#000080") except Exception: pass self.bin_tab = scrolledtext.ScrolledText(self.notebook, state=tk.DISABLED, wrap=tk.NONE, font=("Consolas", 10)) self.notebook.add(self.bin_tab, text="Binary (Hex)") self.json_tab = scrolledtext.ScrolledText(self.notebook, state=tk.DISABLED, wrap=tk.WORD, font=("Consolas", 10)) self.notebook.add(self.json_tab, text="JSON") def _on_send_target(self): if not self.shared_communicator or not self.shared_communicator.is_open: self._log_to_widget("ERROR: Cannot send target, not connected.", "ERROR") messagebox.showerror("Connection Error", "Communicator is not connected.", parent=self) return try: target_id = self.tgt_id_var.get() range_nm = self.tgt_range_var.get() az_deg = self.tgt_az_var.get() alt_ft = self.tgt_alt_var.get() vel_kn = self.tgt_vel_var.get() hdg_deg = self.tgt_hdg_var.get() is_active = self.tgt_active_var.get() is_traceable = self.tgt_traceable_var.get() is_restart = self.tgt_restart_var.get() vel_fps = vel_kn * KNOTS_TO_FPS initial_waypoint = Waypoint( maneuver_type=ManeuverType.FLY_TO_POINT, target_range_nm=range_nm, target_azimuth_deg=az_deg, target_altitude_ft=alt_ft, target_velocity_fps=vel_fps, target_heading_deg=hdg_deg, ) temp_target = Target(target_id=target_id, trajectory=[initial_waypoint]) temp_target.active = is_active temp_target.traceable = is_traceable temp_target.restart = is_restart temp_target.current_range_nm = range_nm temp_target.current_azimuth_deg = az_deg temp_target.current_velocity_fps = vel_fps temp_target.current_heading_deg = hdg_deg temp_target.current_altitude_ft = alt_ft command_str = command_builder.build_tgtset_from_target_state(temp_target, include_flags=True) command_str = command_str.strip() if not command_str.endswith("\n"): command_str += "\n" self._log_to_widget(f"Built command: {command_str!r}", "INFO") success = self.shared_communicator._send_single_command(command_str) if success: self._log_to_widget(f"Successfully sent command for target {target_id}.", "INFO") else: self._log_to_widget(f"Failed to send command for target {target_id}.", "ERROR") except (ValueError, tk.TclError) as e: error_msg = f"Invalid input value: {e}" self._log_to_widget(f"ERROR: {error_msg}", "ERROR") messagebox.showerror("Input Error", error_msg, parent=self) except Exception as e: self.logger.exception("An unexpected error occurred in _on_send_target.") self._log_to_widget(f"ERROR: {e}", "CRITICAL") messagebox.showerror("Unexpected Error", str(e), parent=self) def _create_image_tab(self, title: str) -> Dict: frame = ttk.Frame(self.notebook) image_container = ttk.Frame(frame, width=self.image_area_size, height=self.image_area_size, relief=tk.SUNKEN) image_container.pack(pady=5, padx=5) image_container.pack_propagate(False) image_label = ttk.Label(image_container, text=f"Waiting for {title}...", anchor=tk.CENTER) image_label.pack(fill=tk.BOTH, expand=True) hex_view = scrolledtext.ScrolledText(frame, height=8, state=tk.DISABLED, wrap=tk.NONE, font=("Consolas", 9)) hex_view.pack(fill=tk.BOTH, expand=True, pady=5, padx=5) return { "frame": frame, "image_label": image_label, "hex_view": hex_view, "image_container": image_container, } def _open_image_size_dialog(self): dlg = tk.Toplevel(self) dlg.title("Image Size") dlg.transient(self) dlg.grab_set() ttk.Label(dlg, text="Image area size (px):").pack(padx=10, pady=(10, 2)) size_var = tk.StringVar(value=str(self.image_area_size)) entry = ttk.Entry(dlg, textvariable=size_var, width=8) entry.pack(padx=10, pady=(0, 10)) btn_frame = ttk.Frame(dlg) btn_frame.pack(padx=10, pady=(0, 10)) def on_save(): try: v = int(size_var.get()) if v <= 0: raise ValueError() except Exception: message = "Please enter a positive integer for image size." try: tk.messagebox.showerror("Invalid value", message, parent=dlg) except Exception: pass return self.image_area_size = v for tab in (getattr(self, "mfd_tab", None), getattr(self, "sar_tab", None)): if tab and "image_container" in tab: tab["image_container"].config(width=v, height=v) gm = getattr(self.master, "config_manager", None) if gm: general = gm.get_general_settings() or {} image_display = general.get("image_display", {}) image_display["size"] = v general["image_display"] = image_display gm.save_general_settings(general) dlg.destroy() def on_cancel(): dlg.destroy() ttk.Button(btn_frame, text="Cancel", command=on_cancel).pack(side=tk.RIGHT, padx=(0, 5)) ttk.Button(btn_frame, text="Save", command=on_save).pack(side=tk.RIGHT) def _on_toggle_connect(self): """Toggles the connection state of the shared communicator.""" if not self.shared_communicator: self._log_to_widget("ERROR: No shared communicator available.", "ERROR") messagebox.showerror("Error", "No shared communicator available.", parent=self) return if self.shared_communicator.is_open: self.shared_communicator.disconnect() else: try: ip = self.ip_var.get() server_port = int(self.server_port_var.get()) local_port = int(self.local_port_var.get()) config = {"ip": ip, "port": server_port, "local_port": local_port} self.shared_communicator.connect(config) except (ValueError, tk.TclError) as e: self._log_to_widget(f"ERROR: Invalid connection settings: {e}", "ERROR") messagebox.showerror("Input Error", f"Invalid connection settings: {e}", parent=self) def _on_send_probe(self): ip = self.ip_var.get() try: port = int(self.server_port_var.get()) except Exception: self._log_to_widget("ERROR: Invalid port number for probe.", "ERROR") return probe_payload = b"SFP_PROBE\n" try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(1.0) sock.sendto(probe_payload, (ip, port)) sock.close() self._log_to_widget(f"Sent probe to {ip}:{port}", "INFO") except Exception as e: self._log_to_widget(f"Failed to send probe to {ip}:{port}: {e}", "ERROR") def _on_send_script(self): if not self.shared_communicator or not self.shared_communicator.is_open: self._log_to_widget("ERROR: Cannot send script, not connected.", "ERROR") return try: command_str = self.script_var.get() self.shared_communicator._send_single_command(command_str) except (ValueError, tk.TclError) as e: self._log_to_widget(f"ERROR: Invalid input for script sending: {e}", "ERROR") def _on_send_simple_command(self, command_str: str): if not self.shared_communicator or not self.shared_communicator.is_open: self._log_to_widget("ERROR: Cannot send command, not connected.", "ERROR") messagebox.showerror("Connection Error", "Communicator is not connected.", parent=self) return False try: success = self.shared_communicator._send_single_command(command_str) if success: self._log_to_widget(f"Successfully sent command: {command_str}", "INFO") else: self._log_to_widget(f"Failed to send command: {command_str}", "ERROR") return success except Exception as e: self.logger.exception("Unexpected error in _on_send_simple_command") self._log_to_widget(f"ERROR: {e}", "ERROR") return False def _on_send_tgtset(self): try: target_id = self.tgt_id_var.get() range_nm = self.tgt_range_var.get() az_deg = self.tgt_az_var.get() alt_ft = self.tgt_alt_var.get() vel_kn = self.tgt_vel_var.get() hdg_deg = self.tgt_hdg_var.get() vel_fps = vel_kn * KNOTS_TO_FPS is_active = self.tgt_active_var.get() is_traceable = self.tgt_traceable_var.get() updates = { "range_nm": f"{range_nm:.2f}", "azimuth_deg": f"{az_deg:.2f}", "velocity_fps": f"{vel_fps:.2f}", "heading_deg": f"{hdg_deg:.2f}", "altitude_ft": f"{alt_ft:.2f}", "active": is_active, "traceable": is_traceable, } command_str = command_builder.build_tgtset_selective(target_id, updates) command_str = command_str.strip() if command_str and not command_str.endswith("\n"): command_str = command_str + "\n" return self._on_send_simple_command(command_str) except (ValueError, tk.TclError) as e: self._log_to_widget(f"ERROR: Invalid input for tgtset: {e}", "ERROR") return False def _on_close(self): self.logger.info("SFP Debug Window closing.") if self.shared_communicator: self.shared_communicator.remove_connection_state_callback(self._update_toggle_state) handlers = self.payload_router.get_handlers() self.shared_communicator.remove_payload_handlers(handlers) self.destroy() def _update_toggle_state(self, connected: bool): """Update the toggle button label and internal UI state to reflect connection.""" try: if connected: self.connect_toggle_btn.config(text="Disconnect") else: self.connect_toggle_btn.config(text="Connect") except Exception: pass def update_toggle_state(self, connected: bool): self._update_toggle_state(connected) def _process_latest_payloads(self): new_payloads = self.payload_router.get_and_clear_latest_payloads() if new_payloads: self._log_to_widget(f"Processing {len(new_payloads)} new payload(s) for flows: {list(new_payloads.keys())}") for flow_id, payload in new_payloads.items(): if flow_id == "MFD" and _IMAGE_LIBS_AVAILABLE: self._display_image_data(payload, self.mfd_tab, "mfd_photo") elif flow_id == "SAR" and _IMAGE_LIBS_AVAILABLE: self._display_image_data(payload, self.sar_tab, "sar_photo") elif flow_id == "BIN": self._display_hex_data(payload, self.bin_tab) elif flow_id == "JSON": self._display_json_data(payload, self.json_tab) elif flow_id == "RIS_STATUS_JSON": try: import json struct = json.loads(payload.decode("utf-8")) if isinstance(payload, (bytes, bytearray)) else payload for iid in self.scenario_tree.get_children(): self.scenario_tree.delete(iid) scenario = struct.get("scenario", {}) if scenario: import math def to_deg(v): return float(v) * (180.0 / math.pi) if isinstance(v, (int, float)) else v def m_s_to_ft_s(v): return float(v) * 3.280839895 if isinstance(v, (int, float)) else v def m_to_ft(v): return float(v) * 3.280839895 if isinstance(v, (int, float)) else v order = [ ("timetag", "timetag", ""), ("platform_azimuth", "platform_azimuth", "°"), ("ant_nav_az", "ant_nav_az", "°"), ("ant_nav_el", "ant_nav_el", "°"), ("flags", "flags", ""), ("mode", "mode", ""), ("vx", "vx", "ft/s"), ("vy", "vy", "ft/s"), ("vz", "vz", "ft/s"), ("baro_altitude", "baro_altitude", "ft"), ("latitude", "latitude", "°"), ("longitude", "longitude", "°"), ("true_heading", "true_heading", "°"), ] view_mode = self.scenario_view_mode.get() if hasattr(self, "scenario_view_mode") else "simplified" dec_simp = int(self.simplified_decimals.get() if hasattr(self, "simplified_decimals") else 4) def fmt_raw_number(v, key_name=None): try: fv = float(v) if key_name in ("latitude", "longitude"): return f"{fv:.8f}" return f"{fv:.6f}" except Exception: return str(v) def fmt_simplified_number(v, unit_str, decimals=4): try: fv = float(v) return f"{fv:.{decimals}f} {unit_str}" if unit_str else f"{fv:.{decimals}f}" except Exception: return str(v) def try_float(v): try: return float(v) except Exception: return None def decimal_deg_to_dms(deg, is_lat=True): try: d = float(deg) except Exception: return str(deg) direction = ("N" if d >= 0 else "S") if is_lat else ("E" if d >= 0 else "W") ad = abs(d) degrees = int(ad) minutes_full = (ad - degrees) * 60 minutes = int(minutes_full) seconds = (minutes_full - minutes) * 60 return f"{degrees}°{minutes}'{seconds:.2f} {direction}" scenario_rows = [] for label, key, unit in order: if key in scenario: val = scenario.get(key) if view_mode == "raw": if key in ("platform_azimuth", "true_heading", "ant_nav_az", "ant_nav_el"): display_val = fmt_raw_number(val, key) elif key in ("vx", "vy", "vz", "baro_altitude"): display_val = fmt_raw_number(val, key) elif key in ("latitude", "longitude"): display_val = fmt_raw_number(val, key) elif key == "flags": try: display_val = f"{int(val)} (0x{int(val):X})" except Exception: display_val = str(val) else: display_val = str(val) else: if key in ("platform_azimuth", "true_heading"): fv = try_float(val) if fv is not None: display_val = fmt_simplified_number(to_deg(fv), "°", dec_simp) else: display_val = str(val) elif key in ("ant_nav_az", "ant_nav_el"): fv = try_float(val) if fv is not None: display_val = fmt_simplified_number(to_deg(fv), "°", dec_simp) else: display_val = str(val) elif key in ("vx", "vy", "vz"): fv = try_float(val) if fv is not None: display_val = fmt_simplified_number(m_s_to_ft_s(fv), "ft/s", dec_simp) else: display_val = str(val) elif key == "baro_altitude": fv = try_float(val) if fv is not None: display_val = fmt_simplified_number(m_to_ft(fv), "ft", dec_simp) else: display_val = str(val) elif key in ("latitude", "longitude"): fv = try_float(val) if fv is not None: display_val = decimal_deg_to_dms(fv, is_lat=(key == "latitude")) else: display_val = str(val) elif key == "flags": try: display_val = f"{int(val)} (0x{int(val):X})" except Exception: display_val = str(val) elif key == "mode": try: midx = int(val) if 0 <= midx < len(self._master_mode_names): name = self._master_mode_names[midx] short = name.replace("_master_mode", "").replace("_mode", "").replace("master", "").strip("_") display_val = short else: display_val = str(midx) except Exception: display_val = str(val) else: display_val = str(val) scenario_rows.append((label, display_val)) for l, v in scenario_rows: self.scenario_tree.insert("", tk.END, values=(f"{l}", v)) for iid in self.ris_tree.get_children(): self.ris_tree.delete(iid) targets = struct.get("targets", []) if isinstance(struct, dict) else [] try: view_mode = self.scenario_view_mode.get() if hasattr(self, "scenario_view_mode") else "simplified" self.ris_tree.heading("heading", text="heading") self.ris_tree.heading("x", text="x") self.ris_tree.heading("y", text="y") self.ris_tree.heading("z", text="z") except Exception: pass view_mode = self.scenario_view_mode.get() if hasattr(self, "scenario_view_mode") else "simplified" dec_simp = int(self.simplified_decimals.get() if hasattr(self, "simplified_decimals") else 4) for t in targets: try: idx = t.get("index", t.get("idx", t.get("#"))) raw_flags = t.get("flags", t.get("flag", 0)) try: flags_val = int(raw_flags) flags_display = f"{flags_val} (0x{flags_val:X})" except Exception: flags_display = str(raw_flags) if view_mode == "raw": hfv = try_float(t.get("heading")) heading_val = f"{hfv:.6f}" if hfv is not None else str(t.get("heading")) xfv, yfv, zfv = try_float(t.get("x")), try_float(t.get("y")), try_float(t.get("z")) x_val = f"{xfv:.6f}" if xfv is not None else t.get("x") y_val = f"{yfv:.6f}" if yfv is not None else t.get("y") z_val = f"{zfv:.6f}" if zfv is not None else t.get("z") vals = (idx, flags_display, heading_val, x_val, y_val, z_val) else: hfv = try_float(t.get("heading")) heading_val = f"{(hfv * 180.0 / math.pi):.{dec_simp}f} °" if hfv is not None else str(t.get("heading")) xfv, yfv, zfv = try_float(t.get("x")), try_float(t.get("y")), try_float(t.get("z")) x_val = f"{xfv:.{dec_simp}f} m" if xfv is not None else str(t.get("x")) y_val = f"{yfv:.{dec_simp}f} m" if yfv is not None else str(t.get("y")) z_val = f"{(zfv * 3.280839895):.{dec_simp}f} ft" if zfv is not None else str(t.get("z")) vals = (idx, flags_display, heading_val, x_val, y_val, z_val) if not isinstance(vals, (list, tuple)) or len(vals) != 6: vals = (idx, flags_display, "", "", "", "") self.ris_tree.insert("", tk.END, values=vals) except Exception as _e: try: self.ris_tree.insert("", tk.END, values=(None, None, str(t.get("heading")), str(t.get("x")), str(t.get("y")), str(t.get("z")))) except Exception: pass try: ppi = getattr(self, "ris_ppi_widget", None) if ppi is not None: ppi_targets = [] METERS_PER_NM = 1852.0 current_range = ppi.range_var.get() if hasattr(ppi, "range_var") else ppi.max_range for t in targets: if int(t.get("flags", 0)) == 0: continue x, y, z = float(t.get("x", 0.0)), float(t.get("y", 0.0)), float(t.get("z", 0.0)) range_nm = ((x**2 + y**2) ** 0.5) / METERS_PER_NM if range_nm > current_range: continue az_deg = math.degrees(math.atan2(y, x)) try: hdg_deg = float(t.get("heading", 0.0)) * (180.0 / math.pi) except Exception: hdg_deg = 0.0 tgt = Target(target_id=int(t.get("index", 0)), trajectory=[], active=True, traceable=True) tgt.current_range_nm = range_nm tgt.current_azimuth_deg = az_deg tgt.current_heading_deg = hdg_deg try: tgt.current_altitude_ft = float(z) * 3.280839895 except Exception: tgt.current_altitude_ft = 0.0 tgt.active = True ppi_targets.append(tgt) try: self._log_to_widget(f"PPI: prepared {len(ppi_targets)} target(s) for display", "DEBUG") if ppi_targets: for pt in ppi_targets[:5]: self._log_to_widget(f"PPI target sample: id={getattr(pt, 'target_id', None)} r_nm={getattr(pt,'current_range_nm',None):.2f} az={getattr(pt,'current_azimuth_deg',None):.2f} hdg={getattr(pt,'current_heading_deg',None):.2f}", "DEBUG") self.update_ppi_targets({"real": ppi_targets}) except Exception: self.logger.exception("Failed to push targets to PPI") except Exception: self.logger.exception("Error while preparing RIS targets for PPI") except Exception: pass self.after(self.GUI_POLL_INTERVAL_MS, self._process_latest_payloads) raw_pkt = self.payload_router.get_and_clear_raw_packet() if raw_pkt: raw_bytes, addr = raw_pkt self._display_raw_packet(raw_bytes, addr) try: self._refresh_history_tree() except Exception: pass def _refresh_history_tree(self): try: hist = self.payload_router.get_history() for iid in self.history_tree.get_children(): self.history_tree.delete(iid) for i, entry in enumerate(reversed(hist)): ts = entry["ts"].strftime("%H:%M:%S.%f")[:-3] flow_name = entry.get("flow_name", "") tid = entry.get("tid", "") size = len(entry.get("raw", b"")) self.history_tree.insert("", tk.END, values=(ts, flow_name, tid, f"{size}B")) except Exception: pass def _on_history_select(self): try: sel = self.history_tree.selection() if not sel: return iid = sel[0] children = list(self.history_tree.get_children()) try: idx = children.index(iid) except ValueError: idx = None hist = list(reversed(self.payload_router.get_history())) if idx is None or idx < 0 or idx >= len(hist): return entry = hist[idx] self._display_raw_packet(entry["raw"], entry["addr"]) except Exception: pass def _on_clear_history(self): try: self.payload_router.clear_history() self._refresh_history_tree() except Exception: pass def _open_history_settings_dialog(self): dlg = tk.Toplevel(self) dlg.title("History Settings") dlg.transient(self) dlg.grab_set() try: hist_size = self.payload_router._history_size persist = self.payload_router._persist except Exception: hist_size, persist = 20, False ttk.Label(dlg, text="History size (entries):").pack(padx=10, pady=(10, 2)) size_var = tk.StringVar(value=str(hist_size)) entry = ttk.Entry(dlg, textvariable=size_var, width=8) entry.pack(padx=10, pady=(0, 10)) persist_var = tk.BooleanVar(value=bool(persist)) ttk.Checkbutton(dlg, text="Persist raw packets to Temp/", variable=persist_var).pack(padx=10, pady=(0, 10)) btn_frame = ttk.Frame(dlg) btn_frame.pack(padx=10, pady=(0, 10), fill=tk.X) def on_save(): try: v = int(size_var.get()) except Exception: try: messagebox.showerror("Invalid value", "Please enter a positive integer for history size.", parent=dlg) except Exception: pass return try: self.payload_router.set_history_size(v) self.payload_router.set_persist(bool(persist_var.get())) except Exception: pass try: gm = getattr(self.master, "config_manager", None) if gm: general = gm.get_general_settings() or {} sfp_debug = general.get("sfp_debug", {}) sfp_debug["history_size"] = v sfp_debug["persist_raw"] = bool(persist_var.get()) general["sfp_debug"] = sfp_debug gm.save_general_settings(general) except Exception: pass dlg.destroy() def on_cancel(): dlg.destroy() ttk.Button(btn_frame, text="Cancel", command=on_cancel).pack(side=tk.RIGHT, padx=(0, 5)) ttk.Button(btn_frame, text="Save", command=on_save).pack(side=tk.RIGHT) def _display_raw_packet(self, raw_bytes: bytes, addr: tuple): try: header_size = SFPHeader.size() if len(raw_bytes) < header_size: raise ValueError("Packet smaller than SFP header") header = SFPHeader.from_buffer_copy(raw_bytes) body = raw_bytes[header_size:] field_list = [ "SFP_MARKER", "SFP_DIRECTION", "SFP_PROT_VER", "SFP_PT_SPARE", "SFP_TAG", "SFP_SRC", "SFP_FLOW", "SFP_TID", "SFP_FLAGS", "SFP_WIN", "SFP_ERR", "SFP_ERR_INFO", "SFP_TOTFRGAS", "SFP_FRAG", "SFP_RECTYPE", "SFP_RECSPARE", "SFP_PLDAP", "SFP_PLEXT", "SFP_RECCOUNTER", "SFP_PLSIZE", "SFP_TOTSIZE", "SFP_PLOFFSET", ] pairs = [] flag_val = None for f in field_list: try: val = getattr(header, f) except Exception: val = "" if f == "SFP_FLAGS": flag_val = val pairs.append((f, f"{val} (0x{val:X})" if isinstance(val, int) else str(val))) continue if isinstance(val, int): pairs.append((f, f"{val} (0x{val:X})")) else: pairs.append((f, str(val))) out_lines = [f"From {addr}\n\nSFP Header:\n\n"] col_width = 36 for i in range(0, len(pairs), 2): left = pairs[i] right = pairs[i + 1] if (i + 1) < len(pairs) else None left_text = f"{left[0]:12s}: {left[1]}" if right: line = f"{left_text:<{col_width}} {right[0]:12s}: {right[1]}" else: line = left_text out_lines.append(line + "\n") flag_defs = [(0, "ACQ_REQ"), (1, "RESENT"), (2, "TRAILER_ACK"), (3, "RESV3"), (4, "RESV4"), (5, "RESV5"), (6, "RESV6"), (7, "ERROR")] out_lines.append(f"SFP_FLAGS : {flag_val} (0x{flag_val:X}) ") for bit, name in flag_defs: out_lines.append(f" [{(name if bool((flag_val >> bit) & 1) else ' ')}]") out_lines.append("\n\nFlags legend:\n") legend_map = {"ACQ_REQ": "Acquisition required/requested", "RESENT": "Fragment resent / please resend", "TRAILER_ACK": "Request trailer acknowledgement", "ERROR": "Packet-level error flag", "RESV3": "Reserved", "RESV4": "Reserved", "RESV5": "Reserved", "RESV6": "Reserved"} for _, name in flag_defs: out_lines.append(f" {name:12s}: {legend_map.get(name, '')}\n") out_lines.append("\nBODY (hex):\n") hex_dump = self._format_hex_dump(body) out_lines.append(hex_dump) full_text = "".join(out_lines) self.raw_tab_text.config(state=tk.NORMAL) self.raw_tab_text.delete("1.0", tk.END) header_block, _, _ = full_text.partition("\nBODY (hex):\n") self.raw_tab_text.insert(tk.END, header_block + "\n", "hdr_field") self.raw_tab_text.config(state=tk.DISABLED) try: self._display_hex_data(body, self.bin_tab) except Exception: self.bin_tab.config(state=tk.NORMAL) self.bin_tab.delete("1.0", tk.END) self.bin_tab.insert("1.0", hex_dump) self.bin_tab.config(state=tk.DISABLED) except Exception as e: text = f"Failed to format raw packet: {e}\n\nRaw dump:\n" text += self._format_hex_dump(raw_bytes) self.raw_tab_text.config(state=tk.NORMAL) self.raw_tab_text.delete("1.0", tk.END) self.raw_tab_text.insert("1.0", text) self.raw_tab_text.config(state=tk.DISABLED) def _display_image_data(self, payload: bytearray, tab_widgets: Dict[str, Any], photo_attr: str): try: if len(payload) < ctypes.sizeof(ImageLeaderData): raise ValueError("Payload smaller than ImageLeaderData header.") leader = ImageLeaderData.from_buffer(payload) h, w, bpp, stride = leader.HEADER_DATA.DY, leader.HEADER_DATA.DX, leader.HEADER_DATA.BPP, leader.HEADER_DATA.STRIDE offset = ctypes.sizeof(ImageLeaderData) if not (h > 0 and w > 0 and bpp in [1, 2] and stride >= w): raise ValueError(f"Invalid image dimensions: {w}x{h}, bpp={bpp}, stride={stride}") dtype = np.uint8 if bpp == 1 else np.uint16 expected_size = stride * h * bpp if (offset + expected_size) > len(payload): offset_fallback = ctypes.sizeof(SFPHeader) + ctypes.sizeof(ImageLeaderData) - ctypes.sizeof(leader.PIXEL_TAG) if (offset_fallback + expected_size) <= len(payload): offset = offset_fallback else: raise ValueError(f"Incomplete image data. Expected {expected_size}, got {len(payload) - offset}") pixel_data_view = np.ndarray(shape=(h, stride), dtype=dtype, buffer=payload, offset=offset) image_data = pixel_data_view[:, :w] display_img_8bit = cv2.normalize(image_data, None, 0, 255, cv2.NORM_MINMAX, cv2.CV_8U) img_pil = Image.fromarray(cv2.cvtColor(display_img_8bit, cv2.COLOR_GRAY2RGB)) resized = self._resize_pil_to_label(img_pil, tab_widgets.get("image_container", tab_widgets["image_label"])) photo = ImageTk.PhotoImage(image=resized) tab_widgets["image_label"].config(image=photo, text="") setattr(self, photo_attr, photo) except Exception as e: self.logger.error(f"Error parsing image payload: {e}") tab_widgets["image_label"].config(image=None, text=f"Error parsing image:\n{e}") setattr(self, photo_attr, None) try: self._display_hex_data(payload, self.bin_tab) except Exception: try: self._display_hex_data(payload, tab_widgets["hex_view"]) except Exception: pass def _resize_pil_to_label(self, img: "Image.Image", label_widget: ttk.Label) -> "Image.Image": try: width, height = label_widget.winfo_width(), label_widget.winfo_height() if width <= 1 or height <= 1: return img img_w, img_h = img.size scale = min(width / img_w, height / img_h) if scale >= 1.0: return img new_w, new_h = max(1, int(img_w * scale)), max(1, int(img_h * scale)) return img.resize((new_w, new_h), Image.LANCZOS) except Exception: return img def _display_hex_data(self, payload: bytearray, widget: scrolledtext.ScrolledText): hex_dump = self._format_hex_dump(payload) widget.config(state=tk.NORMAL) widget.delete("1.0", tk.END) widget.insert("1.0", hex_dump) widget.config(state=tk.DISABLED) def _display_json_data(self, payload: bytearray, widget: scrolledtext.ScrolledText): try: import json text = json.dumps(json.loads(payload.decode("utf-8")), indent=2) except Exception as e: text = f"--- FAILED TO PARSE JSON ---\n{e}\n\n--- RAW HEX DUMP ---" text += self._format_hex_dump(payload) widget.config(state=tk.NORMAL) widget.delete("1.0", tk.END) widget.insert("1.0", text) widget.config(state=tk.DISABLED) def _log_to_widget(self, message: str, level: str = "DEBUG"): self.logger.info(message) self.log_tab.config(state=tk.NORMAL) self.log_tab.insert(tk.END, f"[{level}] {message}\n") self.log_tab.config(state=tk.DISABLED) self.log_tab.see(tk.END) def _on_save_ris_csv(self): try: import csv scenario_rows = [self.scenario_tree.item(iid, "values") for iid in self.scenario_tree.get_children()] rows = [self.ris_tree.item(iid, "values") for iid in self.ris_tree.get_children()] if not scenario_rows and not rows: self._log_to_widget("No RIS data to save.", "INFO"); return project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) temp_dir = os.path.join(project_root, "Temp") os.makedirs(temp_dir, exist_ok=True) ts = datetime.datetime.utcnow().strftime("%Y%m%dT%H%M%S") fname = f"ris_targets_{ts}.csv" path = os.path.join(temp_dir, fname) with open(path, "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) if scenario_rows: writer.writerow(["Scenario Field", "Value"]) for s in scenario_rows: writer.writerow(s) writer.writerow([]) writer.writerow(["index", "flags", "heading", "x", "y", "z"]) for r in rows: writer.writerow(r) self._log_to_widget(f"Saved RIS targets CSV to {path}", "INFO") except Exception as e: self._log_to_widget(f"Failed to save RIS CSV: {e}", "ERROR") def _format_hex_dump(self, data: bytes, length=16) -> str: lines = [] for i in range(0, len(data), length): chunk = data[i : i + length] hex_part = " ".join(f"{b:02X}" for b in chunk) ascii_part = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk) lines.append(f"{i:08X} {hex_part:<{length*3}} |{ascii_part}|") return "\n".join(lines)