775 lines
43 KiB
Python
775 lines
43 KiB
Python
# 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: <service_cache_dir>/<zoom>/<x_tile>/<y_tile>.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}"
|
|
) |