Issue #203: support Overpass.

map_machine/geometry/bounding_box.py

@@ -25,6 +25,16 @@
 )
 
 
+def floor(value: float) -> float:
+    """Round down to 3 digits after the point."""
+    return np.floor(value * 1000.0) / 1000.0
+
+
+def ceil(value: float) -> float:
+    """Round up to 3 digits after the point."""
+    return np.ceil(value * 1000.0) / 1000.0
+
+
 @dataclass
 class BoundingBox:
     """Rectangle that limits the space on the map."""
@@ -156,15 +166,25 @@ def get_format(self) -> str:
         """Get text representation of the bounding box.
 
         Bounding box format is
-        <longitude 1>,<latitude 1>,<longitude 2>,<latitude 2>.  Coordinates are
+        <longitude 1>,<latitude 1>,<longitude 2>,<latitude 2>. Coordinates are
         rounded to three decimal places.
         """
-        left: float = np.floor(self.left * 1000.0) / 1000.0
-        bottom: float = np.floor(self.bottom * 1000.0) / 1000.0
-        right: float = np.ceil(self.right * 1000.0) / 1000.0
-        top: float = np.ceil(self.top * 1000.0) / 1000.0
+        return (
+            f"{floor(self.left):.3f},{floor(self.bottom):.3f},"
+            f"{ceil(self.right):.3f},{ceil(self.top):.3f}"
+        )
+
+    def get_overpass_format(self) -> str:
+        """Get bounding box in Overpass API format.
 
-        return f"{left:.3f},{bottom:.3f},{right:.3f},{top:.3f}"
+        Overpass bounding box format is <south>,<west>,<north>,<east>, which
+        maps to <bottom>,<left>,<top>,<right>. Minimum coordinates are floored
+        and maximum coordinates are ceiled to three decimal places.
+        """
+        return (
+            f"{floor(self.bottom):.3f},{floor(self.left):.3f},"
+            f"{ceil(self.top):.3f},{ceil(self.right):.3f}"
+        )
 
     def update(self, coordinates: np.ndarray) -> None:
         """Make the bounding box cover coordinates."""

map_machine/mapper.py

@@ -32,6 +32,7 @@
     NetworkError,
     find_incomplete_relations,
     get_osm,
+    get_osm_overpass,
     get_overpass_relations,
 )
 from map_machine.osm.osm_reader import OSMData, OSMNode
@@ -412,19 +413,41 @@ def render_map(arguments: argparse.Namespace) -> None:
 
     # Determine files.
 
+    overpass_query: str | None = None
+    overpass_query_path: str | None = getattr(arguments, "overpass_query", None)
+    if overpass_query_path:
+        with Path(overpass_query_path).open(encoding="utf-8") as query_file:
+            overpass_query = query_file.read()
+
     input_file_names: list[Path]
     if arguments.input_file_names:
         input_file_names = list(map(Path, arguments.input_file_names))
     elif bounding_box:
-        try:
+        overpass_cache_path: Path = (
+            cache_path / f"{bounding_box.get_format()}_overpass.osm"
+        )
+        if overpass_cache_path.is_file():
+            input_file_names = [overpass_cache_path]
+        else:
             cache_file_path: Path = (
                 cache_path / f"{bounding_box.get_format()}.osm"
             )
-            get_osm(bounding_box, cache_file_path)
-            input_file_names = [cache_file_path]
-        except NetworkError as error:
-            logger.fatal(error.message)
-            sys.exit(1)
+            try:
+                get_osm(bounding_box, cache_file_path)
+                input_file_names = [cache_file_path]
+            except NetworkError as error:
+                logger.warning(
+                    "OSM API failed (%s), falling back to Overpass API...",
+                    error.message,
+                )
+                try:
+                    get_osm_overpass(
+                        bounding_box, overpass_cache_path, overpass_query
+                    )
+                    input_file_names = [overpass_cache_path]
+                except NetworkError as overpass_error:
+                    logger.fatal(overpass_error.message)
+                    sys.exit(1)
     else:
         fatal(
             "Specify `--input`, `--bounding-box`, `--coordinates`, or `--gpx`."

map_machine/osm/osm_getter.py

@@ -7,6 +7,7 @@
 import logging
 import time
 from dataclasses import dataclass
+from textwrap import dedent
 from typing import TYPE_CHECKING
 
 import urllib3
@@ -26,6 +27,23 @@
 MAX_OSM_MESSAGE_LENGTH: int = 500
 OVERPASS_API_URL: str = "https://overpass-api.de/api/interpreter"
 
+DEFAULT_OVERPASS_FILTER: str = dedent(
+    """
+    (
+        way["building"];
+        way["highway"];
+        way["natural"="water"];
+        way["waterway"];
+        way["natural"="wood"];
+        way["landuse"="forest"];
+        relation["natural"="water"];
+        relation["landuse"="forest"];
+    );
+    (._;>;);
+    out body;
+    """
+)
+
 
 @dataclass
 class NetworkError(Exception):
@@ -158,3 +176,57 @@ def get_overpass_relations(
 
     cache_file.write_bytes(content)
     return content
+
+
+def get_osm_overpass(
+    bounding_box: BoundingBox,
+    cache_file_path: Path,
+    query: str | None = None,
+    *,
+    to_update: bool = False,
+) -> str:
+    """Download OSM data from the Overpass API.
+
+    Uses a simplified filter query to fetch only major map features, which
+    avoids failures when the standard OSM API returns too much data.
+
+    :param bounding_box: borders of the map part to download
+    :param cache_file_path: cache file to store downloaded OSM data
+    :param query: custom Overpass query with ``{{bbox}}`` placeholder; if None,
+        the default filter is used
+    :param to_update: update cache files
+    """
+    if not to_update and cache_file_path.is_file():
+        with cache_file_path.open(encoding="utf-8") as output_file:
+            return output_file.read()
+
+    box: str = bounding_box.get_overpass_format()
+
+    if query is not None:
+        full_query = query.replace("{{bbox}}", box)
+    else:
+        full_query = f"[out:xml][bbox:{box}];\n{DEFAULT_OVERPASS_FILTER}"
+
+    logger.info("Querying Overpass API for bounding box %s...", box)
+
+    try:
+        content: bytes = get_data(OVERPASS_API_URL, {"data": full_query})
+    except NetworkError:
+        logger.warning("Failed to download data from Overpass API.")
+        raise
+
+    if not content or not content.strip().startswith(b"<"):
+        if len(content) < MAX_OSM_MESSAGE_LENGTH:
+            message = (
+                "Unexpected Overpass API response: `"
+                + content.decode("utf-8")
+                + "`."
+            )
+        else:
+            message = "Unexpected Overpass API response."
+        raise NetworkError(message)
+
+    with cache_file_path.open("bw+") as output_file:
+        output_file.write(content)
+
+    return content.decode("utf-8")

map_machine/slippy/tile.py

@@ -25,6 +25,7 @@
     NetworkError,
     find_incomplete_relations,
     get_osm,
+    get_osm_overpass,
     get_overpass_relations,
 )
 from map_machine.osm.osm_reader import OSMData
@@ -141,15 +142,33 @@ def get_extended_bounding_box(self) -> BoundingBox:
             float(point_1[0]),
         ).round()
 
-    def load_osm_data(self, cache_path: Path) -> OSMData:
+    def load_osm_data(
+        self, cache_path: Path, overpass_query: str | None = None
+    ) -> OSMData:
         """Construct map data from extended bounding box.
 
         :param cache_path: directory to store OSM data files
+        :param overpass_query: custom Overpass query with `{{bbox}}`
+            placeholder
         """
-        cache_file_path: Path = (
-            cache_path / f"{self.get_extended_bounding_box().get_format()}.osm"
+        bounding_box: BoundingBox = self.get_extended_bounding_box()
+        overpass_cache_path: Path = (
+            cache_path / f"{bounding_box.get_format()}_overpass.osm"
         )
-        get_osm(self.get_extended_bounding_box(), cache_file_path)
+
+        if overpass_cache_path.is_file():
+            cache_file_path = overpass_cache_path
+        else:
+            cache_file_path = cache_path / f"{bounding_box.get_format()}.osm"
+            try:
+                get_osm(bounding_box, cache_file_path)
+            except NetworkError as error:
+                logger.warning(
+                    "OSM API failed (%s), falling back to Overpass API...",
+                    error.message,
+                )
+                cache_file_path = overpass_cache_path
+                get_osm_overpass(bounding_box, cache_file_path, overpass_query)
 
         osm_data: OSMData = OSMData()
         osm_data.parse_osm_file(cache_file_path)
@@ -178,16 +197,19 @@ def draw(
         configuration: MapConfiguration,
         *,
         use_overpass: bool = True,
+        overpass_query: str | None = None,
     ) -> None:
         """Draw tile to SVG and PNG files.
 
         :param directory_name: output directory for storing tiles
         :param cache_path: directory to store SVG and PNG tiles
         :param configuration: drawing configuration
         :param use_overpass: fetch missing relation data via Overpass API
+        :param overpass_query: custom Overpass query with `{{bbox}}`
+            placeholder
         """
         try:
-            osm_data: OSMData = self.load_osm_data(cache_path)
+            osm_data: OSMData = self.load_osm_data(cache_path, overpass_query)
         except NetworkError as error:
             msg = f"Map is not loaded. {error.message}"
             raise NetworkError(msg) from error
@@ -300,12 +322,35 @@ def from_bounding_box(
 
         return cls(tiles, tile_1, tile_2, zoom_level, extended_bounding_box)
 
-    def load_osm_data(self, cache_path: Path) -> OSMData:
-        """Load OpenStreetMap data."""
-        cache_file_path: Path = (
-            cache_path / f"{self.bounding_box.get_format()}.osm"
+    def load_osm_data(
+        self, cache_path: Path, overpass_query: str | None = None
+    ) -> OSMData:
+        """Load OpenStreetMap data.
+
+        :param cache_path: directory for caching OSM data files
+        :param overpass_query: custom Overpass query with `{{bbox}}` placeholder
+        """
+        overpass_cache_path: Path = (
+            cache_path / f"{self.bounding_box.get_format()}_overpass.osm"
         )
-        get_osm(self.bounding_box, cache_file_path)
+
+        if overpass_cache_path.is_file():
+            cache_file_path = overpass_cache_path
+        else:
+            cache_file_path = (
+                cache_path / f"{self.bounding_box.get_format()}.osm"
+            )
+            try:
+                get_osm(self.bounding_box, cache_file_path)
+            except NetworkError as error:
+                logger.warning(
+                    "OSM API failed (%s), falling back to Overpass API...",
+                    error.message,
+                )
+                cache_file_path = overpass_cache_path
+                get_osm_overpass(
+                    self.bounding_box, cache_file_path, overpass_query
+                )
 
         osm_data: OSMData = OSMData()
         osm_data.parse_osm_file(cache_file_path)
@@ -517,6 +562,12 @@ def generate_tiles(options: argparse.Namespace) -> None:
 
     use_overpass: bool = not getattr(options, "no_overpass", False)
 
+    overpass_query: str | None = None
+    overpass_query_path: str | None = getattr(options, "overpass_query", None)
+    if overpass_query_path:
+        with Path(overpass_query_path).open(encoding="utf-8") as query_file:
+            overpass_query = query_file.read()
+
     if options.input_file_name:
         osm_data = OSMData()
         osm_data.parse_osm_file(Path(options.input_file_name))
@@ -548,7 +599,7 @@ def generate_tiles(options: argparse.Namespace) -> None:
             np.array(coordinates), min_zoom_level
         )
         try:
-            osm_data = min_tile.load_osm_data(cache_path)
+            osm_data = min_tile.load_osm_data(cache_path, overpass_query)
         except NetworkError as error:
             message = f"Map is not loaded. {error.message}"
             raise NetworkError(message) from error
@@ -577,6 +628,7 @@ def generate_tiles(options: argparse.Namespace) -> None:
             cache_path,
             configuration,
             use_overpass=use_overpass,
+            overpass_query=overpass_query,
         )
 
     elif options.bounding_box:
@@ -589,7 +641,7 @@ def generate_tiles(options: argparse.Namespace) -> None:
 
         min_tiles: Tiles = Tiles.from_bounding_box(bounding_box, min_zoom_level)
         try:
-            osm_data = min_tiles.load_osm_data(cache_path)
+            osm_data = min_tiles.load_osm_data(cache_path, overpass_query)
         except NetworkError as error:
             message = f"Map is not loaded. {error.message}"
             raise NetworkError(message) from error

map_machine/ui/cli.py

@@ -278,6 +278,14 @@ def add_tile_arguments(parser: argparse.ArgumentParser) -> None:
         action="store_true",
         default=False,
     )
+    parser.add_argument(
+        "--overpass-query",
+        metavar="<path>",
+        help=(
+            "path to a custom Overpass query file; use {{bbox}} as a "
+            "placeholder for the bounding box"
+        ),
+    )
     parser.add_argument(
         "-c",
         "--coordinates",
@@ -370,6 +378,14 @@ def add_render_arguments(parser: argparse.ArgumentParser) -> None:
         action="store_true",
         default=False,
     )
+    parser.add_argument(
+        "--overpass-query",
+        metavar="<path>",
+        help=(
+            "path to a custom Overpass query file; use {{bbox}} as a "
+            "placeholder for the bounding box"
+        ),
+    )
     parser.add_argument(
         "-i",
         "--input",