# flightmonitor/map/map_services.py import abc import logging from urllib.parse import urlparse from typing import Optional, Tuple, Dict, Any try: from flightmonitor.utils.logger import get_logger logger = get_logger(__name__) except ImportError: logger = logging.getLogger(__name__) logger.warning( "MapServices: Could not import application logger. Using standard logging." ) class BaseMapService(abc.ABC): """ Abstract Base Class for map tile service providers. """ DEFAULT_TILE_PIXEL_SIZE: int = 256 DEFAULT_MAX_ZOOM_LEVEL: int = 19 def __init__( self, service_api_key: Optional[str] = None, tile_pixel_dim: int = DEFAULT_TILE_PIXEL_SIZE, max_supported_zoom: int = DEFAULT_MAX_ZOOM_LEVEL, ) -> None: self._service_log_prefix = f"[{self.__class__.__name__}]" logger.debug(f"{self._service_log_prefix} Initializing base map service.") self.api_key: Optional[str] = service_api_key self.tile_size: int = tile_pixel_dim self.max_zoom: int = max_supported_zoom if not (isinstance(self.tile_size, int) and self.tile_size > 0): logger.warning( f"{self._service_log_prefix} Invalid tile_size '{self.tile_size}'. " f"Using default: {self.DEFAULT_TILE_PIXEL_SIZE}px." ) self.tile_size = self.DEFAULT_TILE_PIXEL_SIZE if not (isinstance(self.max_zoom, int) and 0 <= self.max_zoom <= 25): logger.warning( f"{self._service_log_prefix} Invalid max_zoom '{self.max_zoom}'. " f"Using default: {self.DEFAULT_MAX_ZOOM_LEVEL}." ) self.max_zoom = self.DEFAULT_MAX_ZOOM_LEVEL @property @abc.abstractmethod def name(self) -> str: pass @property @abc.abstractmethod def attribution(self) -> str: pass @abc.abstractmethod def get_tile_url(self, z: int, x: int, y: int) -> Optional[str]: pass def is_zoom_level_valid(self, zoom_level: int) -> bool: is_valid = 0 <= zoom_level <= self.max_zoom if not is_valid: logger.warning( f"{self._service_log_prefix} Requested zoom level {zoom_level} is outside " f"the valid range [0, {self.max_zoom}] for this service." ) return is_valid def _is_generated_url_structurally_valid(self, url_string: str) -> bool: if not url_string: logger.error(f"{self._service_log_prefix} Generated URL is empty.") return False try: parsed_url = urlparse(url_string) has_scheme_and_netloc = bool(parsed_url.scheme and parsed_url.netloc) if not has_scheme_and_netloc: logger.error( f"{self._service_log_prefix} Generated URL '{url_string}' appears malformed (missing scheme or netloc)." ) return has_scheme_and_netloc except Exception as e_url_parse: logger.error( f"{self._service_log_prefix} Error parsing generated URL '{url_string}': {e_url_parse}", exc_info=True, ) return False def __repr__(self) -> str: return f"<{self.__class__.__name__}(Name: '{self.name}', MaxZoom: {self.max_zoom}, TileSize: {self.tile_size})>" class OpenStreetMapService(BaseMapService): """ Concrete implementation for fetching map tiles from OpenStreetMap (OSM). """ SERVICE_IDENTIFIER_NAME: str = "osm" SERVICE_ATTRIBUTION_TEXT: str = ( "© OpenStreetMap contributors (openstreetmap.org/copyright)" ) TILE_URL_TEMPLATE: str = "https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png" OSM_MAX_ZOOM_LEVEL: int = 19 SUBDOMAINS: Tuple[str, ...] = ("a", "b", "c") _subdomain_index: int = 0 def __init__(self) -> None: super().__init__( service_api_key=None, max_supported_zoom=self.OSM_MAX_ZOOM_LEVEL, ) logger.info(f"{self._service_log_prefix} OpenStreetMap service instance ready.") @property def name(self) -> str: return self.SERVICE_IDENTIFIER_NAME @property def attribution(self) -> str: return self.SERVICE_ATTRIBUTION_TEXT def get_tile_url(self, z: int, x: int, y: int) -> Optional[str]: if not self.is_zoom_level_valid(z): return None subdomain = self.SUBDOMAINS[ OpenStreetMapService._subdomain_index % len(self.SUBDOMAINS) ] OpenStreetMapService._subdomain_index += 1 try: tile_url = self.TILE_URL_TEMPLATE.format(s=subdomain, z=z, x=x, y=y) if not self._is_generated_url_structurally_valid(tile_url): return None logger.debug( f"{self._service_log_prefix} Generated URL for ({z},{x},{y}): {tile_url}" ) return tile_url except Exception as e_url_format: logger.error( f"{self._service_log_prefix} Error formatting tile URL for ({z},{x},{y}): {e_url_format}", exc_info=True, ) return None # --- Factory Function to Get Map Service Instances --- def get_map_service_instance( service_name_key: str, api_key_value: Optional[str] = None ) -> Optional[BaseMapService]: """ Factory function to create and return an instance of a specific map service. """ log_prefix_factory = "[MapServiceFactory]" normalized_service_name = service_name_key.lower().strip() logger.debug( f"{log_prefix_factory} Requesting map service instance for '{normalized_service_name}'." ) if normalized_service_name == OpenStreetMapService.SERVICE_IDENTIFIER_NAME: return OpenStreetMapService() else: logger.error( f"{log_prefix_factory} Unknown map service name specified: '{service_name_key}'." ) return None