# geoelevation/map_viewer/map_manager.py """ Manages the retrieval, caching, and stitching of map tiles from a selected map service. This module interacts with a map service provider (implementing BaseMapService), handles local disk caching of tiles to support offline use and reduce network requests, and assembles individual tiles into a complete map image. """ # Standard library imports import logging import os import time import threading from pathlib import Path # For robust path manipulation from typing import Tuple, Optional, List, Dict # Ensure these are available import io # To handle byte stream from requests as an image import shutil # For cache clearing operations # Third-party imports try: import requests # For downloading map tiles REQUESTS_AVAILABLE = True except ImportError: requests = None # type: ignore REQUESTS_AVAILABLE = False logging.error("MapTileManager: 'requests' library not found. Online tile fetching will fail.") try: # MODIFIED: Import ImageDraw along with Image from PIL. # WHY: ImageDraw is required for drawing on placeholder images and potentially other image manipulation tasks. # HOW: Added ImageDraw to the import list from PIL. from PIL import Image, ImageDraw ImageType = Image.Image # type: ignore PIL_AVAILABLE_MANAGER = True except ImportError: Image = None # type: ignore ImageDraw = None # type: ignore # Define as None if import fails ImageType = None # type: ignore logging.error("MapTileManager: 'Pillow' library not found. Image operations will fail.") # Local application/package imports # Assumes map_services is in the same subpackage 'map_viewer' from .map_services import BaseMapService # Module-level logger logger = logging.getLogger(__name__) # Default values for the manager if not provided or configured DEFAULT_MAP_TILE_CACHE_ROOT_DIR = "map_tile_cache" # Root for all service caches DEFAULT_ENABLE_ONLINE_FETCHING = True DEFAULT_NETWORK_TIMEOUT_SECONDS = 10 # Increased timeout slightly DEFAULT_DOWNLOAD_RETRY_DELAY_SECONDS = 2 DEFAULT_MAX_DOWNLOAD_RETRIES = 2 DEFAULT_PLACEHOLDER_COLOR_RGB = (220, 220, 220) # Light grey placeholder class MapTileManager: """ Manages fetching, caching, and assembling map tiles for a given map service. Requires 'requests' and 'Pillow' libraries to be installed. """ def __init__( self, map_service: BaseMapService, cache_root_directory: Optional[str] = None, enable_online_tile_fetching: Optional[bool] = None, # MODIFIED: Added tile_pixel_size parameter to the constructor. # WHY: To allow the caller (GeoElevationMapViewer) to explicitly specify the tile size # based on the selected map service configuration. # HOW: Added the parameter with an Optional[int] type hint and default None. tile_pixel_size: Optional[int] = None ) -> None: """ Initializes the MapTileManager. Args: map_service_instance: An instance of a map service provider (e.g., OpenStreetMapService). cache_root_directory: The root directory for caching tiles from all services. If None, uses DEFAULT_MAP_TILE_CACHE_ROOT_DIR. A subdirectory for the specific service will be created. enable_online_tile_fetching: Whether to download tiles if not found in cache. If None, uses DEFAULT_ENABLE_ONLINE_FETCHING. tile_pixel_size: The pixel dimension (width/height) of map tiles for this manager. If None, the size is taken from the map_service instance. Raises: TypeError: If map_service_instance is not a valid BaseMapService instance. ImportError: If 'requests' or 'Pillow' libraries are not installed. ValueError: If a tile_pixel_size is provided but invalid. """ logger.info("Initializing MapTileManager...") if not REQUESTS_AVAILABLE: raise ImportError("'requests' library is required by MapTileManager but not found.") # MODIFIED: Check for ImageDraw availability as well if Pillow is expected. # WHY: Drawing on placeholders requires ImageDraw. # HOW: Added ImageDraw check. if not (PIL_AVAILABLE_MANAGER and ImageDraw is not None): raise ImportError("'Pillow' library (or its drawing module ImageDraw) is required by MapTileManager but not found.") if not isinstance(map_service, BaseMapService): logger.critical("Invalid map_service_instance provided. Must be an instance of BaseMapService.") raise TypeError("map_service_instance must be an instance of BaseMapService.") self.map_service: BaseMapService = map_service self.service_identifier_name: str = self.map_service.name # MODIFIED: Set the tile_size attribute using the provided parameter or the service's size. # WHY: The manager needs to know the pixel dimensions of the tiles it handles for stitching and placeholder creation. # HOW: Check if tile_pixel_size was provided; if so, validate and use it. Otherwise, use the size from the map_service instance. if tile_pixel_size is not None: if not isinstance(tile_pixel_size, int) or tile_pixel_size <= 0: logger.error(f"Invalid provided tile_pixel_size: {tile_pixel_size}. Must be a positive integer.") # Fallback to service size or raise error? Let's raise for clarity if a bad value is explicitly passed. raise ValueError(f"Invalid provided tile_pixel_size: {tile_pixel_size}. Must be a positive integer.") self.tile_size: int = tile_pixel_size logger.info(f"Map tile size explicitly set to {self.tile_size}px.") else: # Use the size from the provided map service instance self.tile_size: int = self.map_service.tile_size logger.info(f"Map tile size inherited from service '{self.map_service.name}': {self.tile_size}px.") # Determine cache directory path effective_cache_root_dir = cache_root_directory if cache_root_directory is not None \ else DEFAULT_MAP_TILE_CACHE_ROOT_DIR self.service_specific_cache_dir: Path = Path(effective_cache_root_dir) / self.service_identifier_name logger.info(f"Service-specific cache directory set to: {self.service_specific_cache_dir}") # Determine online fetching status self.is_online_fetching_enabled: bool = enable_online_tile_fetching \ if enable_online_tile_fetching is not None else DEFAULT_ENABLE_ONLINE_FETCHING logger.info(f"Online tile fetching enabled: {self.is_online_fetching_enabled}") # Network request parameters self.http_user_agent: str = "GeoElevationMapViewer/0.1 (Python Requests)" self.http_request_headers: Dict[str, str] = {"User-Agent": self.http_user_agent} self.http_request_timeout_seconds: int = DEFAULT_NETWORK_TIMEOUT_SECONDS self.download_max_retries: int = DEFAULT_MAX_DOWNLOAD_RETRIES self.download_retry_delay_seconds: int = DEFAULT_DOWNLOAD_RETRY_DELAY_SECONDS self._ensure_service_cache_directory_exists() self._cache_access_lock = threading.Lock() # For thread-safe cache operations logger.info( f"MapTileManager initialized for service '{self.service_identifier_name}'. Online: {self.is_online_fetching_enabled}" ) def _ensure_service_cache_directory_exists(self) -> None: """Creates the service-specific cache directory if it doesn't exist.""" try: self.service_specific_cache_dir.mkdir(parents=True, exist_ok=True) logger.debug(f"Cache directory verified/created: {self.service_specific_cache_dir}") except OSError as e_mkdir: logger.error( f"Failed to create cache directory '{self.service_specific_cache_dir}': {e_mkdir}" ) except Exception as e_unexpected_mkdir: logger.exception( f"Unexpected error ensuring cache directory exists: {e_unexpected_mkdir}" ) def _get_tile_cache_file_path(self, z: int, x: int, y: int) -> Path: """ Constructs the full local file path for a specific map tile. The structure is: ///.png """ return self.service_specific_cache_dir / str(z) / str(x) / f"{y}.png" def _load_tile_from_cache( self, tile_cache_path: Path, tile_coordinates_log_str: str ) -> Optional[ImageType]: """Attempts to load a tile image from a cache file (thread-safe read).""" logger.debug(f"Checking cache for tile {tile_coordinates_log_str} at {tile_cache_path}") try: with self._cache_access_lock: # Protect file system access if tile_cache_path.is_file(): logger.info(f"Cache hit for tile {tile_coordinates_log_str}. Loading from disk.") pil_image = Image.open(tile_cache_path) # type: ignore pil_image.load() # Load image data into memory to release file lock sooner # Ensure consistency by converting to RGB if needed # MODIFIED: Ensure consistency by converting to RGB or RGBA depending on service need (currently RGB). # WHY: Consistent format is important for image processing and display. # HOW: Convert to RGB. if pil_image.mode != "RGB": logger.debug( f"Converting cached image {tile_coordinates_log_str} from mode {pil_image.mode} to RGB." ) pil_image = pil_image.convert("RGB") return pil_image else: logger.debug(f"Cache miss for tile {tile_coordinates_log_str}.") return None except IOError as e_io_cache: logger.error( f"IOError reading cached tile {tile_cache_path}: {e_io_cache}. Treating as cache miss." ) return None except Exception as e_cache_unexpected: logger.exception( f"Unexpected error accessing cache file {tile_cache_path}: {e_cache_unexpected}" ) return None def _download_and_save_tile_to_cache( self, zoom_level: int, tile_x: int, tile_y: int, tile_cache_path: Path, tile_coordinates_log_str: str ) -> Optional[ImageType]: """Attempts to download a tile, process it, and save it to the cache.""" if not self.is_online_fetching_enabled: logger.debug(f"Online fetching disabled. Cannot download tile {tile_coordinates_log_str}.") return None tile_download_url = self.map_service.get_tile_url(zoom_level, tile_x, tile_y) if not tile_download_url: logger.error(f"Failed to get URL for tile {tile_coordinates_log_str} from service.") return None logger.info(f"Downloading tile {tile_coordinates_log_str} from: {tile_download_url}") downloaded_pil_image: Optional[ImageType] = None for attempt_num in range(self.download_max_retries + 1): try: response = requests.get( # type: ignore tile_download_url, headers=self.http_request_headers, timeout=self.http_request_timeout_seconds, stream=True # Efficient for binary content ) response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx) image_binary_data = response.content if not image_binary_data: logger.warning(f"Downloaded empty content for tile {tile_coordinates_log_str}.") break # Stop retrying if server sends empty response # Process downloaded image data and save to cache try: pil_image = Image.open(io.BytesIO(image_binary_data)) # type: ignore pil_image.load() # MODIFIED: Convert downloaded image to RGB mode before saving/returning. # WHY: Consistency in image format within the manager. # HOW: Added .convert("RGB"). if pil_image.mode != "RGB": pil_image = pil_image.convert("RGB") # Optional: Resize downloaded tile if its size doesn't match self.tile_size # This would be needed if the service URL returns tiles of different sizes, # which is uncommon for standard XYZ services, but could happen. # For standard services, the service.tile_size should be correct. # if pil_image.size != (self.tile_size, self.tile_size): # logger.warning(f"Downloaded tile size {pil_image.size} doesn't match expected {self.tile_size}. Resizing.") # pil_image = pil_image.resize((self.tile_size, self.tile_size), Image.Resampling.LANCZOS) logger.debug(f"Tile {tile_coordinates_log_str} downloaded (Attempt {attempt_num + 1}).") self._save_image_to_cache_file(tile_cache_path, pil_image) downloaded_pil_image = pil_image break # Success, exit retry loop except (IOError, Image.UnidentifiedImageError) as e_img_proc: # type: ignore logger.error( f"Failed to process image data for {tile_coordinates_log_str}: {e_img_proc}" ) break # Don't retry if image data is corrupt except Exception as e_proc_unexpected: logger.exception( f"Unexpected error processing downloaded image {tile_coordinates_log_str}: {e_proc_unexpected}" ) break except requests.exceptions.Timeout: # type: ignore logger.warning( f"Timeout downloading tile {tile_coordinates_log_str} (Attempt {attempt_num + 1})." ) except requests.exceptions.RequestException as e_req: # type: ignore status = getattr(e_req.response, "status_code", "N/A") logger.warning( f"Request error for tile {tile_coordinates_log_str} (Status: {status}, Attempt {attempt_num + 1}): {e_req}" ) if status == 404: break # No point retrying a 404 Not Found except Exception as e_dl_unexpected: logger.exception( f"Unexpected error downloading tile {tile_coordinates_log_str} (Attempt {attempt_num + 1}): {e_dl_unexpected}" ) break # Stop on other unexpected download errors if attempt_num < self.download_max_retries: logger.debug(f"Waiting {self.download_retry_delay_seconds}s before retrying for {tile_coordinates_log_str}.") time.sleep(self.download_retry_delay_seconds) if downloaded_pil_image is None: logger.error(f"Failed to download tile {tile_coordinates_log_str} after all retries.") return downloaded_pil_image def _save_image_to_cache_file(self, tile_cache_path: Path, pil_image: ImageType) -> None: """Saves a PIL Image object to a file in the cache (thread-safe write).""" with self._cache_access_lock: # Protect file system access try: # Ensure parent directory for the tile file exists tile_cache_path.parent.mkdir(parents=True, exist_ok=True) # Use 'png' format explicitly as it's lossless and common for map tiles pil_image.save(tile_cache_path, format='PNG') # MODIFIED: Explicitly save as PNG. WHY: Standard format. logger.debug(f"Saved tile to cache: {tile_cache_path}") except IOError as e_io_save: logger.error(f"IOError saving tile to cache {tile_cache_path}: {e_io_save}") except Exception as e_save_unexpected: logger.exception( f"Unexpected error saving tile to cache {tile_cache_path}: {e_save_unexpected}" ) def get_tile_image( self, zoom_level: int, tile_x: int, tile_y: int, force_online_refresh: bool = False ) -> Optional[ImageType]: """ Retrieves a map tile image, using local cache first. If not cached or refresh is forced, attempts to download (if enabled). Returns a placeholder image if the tile cannot be retrieved. Args: zoom_level: The zoom level of the tile. tile_x: The X coordinate of the tile. tile_y: The Y coordinate of the tile. force_online_refresh: If True, bypasses cache and attempts download. Returns: A PIL Image object of the tile, or a placeholder image on failure. Returns None only if placeholder creation itself fails critically. """ tile_coords_log_str = f"({zoom_level},{tile_x},{tile_y})" logger.debug(f"Requesting tile image for {tile_coords_log_str}") # MODIFIED: Check if the zoom level is valid for the map service. # WHY: Avoid requesting tiles for invalid zoom levels from the service or cache. # HOW: Added a check using self.map_service.is_zoom_level_valid. if not self.map_service.is_zoom_level_valid(zoom_level): logger.error(f"Invalid zoom level {zoom_level} for map service '{self.service_identifier_name}'. Cannot get tile.") # Return a placeholder for invalid zoom levels return self._create_placeholder_tile_image(f"Invalid Zoom {zoom_level}") tile_cache_file = self._get_tile_cache_file_path(zoom_level, tile_x, tile_y) retrieved_image: Optional[ImageType] = None if not force_online_refresh: retrieved_image = self._load_tile_from_cache(tile_cache_file, tile_coords_log_str) if retrieved_image is None: # Cache miss or force_refresh retrieved_image = self._download_and_save_tile_to_cache( zoom_level, tile_x, tile_y, tile_cache_file, tile_coords_log_str ) if retrieved_image is None: # All attempts failed logger.warning(f"Failed to retrieve tile {tile_coords_log_str}. Using placeholder.") # MODIFIED: Pass tile coordinates to placeholder for debugging/visual info. # WHY: Helps identify which specific tile failed when looking at the stitched map. # HOW: Pass a string identifier to the placeholder creation function. retrieved_image = self._create_placeholder_tile_image(tile_coords_log_str) if retrieved_image is None: # Should be rare if Pillow is working logger.critical("Failed to create even a placeholder tile. Returning None.") return retrieved_image def stitch_map_image( self, zoom_level: int, x_tile_range: Tuple[int, int], # (min_x, max_x) y_tile_range: Tuple[int, int] # (min_y, max_y) ) -> Optional[ImageType]: """ Retrieves and stitches multiple map tiles to form a larger composite map image. Args: zoom_level: The zoom level for all tiles. x_tile_range: Inclusive start and end X tile coordinates (min_x, max_x). y_tile_range: Inclusive start and end Y tile coordinates (min_y, max_y). Returns: A PIL Image object of the stitched map, or None if a critical error occurs. Missing individual tiles are replaced by placeholders. """ logger.info( f"Request to stitch map: Zoom {zoom_level}, X-Range {x_tile_range}, Y-Range {y_tile_range}" ) min_tile_x, max_tile_x = x_tile_range min_tile_y, max_tile_y = y_tile_range # Basic validation of ranges if not (min_tile_x <= max_tile_x and min_tile_y <= max_tile_y): logger.error(f"Invalid tile ranges for stitching: X={x_tile_range}, Y={y_tile_range}") return None # MODIFIED: Use the tile_size attribute of the manager. # WHY: Consistency. The manager's size should be used, not necessarily the service's size again here. # HOW: Changed self.map_service.tile_size to self.tile_size. single_tile_pixel_size = self.tile_size if single_tile_pixel_size <= 0: logger.error(f"Invalid tile size ({single_tile_pixel_size}) stored in manager. Cannot stitch.") # MODIFIED: Return placeholder instead of None on invalid tile size. # WHY: Provide a visual indication that stitching failed due to config, rather than a blank window. # HOW: Create and return a large placeholder image. try: # Ensure Pillow/ImageDraw are available for placeholder creation if PIL_AVAILABLE_MANAGER and ImageDraw is not None: # type: ignore # Create a large placeholder image (e.g., 3x3 tiles size) placeholder_img = Image.new("RGB", (256 * 3, 256 * 3), DEFAULT_PLACEHOLDER_COLOR_RGB) # Use a fixed reasonable size for error image draw = ImageDraw.Draw(placeholder_img) # type: ignore # Add error text error_text = f"Stitch Failed\nInvalid Tile Size:\n{single_tile_pixel_size}" # This simple text drawing assumes basic PIL text capabilities try: # Try drawing with a font loaded in image_processor from geoelevation.image_processor import DEFAULT_FONT # Assume font loading logic in image_processor font_to_use = DEFAULT_FONT # type: ignore # Use font loaded in image_processor if font_to_use: # Calculate text size and position using the font # Note: textbbox requires Pillow >= 8.0 try: text_bbox = draw.textbbox((0,0), error_text, font=font_to_use, spacing=2) # type: ignore text_width = text_bbox[2] - text_bbox[0] text_height = text_bbox[3] - text_bbox[1] text_pos = ((placeholder_img.width - text_width) // 2, (placeholder_img.height - text_height) // 2) draw.text(text_pos, error_text, fill="black", font=font_to_use, spacing=2, anchor="lt") # type: ignore # Draw text using font except AttributeError: # Fallback for textbbox if Pillow < 8.0 text_width, text_height = draw.textsize(error_text, font=font_to_use) # type: ignore text_pos = ((placeholder_img.width - text_width) // 2, (placeholder_img.height - text_height) // 2) draw.text(text_pos, error_text, fill="black", font=font_to_use) # type: ignore # Draw text using font fallback except Exception as e_font_draw: logger.warning(f"Error drawing text with font on placeholder: {e_font_draw}. Falling back to simple draw.") draw.text((10, 10), error_text, fill="black") # type: ignore # Simple draw if font drawing fails else: draw.text((10, 10), error_text, fill="black") # type: ignore # Simple draw if no font was loaded except Exception as e_draw: logger.warning(f"Error drawing text on placeholder: {e_draw}. Falling back to simple draw.") draw.text((10, 10), error_text, fill="black") # type: ignore # Final fallback return placeholder_img else: logger.error("Pillow or ImageDraw not available to create placeholder image.") return None # Cannot create placeholder without PIL except Exception as e_placeholder_fail: logger.exception(f"Failed to create large placeholder for stitching error: {e_placeholder_fail}") return None # Return None if placeholder creation fails num_tiles_wide = (max_tile_x - min_tile_x) + 1 num_tiles_high = (max_tile_y - min_tile_y) + 1 total_image_width = num_tiles_wide * single_tile_pixel_size total_image_height = num_tiles_high * single_tile_pixel_size logger.debug( f"Stitched image dimensions: {total_image_width}x{total_image_height} " f"({num_tiles_wide}x{num_tiles_high} tiles of {single_tile_pixel_size}px)" ) # Handle potential excessively large image size request MAX_IMAGE_DIMENSION = 16384 # Arbitrary limit to prevent crashes with massive requests if total_image_width > MAX_IMAGE_DIMENSION or total_image_height > MAX_IMAGE_DIMENSION: logger.error( f"Requested stitched image size ({total_image_width}x{total_image_height}) " f"exceeds maximum allowed dimension ({MAX_IMAGE_DIMENSION}). Aborting stitch." ) # Return placeholder for excessive size request try: if PIL_AVAILABLE_MANAGER and ImageDraw is not None: # type: ignore placeholder_img = Image.new("RGB", (256 * 3, 256 * 3), (255, 100, 100)) # Reddish placeholder draw = ImageDraw.Draw(placeholder_img) # type: ignore error_text = f"Stitch Failed\nImage too large:\n{total_image_width}x{total_image_height}px" try: from geoelevation.image_processor import DEFAULT_FONT font_to_use = DEFAULT_FONT # type: ignore if font_to_use: try: text_bbox = draw.textbbox((0,0), error_text, font=font_to_use, spacing=2) # type: ignore text_w = text_bbox[2] - text_bbox[0] text_h = text_bbox[3] - text_bbox[1] text_pos = ((placeholder_img.width - text_w) // 2, (placeholder_img.height - text_h) // 2) draw.text(text_pos, error_text, fill="black", font=font_to_use, spacing=2, anchor="lt") # type: ignore except AttributeError: text_w, text_h = draw.textsize(error_text, font=font_to_use) # type: ignore text_pos = ((placeholder_img.width - text_w) // 2, (placeholder_img.height - text_h) // 2) draw.text(text_pos, error_text, fill="black", font=font_to_use) # type: ignore except Exception as e_font_draw: logger.warning(f"Error drawing text with font on placeholder: {e_font_draw}. Falling back to simple draw.") draw.text((10, 10), error_text, fill="black") # type: ignore else: draw.text((10, 10), error_text, fill="black") # type: ignore except Exception as e_draw: logger.warning(f"Error drawing text on placeholder: {e_draw}. Falling back to simple draw.") draw.text((10, 10), error_text, fill="black") # type: ignore # Final fallback return placeholder_img else: logger.error("Pillow or ImageDraw not available to create placeholder image.") return None # Cannot create placeholder without PIL except Exception as e_placeholder_fail: logger.exception(f"Failed to create large placeholder for size error: {e_placeholder_fail}") return None # Return None if placeholder fails try: # Create a new blank RGB image to paste tiles onto # MODIFIED: Ensure PIL_AVAILABLE_MANAGER is true before creating Image.new. # WHY: Avoids NameError if PIL import failed. # HOW: Added check. if PIL_AVAILABLE_MANAGER: stitched_map_image = Image.new("RGB", (total_image_width, total_image_height)) # type: ignore else: raise ImportError("Pillow not available to create new image.") except Exception as e_create_blank: logger.exception(f"Failed to create blank image for stitching: {e_create_blank}. Dimensions: {total_image_width}x{total_image_height}") # Return placeholder if blank image creation fails (e.g., out of memory) try: if PIL_AVAILABLE_MANAGER and ImageDraw is not None: # type: ignore placeholder_img = Image.new("RGB", (256 * 3, 256 * 3), (255, 100, 100)) # Reddish placeholder draw = ImageDraw.Draw(placeholder_img) # type: ignore error_text = f"Stitch Failed\nCannot create image:\n{e_create_blank}" try: from geoelevation.image_processor import DEFAULT_FONT font_to_use = DEFAULT_FONT # type: ignore if font_to_use: try: text_bbox = draw.textbbox((0,0), error_text, font=font_to_use, spacing=2) # type: ignore text_w = text_bbox[2] - text_bbox[0] text_h = text_bbox[3] - text_bbox[1] text_pos = ((placeholder_img.width - text_w) // 2, (placeholder_img.height - text_h) // 2) draw.text(text_pos, error_text, fill="black", font=font_to_use, spacing=2, anchor="lt") # type: ignore except AttributeError: text_w, text_h = draw.textsize(error_text, font=font_to_use) # type: ignore text_pos = ((placeholder_img.width - text_w) // 2, (placeholder_img.height - text_h) // 2) draw.text(text_pos, error_text, fill="black", font=font_to_use) # type: ignore except Exception as e_font_draw: logger.warning(f"Error drawing text with font on placeholder: {e_font_draw}. Falling back to simple draw.") draw.text((10, 10), error_text, fill="black") # type: ignore else: draw.text((10, 10), error_text, fill="black") # type: ignore except Exception as e_draw: logger.warning(f"Error drawing text on placeholder: {e_draw}. Falling back to simple draw.") draw.text((10, 10), error_text, fill="black") # type: ignore # Final fallback return placeholder_img else: logger.error("Pillow or ImageDraw not available to create placeholder image.") return None # Cannot create placeholder without PIL except Exception as e_placeholder_fail: logger.exception(f"Failed to create large placeholder for memory error: {e_placeholder_fail}") return None # Return None if placeholder fails # Iterate through the required tile coordinates, fetch, and paste for row_index, current_tile_y in enumerate(range(min_tile_y, max_tile_y + 1)): for col_index, current_tile_x in enumerate(range(min_tile_x, max_tile_x + 1)): tile_image_pil = self.get_tile_image(zoom_level, current_tile_x, current_tile_y) if tile_image_pil is None: # This implies even placeholder creation failed, which is critical. logger.critical( f"Critical error: get_tile_image returned None for ({zoom_level},{current_tile_x},{current_tile_y}). Aborting stitch." ) return None # Abort stitching on critical tile failure # Calculate top-left pixel position to paste this tile paste_position_x = col_index * single_tile_pixel_size paste_position_y = row_index * single_tile_pixel_size logger.debug( f"Pasting tile ({zoom_level},{current_tile_x},{current_tile_y}) at " f"pixel position ({paste_position_x},{paste_position_y})" ) try: # Ensure the tile image is the correct size before pasting # MODIFIED: Check if tile_image_pil is valid before checking its size. # WHY: Avoids AttributeError if tile_image_pil is None (shouldn't happen if get_tile_image handles None, but defensive). # HOW: Added `if tile_image_pil and tile_image_pil.size...`. if tile_image_pil and tile_image_pil.size != (self.tile_size, self.tile_size): # This might happen if the downloaded tile or placeholder was the wrong size. # Resize it to match the expected tile size for stitching consistency. logger.warning(f"Tile image size {tile_image_pil.size} doesn't match expected {self.tile_size}. Resizing for stitch.") # MODIFIED: Check PIL_AVAILABLE_MANAGER before resizing. # WHY: Resize requires PIL. # HOW: Added check. if PIL_AVAILABLE_MANAGER: tile_image_pil = tile_image_pil.resize((self.tile_size, self.tile_size), Image.Resampling.LANCZOS) # type: ignore else: logger.error("Pillow not available, cannot resize tile for stitch.") # Decide fallback: skip pasting this tile or use placeholder? # Leaving it blank might be okay, or replace with a placeholder of correct size. # Let's just leave it blank (skip paste) if resize fails due to missing lib. continue # Skip pasting this tile # MODIFIED: Check if tile_image_pil is still valid before pasting. # WHY: It might have become None if resize failed due to missing PIL. # HOW: Added `if tile_image_pil:`. if tile_image_pil: stitched_map_image.paste(tile_image_pil, (paste_position_x, paste_position_y)) # type: ignore except Exception as e_paste: logger.exception( f"Error pasting tile ({zoom_level},{current_tile_x},{current_tile_y}) " f"at ({paste_position_x},{paste_position_y}): {e_paste}" ) # Continue, leaving that part of the map blank or with placeholder color logger.info(f"Map stitching complete for zoom {zoom_level}, X={x_tile_range}, Y={y_tile_range}.") return stitched_map_image def _create_placeholder_tile_image(self, identifier: str = "N/A") -> Optional[ImageType]: """ Creates and returns a placeholder tile image (e.g., a grey square). Includes optional text identifier on the placeholder. """ # MODIFIED: Added check for ImageDraw availability. # WHY: Drawing on placeholders requires ImageDraw. # HOW: Added check. if not (PIL_AVAILABLE_MANAGER and ImageDraw is not None): logger.warning("Cannot create placeholder tile: Pillow or ImageDraw library not available.") return None try: tile_pixel_size = self.tile_size # Use the manager's stored tile size # Ensure placeholder_color is a valid RGB tuple placeholder_color = DEFAULT_PLACEHOLDER_COLOR_RGB # No need to re-validate color if it's a fixed constant, but defensive check # if not (isinstance(placeholder_color, tuple) and len(placeholder_color) == 3 and # all(isinstance(c, int) and 0 <= c <= 255 for c in placeholder_color)): # logger.warning(f"Invalid placeholder color '{placeholder_color}'. Using default grey.") # placeholder_color = (220, 220, 220) placeholder_img = Image.new("RGB", (tile_pixel_size, tile_pixel_size), color=placeholder_color) # type: ignore draw = ImageDraw.Draw(placeholder_img) # type: ignore # Add text overlay indicating failure and identifier overlay_text = f"Tile Fail\n{identifier}" try: # Attempt to use a font loaded in image_processor for consistency from geoelevation.image_processor import DEFAULT_FONT # Assume font loading logic exists font_to_use = DEFAULT_FONT # type: ignore # Use the shared loaded font # Calculate text position for centering or top-left # Using textbbox for accurate size calculation (requires Pillow >= 8.0) try: # textbbox returns (left, top, right, bottom) relative to the anchor (0,0) text_bbox = draw.textbbox((0,0), overlay_text, font=font_to_use, spacing=2) # type: ignore text_width = text_bbox[2] - text_bbox[0] text_height = text_bbox[3] - text_bbox[1] # Center the text (approx) text_x = (tile_pixel_size - text_width) // 2 text_y = (tile_pixel_size - text_height) // 2 # Draw text with the loaded font, anchored at the top-left of the text bbox draw.text((text_x, text_y), overlay_text, fill="black", font=font_to_use, spacing=2, anchor="lt") # type: ignore except AttributeError: # Fallback for textbbox if Pillow < 8.0 logger.warning("Pillow textbbox not available (Pillow < 8.0). Using textsize fallback.") # textsize might not handle multiline spacing well text_width, text_height = draw.textsize(overlay_text, font=font_to_use) # type: ignore # Add approximated height for multiline if needed if "\n" in overlay_text: line_count = overlay_text.count("\n") + 1 text_height += line_count * 2 # Rough approximation # Center text based on textsize (less accurate for multiline) text_x = (tile_pixel_size - text_width) // 2 text_y = (tile_pixel_size - text_height) // 2 draw.text((text_x, text_y), overlay_text, fill="black", font=font_to_use) # type: ignore # Draw text using font fallback except Exception as e_font_draw: logger.warning(f"Error drawing text with font on placeholder: {e_font_draw}. Falling back to simple draw.") # Fallback to simple draw if font drawing fails draw.text((10, 10), overlay_text, fill="black") # type: ignore # Simple draw near top-left except Exception as e_draw: logger.warning(f"Error drawing text on placeholder: {e_draw}. Falling back to simple draw.") draw.text((10, 10), overlay_text, fill="black") # type: ignore # Final fallback return placeholder_img except Exception as e_placeholder: logger.exception(f"Error creating placeholder tile image: {e_placeholder}") return None def _get_bounds_for_tile_range( self, zoom: int, tile_ranges: Tuple[Tuple[int, int], Tuple[int, int]] # ((min_x, max_x), (min_y, max_y)) ) -> Optional[Tuple[float, float, float, float]]: # (west, south, east, north) """ Calculates the precise geographic bounds covered by a given range of tiles. This method might be better placed in map_utils if mercantile is available there, or kept here if MapTileManager is the primary user of mercantile for this. Requires 'mercantile' library. """ # Check if mercantile is available (it should be if MapTileManager initialized without error) try: import mercantile as local_mercantile # Local import for this method # MODIFIED: Check if mercantile is actually available after import attempt. # WHY: Defend against scenarios where the import succeeds but mercantile is None. # HOW: Add explicit check. if local_mercantile is None: raise ImportError("mercantile is None after import.") except ImportError: logger.error("mercantile library not found, cannot calculate bounds for tile range.") return None try: min_x, max_x = tile_ranges[0] min_y, max_y = tile_ranges[1] # Get bounds of the top-left tile and bottom-right tile # mercantile.bounds(x, y, z) returns (west, south, east, north) top_left_tile_bounds = local_mercantile.bounds(min_x, min_y, zoom) bottom_right_tile_bounds = local_mercantile.bounds(max_x, max_y, zoom) # The overall bounding box is: # West longitude from the top-left tile # South latitude from the bottom-right tile # East longitude from the bottom-right tile # North latitude from the top-left tile overall_west_lon = top_left_tile_bounds.west overall_south_lat = bottom_right_tile_bounds.south overall_east_lon = bottom_right_tile_bounds.east overall_north_lat = top_left_tile_bounds.north return (overall_west_lon, overall_south_lat, overall_east_lon, overall_north_lat) except Exception as e_bounds_calc: logger.exception( f"Error calculating geographic bounds for tile range {tile_ranges} at zoom {zoom}: {e_bounds_calc}" ) return None def clear_entire_service_cache(self) -> None: """Deletes all cached tiles for the current map service.""" logger.info(f"Attempting to clear entire cache for service '{self.service_identifier_name}' at {self.service_specific_cache_dir}") if not self.service_specific_cache_dir.exists(): logger.warning(f"Cache directory '{self.service_specific_cache_dir}' does not exist. Nothing to clear.") return with self._cache_access_lock: # Ensure exclusive access during deletion try: if self.service_specific_cache_dir.is_dir(): shutil.rmtree(self.service_specific_cache_dir) logger.info(f"Successfully cleared cache at {self.service_specific_cache_dir}.") # Recreate the base directory for this service after clearing self._ensure_service_cache_directory_exists() else: logger.warning(f"Cache path '{self.service_specific_cache_dir}' is not a directory.") except OSError as e_os_clear: logger.error(f"OS Error clearing cache at '{self.service_specific_cache_dir}': {e_os_clear}") except Exception as e_clear_unexpected: logger.exception( f"Unexpected error clearing cache '{self.service_specific_cache_dir}': {e_clear_unexpected}" )