# geoelevation/__init__.py """ GeoElevation Package. Provides functionalities to retrieve elevation data for geographic coordinates, a Graphical User Interface (GUI) for interactive exploration, a Command Line Interface (CLI) for quick point elevation lookups, and an API for integration into other Python applications. Key public interfaces: - get_point_elevation(): Function to retrieve elevation for a lat/lon point. - run_gui_application(): Function to launch the main GUI. - GEOELEVATION_DEM_CACHE_DEFAULT: Default cache path for DEM files. - _LibraryProgressWindow: Internal utility for showing progress (exposed for typing/advanced use). """ # Standard library imports import os import sys import tkinter as tk # For _LibraryProgressWindow and parts of run_gui_application from tkinter import ttk # For _LibraryProgressWindow import threading import logging import math # For math.isnan in get_point_elevation from typing import Optional, Tuple, Any # Added Any # --- Configure Logging for Library Use --- # This setup ensures that the library itself doesn't output logs unless the # consuming application configures handlers for the 'geoelevation' logger. library_logger = logging.getLogger(__name__) # __name__ will be 'geoelevation' if not library_logger.hasHandlers(): library_logger.addHandler(logging.NullHandler()) # By default, the library logger will inherit the level of the root logger, # or it can be explicitly set, e.g., library_logger.setLevel(logging.WARNING). # For now, let the consuming application control the level. # --- Import Core Components and Configuration --- # These imports make key classes and constants available and are used by # the public API functions defined in this __init__.py. ## Ensure external `python-map-manager` package (workspace submodule) is importable # This helps top-level modules (e.g. elevation_gui) import `map_manager` early. try: _pkg_root = os.path.dirname(__file__) _candidate = os.path.normpath(os.path.join(_pkg_root, "..", "external", "python-map-manager")) if os.path.isdir(_candidate) and _candidate not in sys.path: # Prepend so it takes precedence over other installations sys.path.insert(0, _candidate) library_logger.debug(f"Added external python-map-manager to sys.path: {_candidate}") except Exception: # Non-fatal: if path manipulation fails, imports will raise as before and fallback handles it. pass try: from .elevation_manager import ElevationManager from .elevation_manager import RASTERIO_AVAILABLE # Critical dependency check # MODIFIED: Import GEOELEVATION_DEM_CACHE_DEFAULT directly from .config # WHY: Centralizes configuration, avoids potential import timing issues. # HOW: Changed from defining a fallback here to importing from the config module. from .config import GEOELEVATION_DEM_CACHE_DEFAULT except ImportError as e_core_pkg_import: # This signifies a critical failure if core components cannot be imported. library_logger.critical( f"Failed to import core components (ElevationManager, RASTERIO_AVAILABLE, or config) " f"for GeoElevation package: {e_core_pkg_import}. The library will likely not function.", exc_info=True ) # Define fallback values/classes so that an application attempting to import # 'geoelevation' doesn't crash immediately on this __init__.py, but can # gracefully detect that the library is non-functional. class ElevationManager: # type: ignore # Dummy class def __init__(self, *args: Any, **kwargs: Any) -> None: pass def get_elevation(self, *args: Any, **kwargs: Any) -> None: return None RASTERIO_AVAILABLE = False GEOELEVATION_DEM_CACHE_DEFAULT = "geoelevation_dem_cache_unavailable" # --- Global Shared ElevationManager for Library Functions --- # This instance is lazily initialized and shared for calls to get_point_elevation # when geoelevation is used as a library, improving efficiency. _shared_library_elevation_manager_instance: Optional[ElevationManager] = None _shared_manager_init_lock = threading.Lock() # Ensures thread-safe initialization def _get_shared_library_elevation_manager() -> ElevationManager: """ Lazily initializes and returns a globally shared ElevationManager instance. This function is thread-safe. Returns: An initialized ElevationManager instance. Raises: RuntimeError: If Rasterio (a critical dependency) is not available, or if the ElevationManager itself fails to initialize. """ global _shared_library_elevation_manager_instance if not RASTERIO_AVAILABLE: error_message = ( "Rasterio library is not installed or could not be imported. " "This library is essential for elevation data retrieval functions. " "Please install Rasterio (e.g., 'pip install rasterio') and ensure it is operational." ) library_logger.error(error_message) raise RuntimeError(error_message) # Use a lock for thread-safe lazy initialization of the shared manager with _shared_manager_init_lock: if _shared_library_elevation_manager_instance is None: library_logger.debug("Initializing shared ElevationManager for library use...") try: # Allow overriding the default DEM cache directory via an environment variable dem_cache_directory_path = os.environ.get( "GEOELEVATION_DEM_CACHE_DIR", GEOELEVATION_DEM_CACHE_DEFAULT ) _shared_library_elevation_manager_instance = ElevationManager( tile_directory=dem_cache_directory_path ) library_logger.info( f"GeoElevation library: Shared ElevationManager initialized. " f"Using DEM Cache Directory: '{dem_cache_directory_path}'" ) except Exception as e_manager_lib_init_fatal: library_logger.error( f"GeoElevation library: CRITICAL - Failed to initialize shared ElevationManager: {e_manager_lib_init_fatal}", exc_info=True ) raise RuntimeError( f"Failed to initialize the core ElevationManager for GeoElevation library: {e_manager_lib_init_fatal}" ) from e_manager_lib_init_fatal return _shared_library_elevation_manager_instance class _LibraryProgressWindow(threading.Thread): """ A simple, non-blocking progress window using Tkinter, designed for use when GeoElevation functions are called as a library and might take time. Runs in its own thread. """ def __init__( self, window_title_text: str = "GeoElevation - Processing", progress_message_text: str = "Accessing elevation data...\nPlease wait." ) -> None: super().__init__(daemon=True) # Daemon threads exit when the main program exits self.window_title: str = window_title_text self.progress_message: str = progress_message_text self.tkinter_root: Optional[tk.Tk] = None self.is_tk_ready_event = threading.Event() # Signals when Tkinter root window is ready self._please_stop_event = threading.Event() # Signals this thread's Tkinter loop to stop def run(self) -> None: """Executes the Tkinter event loop for the progress window in this thread.""" try: self.tkinter_root = tk.Tk() self.tkinter_root.title(self.window_title) # Window styling and positioning win_width = 380 win_height = 120 screen_w = self.tkinter_root.winfo_screenwidth() screen_h = self.tkinter_root.winfo_screenheight() pos_x = int(screen_w / 2 - win_width / 2) pos_y = int(screen_h / 2 - win_height / 2) self.tkinter_root.geometry(f'{win_width}x{win_height}+{pos_x}+{pos_y}') self.tkinter_root.resizable(False, False) # MODIFIED: Set the window to be always on top. # WHY: Ensures the progress window is visible and not hidden by other windows (like the map window). # HOW: Used the `attributes()` method with the `'-topmost', True` option. Wrapped in try-except # to handle potential TclErrors on platforms where this might not be fully supported. try: self.tkinter_root.attributes('-topmost', True) library_logger.debug("Progress window set to be topmost.") except tk.TclError as e_topmost: library_logger.warning(f"Could not set progress window to topmost: {e_topmost}") content_frame = ttk.Frame(self.tkinter_root, padding="15") content_frame.pack(expand=True, fill=tk.BOTH) msg_label = ttk.Label( content_frame, text=self.progress_message, font=("Helvetica", 10), justify=tk.CENTER ) msg_label.pack(pady=(0, 10)) prog_bar = ttk.Progressbar(content_frame, mode='indeterminate', length=300) prog_bar.pack(pady=10) prog_bar.start(25) # Animation interval in milliseconds self.tkinter_root.protocol("WM_DELETE_WINDOW", self._trigger_stop_from_window_close) self.is_tk_ready_event.set() # Signal that the Tkinter window is configured self._check_stop_request_periodically() # Start periodic check for stop signal self.tkinter_root.mainloop() # Blocks this thread until window is closed via quit() except Exception as e_prog_win_thread_run: library_logger.warning( f"GeoElevation ProgressWindow: Error during thread run method: {e_prog_win_thread_run}", exc_info=True ) self.is_tk_ready_event.set() # Ensure event is set to unblock any waiters, even on error finally: # Attempt to clean up Tkinter resources if mainloop exited unexpectedly if hasattr(self, 'tkinter_root') and self.tkinter_root: try: if self.tkinter_root.winfo_exists(): # Mainloop should handle destroy on quit(), but as a fallback: # self.tkinter_root.destroy() pass except tk.TclError as e_tcl_final_cleanup: # Catch Tcl errors if already destroyed if "application has been destroyed" not in str(e_tcl_final_cleanup).lower(): library_logger.warning(f"GeoElevation ProgressWindow: TclError on final cleanup: {e_tcl_final_cleanup}") except Exception as e_destroy_final_prog_win: library_logger.warning(f"GeoElevation ProgressWindow: Exception on final cleanup: {e_destroy_final_prog_win}") library_logger.debug("GeoElevation ProgressWindow: Thread execution finished.") def _check_stop_request_periodically(self) -> None: """Internal method, run via Tkinter's 'after', to check for external stop requests.""" if self._please_stop_event.is_set(): # If stop has been requested if self.tkinter_root and self.tkinter_root.winfo_exists(): try: # This is executed in the Tkinter thread, so it's safe to call quit() self.tkinter_root.quit() # Cleanly exit the mainloop except Exception as e_tk_periodic_quit: library_logger.warning( f"GeoElevation ProgressWindow: Error during Tcl quit in _check_stop_request_periodically: {e_tk_periodic_quit}" ) else: # Reschedule this check if the window still exists if self.tkinter_root and self.tkinter_root.winfo_exists(): try: self.tkinter_root.after(150, self._check_stop_request_periodically) # Check again in 150ms except tk.TclError: # Can occur if root is destroyed while 'after' is pending pass # Window is gone, no need to reschedule def _trigger_stop_from_window_close(self) -> None: """Callback for WM_DELETE_WINDOW (window 'X' button), runs in Tkinter thread.""" library_logger.debug("GeoElevation ProgressWindow: WM_DELETE_WINDOW (close button) invoked by user.") self._please_stop_event.set() # Signal the stop event # The _check_stop_request_periodically method will then see this and call root.quit() def request_progress_window_stop(self) -> None: """Public method callable from any thread to request the progress window to stop.""" library_logger.debug("GeoElevation ProgressWindow: Stop requested from an external thread.") self._please_stop_event.set() # The actual closure is handled by the periodic check within the Tkinter thread. # --- Public API Functions Exposed by the Package --- def get_point_elevation( latitude: float, longitude: float, # MODIFIED: Renamed parameter back to show_progress_dialog for backward compatibility. # WHY: The external tool expects this parameter name. # HOW: Changed the parameter name in the function signature. show_progress_dialog: bool = False, progress_dialog_custom_message: str = "Retrieving elevation data from remote sources.\nThis may take a moment..." ) -> Optional[float]: """ Retrieves the elevation for a given geographic point (latitude, longitude). Args: latitude (float): The latitude of the point. longitude (float): The longitude of the point. show_progress_dialog (bool, optional): If True, attempts to display a simple progress window. Defaults to False. (May only work in environments where Tkinter can be safely run in a background thread, like the main process). progress_dialog_custom_message (str, optional): Custom message for the progress dialog. Defaults to a standard message. Returns: Optional[float]: The elevation in meters, float('nan') if the point is on a NoData area, or None if data is unavailable or an error occurs. Raises: RuntimeError: If core dependencies like Rasterio are not available. ValueError: If input coordinates are invalid. """ progress_window_instance: Optional[_LibraryProgressWindow] = None try: elevation_manager_instance = _get_shared_library_elevation_manager() if not (-90.0 <= latitude < 90.0): # Basic validation library_logger.error(f"GeoElevation API: Invalid latitude provided: {latitude}") return None if not (-180.0 <= longitude < 180.0): # Basic validation library_logger.error(f"GeoElevation API: Invalid longitude provided: {longitude}") return None # MODIFIED: Use the renamed parameter show_progress_dialog. # WHY: The variable name inside the function must match the parameter name in the signature. # HOW: Changed the variable name here. if show_progress_dialog: # MODIFIED: Check if _LibraryProgressWindow class is available before instantiating. # WHY: Prevent NameError if the import failed due to missing dependencies. # HOW: Added `if _LibraryProgressWindow is not None`. if _LibraryProgressWindow is not None: library_logger.debug("GeoElevation API: Progress dialog requested. Preparing window.") try: progress_window_instance = _LibraryProgressWindow( progress_message_text=progress_dialog_custom_message ) progress_window_instance.start() # Wait for the Tkinter window to signal it's ready, with a timeout # This wait is important because the main thread will proceed to a potentially blocking call. tk_ready_signal = progress_window_instance.is_tk_ready_event.wait(timeout=2.5) # Slightly longer timeout if not tk_ready_signal: library_logger.warning( "GeoElevation API: Progress dialog did not signal readiness within timeout. Continuing without it." ) # If it didn't become ready, ensure we don't try to stop a non-existent/broken window later # Signal it to stop, attempt a brief join, and set to None so finally block doesn't interact. try: progress_window_instance.request_progress_window_stop() except Exception: pass try: progress_window_instance.join(timeout=0.5) except Exception: pass progress_window_instance = None # Treat as if not started for finally block except Exception as e_prog_win_api: library_logger.error(f"GeoElevation API Error: Failed to start progress window: {e_prog_win_api}", exc_info=True) progress_window_instance = None # Ensure reference is None on failure else: library_logger.warning("GeoElevation API: Cannot show progress dialog: _LibraryProgressWindow class is not available.") library_logger.info(f"GeoElevation API: Requesting elevation for Lat: {latitude}, Lon: {longitude}") # The call to get_elevation might block if downloading data. retrieved_elevation_value = elevation_manager_instance.get_elevation(latitude, longitude) if retrieved_elevation_value is None: library_logger.warning(f"GeoElevation API: Elevation data unavailable for Lat: {latitude}, Lon: {longitude}") elif isinstance(retrieved_elevation_value, float) and math.isnan(retrieved_elevation_value): library_logger.info(f"GeoElevation API: Point Lat: {latitude}, Lon: {longitude} is on a NoData area.") else: library_logger.info( f"GeoElevation API: Elevation for Lat: {latitude}, Lon: {longitude} is {retrieved_elevation_value:.2f}m" ) return retrieved_elevation_value except RuntimeError: # Specifically from _get_shared_library_elevation_manager library_logger.error("GeoElevation API: Critical runtime error (e.g., Rasterio missing).", exc_info=False) # Log less for expected runtime errors raise # Re-raise critical initialization errors to the caller except ValueError: # Propagated from ElevationManager for invalid coordinates for tile naming etc. library_logger.error("GeoElevation API: Invalid value encountered during elevation retrieval.", exc_info=False) raise except Exception as e_api_call_unhandled: library_logger.error( f"GeoElevation API: Unexpected error in get_point_elevation for ({latitude},{longitude}): {e_api_call_unhandled}", exc_info=True ) return None # Return None for other unexpected errors finally: # MODIFIED: Check if progress_window_instance is not None before calling methods. # WHY: The instance might be None if it failed to start. # HOW: Added the check `if progress_window_instance:`. if progress_window_instance: if progress_window_instance.is_alive(): library_logger.debug("GeoElevation API: Signaling progress dialog to stop after operation.") progress_window_instance.request_progress_window_stop() # Wait for the progress thread to finish processing the stop request and exit its mainloop. # Use a timeout to prevent the main thread from hanging indefinitely if the progress thread encounters an issue. progress_window_instance.join(timeout=3.0) # Wait for thread to finish if progress_window_instance.is_alive(): library_logger.warning( "GeoElevation API: Progress dialog thread did not terminate cleanly after join." ) else: library_logger.debug("GeoElevation API: Progress dialog was not alive or already stopped. No stop signal sent.") def run_gui_application() -> None: """ Launches the main GeoElevation Graphical User Interface (GUI). (Full docstring from previous version) """ library_logger.info("GeoElevation: Attempting to launch GUI application via API call...") try: # Local import of ElevationApp to avoid top-level circular dependencies # and to ensure it's only imported when GUI is explicitly requested. from .elevation_gui import ElevationApp except ImportError as e_gui_module_import: library_logger.critical( f"GeoElevation: Failed to import ElevationApp GUI components: {e_gui_module_import}. GUI cannot start.", exc_info=True ) print("ERROR: GeoElevation GUI could not be launched due to missing application components.", file=sys.stderr) return # Exit this function if GUI components cannot be imported except tk.TclError as e_tkinter_precheck: # Pre-check for basic Tkinter availability library_logger.critical( f"GeoElevation: Tkinter TclError occurred, possibly indicating an issue with the Tk environment: {e_tkinter_precheck}. GUI may not be available.", exc_info=True ) print(f"ERROR: A Tkinter problem prevents the GUI from launching: {e_tkinter_precheck}", file=sys.stderr) return # `multiprocessing.freeze_support()` is essential for packaged applications # (e.g., created with PyInstaller) that use multiprocessing, especially on Windows. # It should be called once at the beginning of the program if it's frozen. # Calling it here ensures it's run before any GUI-related processes might start. import multiprocessing # Import here, only when GUI is to be run multiprocessing.freeze_support() root_main_tk_window = tk.Tk() try: gui_application_instance = ElevationApp(parent_widget=root_main_tk_window) root_main_tk_window.mainloop() # Start the Tkinter event loop for the GUI except Exception as e_gui_runtime_error: library_logger.critical( f"GeoElevation: Unhandled exception during GUI application execution: {e_gui_runtime_error}", exc_info=True ) # Attempt to display a simple error dialog as a last resort try: fallback_error_root = tk.Tk() fallback_error_root.withdraw() # Hide the empty root window from tkinter import messagebox # Local import for this specific use messagebox.showerror( "Fatal GUI Error", f"GeoElevation GUI encountered a fatal error and cannot continue:\n{e_gui_runtime_error}\n\n" "Please check the application console or logs for more detailed information.", parent=None # No parent if the main window might be compromised or already destroyed ) fallback_error_root.destroy() except Exception: # If even the fallback error dialog fails, print to stderr print( f"FATAL GUI ERROR: {e_gui_runtime_error}\n" "(Additionally, failed to display a Tkinter error dialog for this error)", file=sys.stderr ) # Consider sys.exit(1) here if this is the main application entry point and GUI fails. # If called as a library function, re-raising might be better. # --- Define __all__ for the package's public API --- # This list controls what `from geoelevation import *` imports. # It's good practice to be explicit about the public interface. __all__ = [ "get_point_elevation", # Primary API function for elevation "run_gui_application", # Primary API function to launch the GUI "ElevationManager", # Expose for advanced users or type hinting "RASTERIO_AVAILABLE", # Allow checking for this critical dependency "GEOELEVATION_DEM_CACHE_DEFAULT", # Expose the default DEM cache path constant "_LibraryProgressWindow", # Expose the progress window class for library users # Note: deg_to_dms_string is imported but not added to __all__ as it's a utility, # typically accessed via the map_viewer subpackage or directly if needed, # but not considered a primary public API function of the top-level package. # Similarly, bounds functions are not added to __all__. ] library_logger.debug("GeoElevation package (__init__.py) has been loaded and configured.")