Avoid generating empty output files

pyhgtmap/contour.py

@@ -1,10 +1,12 @@
-from typing import Callable, List, Optional, Tuple, cast
+from typing import List, Optional, Tuple, cast
 
 import contourpy
 import numpy
 import numpy.typing
 from pybind11_rdp import rdp
 
+from pyhgtmap.hgt import TransformFunType
+
 
 def simplify_path(
     input_path: numpy.ndarray, rdp_epsilon: Optional[float] = None
@@ -32,14 +34,14 @@ def __init__(
         self,
         cntr: contourpy.ContourGenerator,
         max_nodes_per_way,
-        transform,
+        transform: Optional[TransformFunType],
         polygon=None,
         rdp_epsilon=None,
     ) -> None:
         self.cntr: contourpy.ContourGenerator = cntr
         self.max_nodes_per_way = max_nodes_per_way
         self.polygon = polygon
-        self.transform = transform
+        self.transform: Optional[TransformFunType] = transform
         self.rdp_epsilon = rdp_epsilon
 
     def _cutBeginning(self, p):
@@ -136,7 +138,7 @@ def build_contours(
     y: numpy.typing.ArrayLike,
     z: numpy.typing.ArrayLike,
     max_nodes_per_way: int,
-    transform: Callable,
+    transform: Optional[TransformFunType],
     polygon,
     rdp_epsilon,
 ) -> ContoursGenerator:

pyhgtmap/hgt/__init__.py

@@ -1,11 +1,22 @@
-from typing import Tuple
+from typing import Callable, Iterable, Optional, Tuple
+
+# Coordinates transformation function prototype
+TransformFunType = Callable[
+    [Iterable[Tuple[float, float]]], Iterable[Tuple[float, float]]
+]
 
 
 def makeBBoxString(bbox: Tuple[float, float, float, float]) -> str:
     return "{{0:s}}lon{0[0]:.2f}_{0[2]:.2f}lat{0[1]:.2f}_{0[3]:.2f}".format(bbox)
 
 
-def transformLonLats(minLon, minLat, maxLon, maxLat, transform):
+def transformLonLats(
+    minLon: float,
+    minLat: float,
+    maxLon: float,
+    maxLat: float,
+    transform: Optional[TransformFunType],
+) -> Tuple[float, float, float, float]:
     if transform is None:
         return minLon, minLat, maxLon, maxLat
     else:

pyhgtmap/hgt/file.py

@@ -1,15 +1,15 @@
 import logging
 import os
 import sys
-from typing import Callable, List, Optional, Tuple, cast
+from typing import Iterable, List, Optional, Tuple, cast
 
 import numpy
 import numpy.typing
 import shapely
 from matplotlib.path import Path as PolygonPath
 from scipy import ndimage
 
-from pyhgtmap.hgt import transformLonLats
+from pyhgtmap.hgt import TransformFunType, transformLonLats
 
 from .tile import hgtTile
 
@@ -114,10 +114,6 @@ def parseHgtFilename(
     return minLon + corrx, minLat + corry, maxLon + corrx, maxLat + corry
 
 
-# Coordinates transformation function prototype
-TransformFunType = Callable[[List[Tuple[float, float]]], List[Tuple[float, float]]]
-
-
 def getTransform(o, reverse=False) -> Optional[TransformFunType]:
     try:
         from osgeo import osr
@@ -137,7 +133,9 @@ def getTransform(o, reverse=False) -> Optional[TransformFunType]:
         else:
             t = osr.CoordinateTransformation(o, n)
 
-        def transform(points: List[Tuple[float, float]]) -> List[Tuple[float, float]]:
+        def transform(
+            points: Iterable[Tuple[float, float]]
+        ) -> Iterable[Tuple[float, float]]:
             return [
                 p[:2]
                 for p in t.TransformPoints(points)
@@ -236,7 +234,7 @@ def polygon_mask(
     x_data: numpy.ndarray,
     y_data: numpy.ndarray,
     polygons: List[List[Tuple[float, float]]],
-    transform,
+    transform: Optional[TransformFunType],
 ) -> numpy.ndarray:
     """return a mask on self.zData corresponding to all polygons in self.polygons.
     <xData> is meant to be a 1-D array of longitude values, <yData> a 1-D array of
@@ -246,12 +244,14 @@ def polygon_mask(
     which is the projection used within polygon files.
     """
     X, Y = numpy.meshgrid(x_data, y_data)
-    xyPoints = numpy.vstack(([X.T], [Y.T])).T.reshape(len(x_data) * len(y_data), 2)
+    xyPoints: Iterable[tuple[float, float]] = numpy.vstack(([X.T], [Y.T])).T.reshape(
+        len(x_data) * len(y_data), 2
+    )
 
     # To improve performances, clip original polygons to current data boundaries.
     # Slightly expand the bounding box, as PolygonPath.contains_points result is undefined for points on boundary
     # https://matplotlib.org/stable/api/path_api.html#matplotlib.path.Path.contains_point
-    bbox_points: List[Tuple[float, float]] = [
+    bbox_points: Iterable[Tuple[float, float]] = [
         (x_data.min() - BBOX_EXPAND_EPSILON, y_data.min() - BBOX_EXPAND_EPSILON),
         (x_data.min() - BBOX_EXPAND_EPSILON, y_data.max() + BBOX_EXPAND_EPSILON),
         (x_data.max() + BBOX_EXPAND_EPSILON, y_data.max() + BBOX_EXPAND_EPSILON),

pyhgtmap/hgt/processor.py

@@ -114,6 +114,10 @@ def process_tile_internal(self, file_name: str, tile: hgtTile) -> None:
                 rdp_epsilon=self.options.rdpEpsilon,
             )
 
+            if not tile_contours.nb_nodes:
+                logger.info("%s doesn't contain any node, skipping.", tile)
+                return
+
             # Update counters shared among parallel processes
             # This is the actual critical section, to avoid duplicated node IDs
             logger.debug("Pending next_node_id_lock")

pyhgtmap/hgt/tile.py

@@ -53,9 +53,7 @@ def __init__(self, tile_data: Dict[str, Any]):
 
     def get_stats(self) -> str:
         """Get some statistics about the tile."""
-        minLon, minLat, maxLon, maxLat = transformLonLats(
-            self.minLon, self.minLat, self.maxLon, self.maxLat, self.transform
-        )
+        minLon, minLat, maxLon, maxLat = self.bbox()
         result = (
             f"tile with {self.numOfRows:d} x {self.numOfCols:d} points, "
             f"bbox: ({minLon:.2f}, {minLat:.2f}, {maxLon:.2f}, {maxLat:.2f})"
@@ -67,6 +65,10 @@ def printStats(self) -> None:
         """prints some statistics about the tile."""
         print(f"\n{self.get_stats()}")
 
+    def __str__(self) -> str:
+        bbox = self.bbox()
+        return f"Tile ({bbox[0]:.2f}, {bbox[1]:.2f}, {bbox[2]:.2f}, {bbox[3]:.2f})"
+
     def getElevRange(self) -> Tuple[int, int]:
         """returns minEle, maxEle of the current tile.
 

tests/hgt/test_processor.py

@@ -1,5 +1,6 @@
 import glob
 import itertools
+import logging
 import multiprocessing
 import os
 import shutil
@@ -9,14 +10,15 @@
 from types import SimpleNamespace
 from typing import Callable, Generator, List, NamedTuple, Tuple
 from unittest import mock
-from unittest.mock import Mock
+from unittest.mock import MagicMock, Mock
 
 import npyosmium
 import npyosmium.io
 import npyosmium.osm
 import pytest
 
 from pyhgtmap.hgt.processor import HgtFilesProcessor
+from pyhgtmap.hgt.tile import TileContours
 
 from .. import TEST_DATA_PATH
 
@@ -327,3 +329,27 @@ def test_way_id_overflow(default_options: SimpleNamespace) -> None:
         )
         assert processor.get_and_inc_counter(processor.next_way_id, 1) == 2147483647
         assert processor.get_and_inc_counter(processor.next_way_id, 1) == 2147483648
+
+    @staticmethod
+    def test_process_tile_internal_empty_contour(
+        default_options: SimpleNamespace, caplog
+    ) -> None:
+        """Ensure no empty output file is generated when there's no contour."""
+        processor = HgtFilesProcessor(
+            1, node_start_id=100, way_start_id=200, options=default_options
+        )
+        # Empty tile
+        tile_contours = TileContours(nb_nodes=0, nb_ways=0, contours={})
+        tile_mock = MagicMock()
+        tile_mock.get_contours.return_value = tile_contours
+        tile_mock.__str__.return_value = "Tile (28.00, 42.50, 29.00, 43.00)"  # type: ignore
+        with tempfile.TemporaryDirectory() as tempdir_name:
+            with cwd(tempdir_name):
+                caplog.set_level(logging.INFO)
+                processor.process_tile_internal("empty.pbf", tile_mock)
+                # NO file must be generated
+                assert not os.path.exists("empty.pbf")
+                assert (
+                    "Tile (28.00, 42.50, 29.00, 43.00) doesn't contain any node, skipping."
+                    in caplog.text
+                )