S1005403_RisCC/target_simulator/gui/main_view.py
VALLONGOL 3b447d7fdb fix save/new/load scenario
minor fix in gui
2025-10-07 14:40:52 +02:00

461 lines
23 KiB
Python

# target_simulator/gui/main_view.py
"""
Main view of the application, containing the primary window and widgets.
"""
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
from typing import Optional, Dict, Any
from .ppi_display import PPIDisplay
from .settings_window import SettingsWindow
from .connection_settings_window import ConnectionSettingsWindow
from .radar_config_window import RadarConfigWindow
from .scenario_controls_frame import ScenarioControlsFrame
from .target_list_frame import TargetListFrame
from .add_target_window import AddTargetWindow
from core.communicator_interface import CommunicatorInterface
from core.serial_communicator import SerialCommunicator
from core.tftp_communicator import TFTPCommunicator
from core.models import Scenario
from utils.logger import get_logger, shutdown_logging_system
from utils.config_manager import ConfigManager
from target_simulator.utils.logger import get_logger, shutdown_logging_system
from target_simulator.utils.config_manager import ConfigManager
class MainView(tk.Tk):
def _on_new_scenario(self, scenario_name):
"""
Handles creation of a new scenario:
- If scenario name exists, notify user and load it.
- If not, create new scenario, update combobox, select, and clear targets.
"""
scenario_names = self.config_manager.get_scenario_names()
if scenario_name in scenario_names:
messagebox.showinfo(
"Duplicate Scenario",
f"Scenario '{scenario_name}' already exists. Loading saved data.",
parent=self
)
self._on_load_scenario(scenario_name)
# Select in comboboxes
self.scenario_controls.update_scenario_list(scenario_names, scenario_name)
if hasattr(self, 'sim_scenario_combobox'):
self.sim_scenario_combobox['values'] = scenario_names
self.sim_scenario_combobox.set(scenario_name)
return
# Create new scenario
self.scenario = Scenario(name=scenario_name)
self.current_scenario_name = scenario_name
# Clear targets in UI
self.target_list.update_target_list([])
# Update comboboxes
scenario_names.append(scenario_name)
self.scenario_controls.update_scenario_list(scenario_names, scenario_name)
if hasattr(self, 'sim_scenario_combobox'):
self.sim_scenario_combobox['values'] = scenario_names
self.sim_scenario_combobox.set(scenario_name)
self._update_window_title()
self.logger.info(f"Created new scenario '{scenario_name}'. Ready for target input.")
messagebox.showinfo(
"New Scenario",
f"Scenario '{scenario_name}' created. Add targets to begin.",
parent=self
)
"""The main application window."""
def __init__(self):
super().__init__()
self.logger = get_logger(__name__)
self.config_manager = ConfigManager()
settings = self.config_manager.get_general_settings()
self.scan_limit = settings.get('scan_limit', 60)
self.max_range = settings.get('max_range', 100)
self.connection_config = self.config_manager.get_connection_settings()
self.title("Radar Target Simulator")
self.geometry(settings.get('geometry', '1200x1024'))
self.minsize(1024, 1024)
self.target_communicator = None
self.lru_communicator = None
self.current_scenario_name = None
self._create_main_layout()
self._create_menubar()
self._create_statusbar()
self._initialize_communicator() # Attempt to connect with saved settings
def _create_main_layout(self):
v_pane = ttk.PanedWindow(self, orient=tk.VERTICAL)
v_pane.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.h_pane = ttk.PanedWindow(v_pane, orient=tk.HORIZONTAL)
v_pane.add(self.h_pane, weight=4)
# --- Right Pane (PPI) ---
self.ppi_widget = PPIDisplay(self.h_pane, max_range_nm=self.max_range, scan_limit_deg=self.scan_limit)
self.h_pane.add(self.ppi_widget, weight=2)
# Add Connect button next to PPI controls
# Find the controls_frame inside PPIDisplay and add the button
controls_frame = self.ppi_widget.children.get('!frame')
if controls_frame:
connect_btn = ttk.Button(controls_frame, text="Connect", command=self._on_connect_button)
connect_btn.pack(side=tk.RIGHT, padx=10)
# --- Bottom Pane (Logs) ---
log_frame_container = ttk.LabelFrame(v_pane, text="Logs")
v_pane.add(log_frame_container, weight=1)
self.log_text_widget = scrolledtext.ScrolledText(log_frame_container, state=tk.DISABLED, wrap=tk.WORD, font=("Consolas", 9))
self.log_text_widget.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
# --- Left Pane ---
left_pane_container = ttk.Frame(self.h_pane)
self.h_pane.add(left_pane_container, weight=1)
# --- Notebook for left pane ---
left_notebook = ttk.Notebook(left_pane_container)
left_notebook.pack(fill=tk.BOTH, expand=True)
# --- TAB 2: SCENARIO CONFIG ---
if not hasattr(self, 'scenario_tab'):
self.scenario_tab = ttk.Frame(left_notebook)
left_notebook.add(self.scenario_tab, text="Scenario Config")
self.scenario_controls = ScenarioControlsFrame(self.scenario_tab,
main_view=self,
load_scenario_command=self._on_load_scenario,
save_as_command=self._on_save_scenario_as,
delete_command=self._on_delete_scenario,
new_scenario_command=self._on_new_scenario)
self.scenario_controls.pack(fill=tk.X, expand=False, padx=5, pady=(5, 5))
self.target_list = TargetListFrame(self.scenario_tab, targets_changed_callback=self._on_targets_changed)
self.target_list.pack(fill=tk.BOTH, expand=True, padx=5)
# --- TAB 3: SIMULATION ---
if not hasattr(self, 'simulation_tab'):
self.simulation_tab = ttk.Frame(left_notebook)
left_notebook.add(self.simulation_tab, text="Simulation")
sim_controls_frame = ttk.LabelFrame(self.simulation_tab, text="Simulation Controls")
sim_controls_frame.pack(fill=tk.X, padx=5, pady=5, anchor='n')
ttk.Label(sim_controls_frame, text="Scenario:").pack(side=tk.LEFT, padx=(5, 5), pady=5)
self.sim_scenario_combobox = ttk.Combobox(
sim_controls_frame,
textvariable=self.scenario_controls.current_scenario,
state="readonly"
)
self.sim_scenario_combobox.pack(side=tk.LEFT, expand=True, fill=tk.X, pady=5)
self.sim_scenario_combobox.bind("<<ComboboxSelected>>", lambda event: self._on_load_scenario(self.sim_scenario_combobox.get()))
send_button = ttk.Button(sim_controls_frame, text="Send to Radar", command=self._on_send_scenario)
send_button.pack(side=tk.LEFT, padx=5, pady=5)
# --- TAB 4: LRU SIMULATION ---
if not hasattr(self, 'lru_tab'):
self.lru_tab = ttk.Frame(left_notebook)
left_notebook.add(self.lru_tab, text="LRU Simulation")
cooling_frame = ttk.LabelFrame(self.lru_tab, text="Cooling Unit Status")
cooling_frame.pack(fill=tk.X, padx=5, pady=5, anchor='n')
ttk.Label(cooling_frame, text="Status:").pack(side=tk.LEFT, padx=5, pady=5)
self.cooling_status_var = tk.StringVar(value="OK")
cooling_combo = ttk.Combobox(
cooling_frame,
textvariable=self.cooling_status_var,
values=["OK", "OVERHEATING", "FAULT"],
state="readonly"
)
cooling_combo.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=5, pady=5)
power_frame = ttk.LabelFrame(self.lru_tab, text="Power Supply Unit Status")
power_frame.pack(fill=tk.X, padx=5, pady=5, anchor='n')
ttk.Label(power_frame, text="Status:").pack(side=tk.LEFT, padx=5, pady=5)
self.power_status_var = tk.StringVar(value="OK")
power_combo = ttk.Combobox(
power_frame,
textvariable=self.power_status_var,
values=["OK", "LOW_VOLTAGE", "FAULT"],
state="readonly"
)
power_combo.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=5, pady=5)
lru_action_frame = ttk.Frame(self.lru_tab)
lru_action_frame.pack(fill=tk.X, padx=5, pady=10, anchor='n')
self.lru_message_var = tk.StringVar()
ttk.Entry(lru_action_frame, textvariable=self.lru_message_var, state="readonly").pack(side=tk.LEFT, expand=True, fill=tk.X, padx=(0, 5))
send_lru_button = ttk.Button(lru_action_frame, text="Send LRU Status", command=self._on_send_lru_status)
send_lru_button.pack(side=tk.LEFT)
# Carica gli scenari solo dopo aver creato scenario_controls
self._load_scenarios_into_ui()
# Rimuovi la voce di menu per la configurazione
def _on_targets_changed(self, targets):
# Called by TargetListFrame when targets are changed
self.ppi_widget.update_targets(targets)
def _create_statusbar(self):
status_bar = ttk.Frame(self, relief=tk.SUNKEN)
status_bar.pack(side=tk.BOTTOM, fill=tk.X)
ttk.Label(status_bar, text="Target:").pack(side=tk.LEFT, padx=(5, 2))
self.target_status_canvas = tk.Canvas(status_bar, width=16, height=16, highlightthickness=0)
self.target_status_canvas.pack(side=tk.LEFT, padx=(0, 10))
self._draw_status_indicator(self.target_status_canvas, "#e74c3c") # Default to red
ttk.Label(status_bar, text="LRU:").pack(side=tk.LEFT, padx=(5, 2))
self.lru_status_canvas = tk.Canvas(status_bar, width=16, height=16, highlightthickness=0)
self.lru_status_canvas.pack(side=tk.LEFT, padx=(0, 10))
self._draw_status_indicator(self.lru_status_canvas, "#e74c3c") # Default to red
self.status_var = tk.StringVar(value="Ready")
ttk.Label(status_bar, textvariable=self.status_var, anchor=tk.W).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
def _on_connect_button(self):
try:
self.logger.info("Connection requested by user.")
self._initialize_communicator()
except Exception as e:
self.logger.error(f"Error during connection: {e}")
self._log_message(f"Error during connection: {e}")
def _log_message(self, message):
self.log_text_widget.configure(state=tk.NORMAL)
self.log_text_widget.insert(tk.END, message + "\n")
self.log_text_widget.configure(state=tk.DISABLED)
self.log_text_widget.see(tk.END)
def _draw_status_indicator(self, canvas, color):
canvas.delete("all")
canvas.create_oval(2, 2, 14, 14, fill=color, outline="black")
def _create_menubar(self):
menubar = tk.Menu(self)
self.config(menu=menubar)
# Solo Settings Menu
settings_menu = tk.Menu(menubar, tearoff=0)
menubar.add_cascade(label="Settings", menu=settings_menu)
settings_menu.add_command(label="Connection...", command=self._open_settings)
settings_menu.add_command(label="Radar Config...", command=self._open_radar_config)
def _initialize_communicator(self):
"""Creates and connects the appropriate communicators based on config."""
# Disconnect if already connected
if self.target_communicator and self.target_communicator.is_open:
self.target_communicator.disconnect()
if self.lru_communicator and self.lru_communicator.is_open:
self.lru_communicator.disconnect()
# --- Initialize Target Communicator ---
# --- Initialize Target Communicator ---
target_cfg = self.connection_config.get("target", {})
target_comm_type = target_cfg.get("type")
self.logger.info(f"Initializing Target communicator of type: {target_comm_type}")
self.target_communicator = None
config_data = None
if target_comm_type == "serial":
config_data = target_cfg.get("serial", {})
port = config_data.get("port")
baudrate = config_data.get("baudrate")
if isinstance(port, str) and port and isinstance(baudrate, int):
self.target_communicator = SerialCommunicator()
if self.target_communicator.connect(config_data):
self._draw_status_indicator(self.target_status_canvas, "#2ecc40") # green
self.status_var.set("Target: Connected")
else:
self._draw_status_indicator(self.target_status_canvas, "#e74c3c") # red
self.status_var.set("Target: Disconnected")
else:
self.logger.error(f"Skipped Target serial connection: invalid config {config_data}")
self._draw_status_indicator(self.target_status_canvas, "#e74c3c")
self.status_var.set("Target: Invalid config")
elif target_comm_type == "tftp":
config_data = target_cfg.get("tftp", {})
ip = config_data.get("ip")
port = config_data.get("port")
if isinstance(ip, str) and ip and isinstance(port, int):
self.target_communicator = TFTPCommunicator()
if self.target_communicator.connect(config_data):
self._draw_status_indicator(self.target_status_canvas, "#2ecc40") # green
self.status_var.set("Target: Connected")
else:
self._draw_status_indicator(self.target_status_canvas, "#e74c3c") # red
self.status_var.set("Target: Disconnected")
else:
self.logger.error(f"Skipped Target TFTP connection: invalid config {config_data}")
self._draw_status_indicator(self.target_status_canvas, "#e74c3c")
self.status_var.set("Target: Invalid config")
# --- Initialize LRU Communicator ---
lru_cfg = self.connection_config.get("lru", {})
lru_comm_type = lru_cfg.get("type")
self.logger.info(f"Initializing LRU communicator of type: {lru_comm_type}")
self.lru_communicator = None
lru_config_data = None
if lru_comm_type == "serial":
lru_config_data = lru_cfg.get("serial", {})
port = lru_config_data.get("port")
baudrate = lru_config_data.get("baudrate")
if isinstance(port, str) and port and isinstance(baudrate, int):
self.lru_communicator = SerialCommunicator()
if self.lru_communicator.connect(lru_config_data):
self._draw_status_indicator(self.lru_status_canvas, "#2ecc40") # green
else:
self._draw_status_indicator(self.lru_status_canvas, "#e74c3c") # red
else:
self.logger.error(f"Skipped LRU serial connection: invalid config {lru_config_data}")
self._draw_status_indicator(self.lru_status_canvas, "#e74c3c")
elif lru_comm_type == "tftp":
lru_config_data = lru_cfg.get("tftp", {})
ip = lru_config_data.get("ip")
port = lru_config_data.get("port")
if isinstance(ip, str) and ip and isinstance(port, int):
self.lru_communicator = TFTPCommunicator()
if self.lru_communicator.connect(lru_config_data):
self._draw_status_indicator(self.lru_status_canvas, "#2ecc40") # green
else:
self._draw_status_indicator(self.lru_status_canvas, "#e74c3c") # red
else:
self.logger.error(f"Skipped LRU TFTP connection: invalid config {lru_config_data}")
self._draw_status_indicator(self.lru_status_canvas, "#e74c3c")
def update_connection_settings(self, new_config: Dict[str, Any]):
"""Callback from SettingsWindow to apply and save new connection settings."""
self.logger.info(f"Updating connection settings: {new_config}")
self.connection_config = new_config
self.config_manager.save_connection_settings(new_config)
self._initialize_communicator() # Re-initialize with new settings
def _open_settings(self):
self.logger.info("Opening connection settings window.")
ConnectionSettingsWindow(self, self.config_manager, self.connection_config)
def _on_send_scenario(self):
if self.target_communicator and self.target_communicator.is_open:
if self.scenario and self.scenario.get_all_targets():
self.target_communicator.send_scenario(self.scenario)
else:
messagebox.showinfo("Empty Scenario", "Cannot send an empty scenario.", parent=self)
else:
messagebox.showerror("Not Connected", "Target communicator is not connected. Please check settings.", parent=self)
def _on_send_lru_status(self):
cooling_status = self.cooling_status_var.get()
power_status = self.power_status_var.get()
# Assume a simple CSV-like format for the message
message = f"LRU_STATUS,COOLING={cooling_status},POWER={power_status}"
self.lru_message_var.set(message)
self.logger.info(f"Formatted LRU status message: {message}")
if self.lru_communicator and self.lru_communicator.is_open:
if hasattr(self.lru_communicator, 'send_raw_message'):
self.lru_communicator.send_raw_message(message)
self.logger.info("Sent LRU status message to radar.")
messagebox.showinfo("LRU Status Sent", f"Message sent:\n{message}", parent=self)
else:
self.logger.warning("Communicator does not have a 'send_raw_message' method.")
messagebox.showwarning("Not Supported", "The current communicator does not support sending raw messages.", parent=self)
else:
self.logger.warning("Attempted to send LRU status while disconnected.")
messagebox.showerror("Not Connected", "LRU communicator is not connected. Please check settings.", parent=self)
# --- Other methods remain largely the same ---
def _update_window_title(self):
title = "Radar Target Simulator"
if self.current_scenario_name:
title = f"{self.current_scenario_name} - {title}"
self.title(title)
def _update_all_views(self):
self._update_window_title()
all_targets = self.scenario.get_all_targets()
self.target_list.update_target_list(all_targets)
self.ppi_widget.update_targets(all_targets)
self.logger.info(f"Views updated for scenario '{self.current_scenario_name}'. It has {len(all_targets)} target(s).")
def _load_scenarios_into_ui(self):
scenario_names = self.config_manager.get_scenario_names()
self.scenario_controls.update_scenario_list(scenario_names, self.current_scenario_name)
# Also update the values for the simulation tab combobox
if hasattr(self, 'sim_scenario_combobox'):
self.sim_scenario_combobox['values'] = scenario_names
# Rimosso: la logica di aggiunta target è ora nel TargetListFrame
def _on_load_scenario(self, scenario_name: str):
self.logger.info(f"Loading scenario: {scenario_name}")
scenario_data = self.config_manager.get_scenario(scenario_name)
if scenario_data:
try:
self.scenario = Scenario.from_dict(scenario_data)
self.current_scenario_name = scenario_name
self._update_all_views()
except Exception as e:
self.logger.error(f"Failed to parse scenario data for '{scenario_name}': {e}")
messagebox.showerror("Load Error", f"Could not load scenario '{scenario_name}'.\n{e}")
else:
self.logger.warning(f"Attempted to load a non-existent scenario: {scenario_name}")
def _on_save_scenario_as(self, scenario_name: str):
self.logger.info(f"Saving current state as new scenario: {scenario_name}")
scenario_data = self.scenario.to_dict()
self.config_manager.save_scenario(scenario_name, scenario_data)
self.current_scenario_name = scenario_name
self._load_scenarios_into_ui() # Refresh list
self._update_window_title()
messagebox.showinfo("Success", f"Scenario '{scenario_name}' saved successfully.", parent=self)
def _on_save_scenario(self, scenario_name: str):
"""
Save the current scenario and its targets (from the target list) under the selected name.
"""
self.logger.info(f"Saving scenario: {scenario_name}")
# Update scenario's targets from the target list
self.scenario.targets = {t.target_id: t for t in self.target_list.get_targets()}
scenario_data = self.scenario.to_dict()
self.config_manager.save_scenario(scenario_name, scenario_data)
self.current_scenario_name = scenario_name
self._load_scenarios_into_ui() # Refresh list
self._update_window_title()
messagebox.showinfo("Success", f"Scenario '{scenario_name}' saved successfully.", parent=self)
def _on_delete_scenario(self, scenario_name: str):
self.logger.info(f"Deleting scenario: {scenario_name}")
self.config_manager.delete_scenario(scenario_name)
self.logger.info(f"Scenario '{scenario_name}' deleted.")
if self.current_scenario_name == scenario_name:
self.scenario = Scenario()
self.current_scenario_name = None
self._update_all_views()
self._load_scenarios_into_ui()
messagebox.showinfo("Deleted", f"Scenario '{scenario_name}' has been deleted.", parent=self)
def _open_radar_config(self):
self.logger.info("Opening radar config window.")
dialog = RadarConfigWindow(self, current_scan_limit=self.scan_limit, current_max_range=self.max_range)
if dialog.scan_limit is not None and dialog.max_range is not None:
self.scan_limit = dialog.scan_limit
self.max_range = dialog.max_range
self.logger.info(f"Scan limit set to: ±{self.scan_limit} degrees")
self.logger.info(f"Max range set to: {self.max_range} NM")
self.ppi_widget.destroy()
self.ppi_widget = PPIDisplay(self.h_pane, max_range_nm=self.max_range, scan_limit_deg=self.scan_limit)
self.h_pane.add(self.ppi_widget, weight=2)
self._update_all_views()
def _on_closing(self):
self.logger.info("Application shutting down.")
# Save general settings
settings_to_save = {
'scan_limit': self.scan_limit,
'max_range': self.max_range,
'geometry': self.winfo_geometry(),
'last_selected_scenario': self.current_scenario_name
}
self.config_manager.save_general_settings(settings_to_save)
self.config_manager.save_connection_settings(self.connection_config)
if self.target_communicator and self.target_communicator.is_open:
self.target_communicator.disconnect()
if self.lru_communicator and self.lru_communicator.is_open:
self.lru_communicator.disconnect()
shutdown_logging_system()
self.destroy()