# FlightMonitor/controller/app_controller.py import threading import json import os import time import sys import subprocess from queue import Queue, Empty as QueueEmpty from typing import List, Optional, Dict, Any, TYPE_CHECKING from tkinter import messagebox, simpledialog # Import dei processori e degli adapter from flightmonitor.controller.historical_data_processor import HistoricalDataProcessor from flightmonitor.controller.live_data_processor import LiveDataProcessor from flightmonitor.controller.playback_data_processor import PlaybackDataProcessor from flightmonitor.data.opensky_live_adapter import OpenSkyLiveAdapter, AdapterMessage from flightmonitor.data.opensky_historical_adapter import OpenSkyHistoricalAdapter from flightmonitor.data.playback_adapter import PlaybackAdapter # Import dei componenti di base e di configurazione from flightmonitor.data import config as app_config from flightmonitor.utils.logger import get_logger from flightmonitor.data.storage import DataStorage from flightmonitor.data.aircraft_database_manager import AircraftDatabaseManager from flightmonitor.utils.gui_utils import ( GUI_STATUS_OK, GUI_STATUS_WARNING, GUI_STATUS_ERROR, GUI_STATUS_FETCHING, ) from flightmonitor.controller.aircraft_db_importer import AircraftDBImporter from flightmonitor.controller.map_command_handler import MapCommandHandler from flightmonitor.controller.cleanup_manager import CleanupManager from flightmonitor.controller.raw_data_logger import RawDataLogger from flightmonitor.data.area_profile_manager import AreaProfileManager if TYPE_CHECKING: from flightmonitor.gui.main_window import MainWindow from flightmonitor.gui.dialogs.import_progress_dialog import ImportProgressDialog from flightmonitor.gui.dialogs.full_flight_details_window import ( FullFlightDetailsWindow, ) from flightmonitor.data.common_models import CanonicalFlightState module_logger = get_logger(__name__) ADAPTER_JOIN_TIMEOUT_SECONDS = 5.0 SETTINGS_FILENAME = "app_settings.json" class AppController: """ Orchestrates the application's logic, acting as an intermediary between the GUI (View) and the data layer (Model). """ def __init__(self): self.main_window: Optional["MainWindow"] = None # Live Monitoring State self.live_adapter_thread: Optional[OpenSkyLiveAdapter] = None self.is_live_monitoring_active: bool = False self.live_data_queue: Optional[Queue[AdapterMessage]] = None self.live_data_processor: Optional[LiveDataProcessor] = None self._current_live_session_id: Optional[int] = None # Historical Download State self.historical_adapter_thread: Optional[OpenSkyHistoricalAdapter] = None self.is_historical_download_active: bool = False self.historical_data_queue: Optional[Queue[AdapterMessage]] = None self.historical_data_processor: Optional[HistoricalDataProcessor] = None # Playback State self.playback_adapter_thread: Optional[PlaybackAdapter] = None self.is_playback_active: bool = False self.playback_data_queue: Optional[Queue[AdapterMessage]] = None self.playback_data_processor: Optional[PlaybackDataProcessor] = None # Common attributes self._current_scan_params: Optional[Dict[str, Any]] = None self._active_bounding_box: Optional[Dict[str, float]] = None self.data_storage: Optional[DataStorage] = None self.aircraft_db_manager: Optional[AircraftDatabaseManager] = None self.aircraft_db_importer: Optional[AircraftDBImporter] = None self.map_command_handler: Optional[MapCommandHandler] = None self.cleanup_manager: Optional[CleanupManager] = None self.profile_manager: Optional[AreaProfileManager] = None self.raw_data_logger: Optional[RawDataLogger] = RawDataLogger() self.active_detail_window_icao: Optional[str] = None self.active_detail_window_ref: Optional["FullFlightDetailsWindow"] = None self._app_settings: Dict[str, Any] = {} # Initialization of data layer components try: self.data_storage = DataStorage() module_logger.info("DataStorage initialized successfully.") except Exception as e: module_logger.critical( f"CRITICAL: Failed to initialize DataStorage: {e}", exc_info=True ) self.data_storage = None try: self.aircraft_db_manager = AircraftDatabaseManager() module_logger.info("AircraftDatabaseManager initialized successfully.") except Exception as e: module_logger.critical( f"CRITICAL: Failed to initialize AircraftDatabaseManager: {e}", exc_info=True, ) self.aircraft_db_manager = None try: self.profile_manager = AreaProfileManager() module_logger.info("AreaProfileManager initialized successfully.") except Exception as e: module_logger.error( f"Failed to initialize AreaProfileManager: {e}", exc_info=True ) module_logger.info("AppController initialized.") def _load_app_settings(self): default_settings = { "raw_logging_enabled": app_config.RAW_DATA_LOGGING_ENABLED, "raw_logging_directory": app_config.RAW_DATA_LOGGING_DEFAULT_DIR, } try: if os.path.exists(SETTINGS_FILENAME): with open(SETTINGS_FILENAME, "r") as f: loaded_settings = json.load(f) self._app_settings = {**default_settings, **loaded_settings} module_logger.info("Application settings loaded successfully.") else: self._app_settings = default_settings module_logger.info("Settings file not found. Using default settings.") except (json.JSONDecodeError, IOError) as e: module_logger.error( f"Failed to load settings from {SETTINGS_FILENAME}: {e}. Using defaults." ) self._app_settings = default_settings def _save_app_settings(self): if not self.main_window: return logging_settings = ( self.main_window.function_notebook_panel.get_logging_panel_settings() ) if logging_settings: self._app_settings["raw_logging_enabled"] = logging_settings.get( "enabled", False ) self._app_settings["raw_logging_directory"] = logging_settings.get( "directory", "" ) try: with open(SETTINGS_FILENAME, "w") as f: json.dump(self._app_settings, f, indent=4) module_logger.info(f"Application settings saved to {SETTINGS_FILENAME}.") except IOError as e: module_logger.error(f"Failed to save settings to {SETTINGS_FILENAME}: {e}") def set_main_window(self, main_window_instance: "MainWindow"): self.main_window = main_window_instance module_logger.debug(f"Main window instance set in AppController.") if self.aircraft_db_manager and self.main_window: self.aircraft_db_importer = AircraftDBImporter( self.aircraft_db_manager, self.main_window ) module_logger.info("AircraftDBImporter initialized successfully.") else: module_logger.warning("AircraftDBImporter could not be initialized.") self.live_data_queue = Queue(maxsize=200) self.historical_data_queue = Queue(maxsize=500) self.playback_data_queue = Queue(maxsize=500) self.live_data_processor = LiveDataProcessor(self, self.live_data_queue) self.historical_data_processor = HistoricalDataProcessor( self, self.historical_data_queue ) self.playback_data_processor = PlaybackDataProcessor( self, self.playback_data_queue ) self.map_command_handler = MapCommandHandler(self) self.cleanup_manager = CleanupManager(self) module_logger.info("Controller sub-components initialized.") self._load_app_settings() if self.main_window and hasattr( self.main_window.function_notebook_panel, "update_logging_panel_settings" ): logging_settings_to_apply = { "enabled": self._app_settings.get("raw_logging_enabled"), "directory": self._app_settings.get("raw_logging_directory"), } self.main_window.function_notebook_panel.update_logging_panel_settings( logging_settings_to_apply ) initial_status_msg = "System Initialized. Ready." initial_status_level = GUI_STATUS_OK if not self.data_storage: err_msg_ds = "Data storage init failed. History will not be saved." initial_status_msg, initial_status_level = err_msg_ds, GUI_STATUS_ERROR if not self.aircraft_db_manager: err_msg_adb = "Aircraft DB init failed. Static details may be unavailable." if initial_status_level == GUI_STATUS_OK: initial_status_msg, initial_status_level = ( err_msg_adb, GUI_STATUS_WARNING, ) else: initial_status_msg += f" {err_msg_adb}" initial_status_level = GUI_STATUS_ERROR if self.main_window and self.main_window.root.winfo_exists(): self.main_window.update_semaphore_and_status( initial_status_level, initial_status_msg ) def start_live_monitoring(self): if not self.main_window: return bounding_box = self.main_window.get_bounding_box_from_gui() if not bounding_box: err_msg = "Invalid or missing Bounding Box. Define monitoring area." self.main_window.show_error_message("Input Error", err_msg) if hasattr(self.main_window, "_reset_gui_to_stopped_state"): self.main_window._reset_gui_to_stopped_state(f"Start failed: {err_msg}") return logging_settings = ( self.main_window.function_notebook_panel.get_logging_panel_settings() ) if logging_settings and logging_settings["enabled"]: log_dir = logging_settings.get("directory") if not log_dir: self.main_window.show_error_message( "Logging Error", "Logging is enabled but no save directory is specified.", ) return if self.raw_data_logger and not self.raw_data_logger.start_logging_session( log_dir, bounding_box ): self.main_window.show_error_message( "Logging Error", "Failed to start raw data logging session. Check logs.", ) return if hasattr(self.main_window.function_notebook_panel, "clear_all_panel_data"): self.main_window.function_notebook_panel.clear_all_panel_data() if hasattr(self.main_window, "clear_all_views_data"): self.main_window.clear_all_views_data() if hasattr(self.main_window, "set_monitoring_button_states"): self.main_window.set_monitoring_button_states(True) if self.is_live_monitoring_active: module_logger.warning("Live monitoring requested but already active.") return module_logger.info( f"Controller: Starting live monitoring for bbox: {bounding_box}" ) self._active_bounding_box = bounding_box if self.main_window.map_manager_instance: self.main_window.map_manager_instance.set_target_bbox(bounding_box) if self.live_data_queue: while not self.live_data_queue.empty(): try: self.live_data_queue.get_nowait() except QueueEmpty: break if not self.live_data_processor: module_logger.critical("Controller: LiveDataProcessor not initialized.") if hasattr(self.main_window, "_reset_gui_to_stopped_state"): self.main_window._reset_gui_to_stopped_state( "Start failed: Internal error." ) return self.live_adapter_thread = OpenSkyLiveAdapter( self.live_data_queue, self._active_bounding_box, app_config.LIVE_POLLING_INTERVAL_SECONDS, ) self.is_live_monitoring_active = True self.live_adapter_thread.start() self.live_data_processor.start_processing_queue() if self.aircraft_db_manager and self._active_bounding_box: session_params = { "start_time": time.time(), "end_time": time.time(), "sampling_interval_sec": None, "scan_rate_sec": app_config.LIVE_POLLING_INTERVAL_SECONDS, } self._current_live_session_id = self.aircraft_db_manager.add_scan_history( session_params, self._active_bounding_box, status="running" ) def stop_live_monitoring(self, from_error: bool = False): if not self.is_live_monitoring_active and not from_error: if self.main_window and hasattr( self.main_window, "_reset_gui_to_stopped_state" ): self.main_window._reset_gui_to_stopped_state( "Monitoring already stopped." ) return module_logger.info( f"Controller: Stopping live monitoring (from_error={from_error})." ) self.is_live_monitoring_active = False if ( self.aircraft_db_manager and self._current_live_session_id is not None and self._active_bounding_box ): session_params = { "start_time": 0, "end_time": time.time(), "sampling_interval_sec": None, "scan_rate_sec": app_config.LIVE_POLLING_INTERVAL_SECONDS, } status = "failed" if from_error else "completed" self.aircraft_db_manager.add_scan_history( session_params, self._active_bounding_box, status, scan_id=self._current_live_session_id, ) self._current_live_session_id = None if self.live_data_processor: self.live_data_processor.stop_processing_queue() if self.live_adapter_thread and self.live_adapter_thread.is_alive(): self.live_adapter_thread.stop() self.live_adapter_thread.join(ADAPTER_JOIN_TIMEOUT_SECONDS) self.live_adapter_thread = None if self.raw_data_logger and self.raw_data_logger.is_active: self.raw_data_logger.stop_logging_session() if self.main_window and hasattr( self.main_window, "_reset_gui_to_stopped_state" ): msg = ( "Monitoring stopped due to an error." if from_error else "Monitoring stopped." ) self.main_window._reset_gui_to_stopped_state(msg) def process_raw_data_logging( self, raw_json_string: str, canonical_data: List["CanonicalFlightState"] ): if self.raw_data_logger and self.raw_data_logger.is_active: self.raw_data_logger.log_raw_data(raw_json_string) def update_live_summary_table(self, timestamp: float, aircraft_count: int): if self.raw_data_logger and self.raw_data_logger.is_active: self.raw_data_logger.log_summary_data(timestamp, aircraft_count) if self.main_window and hasattr( self.main_window.function_notebook_panel, "data_logging_panel" ): panel = self.main_window.function_notebook_panel.data_logging_panel if panel: panel.add_summary_entry(timestamp, aircraft_count) panel.update_last_query_result(timestamp, aircraft_count) def open_log_directory(self, directory_path: str): if not directory_path: self.main_window.show_error_message("Error", "No directory path specified.") return abs_path = os.path.abspath(directory_path) if not os.path.isdir(abs_path): self.main_window.show_error_message( "Error", f"Directory does not exist:\n{abs_path}" ) return try: if sys.platform == "win32": os.startfile(abs_path) elif sys.platform == "darwin": subprocess.Popen(["open", abs_path]) else: subprocess.Popen(["xdg-open", abs_path]) module_logger.info(f"Opening log directory: {abs_path}") except Exception as e: module_logger.error(f"Failed to open directory '{abs_path}': {e}") self.main_window.show_error_message( "Error", f"Could not open directory:\n{e}" ) def start_historical_download(self): if not self.main_window or not hasattr( self.main_window.function_notebook_panel, "historical_panel" ): module_logger.error( "Controller: Main window or historical panel not available." ) return if self.is_historical_download_active: module_logger.warning( "Historical download requested but is already active." ) return historical_panel = self.main_window.function_notebook_panel.historical_panel if not historical_panel: module_logger.error("Historical download panel not found in GUI.") return params = historical_panel.get_download_parameters() if not params: self.main_window.show_error_message( "Input Error", "Invalid download parameters." ) return bbox = self.main_window.get_bounding_box_from_gui() if not bbox: self.main_window.show_error_message( "Input Error", "A valid Bounding Box must be defined." ) return logging_settings = ( self.main_window.function_notebook_panel.get_logging_panel_settings() ) if logging_settings and logging_settings["enabled"]: log_dir = logging_settings.get("directory") if not log_dir: self.main_window.show_error_message( "Logging Error", "Logging is enabled but no save directory is specified.", ) return if self.raw_data_logger and not self.raw_data_logger.start_logging_session( log_dir, bbox ): self.main_window.show_error_message( "Logging Error", "Failed to start raw data logging session. Check logs.", ) return if self.aircraft_db_manager: overlapping_scans = self.aircraft_db_manager.find_overlapping_scans( params["start_time"], params["end_time"], bbox ) if overlapping_scans and not messagebox.askyesno( "Confirm Download", "A similar scan exists. Download again?" ): module_logger.info("Historical download cancelled by user.") if self.raw_data_logger and self.raw_data_logger.is_active: self.raw_data_logger.stop_logging_session() return module_logger.info( f"Controller: Starting historical download with params: {params} for bbox: {bbox}" ) self.is_historical_download_active = True self._current_scan_params = params self._active_bounding_box = bbox historical_panel.set_controls_state(is_downloading=True) if hasattr(self.main_window.function_notebook_panel, "clear_all_panel_data"): self.main_window.function_notebook_panel.clear_all_panel_data() self.main_window.clear_all_views_data() self.historical_adapter_thread = OpenSkyHistoricalAdapter( self.historical_data_queue, bbox, params["start_time"], params["end_time"], params["sampling_interval_sec"], params["scan_rate_sec"], ) self.historical_adapter_thread.start() self.historical_data_processor.start_processing() self.main_window.update_semaphore_and_status( GUI_STATUS_FETCHING, "Historical download started." ) def stop_historical_download( self, from_error: bool = False, finished_normally: bool = False ): if not self.is_historical_download_active and not from_error: module_logger.info("Historical download stop requested, but not active.") return module_logger.info( f"Controller: Stopping historical download (from_error={from_error}, finished_normally={finished_normally})." ) if ( self.aircraft_db_manager and self._current_scan_params and self._active_bounding_box ): status = ( "completed" if finished_normally else ("failed" if from_error else "aborted") ) self.aircraft_db_manager.add_scan_history( self._current_scan_params, self._active_bounding_box, status ) self.is_historical_download_active = False self._current_scan_params = None if self.historical_data_processor: self.historical_data_processor.stop_processing() if self.historical_adapter_thread and self.historical_adapter_thread.is_alive(): if hasattr(self.historical_adapter_thread, "stop"): self.historical_adapter_thread.stop() self.historical_adapter_thread.join(ADAPTER_JOIN_TIMEOUT_SECONDS) self.historical_adapter_thread = None if self.raw_data_logger and self.raw_data_logger.is_active: self.raw_data_logger.stop_logging_session() if self.main_window and hasattr( self.main_window.function_notebook_panel, "historical_panel" ): historical_panel = self.main_window.function_notebook_panel.historical_panel if historical_panel: historical_panel.set_controls_state(is_downloading=False) status_msg = ( "Historical download finished." if finished_normally else ("stopped due to an error." if from_error else "stopped.") ) self.main_window.update_semaphore_and_status( GUI_STATUS_OK if finished_normally else GUI_STATUS_WARNING, f"Historical download {status_msg}", ) def start_playback(self, session_info: Dict[str, Any]): if self.is_playback_active: module_logger.warning("Playback requested but is already active.") return if not self.data_storage or not self.playback_data_processor or not self.playback_data_queue: module_logger.error("DataStorage, PlaybackDataProcessor, or queue not initialized.") return self.stop_live_monitoring() self.stop_historical_download() date_str = session_info.get("date_str") start_ts = session_info.get("start_timestamp") end_ts = session_info.get("end_timestamp") bbox = { "lat_min": session_info.get("lat_min"), "lon_min": session_info.get("lon_min"), "lat_max": session_info.get("lat_max"), "lon_max": session_info.get("lon_max"), } if not all([date_str, start_ts is not None, end_ts is not None, all(v is not None for v in bbox.values())]): module_logger.error(f"Incomplete session info for playback: {session_info}") if self.main_window: self.main_window.show_error_message( "Playback Error", "Incomplete session data." ) return module_logger.info( f"Controller: Starting playback for session ID {session_info.get('scan_id')} on {date_str}." ) self.is_playback_active = True self._active_bounding_box = bbox if self.main_window: self.main_window.clear_all_views_data() if self.main_window.map_manager_instance: self.main_window.map_manager_instance.set_target_bbox(bbox) self.playback_adapter_thread = PlaybackAdapter( self.playback_data_queue, self.data_storage, date_str, start_ts, end_ts, bbox, ) self.playback_adapter_thread.start() self.playback_data_processor.start_processing() if self.main_window: self.main_window.update_semaphore_and_status( GUI_STATUS_FETCHING, "Playback started." ) playback_panel = self.main_window.function_notebook_panel.playback_panel if playback_panel: playback_panel.set_controls_state(is_playing=True, is_paused=False) def stop_playback(self, from_error: bool = False, finished_normally: bool = False): if not self.is_playback_active: return module_logger.info( f"Controller: Stopping playback (from_error={from_error}, finished_normally={finished_normally})." ) self.is_playback_active = False if self.playback_data_processor: self.playback_data_processor.stop_processing() if self.playback_adapter_thread and self.playback_adapter_thread.is_alive(): self.playback_adapter_thread.stop() self.playback_adapter_thread.join(ADAPTER_JOIN_TIMEOUT_SECONDS) self.playback_adapter_thread = None if self.main_window: status = ( GUI_STATUS_OK if finished_normally else (GUI_STATUS_ERROR if from_error else GUI_STATUS_WARNING) ) msg = ( "Playback finished." if finished_normally else ("Playback failed." if from_error else "Playback stopped.") ) self.main_window.update_semaphore_and_status(status, msg) playback_panel = self.main_window.function_notebook_panel.playback_panel if playback_panel: playback_panel.set_controls_state(is_playing=False) def pause_playback(self): if self.playback_adapter_thread and self.is_playback_active: self.playback_adapter_thread.pause() if self.main_window: playback_panel = self.main_window.function_notebook_panel.playback_panel if playback_panel: playback_panel.set_controls_state(is_playing=True, is_paused=True) def resume_playback(self): if self.playback_adapter_thread and self.is_playback_active: self.playback_adapter_thread.resume() if self.main_window: playback_panel = self.main_window.function_notebook_panel.playback_panel if playback_panel: playback_panel.set_controls_state(is_playing=True, is_paused=False) def set_playback_speed(self, speed: float): if self.playback_adapter_thread and self.is_playback_active: self.playback_adapter_thread.set_speed(speed) def seek_playback_to_time(self, timestamp: float): if self.playback_adapter_thread and self.is_playback_active: self.playback_adapter_thread.seek_to_time(timestamp) def on_application_exit(self): self._save_app_settings() if self.is_live_monitoring_active: self.stop_live_monitoring() if self.is_historical_download_active: self.stop_historical_download() if self.is_playback_active: self.stop_playback() if self.raw_data_logger and self.raw_data_logger.is_active: self.raw_data_logger.stop_logging_session() if self.cleanup_manager: self.cleanup_manager.on_application_exit() else: module_logger.critical("CleanupManager not initialized.") def on_function_tab_changed(self, tab_text: str): module_logger.info(f"Controller notified of function tab change to: {tab_text}") if self.is_live_monitoring_active: self.stop_live_monitoring() if self.is_historical_download_active: self.stop_historical_download() if self.is_playback_active: self.stop_playback() if self.main_window: self.main_window.clear_all_views_data() if hasattr( self.main_window.function_notebook_panel, "clear_all_panel_data" ): self.main_window.function_notebook_panel.clear_all_panel_data() placeholders = { "Live Monitor": "Define area and press Start Live Monitoring.", "Historical Download": "Set parameters and press Start Download.", "Playback": "Scan for recordings to start playback.", } placeholder = placeholders.get(tab_text, "Select a mode to begin.") if hasattr(self.main_window, "_update_map_placeholder"): self.main_window._update_map_placeholder(placeholder) if tab_text == "Playback": panel = self.main_window.function_notebook_panel.playback_panel if panel: panel.update_available_dates() def import_aircraft_database_from_file_with_progress( self, csv_filepath: str, progress_dialog_ref: "ImportProgressDialog" ): if not self.aircraft_db_importer: module_logger.error("AircraftDBImporter not initialized.") if progress_dialog_ref and progress_dialog_ref.winfo_exists(): if self.main_window and self.main_window.root.winfo_exists(): self.main_window.root.after( 0, lambda: progress_dialog_ref.import_finished( False, "Error: Import function not initialized." ), ) return self.aircraft_db_importer.import_aircraft_database_with_progress( csv_filepath, progress_dialog_ref ) def get_historical_track_for_icao(self, icao24: str) -> List[Dict[str, Any]]: if not self.data_storage or not icao24: return [] try: from datetime import datetime, timezone current_utc_dt = datetime.now(timezone.utc) track_states = self.data_storage.get_flight_track_for_icao_on_date( icao24, current_utc_dt ) if track_states: return [state.to_dict() for state in track_states] except Exception as e: module_logger.error( f"Controller: Error retrieving historical track for {icao24}: {e}", exc_info=True, ) return [] def request_detailed_flight_info(self, icao24: str): if not self.main_window: return normalized_icao24 = icao24.lower().strip() if not normalized_icao24: if hasattr(self.main_window, "update_selected_flight_details"): self.main_window.update_selected_flight_details(None) return combined_details = {"icao24": normalized_icao24} if self.main_window.map_manager_instance: map_mgr = self.main_window.map_manager_instance with map_mgr._map_data_lock: if self.is_playback_active: flight_list = list(map_mgr._active_aircraft_states.values()) else: flight_list = map_mgr._current_flights_to_display_gui for state in flight_list: if state.icao24 == normalized_icao24: combined_details.update(state.to_dict()) break if self.aircraft_db_manager: static_data = self.aircraft_db_manager.get_aircraft_details( normalized_icao24 ) if static_data: for k, v in static_data.items(): if k not in combined_details or combined_details[k] is None: combined_details[k] = v if hasattr(self.main_window, "update_selected_flight_details"): self.main_window.update_selected_flight_details(combined_details) def request_and_show_full_flight_details(self, icao24: str): normalized_icao24 = icao24.lower().strip() module_logger.info( f"Controller: Requesting full details for ICAO24: {normalized_icao24}" ) if not ( self.main_window and self.main_window.root and self.main_window.root.winfo_exists() ): return if not normalized_icao24: if hasattr(self.main_window, "show_info_message"): self.main_window.show_info_message( "Flight Details", "No ICAO24 provided." ) return if ( self.active_detail_window_ref and self.active_detail_window_ref.winfo_exists() ): if self.active_detail_window_icao == normalized_icao24: self.active_detail_window_ref.lift() self.active_detail_window_ref.focus_set() else: try: self.active_detail_window_ref.destroy() except tk.TclError: pass static_data, live_data, full_track_data_list = None, None, [] if self.aircraft_db_manager: static_data = self.aircraft_db_manager.get_aircraft_details( normalized_icao24 ) if self.main_window and self.main_window.map_manager_instance: map_mgr = self.main_window.map_manager_instance with map_mgr._map_data_lock: if self.is_playback_active: flight_list = list(map_mgr._active_aircraft_states.values()) else: flight_list = map_mgr._current_flights_to_display_gui for state in flight_list: if state.icao24 == normalized_icao24: live_data = state.to_dict() break if self.data_storage: try: from datetime import datetime, timezone date_to_query = datetime.now(timezone.utc) if self.is_playback_active and self.playback_adapter_thread: date_to_query = datetime.strptime(self.playback_adapter_thread.session_date_str, "%Y-%m-%d") track_states = self.data_storage.get_flight_track_for_icao_on_date( normalized_icao24, date_to_query ) if track_states: full_track_data_list = [state.to_dict() for state in track_states] except Exception as e_track: module_logger.error( f"Error retrieving track for {normalized_icao24}: {e_track}", exc_info=True, ) try: from ..gui.dialogs.full_flight_details_window import FullFlightDetailsWindow details_win = FullFlightDetailsWindow( self.main_window.root, normalized_icao24, self ) self.active_detail_window_ref, self.active_detail_window_icao = ( details_win, normalized_icao24, ) if self.main_window: self.main_window.full_flight_details_window = details_win details_win.update_details(static_data, live_data, full_track_data_list) except Exception as e_show: module_logger.error( f"Error showing full details for {normalized_icao24}: {e_show}", exc_info=True, ) if hasattr(self.main_window, "show_error_message"): self.main_window.show_error_message( "Error", f"Could not display full details: {e_show}" ) self.active_detail_window_ref, self.active_detail_window_icao = None, None if self.main_window: self.main_window.full_flight_details_window = None def details_window_closed(self, closed_icao24: str): if self.cleanup_manager: self.cleanup_manager.details_window_closed(closed_icao24) def on_map_left_click(self, lat, lon, cx, cy, sx, sy): if self.map_command_handler: self.map_command_handler.on_map_left_click(lat, lon, cx, cy, sx, sy) def on_map_right_click(self, lat, lon, sx, sy): if self.map_command_handler: self.map_command_handler.on_map_right_click(lat, lon, sx, sy) def recenter_map_at_coords(self, lat, lon): if self.map_command_handler: self.map_command_handler.recenter_map_at_coords(lat, lon) def set_bbox_around_coords(self, lat, lon, size_km): if self.map_command_handler: self.map_command_handler.set_bbox_around_coords(lat, lon, size_km) def update_bbox_gui_fields(self, bbox_dict): if self.main_window: self.main_window.update_bbox_gui_fields(bbox_dict) def update_general_map_info(self): if self.map_command_handler: self.map_command_handler.update_general_map_info() def map_zoom_in(self): if self.map_command_handler: self.map_command_handler.map_zoom_in() def map_zoom_out(self): if self.map_command_handler: self.map_command_handler.map_zoom_out() def map_pan_direction(self, direction): if self.map_command_handler: self.map_command_handler.map_pan_direction(direction) def map_center_on_coords_and_fit_patch(self, lat, lon, patch_km): if self.map_command_handler: self.map_command_handler.map_center_on_coords_and_fit_patch( lat, lon, patch_km ) def set_map_track_length(self, length): if self.main_window and self.main_window.map_manager_instance: self.main_window.map_manager_instance.set_max_track_points(length) def get_profile_names(self) -> List[str]: if self.profile_manager: return self.profile_manager.get_profile_names() return [] def load_area_profile(self, profile_name: str): if not self.profile_manager: return profile_data = self.profile_manager.get_profile_data(profile_name) if profile_data and self.main_window: self.main_window.update_bbox_gui_fields(profile_data) if self.main_window.map_manager_instance: self.main_window.map_manager_instance.set_target_bbox(profile_data) def save_current_area_as_profile(self): if not self.profile_manager or not self.main_window: return current_bbox = self.main_window.get_bounding_box_from_gui() if not current_bbox: self.main_window.show_error_message( "Save Error", "The current Bounding Box values are invalid." ) return profile_name = simpledialog.askstring( "Save Profile", "Enter a name for the new area profile:", parent=self.main_window.root, ) if not profile_name or not profile_name.strip(): return profile_name = profile_name.strip() if self.profile_manager.save_profile(profile_name, current_bbox): if hasattr(self.main_window.function_notebook_panel, "update_profile_list"): self.main_window.function_notebook_panel.update_profile_list() self.main_window.function_notebook_panel.set_selected_profile( profile_name ) self.main_window.show_info_message( "Success", f"Profile '{profile_name}' saved successfully." ) else: self.main_window.show_error_message( "Save Error", f"Could not save profile '{profile_name}'." ) def delete_area_profile(self, profile_name: str): if not self.profile_manager or not self.main_window: return if self.profile_manager.delete_profile(profile_name): if hasattr(self.main_window.function_notebook_panel, "update_profile_list"): self.main_window.function_notebook_panel.update_profile_list() self.main_window.show_info_message( "Success", f"Profile '{profile_name}' deleted." ) else: self.main_window.show_error_message( "Delete Error", f"Could not delete profile '{profile_name}'." ) def get_available_recording_dates(self) -> List[str]: if self.data_storage: return self.data_storage.get_available_recording_dates() return [] def get_sessions_for_date(self, date_str: str) -> List[Dict[str, Any]]: all_sessions = [] if self.aircraft_db_manager: explicit_sessions = self.aircraft_db_manager.get_sessions_for_date( date_str ) for sess in explicit_sessions: sess_dict = dict(sess) sess_dict["type"] = "Explicit" sess_dict["date_str"] = date_str all_sessions.append(sess_dict) if self.data_storage: summary = self.data_storage.get_daily_recording_summary(date_str) if summary and summary.get("time_range") and summary.get("bbox"): time_range = summary["time_range"] bbox = summary["bbox"] inferred_session = { "scan_id": f"inferred-{date_str}", "start_timestamp": time_range[0], "end_timestamp": time_range[1], "lat_min": bbox["lat_min"], "lon_min": bbox["lon_min"], "lat_max": bbox["lat_max"], "lon_max": bbox["lon_max"], "type": "Inferred (Full Day)", "status": "completed", "date_str": date_str, } all_sessions.append(inferred_session) if all_sessions: all_sessions.sort(key=lambda s: s.get("start_timestamp", 0), reverse=True) return all_sessions def get_daily_recordings_summary(self) -> List[Dict[str, Any]]: summaries = [] if not self.data_storage: return [] dates = self.data_storage.get_available_recording_dates() for date_str in dates: summary = self.data_storage.get_daily_recording_summary(date_str) if summary: summary["date"] = date_str summaries.append(summary) return summaries def delete_daily_recording(self, date_str: str) -> bool: if self.data_storage: return self.data_storage.delete_daily_db(date_str) return False def get_all_scan_sessions(self) -> List[Dict[str, Any]]: if self.aircraft_db_manager: return self.aircraft_db_manager.get_all_scan_sessions() return [] def delete_scan_session(self, scan_id: int) -> bool: if self.aircraft_db_manager: return self.aircraft_db_manager.delete_scan_session(scan_id) return False def get_aircraft_database_size_info(self) -> str: """ Retrieves the size of the aircraft database as a formatted string. Returns: A string representing the size (e.g., "15.2 MB") or "N/A". """ if self.aircraft_db_manager: size_bytes = self.aircraft_db_manager.get_database_size_bytes() if size_bytes < 1024: return f"{size_bytes} Bytes" elif size_bytes < 1024**2: return f"{size_bytes / 1024:.2f} KB" elif size_bytes < 1024**3: return f"{size_bytes / 1024**2:.2f} MB" else: return f"{size_bytes / 1024**3:.2f} GB" return "N/A" def get_map_tile_cache_size_info(self) -> str: """ Retrieves the size of the map tile cache as a formatted string. Returns: A string representing the size (e.g., "50.7 MB") or "N/A". """ if ( self.main_window and self.main_window.map_manager_instance and self.main_window.map_manager_instance.tile_manager ): size_bytes = self.main_window.map_manager_instance.tile_manager.get_cache_size_bytes() if size_bytes < 1024: return f"{size_bytes} Bytes" elif size_bytes < 1024**2: return f"{size_bytes / 1024:.2f} KB" elif size_bytes < 1024**3: return f"{size_bytes / 1024**2:.2f} MB" else: return f"{size_bytes / 1024**3:.2f} GB" return "N/A" def clear_aircraft_database(self) -> bool: """Clears the aircraft details table.""" if self.aircraft_db_manager: return self.aircraft_db_manager.clear_aircraft_details_table() return False def clear_scan_history(self) -> bool: """Clears the scan history table.""" if self.aircraft_db_manager: return self.aircraft_db_manager.clear_scan_history_table() return False def clear_map_tile_cache(self) -> bool: """Clears the map tile cache.""" if ( self.main_window and self.main_window.map_manager_instance and self.main_window.map_manager_instance.tile_manager ): self.main_window.map_manager_instance.tile_manager.clear_entire_service_cache() return True module_logger.warning("Map tile manager not available to clear cache.") return False