SXXXXXXX_FlightMonitor/flightmonitor/map/map_services.py

# flightmonitor/map/map_services.py

import abc
import logging
from urllib.parse import urlparse
from typing import Optional, Tuple, Dict, Any

try:
    from ..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