VALLONGOL/SXXXXXXX_GeoElevation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
e7ca4855ab
SXXXXXXX_GeoElevation/geoelevation/elevation_gui.py
VALLONGOL e7ca4855ab import corrected
2025-05-13 15:11:07 +02:00

678 lines
40 KiB
Python
Raw Blame History

# geoelevation/elevation_gui.py
"""
Provides the main graphical user interface (GUI) for the GeoElevation application.
(Resto della docstring come prima)
"""
# Standard library imports
import tkinter as tk
from tkinter import ttk
from tkinter import messagebox
import logging
import math
import multiprocessing
import threading
import os
import sys
import queue
from typing import Optional, Tuple, List, Dict, TYPE_CHECKING, Any
# Local application/package imports
from geoelevation.elevation_manager import ElevationManager
from geoelevation.elevation_manager import RASTERIO_AVAILABLE
from geoelevation.image_processor import load_prepare_single_browse
from geoelevation.image_processor import create_composite_area_image
from geoelevation.image_processor import PIL_AVAILABLE
from geoelevation.visualizer import show_image_matplotlib
from geoelevation.visualizer import show_3d_matplotlib
from geoelevation.visualizer import MATPLOTLIB_AVAILABLE
from geoelevation.visualizer import SCIPY_AVAILABLE
# MODIFIED: Corrected import of default cache directory constant from the geoelevation package's public API
# WHY: The previous import was attempting to use the internal name `DEFAULT_CACHE_DIR` which is not exposed
# by geoelevation/__init__.py. The correct name exposed in __init__.py (and sourced from config.py)
# is GEOELEVATION_DEM_CACHE_DEFAULT. This change aligns the import with the package's defined public API.
# HOW: Changed the import statement to directly import GEOELEVATION_DEM_CACHE_DEFAULT. The alias is no longer needed
# in the import statement itself, as the imported name is already the desired one. The existing try...except
# block correctly handles the case where this specific constant might not be available from the package.
try:
from geoelevation import GEOELEVATION_DEM_CACHE_DEFAULT
except ImportError:
# This fallback is if the main __init__.py itself has issues exporting the constant,
# which shouldn't happen if __init__.py and config.py are correct.
GEOELEVATION_DEM_CACHE_DEFAULT = "elevation_data_cache_gui_fallback_critical"
logging.getLogger(__name__).critical(
"elevation_gui.py: CRITICAL - Could not import GEOELEVATION_DEM_CACHE_DEFAULT from geoelevation package. "
f"Using fallback: {GEOELEVATION_DEM_CACHE_DEFAULT}"
)
try:
import cv2
MAP_VIEWER_SYSTEM_AVAILABLE = True
except ImportError as e_map_import_gui:
logging.warning(
f"GUI: Map viewer components (e.g. OpenCV) not found, map functionality will be disabled: {e_map_import_gui}"
)
cv2 = None # type: ignore
MAP_VIEWER_SYSTEM_AVAILABLE = False
if TYPE_CHECKING:
import numpy as np_typing
logger = logging.getLogger(__name__)
# --- MULTIPROCESSING TARGET FUNCTIONS ---
# (Queste funzioni rimangono identiche alla versione precedente completa che ti ho fornito)
def process_target_show_image(image_path: str, tile_name: str, window_title: str) -> None:
try:
from geoelevation.image_processor import load_prepare_single_browse as local_lpst
from geoelevation.image_processor import PIL_AVAILABLE as local_pil_ok
from geoelevation.visualizer import show_image_matplotlib as local_sim
from geoelevation.visualizer import MATPLOTLIB_AVAILABLE as local_mpl_ok
import os as local_os
if not (local_pil_ok and local_mpl_ok):
print("PROCESS ERROR (show_image): Pillow/Matplotlib missing in child process.")
return
prepared_image = local_lpst(image_path, tile_name)
if prepared_image:
print(f"PROCESS (show_image): Showing '{window_title}'")
local_sim(prepared_image, window_title)
else:
print(f"PROCESS ERROR (show_image): Could not prepare {local_os.path.basename(image_path)}")
except Exception as e_proc_img:
print(f"PROCESS ERROR in process_target_show_image: {e_proc_img}")
import traceback as tb
tb.print_exc()
def process_target_show_3d(
hgt_data: Optional["np_typing.ndarray"],
plot_title: str,
initial_subsample: int,
smooth_sigma: Optional[float],
interpolation_factor: int,
plot_grid_points: int
) -> None:
try:
from geoelevation.visualizer import show_3d_matplotlib as local_s3d
from geoelevation.visualizer import MATPLOTLIB_AVAILABLE as local_mpl_ok
from geoelevation.visualizer import SCIPY_AVAILABLE as local_scipy_ok
if not local_mpl_ok:
print("PROCESS ERROR (show_3d): Matplotlib missing in child process.")
return
effective_smooth_sigma = smooth_sigma
effective_interpolation_factor = interpolation_factor
if (interpolation_factor > 1 or smooth_sigma is not None) and not local_scipy_ok:
print("PROCESS WARNING (show_3d): SciPy missing in child. Disabling 3D plot smoothing/interpolation.")
effective_smooth_sigma = None
effective_interpolation_factor = 1
if hgt_data is not None:
print(f"PROCESS (show_3d): Plotting '{plot_title}' (InitialSub:{initial_subsample}, Smooth:{effective_smooth_sigma}, Interp:{effective_interpolation_factor}x, PlotGridTarget:{plot_grid_points})")
local_s3d(hgt_data, plot_title, initial_subsample, effective_smooth_sigma, effective_interpolation_factor, plot_grid_points)
else:
print("PROCESS ERROR (show_3d): No HGT data array received for plotting.")
except Exception as e_proc_3d:
print(f"PROCESS ERROR in process_target_show_3d: {e_proc_3d}")
import traceback as tb
tb.print_exc()
def process_target_create_show_area(tile_info_list_data: List[Dict], window_title_str: str) -> None:
try:
from geoelevation.image_processor import create_composite_area_image as local_ccai
from geoelevation.image_processor import PIL_AVAILABLE as local_pil_ok
from geoelevation.visualizer import show_image_matplotlib as local_sim
from geoelevation.visualizer import MATPLOTLIB_AVAILABLE as local_mpl_ok
if not (local_pil_ok and local_mpl_ok):
print("PROCESS ERROR (show_area): Pillow/Matplotlib libraries are missing in the child process.")
return
print("PROCESS (show_area): Creating composite area image...")
composite_pil_image = local_ccai(tile_info_list_data)
if composite_pil_image:
print(f"PROCESS (show_area): Displaying composite image '{window_title_str}'")
local_sim(composite_pil_image, window_title_str)
else:
print("PROCESS ERROR (show_area): Failed to create composite area image.")
except Exception as e_proc_area:
print(f"PROCESS ERROR in process_target_create_show_area: {e_proc_area}")
import traceback as tb
tb.print_exc()
def run_map_viewer_process_target(
map_interaction_q: multiprocessing.Queue,
operation_mode: str,
center_latitude: Optional[float],
center_longitude: Optional[float],
area_bounding_box: Optional[Tuple[float, float, float, float]],
dem_data_cache_dir: str,
display_scale_factor: float
) -> None:
child_logger = logging.getLogger("GeoElevationMapViewerChildProcess")
if not child_logger.handlers:
stream_handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)
child_logger.addHandler(stream_handler)
child_logger.setLevel(logging.INFO)
child_logger.propagate = False
child_map_viewer_instance: Optional[Any] = None
try:
from geoelevation.map_viewer.geo_map_viewer import GeoElevationMapViewer as ChildGeoMapViewer
from geoelevation.elevation_manager import ElevationManager as ChildElevationManager
import cv2 as child_cv2
child_map_system_ok = True
except ImportError as e_child_imp_map_corrected:
child_logger.error(f"Map viewer components or OpenCV not found in child: {e_child_imp_map_corrected}")
child_map_system_ok = False
if not child_map_system_ok:
return
try:
child_logger.info(f"Initializing for mode '{operation_mode}', display scale: {display_scale_factor:.2f}")
local_em = ChildElevationManager(tile_directory=dem_data_cache_dir)
child_map_viewer_instance = ChildGeoMapViewer(
elevation_manager_instance=local_em,
gui_output_communication_queue=map_interaction_q,
initial_display_scale=display_scale_factor
)
if operation_mode == "point" and center_latitude is not None and center_longitude is not None:
child_map_viewer_instance.display_map_for_point(center_latitude, center_longitude)
elif operation_mode == "area" and area_bounding_box:
child_map_viewer_instance.display_map_for_area(area_bounding_box)
else:
child_logger.error(f"Invalid mode ('{operation_mode}') or missing parameters.")
if hasattr(child_map_viewer_instance, 'shutdown'): child_map_viewer_instance.shutdown()
return
child_logger.info("Display method called. Map window should be active.")
is_map_active = True
while is_map_active:
if child_map_viewer_instance.map_display_window_controller:
if child_map_viewer_instance.map_display_window_controller.is_window_alive():
child_cv2.waitKey(100)
else:
child_logger.info("Map window reported as not alive by is_window_alive().")
is_map_active = False
else:
child_logger.info("map_display_window_controller is None. Assuming window closed.")
is_map_active = False
child_logger.info("Map window loop in child process exited.")
except Exception as e_map_proc_child_fatal_final:
child_logger.error(f"Fatal error in map viewer process target: {e_map_proc_child_fatal_final}", exc_info=True)
finally:
if child_map_viewer_instance and hasattr(child_map_viewer_instance, 'shutdown'):
child_map_viewer_instance.shutdown()
child_logger.info("Terminated child map viewer process.")
# --- END MULTIPROCESSING TARGET FUNCTIONS ---
class ElevationApp:
"""Main application class for the GeoElevation Tool GUI."""
def __init__(
self,
parent_widget: tk.Tk,
elevation_manager_instance: Optional[ElevationManager] = None
) -> None:
"""
Initializes the main GUI application window and components.
Args:
parent_widget: The root Tkinter window.
elevation_manager_instance: Optional existing ElevationManager instance.
If None, a new one is created.
"""
self.root: tk.Tk = parent_widget
self.elevation_manager: Optional[ElevationManager] = elevation_manager_instance
# MODIFIED: Use the correctly imported GEOELEVATION_DEM_CACHE_DEFAULT constant
# WHY: This constant holds the path to the default DEM cache directory
# as configured in the config.py module and exposed by __init__.py.
# HOW: Replaced the hardcoded or previously mis-aliased variable name
# with the correct imported constant. The try...except block ensures
# a fallback is used if the import somehow fails.
default_dem_cache = GEOELEVATION_DEM_CACHE_DEFAULT
if self.elevation_manager is None:
try:
if not RASTERIO_AVAILABLE:
logger.warning("Rasterio library not found. Elevation functions limited.")
self.elevation_manager = ElevationManager(
tile_directory=default_dem_cache
)
except Exception as e_manager_init:
logger.critical(f"Failed to initialize ElevationManager: {e_manager_init}", exc_info=True)
messagebox.showerror("Init Error", f"Could not start Elevation Manager:\n{e_manager_init}", parent=self.root)
self.elevation_manager = None
self.root.title("GeoElevation Tool")
self.root.minsize(480, 520)
self.last_valid_point_coords: Optional[Tuple[float, float]] = None
self.last_area_coords: Optional[Tuple[float, float, float, float]] = None
self.is_processing_task: bool = False
self.map_display_scale_factor_var: tk.DoubleVar = tk.DoubleVar(value=0.5)
self.scale_options_map: Dict[str, float] = {
"25% (1:4)": 0.25,
"33% (1:3)": 0.33333,
"50% (1:2)": 0.5,
"75% (3:4)": 0.75,
"100% (Original)": 1.0,
"150% (3:2)": 1.5,
"200% (2:1)": 2.0
}
self.map_viewer_process_handle: Optional[multiprocessing.Process] = None
self.map_interaction_message_queue: Optional[multiprocessing.Queue] = None
if MAP_VIEWER_SYSTEM_AVAILABLE:
try:
self.map_interaction_message_queue = multiprocessing.Queue()
except Exception as e_gui_queue:
logger.error(f"GUI: Failed to create multiprocessing queue for map viewer: {e_gui_queue}")
self._build_gui_layout()
self._apply_initial_widget_states()
if self.map_interaction_message_queue:
self.root.after(100, self._process_map_interaction_queue_messages)
self.root.protocol("WM_DELETE_WINDOW", self._on_application_closing)
def _get_scale_display_text_from_value(self, target_scale_value: float) -> str:
closest_text = ""
smallest_difference = float('inf')
default_display_text = "50% (1:2)"
for display_text, scale_val in self.scale_options_map.items():
difference = abs(scale_val - target_scale_value)
if difference < smallest_difference:
smallest_difference = difference
closest_text = display_text
if closest_text and smallest_difference < 0.01:
return closest_text
else:
for text_option, value_option in self.scale_options_map.items():
if abs(value_option - 0.5) < 0.001:
return text_option
return default_display_text
def _build_gui_layout(self) -> None:
main_app_frame = ttk.Frame(self.root, padding="10")
main_app_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
self.root.columnconfigure(0, weight=1)
self.root.rowconfigure(0, weight=1)
point_coords_frame = ttk.LabelFrame(main_app_frame, text="Get Elevation for Point", padding="10")
point_coords_frame.grid(row=0, column=0, padx=5, pady=5, sticky=(tk.W, tk.E))
point_coords_frame.columnconfigure(1, weight=1)
ttk.Label(point_coords_frame, text="Latitude:").grid(row=0, column=0, sticky=tk.W, padx=5, pady=3)
self.latitude_entry = ttk.Entry(point_coords_frame, width=15)
self.latitude_entry.grid(row=0, column=1, sticky=(tk.W, tk.E), padx=5, pady=3)
self.latitude_entry.insert(0, "45.0")
ttk.Label(point_coords_frame, text="Longitude:").grid(row=1, column=0, sticky=tk.W, padx=5, pady=3)
self.longitude_entry = ttk.Entry(point_coords_frame, width=15)
self.longitude_entry.grid(row=1, column=1, sticky=(tk.W, tk.E), padx=5, pady=3)
self.longitude_entry.insert(0, "7.0")
self.get_elevation_button = ttk.Button(point_coords_frame, text="Get Elevation", command=self._trigger_get_elevation_task)
self.get_elevation_button.grid(row=2, column=0, columnspan=2, pady=5, sticky=(tk.W, tk.E))
self.point_result_label = ttk.Label(point_coords_frame, text="Result: ", wraplength=400, justify=tk.LEFT)
self.point_result_label.grid(row=3, column=0, columnspan=2, sticky=tk.W, pady=5)
point_actions_subframe = ttk.Frame(point_coords_frame)
point_actions_subframe.grid(row=4, column=0, columnspan=2, pady=(5,0), sticky=(tk.W, tk.E))
point_actions_subframe.columnconfigure(0, weight=1)
point_actions_subframe.columnconfigure(1, weight=1)
point_actions_subframe.columnconfigure(2, weight=1)
self.show_2d_browse_button = ttk.Button(point_actions_subframe, text="Browse Image (2D)", command=self._trigger_2d_browse_display)
self.show_2d_browse_button.grid(row=0, column=0, padx=2, sticky=(tk.W,tk.E))
self.show_3d_dem_button = ttk.Button(point_actions_subframe, text="DEM Tile (3D)", command=self._trigger_3d_dem_display)
self.show_3d_dem_button.grid(row=0, column=1, padx=2, sticky=(tk.W,tk.E))
self.view_map_for_point_button = ttk.Button(point_actions_subframe, text="View Point on Map", command=self._trigger_view_map_for_point)
self.view_map_for_point_button.grid(row=0, column=2, padx=2, sticky=(tk.W,tk.E))
area_download_frame = ttk.LabelFrame(main_app_frame, text="Pre-Download Tiles for Area", padding="10")
area_download_frame.grid(row=1, column=0, padx=5, pady=5, sticky=(tk.W, tk.E))
area_download_frame.columnconfigure(1, weight=1)
area_download_frame.columnconfigure(3, weight=1)
ttk.Label(area_download_frame, text="Min Lat:").grid(row=0, column=0, sticky=tk.W, padx=5, pady=3)
self.min_latitude_entry = ttk.Entry(area_download_frame, width=10)
self.min_latitude_entry.grid(row=0, column=1, sticky=(tk.W,tk.E), padx=5, pady=3)
self.min_latitude_entry.insert(0, "44.0")
ttk.Label(area_download_frame, text="Max Lat:").grid(row=0, column=2, sticky=tk.W, padx=(10,5), pady=3)
self.max_latitude_entry = ttk.Entry(area_download_frame, width=10)
self.max_latitude_entry.grid(row=0, column=3, sticky=(tk.W,tk.E), padx=5, pady=3)
self.max_latitude_entry.insert(0, "45.0")
ttk.Label(area_download_frame, text="Min Lon:").grid(row=1, column=0, sticky=tk.W, padx=5, pady=3)
self.min_longitude_entry = ttk.Entry(area_download_frame, width=10)
self.min_longitude_entry.grid(row=1, column=1, sticky=(tk.W,tk.E), padx=5, pady=3)
self.min_longitude_entry.insert(0, "7.0")
ttk.Label(area_download_frame, text="Max Lon:").grid(row=1, column=2, sticky=tk.W, padx=(10,5), pady=3)
self.max_longitude_entry = ttk.Entry(area_download_frame, width=10)
self.max_longitude_entry.grid(row=1, column=3, sticky=(tk.W,tk.E), padx=5, pady=3)
self.max_longitude_entry.insert(0, "8.0")
self.download_area_button = ttk.Button(area_download_frame, text="Download Area Tiles", command=self._trigger_download_area_task)
self.download_area_button.grid(row=2, column=0, columnspan=4, pady=10, sticky=(tk.W,tk.E))
self.show_area_composite_button = ttk.Button(area_download_frame, text="Show Area Composite (2D)", command=self._trigger_area_composite_display)
self.show_area_composite_button.grid(row=3, column=0, columnspan=4, pady=5, sticky=(tk.W,tk.E))
self.view_map_for_area_button = ttk.Button(area_download_frame, text="View Area on Map", command=self._trigger_view_map_for_area)
self.view_map_for_area_button.grid(row=4, column=0, columnspan=4, pady=5, sticky=(tk.W,tk.E))
self.area_download_status_label = ttk.Label(area_download_frame, text="Status: Idle", wraplength=400, justify=tk.LEFT)
self.area_download_status_label.grid(row=5, column=0, columnspan=4, sticky=tk.W, pady=5)
map_display_options_frame = ttk.LabelFrame(main_app_frame, text="Map Display Options", padding="10")
map_display_options_frame.grid(row=2, column=0, padx=5, pady=5, sticky=(tk.W, tk.E))
map_display_options_frame.columnconfigure(1, weight=1)
ttk.Label(map_display_options_frame, text="Map Display Scale:").grid(row=0, column=0, sticky=tk.W, padx=5, pady=3)
scale_option_display_texts = list(self.scale_options_map.keys())
initial_combobox_display_text = self._get_scale_display_text_from_value(self.map_display_scale_factor_var.get())
self.map_scale_combobox = ttk.Combobox(
map_display_options_frame,
textvariable=tk.StringVar(value=initial_combobox_display_text),
values=scale_option_display_texts,
state="readonly",
width=20
)
self.map_scale_combobox.grid(row=0, column=1, sticky=(tk.W, tk.E), padx=5, pady=3)
self.map_scale_combobox.bind("<<ComboboxSelected>>", self._on_map_display_scale_changed)
map_click_info_frame = ttk.LabelFrame(main_app_frame, text="Map Click Information", padding="10")
map_click_info_frame.grid(row=3, column=0, padx=5, pady=5, sticky=(tk.W, tk.E))
map_click_info_frame.columnconfigure(1, weight=1)
ttk.Label(map_click_info_frame, text="Clicked Latitude:").grid(row=0, column=0, sticky=tk.W, padx=5, pady=2)
self.map_click_latitude_label = ttk.Label(map_click_info_frame, text="N/A")
self.map_click_latitude_label.grid(row=0, column=1, sticky=(tk.W, tk.E), padx=5, pady=2)
ttk.Label(map_click_info_frame, text="Clicked Longitude:").grid(row=1, column=0, sticky=tk.W, padx=5, pady=2)
self.map_click_longitude_label = ttk.Label(map_click_info_frame, text="N/A")
self.map_click_longitude_label.grid(row=1, column=1, sticky=(tk.W, tk.E), padx=5, pady=2)
ttk.Label(map_click_info_frame, text="Elevation at Click:").grid(row=2, column=0, sticky=tk.W, padx=5, pady=2)
self.map_click_elevation_label = ttk.Label(map_click_info_frame, text="N/A")
self.map_click_elevation_label.grid(row=2, column=1, sticky=(tk.W, tk.E), padx=5, pady=2)
main_app_frame.columnconfigure(0, weight=1)
main_app_frame.rowconfigure(0, weight=0)
main_app_frame.rowconfigure(1, weight=0)
main_app_frame.rowconfigure(2, weight=0)
main_app_frame.rowconfigure(3, weight=0)
if not MAP_VIEWER_SYSTEM_AVAILABLE:
map_display_options_frame.grid_remove()
map_click_info_frame.grid_remove()
def _apply_initial_widget_states(self) -> None:
if self.elevation_manager is None:
self.get_elevation_button.config(state=tk.DISABLED)
self.download_area_button.config(state=tk.DISABLED)
self.point_result_label.config(text="Result: Elevation Manager Init Failed.")
self.area_download_status_label.config(text="Status: Elevation Manager Init Failed.")
self.show_2d_browse_button.config(state=tk.DISABLED)
self.show_3d_dem_button.config(state=tk.DISABLED)
self.show_area_composite_button.config(state=tk.DISABLED)
point_map_initial_state = tk.DISABLED
area_map_initial_state = tk.DISABLED
scale_combo_state = tk.DISABLED
point_map_text = "Map (Sys N/A)"
area_map_text = "Map (Sys N/A)"
if MAP_VIEWER_SYSTEM_AVAILABLE:
point_map_initial_state = tk.NORMAL # Enabled if system ok, but needs point data
# area_map_initial_state remains DISABLED until area data is ready
scale_combo_state = "readonly"
point_map_text = "View Point on Map"
area_map_text = "View Area on Map"
self.view_map_for_point_button.config(state=point_map_initial_state, text=point_map_text)
self.view_map_for_area_button.config(state=area_map_initial_state, text=area_map_text)
self.map_scale_combobox.config(state=scale_combo_state)
if not (MATPLOTLIB_AVAILABLE and PIL_AVAILABLE):
self.show_2d_browse_button.config(text="Browse (Libs N/A)")
self.show_area_composite_button.config(text="Composite (Libs N/A)")
if not MATPLOTLIB_AVAILABLE:
self.show_3d_dem_button.config(text="DEM (MPL N/A)")
def _on_map_display_scale_changed(self, event: Optional[tk.Event] = None) -> None:
selected_display_text = self.map_scale_combobox.get()
new_numeric_scale = self.scale_options_map.get(selected_display_text)
if new_numeric_scale is not None:
self.map_display_scale_factor_var.set(new_numeric_scale)
logger.info(f"Map display scale factor set to: {new_numeric_scale:.3f} ({selected_display_text})")
# MODIFIED: Add a messagebox to inform the user about the scale change application.
# WHY: To clarify that the scale change applies to the *next* map view.
# HOW: Display an informational messagebox.
messagebox.showinfo(
"Map Scale Changed",
f"Map display scale set to {selected_display_text}.\n"
"This will apply the next time a map view is opened.",
parent=self.root
)
else:
logger.warning(f"Invalid map scale option selected: '{selected_display_text}'")
def _set_busy_state(self, is_busy: bool) -> None:
self.is_processing_task = is_busy
new_widget_state = tk.DISABLED if is_busy else tk.NORMAL
if self.elevation_manager:
self.get_elevation_button.config(state=new_widget_state)
self.download_area_button.config(state=new_widget_state)
def _validate_point_coordinates(self, lat_s: str, lon_s: str) -> Optional[Tuple[float, float]]:
try:
if not lat_s: raise ValueError("Latitude empty.")
lat_v = float(lat_s.strip())
if not (-90.0 <= lat_v < 90.0): raise ValueError("Latitude out of [-90, 90).")
if not lon_s: raise ValueError("Longitude empty.")
lon_v = float(lon_s.strip())
if not (-180.0 <= lon_v < 180.0): raise ValueError("Longitude out of [-180, 180).")
return lat_v, lon_v
except ValueError as e: logger.error(f"Invalid coordinate: {e}"); messagebox.showerror("Input Error", f"Invalid coordinate:\n{e}", parent=self.root); return None
def _validate_area_boundary_coordinates(self) -> Optional[Tuple[float, float, float, float]]:
try:
min_ls,max_ls = self.min_latitude_entry.get().strip(), self.max_latitude_entry.get().strip()
min_os,max_os = self.min_longitude_entry.get().strip(), self.max_longitude_entry.get().strip()
if not all([min_ls, max_ls, min_os, max_os]): raise ValueError("All bounds must be filled.")
min_l,max_l,min_o,max_o = float(min_ls),float(max_ls),float(min_os),float(max_os)
if not (-90<=min_l<90 and -90<=max_l<90 and -180<=min_o<180 and -180<=max_o<180):
raise ValueError("Coordinates out of valid range.")
if min_l >= max_l: raise ValueError("Min Lat >= Max Lat.")
if min_o >= max_o: raise ValueError("Min Lon >= Max Lon.")
return min_l, min_o, max_l, max_o
except ValueError as e: logger.error(f"Invalid area: {e}"); messagebox.showerror("Input Error", f"Invalid area:\n{e}", parent=self.root); return None
def _trigger_get_elevation_task(self) -> None:
if self.is_processing_task or not self.elevation_manager: return
coords = self._validate_point_coordinates(self.latitude_entry.get(), self.longitude_entry.get())
if not coords: return
lat, lon = coords
self._set_busy_state(True)
self.point_result_label.config(text="Result: Requesting...")
self.show_2d_browse_button.config(state=tk.DISABLED)
self.show_3d_dem_button.config(state=tk.DISABLED)
if MAP_VIEWER_SYSTEM_AVAILABLE: self.view_map_for_point_button.config(state=tk.DISABLED)
self.last_valid_point_coords = None
self.root.update_idletasks()
try:
elev = self.elevation_manager.get_elevation(lat, lon)
res_txt = "Result: "
if elev is None: res_txt += "Data unavailable."; messagebox.showwarning("Info", "Could not retrieve elevation.", parent=self.root)
elif math.isnan(elev): res_txt += "Point on NoData area."; self.last_valid_point_coords = (lat, lon)
else: res_txt += f"Elevation {elev:.2f}m"; self.last_valid_point_coords = (lat, lon)
self.point_result_label.config(text=res_txt)
if self.last_valid_point_coords:
if MATPLOTLIB_AVAILABLE and PIL_AVAILABLE: self.show_2d_browse_button.config(state=tk.NORMAL)
if MATPLOTLIB_AVAILABLE: self.show_3d_dem_button.config(state=tk.NORMAL)
if MAP_VIEWER_SYSTEM_AVAILABLE: self.view_map_for_point_button.config(state=tk.NORMAL)
except Exception as e: logger.exception("GUI Error: get_elevation"); messagebox.showerror("Error", f"Error:\n{e}", parent=self.root); self.point_result_label.config(text="Result: Error.")
finally: self._set_busy_state(False)
def _trigger_download_area_task(self) -> None:
if self.is_processing_task or not self.elevation_manager: return
bounds = self._validate_area_boundary_coordinates()
if not bounds: return
self.last_area_coords = bounds
self._set_busy_state(True)
self.area_download_status_label.config(text="Status: Starting...")
self.show_area_composite_button.config(state=tk.DISABLED)
if MAP_VIEWER_SYSTEM_AVAILABLE: self.view_map_for_area_button.config(state=tk.DISABLED)
self.root.update_idletasks()
dl_thread = threading.Thread(target=self._perform_background_area_download, args=bounds, daemon=True)
dl_thread.start()
def _perform_background_area_download(self, min_l: float, min_o: float, max_l: float, max_o: float) -> None:
status_str, success_bool, p_c, o_c = "Status: Unknown error.", False, 0, 0
try:
self.root.after(0, lambda: self.area_download_status_label.config(text="Status: Downloading..."))
if self.elevation_manager:
p_c,o_c = self.elevation_manager.download_area(min_l,min_o,max_l,max_o)
status_str = f"Status: Complete. Processed {p_c}, Obtained {o_c} HGT."
success_bool = True
else: status_str = "Status: Error - Manager N/A."
except Exception as e: logger.exception("GUI Error: area download task"); status_str = f"Status: Error: {type(e).__name__}"
finally: self.root.after(0, self._area_download_task_complete_ui_update, status_str, success_bool, p_c, o_c)
def _area_download_task_complete_ui_update(self, msg: str, success: bool, proc: int, obt: int) -> None:
self.area_download_status_label.config(text=msg)
self._set_busy_state(False)
if success:
summary = f"Processed {proc} tiles.\nObtained {obt} HGT files."
messagebox.showinfo("Download Complete", summary, parent=self.root)
if MATPLOTLIB_AVAILABLE and PIL_AVAILABLE: self.show_area_composite_button.config(state=tk.NORMAL)
if MAP_VIEWER_SYSTEM_AVAILABLE and obt > 0: self.view_map_for_area_button.config(state=tk.NORMAL)
else:
err_brief = msg.split(":")[-1].strip(); messagebox.showerror("Download Error", f"Failed: {err_brief}\nCheck logs.", parent=self.root)
self.show_area_composite_button.config(state=tk.DISABLED)
if MAP_VIEWER_SYSTEM_AVAILABLE: self.view_map_for_area_button.config(state=tk.DISABLED)
def _start_visualization_process(self, func: callable, args_t: tuple, name_id: str) -> None:
try:
proc = multiprocessing.Process(target=func, args=args_t, daemon=True, name=name_id)
proc.start()
logger.info(f"Started process '{name_id}' (PID: {proc.pid}).")
except Exception as e: logger.exception(f"Failed to start '{name_id}'."); messagebox.showerror("Process Error", f"Could not start {name_id}:\n{e}", parent=self.root)
def _trigger_2d_browse_display(self) -> None:
if not (MATPLOTLIB_AVAILABLE and PIL_AVAILABLE): messagebox.showwarning("Deps Error", "MPL/PIL N/A.", parent=self.root); return
if not self.last_valid_point_coords or not self.elevation_manager: messagebox.showinfo("Info", "Get elevation first.", parent=self.root); return
lat, lon = self.last_valid_point_coords
info = self.elevation_manager.get_tile_info(lat, lon)
if info and info.get("browse_available") and info.get("browse_image_path"):
args = (info["browse_image_path"], info.get("tile_base_name","?"), f"Browse: {info.get('tile_base_name','?').upper()}")
self._start_visualization_process(process_target_show_image, args, "2DBrowse")
else: messagebox.showinfo("Image Info", "Browse image N/A.", parent=self.root)
def _trigger_3d_dem_display(self) -> None:
if not MATPLOTLIB_AVAILABLE: messagebox.showwarning("Deps Error", "MPL N/A.", parent=self.root); return
if not self.last_valid_point_coords or not self.elevation_manager: messagebox.showinfo("Info", "Get elevation first.", parent=self.root); return
lat, lon = self.last_valid_point_coords
data = self.elevation_manager.get_hgt_data(lat, lon)
if data is not None:
info = self.elevation_manager.get_tile_info(lat, lon)
t_name = info.get("tile_base_name","?").upper() if info else "?"
p_title = f"3D View: Tile {t_name}"
cfg_is, cfg_ss, cfg_if, cfg_pgp = 1, 0.5, 3, 300
if not SCIPY_AVAILABLE: cfg_ss, cfg_if = None, 1; logger.warning("SciPy N/A. Advanced 3D plot disabled.")
args = (data, p_title, cfg_is, cfg_ss, cfg_if, cfg_pgp)
self._start_visualization_process(process_target_show_3d, args, "3DDEM")
else: messagebox.showerror("3D Data Error", "Could not retrieve HGT data.", parent=self.root)
def _trigger_area_composite_display(self) -> None:
if not (MATPLOTLIB_AVAILABLE and PIL_AVAILABLE): messagebox.showwarning("Deps Error", "MPL/PIL N/A.", parent=self.root); return
if not self.last_area_coords or not self.elevation_manager: messagebox.showinfo("Info", "Download area first.", parent=self.root); return
min_l,min_o,max_l,max_o = self.last_area_coords
info_list = self.elevation_manager.get_area_tile_info(min_l,min_o,max_l,max_o)
if not info_list: messagebox.showinfo("Area Info", "No tile info found.", parent=self.root); return
title = f"Area: Lat [{min_l:.1f}-{max_l:.1f}], Lon [{min_o:.1f}-{max_o:.1f}]"
args = (info_list, title)
self._start_visualization_process(process_target_create_show_area, args, "AreaComposite")
def _start_map_viewer_process(self, base_args_tuple: tuple) -> None:
if not self.map_interaction_message_queue:
logger.error("Cannot start map viewer: interaction queue missing."); messagebox.showerror("Internal Error", "Map comms not ready.", parent=self.root); return
try:
scale_val = self.map_display_scale_factor_var.get()
full_args = base_args_tuple + (scale_val,)
self.map_viewer_process_handle = multiprocessing.Process(target=run_map_viewer_process_target, args=full_args, daemon=True, name="GeoElevationMapViewer")
self.map_viewer_process_handle.start()
logger.info(f"Started map viewer PID: {self.map_viewer_process_handle.pid}, Scale: {scale_val:.3f}.")
self.root.after(100, self._process_map_interaction_queue_messages)
except Exception as e: logger.exception("Failed to start map viewer."); messagebox.showerror("Process Error",f"Could not start map viewer:\n{e}",parent=self.root)
def _trigger_view_map_for_point(self) -> None:
if not MAP_VIEWER_SYSTEM_AVAILABLE: messagebox.showerror("Feature N/A", "Map viewer system N/A.", parent=self.root); return
if not self.last_valid_point_coords: messagebox.showinfo("Info", "Get elevation first.", parent=self.root); return
if not self.map_interaction_message_queue: messagebox.showerror("Internal Error", "Map queue N/A.", parent=self.root); return
if self.map_viewer_process_handle and self.map_viewer_process_handle.is_alive(): messagebox.showinfo("Info", "Map already open.", parent=self.root); return
lat, lon = self.last_valid_point_coords
# MODIFIED: Use the correctly imported GEOELEVATION_DEM_CACHE_DEFAULT if elevation_manager is None
# WHY: Consistency in using the configured default cache path if the manager wasn't initialized
# due to an error but we still attempt to start the map viewer.
# HOW: Added a check for self.elevation_manager and used the constant if it's None.
dem_cache = self.elevation_manager.tile_directory if self.elevation_manager else GEOELEVATION_DEM_CACHE_DEFAULT
base_args = (self.map_interaction_message_queue, "point", lat, lon, None, dem_cache)
self._start_map_viewer_process(base_args)
def _trigger_view_map_for_area(self) -> None:
if not MAP_VIEWER_SYSTEM_AVAILABLE: messagebox.showerror("Feature N/A", "Map viewer system N/A.", parent=self.root); return
if not self.last_area_coords: messagebox.showinfo("Info", "Download area first.", parent=self.root); return
if not self.map_interaction_message_queue: messagebox.showerror("Internal Error", "Map queue N/A.", parent=self.root); return
if self.map_viewer_process_handle and self.map_viewer_process_handle.is_alive(): messagebox.showinfo("Info", "Map already open.", parent=self.root); return
min_l, min_o, max_l, max_o = self.last_area_coords
bbox = (min_o, min_l, max_o, max_l)
# MODIFIED: Use the correctly imported GEOELEVATION_DEM_CACHE_DEFAULT if elevation_manager is None
# WHY: Consistency in using the configured default cache path if the manager wasn't initialized
# due to an error but we still attempt to start the map viewer.
# HOW: Added a check for self.elevation_manager and used the constant if it's None.
dem_cache = self.elevation_manager.tile_directory if self.elevation_manager else GEOELEVATION_DEM_CACHE_DEFAULT
base_args = (self.map_interaction_message_queue, "area", None, None, bbox, dem_cache)
self._start_map_viewer_process(base_args)
def _process_map_interaction_queue_messages(self) -> None:
if not self.map_interaction_message_queue: return
try:
while not self.map_interaction_message_queue.empty():
msg = self.map_interaction_message_queue.get_nowait()
msg_t = msg.get("type")
if msg_t == "map_click_data":
lat,lon,elev_s = msg.get("latitude"), msg.get("longitude"), msg.get("elevation_str", "Error")
lat_txt,lon_txt = (f"{lat:.5f}" if lat is not None else "N/A"), (f"{lon:.5f}" if lon is not None else "N/A")
self.map_click_latitude_label.config(text=lat_txt)
self.map_click_longitude_label.config(text=lon_txt)
self.map_click_elevation_label.config(text=elev_s)
except queue.Empty: pass
except Exception as e: logger.error(f"Error processing map queue: {e}")
finally:
if self.map_interaction_message_queue: self.root.after(250, self._process_map_interaction_queue_messages)
def _on_application_closing(self) -> None:
logger.info("Main GUI application closing...")
if self.map_viewer_process_handle and self.map_viewer_process_handle.is_alive():
logger.info("Terminating active map viewer process...")
self.map_viewer_process_handle.terminate()
self.map_viewer_process_handle.join(timeout=1.5)
if self.map_viewer_process_handle.is_alive():
logger.warning("Map viewer process did not terminate via SIGTERM. Killing."); self.map_viewer_process_handle.kill()
self.map_viewer_process_handle.join(timeout=0.5)
if self.map_interaction_message_queue:
logger.debug("Closing map interaction queue."); self.map_interaction_message_queue.close()
try: self.map_interaction_message_queue.join_thread()
except Exception as e: logger.warning(f"Error joining map queue thread: {e}")
self.root.destroy()
# --- Main Execution Block (for direct script testing) ---
if __name__ == "__main__":
print("Running elevation_gui.py directly for testing purposes...")
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s-%(levelname)s-%(name)s-%(module)s-%(message)s")
logger.info(f"TEST: Rasterio Avail: {RASTERIO_AVAILABLE}, PIL Avail: {PIL_AVAILABLE}, MPL Avail: {MATPLOTLIB_AVAILABLE}, SciPy Avail: {SCIPY_AVAILABLE}, MapViewerSys Avail: {MAP_VIEWER_SYSTEM_AVAILABLE}")
multiprocessing.freeze_support()
test_root = tk.Tk()
app_test_instance: Optional[ElevationApp] = None
try:
app_test_instance = ElevationApp(test_root)
test_root.mainloop()
except Exception as e_test_run:
logger.critical(f"Fatal error in direct test run: {e_test_run}", exc_info=True)
if app_test_instance: app_test_instance._on_application_closing() # Try cleanup
elif test_root.winfo_exists(): test_root.destroy()
sys.exit(1)
Reference in New Issue View Git Blame Copy Permalink