Add new DTM providers and refactor DTM provider base classes

maps4fs/__init__.py

@@ -2,8 +2,11 @@
 from maps4fs.generator.dtm.dtm import DTMProvider
 from maps4fs.generator.dtm.srtm import SRTM30Provider, SRTM30ProviderSettings
 from maps4fs.generator.dtm.usgs import USGSProvider, USGSProviderSettings
-from maps4fs.generator.dtm.nrw import NRWProvider, NRWProviderSettings
-from maps4fs.generator.dtm.bavaria import BavariaProvider, BavariaProviderSettings
+from maps4fs.generator.dtm.nrw import NRWProvider
+from maps4fs.generator.dtm.bavaria import BavariaProvider
+from maps4fs.generator.dtm.niedersachsen import NiedersachsenProvider
+from maps4fs.generator.dtm.hessen import HessenProvider
+from maps4fs.generator.dtm.england import England1MProvider
 from maps4fs.generator.game import Game
 from maps4fs.generator.map import Map
 from maps4fs.generator.settings import (

maps4fs/generator/dtm/base/wcs.py

@@ -0,0 +1,71 @@
+"""This module contains the base WCS provider."""
+
+from abc import abstractmethod
+import os
+
+from owslib.wcs import WebCoverageService
+from tqdm import tqdm
+
+from maps4fs.generator.dtm import utils
+from maps4fs.generator.dtm.dtm import DTMProvider
+
+
+# pylint: disable=too-many-locals
+class WCSProvider(DTMProvider):
+    """Generic provider of WCS sources."""
+
+    _is_base = True
+    _wcs_version = "2.0.1"
+    _source_crs: str = "EPSG:4326"
+    _tile_size: float = 0.02
+
+    @abstractmethod
+    def get_wcs_parameters(self, tile: tuple[float, float, float, float]) -> dict:
+        """Get the parameters for the WCS request.
+
+        Arguments:
+            tile (tuple): The tile to download.
+
+        Returns:
+            dict: The parameters for the WCS request.
+        """
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.shared_tiff_path = os.path.join(self._tile_directory, "shared")
+        os.makedirs(self.shared_tiff_path, exist_ok=True)
+
+    def download_tiles(self) -> list[str]:
+        bbox = self.get_bbox()
+        bbox = utils.transform_bbox(bbox, self._source_crs)
+        tiles = utils.tile_bbox(bbox, self._tile_size)
+
+        all_tif_files = self.download_all_tiles(tiles)
+        return all_tif_files
+
+    def download_all_tiles(self, tiles: list[tuple[float, float, float, float]]) -> list[str]:
+        """Download tiles from the NI provider.
+
+        Arguments:
+            tiles (list): List of tiles to download.
+
+        Returns:
+            list: List of paths to the downloaded GeoTIFF files.
+        """
+        all_tif_files = []
+        wcs = WebCoverageService(
+            self._url,
+            version=self._wcs_version,
+            # auth=Authentication(verify=False),
+            timeout=600,
+        )
+        for tile in tqdm(tiles, desc="Downloading tiles", unit="tile"):
+            file_name = "_".join(map(str, tile)) + ".tif"
+            file_path = os.path.join(self.shared_tiff_path, file_name)
+            if not os.path.exists(file_path):
+                output = wcs.getCoverage(**self.get_wcs_parameters(tile))
+                with open(file_path, "wb") as f:
+                    f.write(output.read())
+
+            all_tif_files.append(file_path)
+        return all_tif_files

maps4fs/generator/dtm/base/wms.py

@@ -0,0 +1,70 @@
+"""This module contains the base WMS provider."""
+
+from abc import abstractmethod
+import os
+
+from owslib.wms import WebMapService
+
+from maps4fs.generator.dtm import utils
+from maps4fs.generator.dtm.dtm import DTMProvider
+
+
+# pylint: disable=too-many-locals
+class WMSProvider(DTMProvider):
+    """Generic provider of WMS sources."""
+
+    _is_base = True
+    _wms_version = "1.3.0"
+    _source_crs: str = "EPSG:4326"
+    _tile_size: float = 0.02
+
+    @abstractmethod
+    def get_wms_parameters(self, tile: tuple[float, float, float, float]) -> dict:
+        """Get the parameters for the WMS request.
+
+        Arguments:
+            tile (tuple): The tile to download.
+
+        Returns:
+            dict: The parameters for the WMS request.
+        """
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.shared_tiff_path = os.path.join(self._tile_directory, "shared")
+        os.makedirs(self.shared_tiff_path, exist_ok=True)
+
+    def download_tiles(self) -> list[str]:
+        bbox = self.get_bbox()
+        bbox = utils.transform_bbox(bbox, self._source_crs)
+        tiles = utils.tile_bbox(bbox, self._tile_size)
+
+        all_tif_files = self.download_all_tiles(tiles)
+        return all_tif_files
+
+    def download_all_tiles(self, tiles: list[tuple[float, float, float, float]]) -> list[str]:
+        """Download tiles from the WMS provider.
+
+        Arguments:
+            tiles (list): List of tiles to download.
+
+        Returns:
+            list: List of paths to the downloaded GeoTIFF files.
+        """
+        all_tif_files = []
+        wms = WebMapService(
+            self._url,
+            version=self._wms_version,
+            # auth=Authentication(verify=False),
+            timeout=600,
+        )
+        for tile in tiles:
+            file_name = "_".join(map(str, tile)) + ".tif"
+            file_path = os.path.join(self.shared_tiff_path, file_name)
+            if not os.path.exists(file_path):
+                output = wms.getmap(**self.get_wms_parameters(tile))
+                with open(file_path, "wb") as f:
+                    f.write(output.read())
+
+            all_tif_files.append(file_path)
+        return all_tif_files

maps4fs/generator/dtm/bavaria.py

@@ -4,14 +4,9 @@
 import os
 from xml.etree import ElementTree as ET
 
-import numpy as np
 import requests
 
-from maps4fs.generator.dtm.dtm import DTMProvider, DTMProviderSettings
-
-
-class BavariaProviderSettings(DTMProviderSettings):
-    """Settings for the Bavaria provider."""
+from maps4fs.generator.dtm.dtm import DTMProvider
 
 
 class BavariaProvider(DTMProvider):
@@ -25,11 +20,10 @@ class BavariaProvider(DTMProvider):
     _region = "DE"
     _icon = "🇩🇪󠁥󠁢󠁹󠁿"
     _resolution = 1
-    _data: np.ndarray | None = None
-    _settings = BavariaProviderSettings
     _author = "[H4rdB4se](https://github.com/H4rdB4se)"
     _is_community = True
     _instructions = None
+    _extents = (50.56, 47.25, 13.91, 8.95)
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)

maps4fs/generator/dtm/dtm.py

@@ -17,6 +17,7 @@
 from rasterio.enums import Resampling
 from rasterio.merge import merge
 from rasterio.warp import calculate_default_transform, reproject
+from tqdm import tqdm
 
 from maps4fs.logger import Logger
 
@@ -47,7 +48,10 @@ class DTMProvider(ABC):
     _contributors: str | None = None
     _is_community: bool = False
     _is_base: bool = False
-    _settings: Type[DTMProviderSettings] | None = None
+    _settings: Type[DTMProviderSettings] | None = DTMProviderSettings
+
+    """Bounding box of the provider in the format (north, south, east, west)."""
+    _extents: tuple[float, float, float, float] | None = None
 
     _instructions: str | None = None
 
@@ -234,7 +238,7 @@ def get_provider_by_code(cls, code: str) -> Type[DTMProvider] | None:
         return None
 
     @classmethod
-    def get_provider_descriptions(cls) -> dict[str, str]:
+    def get_valid_provider_descriptions(cls, lat_lon: tuple[float, float]) -> dict[str, str]:
         """Get descriptions of all providers, where keys are provider codes and
         values are provider descriptions.
 
@@ -243,9 +247,24 @@ def get_provider_descriptions(cls) -> dict[str, str]:
         """
         providers = {}
         for provider in cls.__subclasses__():
-            providers[provider._code] = provider.description()  # pylint: disable=W0212
+            # pylint: disable=W0212
+            if not provider._is_base and provider.inside_bounding_box(lat_lon):
+                providers[provider._code] = provider.description()  # pylint: disable=W0212
         return providers  # type: ignore
 
+    @classmethod
+    def inside_bounding_box(cls, lat_lon: tuple[float, float]) -> bool:
+        """Check if the coordinates are inside the bounding box of the provider.
+
+        Returns:
+            bool: True if the coordinates are inside the bounding box, False otherwise.
+        """
+        lat, lon = lat_lon
+        extents = cls._extents
+        return extents is None or (
+            extents[0] >= lat >= extents[1] and extents[2] >= lon >= extents[3]
+        )
+
     @abstractmethod
     def download_tiles(self) -> list[str]:
         """Download tiles from the provider.
@@ -278,6 +297,8 @@ def get_numpy(self) -> np.ndarray:
         with rasterio.open(tile) as src:
             crs = src.crs
         if crs != "EPSG:4326":
+            print("crs:", crs)
+            print("reprojecting to EPSG:4326")
             self.logger.debug(f"Reprojecting GeoTIFF from {crs} to EPSG:4326...")
             tile = self.reproject_geotiff(tile)
 
@@ -362,7 +383,7 @@ def get_bbox(self) -> tuple[float, float, float, float]:
         west, south, east, north = ox.utils_geo.bbox_from_point(  # type: ignore
             self.coordinates, dist=self.size // 2, project_utm=False
         )
-        bbox = north, south, east, west
+        bbox = float(north), float(south), float(east), float(west)
         return bbox
 
     def download_tif_files(self, urls: list[str], output_path: str) -> list[str]:
@@ -376,7 +397,7 @@ def download_tif_files(self, urls: list[str], output_path: str) -> list[str]:
             list: List of paths to the downloaded GeoTIFF files.
         """
         tif_files: list[str] = []
-        for url in urls:
+        for url in tqdm(urls, desc="Downloading tiles", unit="tile"):
             file_name = os.path.basename(url)
             self.logger.debug("Retrieving TIFF: %s", file_name)
             file_path = os.path.join(output_path, file_name)
@@ -443,7 +464,13 @@ def reproject_geotiff(self, input_tiff: str) -> str:
             # Update the metadata for the target GeoTIFF
             kwargs = src.meta.copy()
             kwargs.update(
-                {"crs": "EPSG:4326", "transform": transform, "width": width, "height": height}
+                {
+                    "crs": "EPSG:4326",
+                    "transform": transform,
+                    "width": width,
+                    "height": height,
+                    "nodata": None,
+                }
             )
 
             # Open the destination GeoTIFF file and reproject
@@ -456,7 +483,7 @@ def reproject_geotiff(self, input_tiff: str) -> str:
                         src_crs=src.crs,
                         dst_transform=transform,
                         dst_crs="EPSG:4326",
-                        resampling=Resampling.nearest,  # Choose resampling method
+                        resampling=Resampling.average,  # Choose resampling method
                     )
 
         self.logger.debug("Reprojected GeoTIFF saved to %s", output_tiff)
@@ -472,10 +499,12 @@ def merge_geotiff(self, input_files: list[str]) -> tuple[str, str]:
         # Open all input GeoTIFF files as datasets
         self.logger.debug("Merging tiff files...")
         datasets = [rasterio.open(file) for file in input_files]
+        print("datasets:", datasets)
 
         # Merge datasets
         crs = datasets[0].crs
         mosaic, out_transform = merge(datasets, nodata=0)
+        print("mosaic:", mosaic)
 
         # Get metadata from the first file and update it for the output
         out_meta = datasets[0].meta.copy()

maps4fs/generator/dtm/england.py

@@ -0,0 +1,31 @@
+"""This module contains provider of England data."""
+
+from maps4fs.generator.dtm.base.wcs import WCSProvider
+from maps4fs.generator.dtm.dtm import DTMProvider
+
+
+class England1MProvider(WCSProvider, DTMProvider):
+    """Provider of England data."""
+
+    _code = "england1m"
+    _name = "England DGM1"
+    _region = "UK"
+    _icon = "🏴󠁧󠁢󠁥󠁮󠁧󠁿"
+    _resolution = 1
+    _author = "[kbrandwijk](https://github.com/kbrandwijk)"
+    _is_community = True
+    _instructions = None
+    _is_base = False
+    _extents = (55.87708724246775, 49.85060473351981, 2.0842821419111135, -7.104775741839742)
+
+    _url = "https://environment.data.gov.uk/geoservices/datasets/13787b9a-26a4-4775-8523-806d13af58fc/wcs"  # pylint: disable=line-too-long
+    _wcs_version = "2.0.1"
+    _source_crs = "EPSG:27700"
+    _tile_size = 1000
+
+    def get_wcs_parameters(self, tile):
+        return {
+            "identifier": ["13787b9a-26a4-4775-8523-806d13af58fc:Lidar_Composite_Elevation_DTM_1m"],
+            "subsets": [("E", str(tile[1]), str(tile[3])), ("N", str(tile[0]), str(tile[2]))],
+            "format": "tiff",
+        }

maps4fs/generator/dtm/hessen.py

@@ -0,0 +1,31 @@
+"""This module contains provider of Hessen data."""
+
+from maps4fs.generator.dtm.base.wcs import WCSProvider
+from maps4fs.generator.dtm.dtm import DTMProvider
+
+
+class HessenProvider(WCSProvider, DTMProvider):
+    """Provider of Hessen data."""
+
+    _code = "hessen"
+    _name = "Hessen DGM1"
+    _region = "DE"
+    _icon = "🇩🇪󠁥"
+    _resolution = 1
+    _author = "[kbrandwijk](https://github.com/kbrandwijk)"
+    _is_community = True
+    _is_base = False
+    _extents = (51.66698, 49.38533, 10.25780, 7.72773)
+
+    _url = "https://inspire-hessen.de/raster/dgm1/ows"
+    _wcs_version = "2.0.1"
+    _source_crs = "EPSG:25832"
+    _tile_size = 1000
+
+    def get_wcs_parameters(self, tile: tuple[float, float, float, float]) -> dict:
+        return {
+            "identifier": ["he_dgm1"],
+            "subsets": [("N", str(tile[0]), str(tile[2])), ("E", str(tile[1]), str(tile[3]))],
+            "format": "image/gtiff",
+            "timeout": 600,
+        }