VALLONGOL/SXXXXXXX_GeoElevation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
d5561c6f8b
SXXXXXXX_GeoElevation/geoelevation/elevation_gui.py
VALLONGOL d5561c6f8b add map viewer
2025-05-13 14:18:48 +02:00

1079 lines
55 KiB
Python
Raw Blame History

# geoelevation/elevation_gui.py
"""
Provides the main graphical user interface (GUI) for the GeoElevation application.
Allows users to get elevation for points, download DEM data for areas,
and visualize this data in 2D (browse images, area composites) and 3D (DEM tiles).
Integrates an optional map viewer functionality to display OpenStreetMap tiles,
apply a user-selected display scale, and interact with the map to get
elevation data for clicked points.
"""
# 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 # Used in __main__ block for test exit
import queue # For GUI communication with map viewer process
from typing import Optional, Tuple, List, Dict, TYPE_CHECKING, Any
# Third-party imports (handled by individual modules if specific)
# Local application/package imports (using absolute paths)
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
from geoelevation import DEFAULT_CACHE_DIR as GEOELEVATION_DEFAULT_DEM_CACHE
# Try to import map viewer components; functionality will be disabled if not found
try:
# We don't instantiate GeoElevationMapViewer directly in this class anymore,
# but its availability (and OpenCV's) determines if map features are enabled.
# We check for cv2 as a proxy for the map viewer system's core dependency.
import cv2 # Attempt to 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 # Ensure cv2 is None if import failed
MAP_VIEWER_SYSTEM_AVAILABLE = False
# Type hinting for NumPy arrays if visualizer or other components use them
if TYPE_CHECKING:
import numpy as np_typing # Use an alias to avoid conflicts if np is mocked
# Configure module-level logger
logger = logging.getLogger(__name__)
# --- MULTIPROCESSING TARGET FUNCTIONS (Module-Level for Pickling) ---
def process_target_show_image(
image_path: str,
tile_name: str,
window_title: str
) -> None:
"""
Target function for multiprocessing: Loads, prepares, and shows a single image.
Imports are local to the process.
"""
try:
# Local imports for the child process
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 # For os.path.basename
if not (local_pil_ok and local_mpl_ok):
# Using print as logging might not be configured in the child process's console
print(
"PROCESS ERROR (show_image): Pillow/Matplotlib libraries are missing in the child process."
)
return
prepared_image = local_lpst(image_path, tile_name)
if prepared_image:
print(f"PROCESS (show_image): Displaying image '{window_title}'")
local_sim(prepared_image, window_title) # This call blocks the child process
else:
print(
f"PROCESS ERROR (show_image): Failed to prepare image {local_os.path.basename(image_path)}"
)
except Exception as e_proc_img:
print(f"PROCESS ERROR occurred in process_target_show_image: {e_proc_img}")
import traceback as tb # Local import for traceback
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:
"""
Target function for multiprocessing: Shows 3D plot, optionally smoothed/interpolated.
Imports are local to the process.
"""
try:
# Local imports for the child process
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
# NumPy is typically imported by the visualizer module if Matplotlib is present
if not local_mpl_ok:
print("PROCESS ERROR (show_3d): Matplotlib library is missing in the child process.")
return
# Adjust parameters if SciPy is not available in the child process for advanced features
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 library missing in child process. "
"Disabling 3D plot smoothing/interpolation."
)
effective_smooth_sigma = None
effective_interpolation_factor = 1
if hgt_data is not None:
print(
f"PROCESS (show_3d): Generating 3D plot '{plot_title}' (InitialSubsample:{initial_subsample}, "
f"SmoothSigma:{effective_smooth_sigma}, InterpolationFactor:{effective_interpolation_factor}x, "
f"PlotGridPoints:{plot_grid_points})"
)
local_s3d( # Call the actual plotting function
hgt_data_array=hgt_data,
title=plot_title,
initial_subsample=initial_subsample,
smooth_sigma=effective_smooth_sigma,
interpolation_factor=effective_interpolation_factor,
plot_grid_points=plot_grid_points
) # This call blocks the child process
else:
print("PROCESS ERROR (show_3d): No HGT data array received for plotting.")
except Exception as e_proc_3d:
print(f"PROCESS ERROR occurred 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:
"""
Target function for multiprocessing: Creates a composite area image and shows it.
Imports are local to the process.
"""
try:
# Local imports for the child process
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) # This blocks child process
else:
print("PROCESS ERROR (show_area): Failed to create composite area image.")
except Exception as e_proc_area:
print(f"PROCESS ERROR occurred 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:
"""
Target function for multiprocessing: Runs the GeoElevationMapViewer.
Initializes ElevationManager and GeoElevationMapViewer in the child process.
"""
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:
# MODIFIED: Access 'map_display_window_controller'
# WHY: To match the attribute name used in the GeoElevationMapViewer class.
# HOW: Changed from 'map_display_window_handler' to 'map_display_window_controller'.
if child_map_viewer_instance.map_display_window_controller: # <<< MODIFICA QUI
if child_map_viewer_instance.map_display_window_controller.is_window_alive(): # <<< MODIFICA QUI
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.")
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 ElevationApp GUI.
"""
self.root: tk.Tk = parent_widget
self.elevation_manager: Optional[ElevationManager] = elevation_manager_instance
if self.elevation_manager is None:
try:
if not RASTERIO_AVAILABLE:
logger.warning("Rasterio library not found. Elevation data functions limited.")
self.elevation_manager = ElevationManager(
tile_directory=GEOELEVATION_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) # Min height increased for new scale options frame
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
# Map display scale factor, managed by a Tkinter DoubleVar
self.map_display_scale_factor_var: tk.DoubleVar = tk.DoubleVar(value=0.5) # Default to 50%
self.scale_options_map: Dict[str, float] = {
"25% (1:4)": 0.25,
"33% (1:3)": 0.33333, # More precision
"50% (1:2)": 0.5,
"75% (3:4)": 0.75,
"100% (Original Size)": 1.0,
"150% (3:2)": 1.5,
"200% (2:1)": 2.0
}
# Multiprocessing resources for map viewer
self.map_viewer_process_handle: Optional[multiprocessing.Process] = None
self.map_interaction_message_queue: Optional[multiprocessing.Queue] = None
if MAP_VIEWER_SYSTEM_AVAILABLE:
try:
# Forcing spawn context can sometimes help on macOS/Windows if default causes issues
# mp_context = multiprocessing.get_context("spawn")
# self.map_interaction_message_queue = mp_context.Queue()
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}")
# If queue creation fails, disable map viewer features that depend on it.
# This state needs to be checked before attempting to use the queue.
self._build_gui_layout()
self._apply_initial_widget_states()
# Start polling the map interaction queue if it was successfully created
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:
"""Finds the display text in scale_options_map closest to target_scale_value."""
closest_text_key = ""
smallest_difference = float('inf')
default_display_text = "50% (1:2)" # Ensure this is a key in scale_options_map
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_key = display_text
if closest_text_key and smallest_difference < 0.01: # Threshold for "close enough"
return closest_text_key
else:
# Fallback to the text matching the default scale value (0.5)
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 # Should not be reached if "50% (1:2)" is in map
def _build_gui_layout(self) -> None:
"""Creates and arranges all GUI widgets."""
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) # Allow main_app_frame to expand
# --- Point Elevation Section ---
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) # Equal weight for buttons
point_actions_subframe.columnconfigure(1, weight=1)
point_actions_subframe.columnconfigure(2, weight=1) # For the third (map) button
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 Section ---
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) # Entry expand
area_download_frame.columnconfigure(3, weight=1) # Entry expand
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 Scale Section ---
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) # Combobox expand
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), # Set initial text
values=scale_option_display_texts,
state="readonly", # User can only select from the list
width=20 # Adjust width as needed
)
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 Section ---
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)) # Now at row 3
map_click_info_frame.columnconfigure(1, weight=1) # Label value expand
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)
# Configure main_app_frame rows (0-indexed)
main_app_frame.columnconfigure(0, weight=1) # Main content column expands
main_app_frame.rowconfigure(0, weight=0) # Point Elevation Frame
main_app_frame.rowconfigure(1, weight=0) # Area Download Frame
main_app_frame.rowconfigure(2, weight=0) # Map Display Options Frame
main_app_frame.rowconfigure(3, weight=0) # Map Click Info Frame
# Initially hide map-specific sections if map viewer system not available
if not MAP_VIEWER_SYSTEM_AVAILABLE:
map_display_options_frame.grid_remove()
map_click_info_frame.grid_remove()
# Button text already updated in _apply_initial_widget_states if system N/A
def _apply_initial_widget_states(self) -> None:
"""Sets initial enabled/disabled states of GUI widgets."""
if self.elevation_manager is None:
# Critical failure: disable primary action buttons
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 Failed to Initialize.")
self.area_download_status_label.config(text="Status: Elevation Manager Failed.")
# Visualization buttons for point - initially disabled, enabled on successful data retrieval
self.show_2d_browse_button.config(state=tk.DISABLED)
self.show_3d_dem_button.config(state=tk.DISABLED)
# Visualization buttons for area - initially disabled, enabled on successful area download
self.show_area_composite_button.config(state=tk.DISABLED)
# Map viewer related buttons and controls
if MAP_VIEWER_SYSTEM_AVAILABLE:
self.view_map_for_point_button.config(state=tk.NORMAL) # Can be clicked if point data is later available
self.view_map_for_area_button.config(state=tk.DISABLED) # Needs area data first
self.map_scale_combobox.config(state="readonly")
else:
self.view_map_for_point_button.config(state=tk.DISABLED, text="Map (Sys N/A)")
self.view_map_for_area_button.config(state=tk.DISABLED, text="Map (Sys N/A)")
self.map_scale_combobox.config(state=tk.DISABLED)
# Update button text if specific visualization libraries are missing
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: # SciPy is an optional enhancement for 3D, Matplotlib is core
self.show_3d_dem_button.config(text="DEM (Matplotlib N/A)")
def _on_map_display_scale_changed(self, event: Optional[tk.Event] = None) -> None:
"""Callback when the map display scale Combobox selection changes."""
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})"
)
# TODO (Future): If a map window is currently open, could send a message
# via the queue to tell the map viewer process to rescale and redisplay.
# This requires the map viewer process to listen for such messages.
else:
logger.warning(f"Invalid map scale option selected from Combobox: '{selected_display_text}'")
def _set_busy_state(self, is_busy: bool) -> None:
"""Enables/disables main action buttons based on processing state."""
self.is_processing_task = is_busy
new_widget_state = tk.DISABLED if is_busy else tk.NORMAL
# Only modify state if the elevation manager is available
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, latitude_str: str, longitude_str: str
) -> Optional[Tuple[float, float]]:
"""Validates latitude and longitude input strings for a point."""
try:
if not latitude_str: raise ValueError("Latitude input cannot be empty.")
latitude_val = float(latitude_str.strip())
if not (-90.0 <= latitude_val < 90.0):
raise ValueError("Latitude must be in [-90, 90).")
if not longitude_str: raise ValueError("Longitude input cannot be empty.")
longitude_val = float(longitude_str.strip())
if not (-180.0 <= longitude_val < 180.0):
raise ValueError("Longitude must be in [-180, 180).")
return latitude_val, longitude_val
except ValueError as e_val_pt:
logger.error(f"Invalid point coordinate input: {e_val_pt}")
messagebox.showerror("Input Error", f"Invalid coordinate:\n{e_val_pt}", parent=self.root)
return None
def _validate_area_boundary_coordinates(self) -> Optional[Tuple[float, float, float, float]]:
"""Validates the four bounding box input strings for an area."""
try:
min_lat_s = self.min_latitude_entry.get().strip()
max_lat_s = self.max_latitude_entry.get().strip()
min_lon_s = self.min_longitude_entry.get().strip()
max_lon_s = self.max_longitude_entry.get().strip()
if not all([min_lat_s, max_lat_s, min_lon_s, max_lon_s]):
raise ValueError("All area boundary fields must be filled.")
min_lat_v = float(min_lat_s)
max_lat_v = float(max_lat_s)
min_lon_v = float(min_lon_s)
max_lon_v = float(max_lon_s)
if not (-90.0 <= min_lat_v < 90.0): raise ValueError("Min Latitude out of range.")
if not (-90.0 <= max_lat_v < 90.0): raise ValueError("Max Latitude out of range.")
if not (-180.0 <= min_lon_v < 180.0): raise ValueError("Min Longitude out of range.")
if not (-180.0 <= max_lon_v < 180.0): raise ValueError("Max Longitude out of range.")
if min_lat_v >= max_lat_v: raise ValueError("Min Latitude must be less than Max Latitude.")
if min_lon_v >= max_lon_v: raise ValueError("Min Longitude must be less than Max Longitude.")
return min_lat_v, min_lon_v, max_lat_v, max_lon_v # Standard order in app
except ValueError as e_val_area:
logger.error(f"Invalid area boundary input: {e_val_area}")
messagebox.showerror("Input Error", f"Invalid area boundary:\n{e_val_area}", parent=self.root)
return None
def _trigger_get_elevation_task(self) -> None:
"""Handles the 'Get Elevation' button click to fetch elevation for a point."""
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 # Validation failed, user already notified
latitude, longitude = coords
self._set_busy_state(True)
self.point_result_label.config(text="Result: Requesting elevation...")
# Disable visualization buttons until task completion
self.show_2d_browse_button.config(state=tk.DISABLED)
self.show_3d_dem_button.config(state=tk.DISABLED)
if MAP_VIEWER_SYSTEM_AVAILABLE: # Keep map button enabled if system is OK, but needs point
self.view_map_for_point_button.config(state=tk.DISABLED) # Disable until point is valid
self.last_valid_point_coords = None # Reset
self.root.update_idletasks() # Ensure UI update
try:
elevation_meters = self.elevation_manager.get_elevation(latitude, longitude)
display_text = "Result: "
if elevation_meters is None:
display_text += "Elevation data unavailable (check logs for details)."
messagebox.showwarning("Info", "Could not retrieve elevation.", parent=self.root)
elif math.isnan(elevation_meters):
display_text += "Point is within tile boundaries but on a NoData area."
self.last_valid_point_coords = (latitude, longitude) # Tile exists
else:
display_text += f"Elevation is {elevation_meters:.2f} meters."
self.last_valid_point_coords = (latitude, longitude)
self.point_result_label.config(text=display_text)
# Update states of visualization buttons based on successful data retrieval
if self.last_valid_point_coords:
if MATPLOTLIB_AVAILABLE and PIL_AVAILABLE:
self.show_2d_browse_button.config(state=tk.NORMAL)
if MATPLOTLIB_AVAILABLE: # SciPy is for advanced options within 3D plot
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_get_elev_task:
logger.exception("Error during GUI 'get_elevation' task.")
messagebox.showerror("Error", f"Unexpected error occurred:\n{e_get_elev_task}", parent=self.root)
self.point_result_label.config(text="Result: Error (see application logs).")
finally:
self._set_busy_state(False)
def _trigger_download_area_task(self) -> None:
"""Handles the 'Download Area Tiles' button click."""
if self.is_processing_task or not self.elevation_manager:
return
bounds = self._validate_area_boundary_coordinates()
if not bounds:
return # Validation failed
# Unpack bounds (min_lat, min_lon, max_lat, max_lon)
self.last_area_coords = bounds
self._set_busy_state(True)
self.area_download_status_label.config(text="Status: Starting download process...")
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()
# Run the download in a separate thread
download_thread = threading.Thread(
target=self._perform_background_area_download,
args=bounds, # Pass the tuple directly
daemon=True
)
download_thread.start()
def _perform_background_area_download(
self, min_lat: float, min_lon: float, max_lat: float, max_lon: float
) -> None:
"""Worker function for background area download, executed in a thread."""
final_status_text = "Status: Unknown error during area download."
download_succeeded = False
num_tiles_processed = 0
num_hgt_obtained = 0
try:
# Update GUI from thread using root.after
self.root.after(
0, lambda: self.area_download_status_label.config(text="Status: Downloading/Verifying area tiles...")
)
if self.elevation_manager:
num_tiles_processed, num_hgt_obtained = self.elevation_manager.download_area(
min_lat, min_lon, max_lat, max_lon
)
final_status_text = (
f"Status: Download complete. Processed {num_tiles_processed} locations. "
f"Obtained/Verified {num_hgt_obtained} HGT files."
)
download_succeeded = True
else:
final_status_text = "Status: Error - Elevation Manager instance not available."
except Exception as e_bg_download:
logger.exception("Error during background area download task.")
final_status_text = f"Status: Error - {type(e_bg_download).__name__} (see logs)."
finally:
# Schedule GUI update back on the main thread
self.root.after(
0, self._area_download_task_complete_ui_update,
final_status_text, download_succeeded, num_tiles_processed, num_hgt_obtained
)
def _area_download_task_complete_ui_update(
self, status_message_text: str, success_flag: bool, processed_count: int, obtained_count: int
) -> None:
"""Updates GUI elements after the area download task is complete."""
self.area_download_status_label.config(text=status_message_text)
self._set_busy_state(False) # Release busy state
if success_flag:
summary_message = (
f"Area data processing complete.\n\n"
f"Checked {processed_count} tile locations.\n"
f"Successfully found or downloaded {obtained_count} HGT files."
)
messagebox.showinfo("Download Complete", summary_message, parent=self.root)
# Enable visualization buttons for the area if dependencies are met
if MATPLOTLIB_AVAILABLE and PIL_AVAILABLE:
self.show_area_composite_button.config(state=tk.NORMAL)
# Enable map view for area if system is OK and some HGT files were relevant
if MAP_VIEWER_SYSTEM_AVAILABLE and obtained_count > 0 :
self.view_map_for_area_button.config(state=tk.NORMAL)
else:
# Extract brief error for the popup from the detailed status message
brief_error_message = status_message_text.split(":")[-1].strip()
messagebox.showerror(
"Download Error",
f"Area download operation failed: {brief_error_message}\nPlease check application logs for details.",
parent=self.root
)
# Ensure area-specific buttons remain disabled on failure
self.show_area_composite_button.config(state=tk.DISABLED)
self.view_map_for_area_button.config(state=tk.DISABLED)
def _start_visualization_process(
self, target_function: callable, arguments_tuple: tuple, process_id_name: str
) -> None:
"""Helper method to start a daemonized multiprocessing.Process for visualizations."""
try:
viz_process = multiprocessing.Process(
target=target_function,
args=arguments_tuple,
daemon=True, # Ensures process exits when the main application exits
name=process_id_name # Helpful for debugging
)
viz_process.start()
logger.info(f"Started visualization process '{process_id_name}' (PID: {viz_process.pid}).")
except Exception as e_viz_proc_start:
logger.exception(f"Failed to start '{process_id_name}' process.")
messagebox.showerror(
"Process Start Error",
f"Could not start the {process_id_name} display process:\n{e_viz_proc_start}",
parent=self.root
)
def _trigger_2d_browse_display(self) -> None:
"""Triggers display of the 2D browse image for the last valid point."""
if not (MATPLOTLIB_AVAILABLE and PIL_AVAILABLE):
messagebox.showwarning("Dependency Missing", "Matplotlib and Pillow are required.", parent=self.root)
return
if not self.last_valid_point_coords or not self.elevation_manager:
messagebox.showinfo("Information", "Get a successful point elevation first.", parent=self.root)
return
latitude, longitude = self.last_valid_point_coords
retrieved_tile_info = self.elevation_manager.get_tile_info(latitude, longitude)
if retrieved_tile_info and \
retrieved_tile_info.get("browse_available") and \
retrieved_tile_info.get("browse_image_path"):
image_file_path = retrieved_tile_info["browse_image_path"]
base_tile_name = retrieved_tile_info.get("tile_base_name", "UnknownTile")
window_display_title = f"Browse Image: {base_tile_name.upper()}"
process_args = (image_file_path, base_tile_name, window_display_title)
self._start_visualization_process(process_target_show_image, process_args, "2DBrowseImageDisplay")
else:
messagebox.showinfo("Image Info", "Browse image is not available for this location.", parent=self.root)
def _trigger_3d_dem_display(self) -> None:
"""Triggers display of the 3D DEM tile for the last valid point."""
if not MATPLOTLIB_AVAILABLE: # Core requirement for 3D plot
messagebox.showwarning("Dependency Missing", "Matplotlib is required for 3D DEM display.", parent=self.root)
return
if not self.last_valid_point_coords or not self.elevation_manager:
messagebox.showinfo("Information", "Get a successful point elevation first.", parent=self.root)
return
latitude, longitude = self.last_valid_point_coords
hgt_numpy_array = self.elevation_manager.get_hgt_data(latitude, longitude)
if hgt_numpy_array is not None:
retrieved_tile_info = self.elevation_manager.get_tile_info(latitude, longitude)
base_tile_name = retrieved_tile_info.get("tile_base_name", "Unknown").upper() if retrieved_tile_info else "Unknown"
plot_display_title = f"3D DEM View: Tile {base_tile_name}"
# Configuration for 3D plot rendering
config_initial_subsample = 1
config_smooth_sigma: Optional[float] = 0.5 # Light smoothing, or None to disable
config_interpolation_factor = 3 # e.g., 2, 3, or 4 for smoother surface
config_plot_grid_points = 300 # Target points for the final rendered grid
# Disable advanced features if SciPy is not available (checked in main thread)
if not SCIPY_AVAILABLE:
if config_smooth_sigma is not None: config_smooth_sigma = None
if config_interpolation_factor > 1: config_interpolation_factor = 1
logger.warning("SciPy not available in main thread. Advanced 3D plot features disabled for child process.")
process_args = (
hgt_numpy_array, plot_display_title, config_initial_subsample,
config_smooth_sigma, config_interpolation_factor, config_plot_grid_points
)
self._start_visualization_process(process_target_show_3d, process_args, "3DDEMTileDisplay")
else:
messagebox.showerror("3D Data Error", "Could not retrieve HGT data array for 3D plot.", parent=self.root)
def _trigger_area_composite_display(self) -> None:
"""Triggers display of the composite 2D browse image for the last downloaded area."""
if not (MATPLOTLIB_AVAILABLE and PIL_AVAILABLE):
messagebox.showwarning("Dependency Missing", "Matplotlib and Pillow are required.", parent=self.root)
return
if not self.last_area_coords or not self.elevation_manager:
messagebox.showinfo("Information", "Download map tiles for an area successfully first.", parent=self.root)
return
min_lat, min_lon, max_lat, max_lon = self.last_area_coords
list_of_tile_info = self.elevation_manager.get_area_tile_info(min_lat, min_lon, max_lat, max_lon)
if not list_of_tile_info:
messagebox.showinfo("Area Info", "No tile information found for the specified area.", parent=self.root)
return
window_display_title = f"Area Composite: Lat [{min_lat:.1f}-{max_lat:.1f}], Lon [{min_lon:.1f}-{max_lon:.1f}]"
process_args = (list_of_tile_info, window_display_title)
self._start_visualization_process(process_target_create_show_area, process_args, "AreaCompositeDisplay")
def _start_map_viewer_process(self, base_process_args_tuple: tuple) -> None:
"""
Helper to start the map viewer process. It retrieves the current display
scale factor and appends it to the arguments for the target function.
"""
if not self.map_interaction_message_queue:
logger.error("Cannot start map viewer: interaction message queue is not initialized.")
messagebox.showerror("Internal Error", "Map communication system is not ready.", parent=self.root)
return
try:
current_display_scale = self.map_display_scale_factor_var.get()
# base_process_args_tuple is (queue, mode, lat, lon, bbox, dem_cache_path)
# Append display_scale_factor as the last argument
full_args_for_map_process = base_process_args_tuple + (current_display_scale,)
self.map_viewer_process_handle = multiprocessing.Process(
target=run_map_viewer_process_target, # Module-level target function
args=full_args_for_map_process,
daemon=True, # Ensures process exits with main app
name="GeoElevationMapViewerProcess"
)
self.map_viewer_process_handle.start()
logger.info(
f"Started map viewer process (PID: {self.map_viewer_process_handle.pid}) "
f"with display scale: {current_display_scale:.3f}."
)
# Ensure queue polling is (re)started if necessary
self.root.after(100, self._process_map_interaction_queue_messages)
except Exception as e_map_viewer_start:
logger.exception("Failed to start the map viewer process.")
messagebox.showerror(
"Process Start Error",
f"Could not start the map viewer process:\n{e_map_viewer_start}",
parent=self.root
)
def _trigger_view_map_for_point(self) -> None:
"""Triggers the display of the map centered on the last valid point."""
if not MAP_VIEWER_SYSTEM_AVAILABLE:
messagebox.showerror("Feature Unavailable", "Map viewer system components are not available.", parent=self.root)
return
if not self.last_valid_point_coords:
messagebox.showinfo("Information", "Please get elevation for a specific point first.", parent=self.root)
return
if not self.map_interaction_message_queue: # Check if queue was created
messagebox.showerror("Internal Error", "Map interaction system is not properly initialized.", parent=self.root)
return
if self.map_viewer_process_handle and self.map_viewer_process_handle.is_alive():
messagebox.showinfo("Information", "Map window is already open. Please close it first to view a new map.", parent=self.root)
return
latitude, longitude = self.last_valid_point_coords
dem_cache_path = self.elevation_manager.tile_directory if self.elevation_manager else GEOELEVATION_DEFAULT_DEM_CACHE
# Arguments for run_map_viewer_process_target (excluding scale factor, added by helper)
base_args = (
self.map_interaction_message_queue,
"point", # operation_mode
latitude,
longitude,
None, # area_bounding_box (None for point mode)
dem_cache_path
)
self._start_map_viewer_process(base_args)
def _trigger_view_map_for_area(self) -> None:
"""Triggers the display of the map covering the last downloaded area."""
if not MAP_VIEWER_SYSTEM_AVAILABLE:
messagebox.showerror("Feature Unavailable", "Map viewer system components are not available.", parent=self.root)
return
if not self.last_area_coords: # This is set after a successful area download
messagebox.showinfo("Information", "Please download map tiles for an area first.", parent=self.root)
return
if not self.map_interaction_message_queue:
messagebox.showerror("Internal Error", "Map interaction system is not properly initialized.", parent=self.root)
return
if self.map_viewer_process_handle and self.map_viewer_process_handle.is_alive():
messagebox.showinfo("Information", "Map window is already open. Please close it first.", parent=self.root)
return
min_lat, min_lon, max_lat, max_lon = self.last_area_coords
# Bounding box format for mercantile: (west, south, east, north)
area_bbox_data = (min_lon, min_lat, max_lon, max_lat)
dem_cache_path = self.elevation_manager.tile_directory if self.elevation_manager else GEOELEVATION_DEFAULT_DEM_CACHE
base_args = (
self.map_interaction_message_queue,
"area", # operation_mode
None, # center_latitude (None for area mode)
None, # center_longitude (None for area mode)
area_bbox_data,
dem_cache_path
)
self._start_map_viewer_process(base_args)
def _process_map_interaction_queue_messages(self) -> None:
"""Periodically checks the queue for messages from the map viewer process."""
if not self.map_interaction_message_queue:
# Queue was not initialized (e.g., MAP_VIEWER_SYSTEM_AVAILABLE was False)
return
try:
while not self.map_interaction_message_queue.empty():
message_data = self.map_interaction_message_queue.get_nowait()
message_type = message_data.get("type")
if message_type == "map_click_data":
lat_val = message_data.get("latitude")
lon_val = message_data.get("longitude")
elevation_text = message_data.get("elevation_str", "Error fetching")
latitude_display_text = f"{lat_val:.5f}" if lat_val is not None else "N/A"
longitude_display_text = f"{lon_val:.5f}" if lon_val is not None else "N/A"
self.map_click_latitude_label.config(text=latitude_display_text)
self.map_click_longitude_label.config(text=longitude_display_text)
self.map_click_elevation_label.config(text=elevation_text)
# Add handling for other message types from map viewer if needed in the future
# else:
# logger.warning(f"Received unknown message type from map queue: {message_type}")
except queue.Empty: # This is expected when the queue has no new messages
pass
except Exception as e_gui_queue_proc:
logger.error(f"Error processing map interaction queue in GUI: {e_gui_queue_proc}")
finally:
# Continue polling if the queue exists. A more robust check might involve
# also checking if self.map_viewer_process_handle is alive, but messages
# could still arrive after the process ends.
if self.map_interaction_message_queue:
self.root.after(250, self._process_map_interaction_queue_messages) # Poll every 250ms
def _on_application_closing(self) -> None:
"""Handles cleanup when the main application window is closed."""
logger.info("Main GUI application is closing. Performing cleanup...")
# Terminate the map viewer process if it's running
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() # Sends SIGTERM
# Wait for a short period for the process to exit
self.map_viewer_process_handle.join(timeout=1.5) # Increased timeout slightly
if self.map_viewer_process_handle.is_alive():
logger.warning("Map viewer process did not terminate via SIGTERM. Attempting SIGKILL.")
self.map_viewer_process_handle.kill() # Force kill
self.map_viewer_process_handle.join(timeout=0.5) # Brief wait for kill to take effect
# Close and join the multiprocessing queue to release resources
if self.map_interaction_message_queue:
logger.debug("Closing and joining map interaction message queue.")
try:
self.map_interaction_message_queue.close()
# join_thread waits for the queue's feeder thread to finish
self.map_interaction_message_queue.join_thread()
except Exception as e_queue_cleanup:
# Log errors but don't prevent application exit
logger.warning(f"Error during map interaction queue cleanup: {e_queue_cleanup}")
self.root.destroy() # Destroy the main Tkinter window
# --- Main Execution Block (for direct script testing of this GUI module) ---
if __name__ == "__main__":
print("Running elevation_gui.py directly for testing purposes...")
# Basic logging setup for testing
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s - %(levelname)s - %(name)s - %(module)s - %(message)s"
)
logger.info("--- Direct Test Run: Dependency Status ---")
logger.info(f"Rasterio Available: {RASTERIO_AVAILABLE}")
logger.info(f"Pillow (PIL) Available: {PIL_AVAILABLE}")
logger.info(f"Matplotlib Available: {MATPLOTLIB_AVAILABLE}")
logger.info(f"SciPy Available: {SCIPY_AVAILABLE}")
logger.info(f"Map Viewer System (OpenCV & Co.) Available: {MAP_VIEWER_SYSTEM_AVAILABLE}")
logger.info("-----------------------------------------")
# Important for multiprocessing if any child processes are spawned during tests
multiprocessing.freeze_support()
test_root_tk_window = tk.Tk()
gui_app_instance: Optional[ElevationApp] = None
try:
gui_app_instance = ElevationApp(test_root_tk_window)
test_root_tk_window.mainloop()
except Exception as e_main_gui_test:
logger.critical(
f"Fatal error during direct test run of ElevationApp: {e_main_gui_test}", exc_info=True
)
try:
# Attempt to show a simple Tkinter error dialog if GUI is somewhat functional
error_dialog_root = tk.Tk()
error_dialog_root.withdraw() # Hide the empty root window
messagebox.showerror(
"Fatal Test Error",
f"Application test run failed critically:\n{e_main_gui_test}",
parent=None # No parent if main window might be broken
)
error_dialog_root.destroy()
except Exception:
pass # Ignore if the error dialog itself fails
# Attempt cleanup if app instance was created
if gui_app_instance and hasattr(gui_app_instance, '_on_application_closing'):
gui_app_instance._on_application_closing() # Call cleanup manually
else:
if test_root_tk_window.winfo_exists(): # Check if window still exists
test_root_tk_window.destroy() # Fallback destroy
sys.exit(1) # Exit with error status for test run
Reference in New Issue View Git Blame Copy Permalink