mamut_routing_tools.osm.fetch

Reliable, purpose-filtered OSM acquisition through Nominatim and Overpass.

Road ways and POI nodes use separate queries. Small extracts download each directly; large extracts are assembled from persistently cached bounded tiles in a disk-backed element store. This avoids Overpass recurse memory failures, unbounded client-side memory growth, and repeated work after an interruption.

fetch_city_bbox

fetch_city_bbox(city: str, country: str = '') -> tuple[float, float, float, float]

Return (min_lat, min_lon, max_lat, max_lon) for a place.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def fetch_city_bbox(city: str, country: str = "") -> tuple[float, float, float, float]:
    """Return ``(min_lat, min_lon, max_lat, max_lon)`` for a place."""
    query = city if not country.strip() else f"{city}, {country}"
    box = _geocode(query)["boundingbox"]
    return float(box[0]), float(box[2]), float(box[1]), float(box[3])

fetch_city_center

fetch_city_center(city: str, country: str = '') -> tuple[float, float]

Return the geocode point, rather than the administrative bbox center.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def fetch_city_center(city: str, country: str = "") -> tuple[float, float]:
    """Return the geocode point, rather than the administrative bbox center."""
    query = city if not country.strip() else f"{city}, {country}"
    result = _geocode(query)
    return float(result["lat"]), float(result["lon"])

build_road_overpass_query

build_road_overpass_query(bbox: str, *, road_classes: tuple[str, ...] | None = tuple(ROAD_CLASSES)) -> str

Fetch only road ways plus skeleton coordinates for referenced nodes.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def build_road_overpass_query(
    bbox: str, *, road_classes: tuple[str, ...] | None = tuple(ROAD_CLASSES)
) -> str:
    """Fetch only road ways plus skeleton coordinates for referenced nodes."""
    return (
        "[out:xml][timeout:120][maxsize:536870912];\n"
        f"{_road_statement(bbox, road_classes)}\n"
        "out body qt;\n"
        ">;\n"
        "out skel qt;\n"
    )

build_amenity_overpass_query

build_amenity_overpass_query(bbox: str, *, poi_categories: tuple[str, ...] | None = tuple(DEFAULT_CATEGORIES)) -> str

Fetch POI nodes, ways and relations, optionally restricted to categories.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def build_amenity_overpass_query(
    bbox: str,
    *,
    poi_categories: tuple[str, ...] | None = tuple(DEFAULT_CATEGORIES),
) -> str:
    """Fetch POI nodes, ways and relations, optionally restricted to categories."""
    return (
        "[out:xml][timeout:75][maxsize:268435456];\n"
        f"{_amenity_statement(bbox, poi_categories)}\n"
        f"{AMENITY_OUT_STATEMENT}\n"
    )

build_overpass_query

build_overpass_query(bbox: str, *, include_amenities: bool | None = None, profile: FetchProfileName | str | None = None, poi_categories: Sequence[str] | None = None) -> str

Build the legacy combined query; downloads use split queries internally.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def build_overpass_query(
    bbox: str,
    *,
    include_amenities: bool | None = None,
    profile: FetchProfileName | str | None = None,
    poi_categories: Sequence[str] | None = None,
) -> str:
    """Build the legacy combined query; downloads use split queries internally."""
    resolved = _resolve_fetch_profile(
        profile,
        include_amenities=include_amenities,
        poi_categories=poi_categories,
    )
    query = (
        "[out:xml][timeout:180][maxsize:1073741824];\n"
        f"{_road_statement(bbox, resolved.road_classes)}\n"
        "out body qt;\n"
        ">;\n"
        "out skel qt;\n"
    )
    if resolved.include_amenities:
        query += (
            f"{_amenity_statement(bbox, resolved.poi_categories)}\n"
            f"{AMENITY_OUT_STATEMENT}\n"
        )
    return query

validate_osm_extract

validate_osm_extract(osm_path: str | Path) -> dict[str, Any]

Validate the structure needed by the road-graph parser.

In particular, an Overpass <remark> document is rejected even though it is well-formed XML and was returned with HTTP 200.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def validate_osm_extract(osm_path: str | Path) -> dict[str, Any]:
    """Validate the structure needed by the road-graph parser.

    In particular, an Overpass ``<remark>`` document is rejected even though
    it is well-formed XML and was returned with HTTP 200.
    """
    path = Path(osm_path)
    if not path.is_file():
        raise FetchError(f"OSM extract does not exist: {path}")

    nodes = 0
    ways = 0
    bounds: dict[str, float] | None = None
    try:
        for _event, elem in ET.iterparse(path, events=("end",)):
            tag = elem.tag.rsplit("}", 1)[-1]
            if tag == "remark":
                detail = " ".join("".join(elem.itertext()).split())
                raise FetchError(
                    f"OSM extract contains an Overpass error remark: "
                    f"{detail or 'unknown error'}"
                )
            if tag == "bounds":
                try:
                    bounds = {
                        "minlat": float(elem.attrib["minlat"]),
                        "minlon": float(elem.attrib["minlon"]),
                        "maxlat": float(elem.attrib["maxlat"]),
                        "maxlon": float(elem.attrib["maxlon"]),
                    }
                except (KeyError, ValueError) as error:
                    raise FetchError(f"OSM extract has invalid bounds: {path}") from error
            elif tag == "node":
                nodes += 1
            elif tag == "way":
                ways += 1
            elem.clear()
    except ET.ParseError as error:
        raise FetchError(f"OSM extract is not complete XML: {path}: {error}") from error

    if bounds is None:
        raise FetchError(f"OSM extract has no <bounds> element: {path}")
    if nodes == 0:
        raise FetchError(f"OSM extract contains no nodes: {path}")
    if ways == 0:
        raise FetchError(f"OSM extract contains no ways: {path}")
    return {
        "path": str(path),
        "bytes": path.stat().st_size,
        "nodes": nodes,
        "ways": ways,
        "bounds": bounds,
    }

merge_amenity_elements_into_osm

merge_amenity_elements_into_osm(osm_path: Path, body: str) -> tuple[int, int, int]

Fold one amenity response into an existing road extract.

Nodes keep the enriching behaviour of merge_nodes_into_osm: a POI body replaces the tagless skeleton copy of the same node. Ways and relations are only ever appended, never substituted -- a road way already in the file is the one holding the <nd> refs the graph is built from, and the amenity copy of it carries a <center> instead. Returns the counts added, per element type.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def merge_amenity_elements_into_osm(osm_path: Path, body: str) -> tuple[int, int, int]:
    """Fold one amenity response into an existing road extract.

    Nodes keep the enriching behaviour of ``merge_nodes_into_osm``: a POI body
    replaces the tagless skeleton copy of the same node. Ways and relations are
    only ever *appended*, never substituted -- a road way already in the file is
    the one holding the ``<nd>`` refs the graph is built from, and the amenity
    copy of it carries a ``<center>`` instead. Returns the counts added, per
    element type.
    """
    nodes_added = merge_nodes_into_osm(
        osm_path, [match.group(0) for match in _NODE_BLOCK_PATTERN.finditer(body)]
    )
    text = osm_path.read_text(encoding="utf-8")
    additions: list[str] = []
    counts: list[int] = []
    for pattern in (_WAY_BLOCK_PATTERN, _RELATION_BLOCK_PATTERN):
        present = {
            _ELEMENT_ID_PATTERN.search(match.group(0)).group(1)  # type: ignore[union-attr]
            for match in pattern.finditer(text)
            if _ELEMENT_ID_PATTERN.search(match.group(0)) is not None
        }
        new_blocks: dict[str, str] = {}
        for match in pattern.finditer(body):
            block = match.group(0)
            id_match = _ELEMENT_ID_PATTERN.search(block)
            if id_match is not None and id_match.group(1) not in present:
                new_blocks[id_match.group(1)] = block
        additions.extend(new_blocks.values())
        counts.append(len(new_blocks))
    if additions:
        close_at = text.rfind("</osm>")
        if close_at < 0:
            raise FetchError(f"Invalid OSM file (missing </osm>): {osm_path}")
        _atomic_write_text(
            osm_path, text[:close_at] + "\n" + "\n".join(additions) + "\n" + text[close_at:]
        )
    return nodes_added, counts[0], counts[1]

fetch_tiled_amenities

fetch_tiled_amenities(min_lat: float, min_lon: float, max_lat: float, max_lon: float, outpath: Path, *, poi_categories: tuple[str, ...] | None = tuple(DEFAULT_CATEGORIES), progress: ProgressCallback | None = None, tile_cache_dir: str | Path | None = None, use_tile_cache: bool = True) -> dict[str, Any]

Backfill amenities into an already downloaded road extract.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def fetch_tiled_amenities(
    min_lat: float,
    min_lon: float,
    max_lat: float,
    max_lon: float,
    outpath: Path,
    *,
    poi_categories: tuple[str, ...] | None = tuple(DEFAULT_CATEGORIES),
    progress: ProgressCallback | None = None,
    tile_cache_dir: str | Path | None = None,
    use_tile_cache: bool = True,
) -> dict[str, Any]:
    """Backfill amenities into an already downloaded road extract."""
    target = Path(outpath)
    cache_root = _resolve_tile_cache_root(
        target,
        tile_cache_dir=tile_cache_dir,
        use_tile_cache=use_tile_cache,
    )
    lat_tiles = split_range(min_lat, max_lat, AMENITY_TILE_LAT_SPAN)
    lon_tiles = split_range(min_lon, max_lon, AMENITY_TILE_LON_SPAN)
    total_tiles = len(lat_tiles) * len(lon_tiles)
    bodies: list[str] = []
    tiles_ok = 0
    cache_hits = 0
    failure_count = 0
    current = 0
    for lat_lo, lat_hi in lat_tiles:
        for lon_lo, lon_hi in lon_tiles:
            bbox = f"({lat_lo},{lon_lo},{lat_hi},{lon_hi})"
            result, cache_hit = _fetch_overpass_tile(
                _amenity_query(bbox, poi_categories),
                cache_root=cache_root,
                kind="amenities",
                read_timeout=120.0,
            )
            current += 1
            cache_hits += int(cache_hit)
            if result.body is None:
                failure_count += len(result.failures)
            else:
                bodies.append(result.body)
                tiles_ok += 1
            if progress is not None:
                progress(
                    {
                        "phase": "amenities",
                        "current": current,
                        "total": total_tiles,
                        "tiles_ok": tiles_ok,
                        "cache_hits": cache_hits,
                    }
                )
    added = ways_added = relations_added = 0
    for body in bodies:
        nodes, ways, relations = merge_amenity_elements_into_osm(outpath, body)
        added += nodes
        ways_added += ways
        relations_added += relations
    return {
        "ok": tiles_ok == total_tiles,
        "tiles_total": total_tiles,
        "tiles_ok": tiles_ok,
        "cache_hits": cache_hits,
        "amenity_nodes_added": added,
        "amenity_ways_added": ways_added,
        "amenity_relations_added": relations_added,
        "failure_count": failure_count,
    }

fetch_tiled_osm

fetch_tiled_osm(min_lat: float, min_lon: float, max_lat: float, max_lon: float, outpath: str | Path, *, include_amenities: bool | None = None, profile: FetchProfileName | str | None = None, poi_categories: Sequence[str] | None = None, progress: ProgressCallback | None = None, tile_cache_dir: str | Path | None = None, use_tile_cache: bool = True) -> dict[str, Any]

Fetch every road tile, deduplicate it on disk, then atomically publish.

Road coverage is all-or-nothing. Amenity tile failures are non-fatal because amenities do not affect road-cache computation.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def fetch_tiled_osm(
    min_lat: float,
    min_lon: float,
    max_lat: float,
    max_lon: float,
    outpath: str | Path,
    *,
    include_amenities: bool | None = None,
    profile: FetchProfileName | str | None = None,
    poi_categories: Sequence[str] | None = None,
    progress: ProgressCallback | None = None,
    tile_cache_dir: str | Path | None = None,
    use_tile_cache: bool = True,
) -> dict[str, Any]:
    """Fetch every road tile, deduplicate it on disk, then atomically publish.

    Road coverage is all-or-nothing. Amenity tile failures are non-fatal
    because amenities do not affect road-cache computation.
    """
    target = Path(outpath)
    target.parent.mkdir(parents=True, exist_ok=True)
    resolved = _resolve_fetch_profile(
        profile,
        include_amenities=include_amenities,
        poi_categories=poi_categories,
    )
    cache_root = _resolve_tile_cache_root(
        target,
        tile_cache_dir=tile_cache_dir,
        use_tile_cache=use_tile_cache,
    )
    lat_tiles = split_range(min_lat, max_lat, ROAD_TILE_LAT_SPAN)
    lon_tiles = split_range(min_lon, max_lon, ROAD_TILE_LON_SPAN)
    road_tiles_total = len(lat_tiles) * len(lon_tiles)
    road_tiles_done = 0
    road_tiles_ok = 0
    road_cache_hits = 0
    road_failures: list[str] = []
    amenity_tiling = _empty_amenity_tiling()

    with tempfile.TemporaryDirectory(prefix=f".{target.name}.tiles.", dir=target.parent) as temp:
        store = _OsmElementStore(Path(temp) / "elements.sqlite3")
        try:
            for lat_lo, lat_hi in lat_tiles:
                for lon_lo, lon_hi in lon_tiles:
                    bbox = f"({lat_lo},{lon_lo},{lat_hi},{lon_hi})"
                    result, cache_hit = _fetch_overpass_tile(
                        build_road_overpass_query(
                            bbox, road_classes=resolved.road_classes
                        ),
                        cache_root=cache_root,
                        kind="roads",
                        read_timeout=180.0,
                    )
                    road_tiles_done += 1
                    road_cache_hits += int(cache_hit)
                    if result.body is None:
                        road_failures.extend(result.failures)
                    else:
                        store.add_body(result.body)
                        road_tiles_ok += 1
                    if progress is not None:
                        progress(
                            {
                                "phase": "roads",
                                "current": road_tiles_done,
                                "total": road_tiles_total,
                                "tiles_ok": road_tiles_ok,
                                "cache_hits": road_cache_hits,
                            }
                        )

            nodes, ways = store.counts()
            road_tiling = {
                "ok": road_tiles_ok == road_tiles_total and nodes > 0 and ways > 0,
                "tiles_total": road_tiles_total,
                "tiles_ok": road_tiles_ok,
                "cache_hits": road_cache_hits,
                "failure_count": len(road_failures),
            }
            if not road_tiling["ok"]:
                if nodes == 0:
                    road_failures.append("tiled road fetch produced no nodes")
                if ways == 0:
                    road_failures.append("tiled road fetch produced no ways")
                return {
                    "ok": False,
                    "profile": resolved.name,
                    "road_tiling": road_tiling,
                    "amenity_tiling": amenity_tiling,
                    "failures": road_failures,
                }

            if resolved.include_amenities:
                amenity_lat_tiles = split_range(
                    min_lat, max_lat, AMENITY_TILE_LAT_SPAN
                )
                amenity_lon_tiles = split_range(
                    min_lon, max_lon, AMENITY_TILE_LON_SPAN
                )
                amenity_total = len(amenity_lat_tiles) * len(amenity_lon_tiles)
                amenity_done = 0
                amenity_ok = 0
                amenity_cache_hits = 0
                amenity_failures = 0
                amenity_nodes_before, _ = store.counts()
                amenity_elements_before = store.element_counts()
                for lat_lo, lat_hi in amenity_lat_tiles:
                    for lon_lo, lon_hi in amenity_lon_tiles:
                        bbox = f"({lat_lo},{lon_lo},{lat_hi},{lon_hi})"
                        result, cache_hit = _fetch_overpass_tile(
                            _amenity_query(bbox, resolved.poi_categories),
                            cache_root=cache_root,
                            kind="amenities",
                            read_timeout=120.0,
                        )
                        amenity_done += 1
                        amenity_cache_hits += int(cache_hit)
                        if result.body is None:
                            amenity_failures += len(result.failures)
                        else:
                            store.add_body(result.body)
                            amenity_ok += 1
                        if progress is not None:
                            progress(
                                {
                                    "phase": "amenities",
                                    "current": amenity_done,
                                    "total": amenity_total,
                                    "tiles_ok": amenity_ok,
                                    "cache_hits": amenity_cache_hits,
                                }
                            )
                amenity_nodes_after, _ = store.counts()
                amenity_elements_after = store.element_counts()
                amenity_tiling = {
                    "ok": amenity_ok == amenity_total,
                    "tiles_total": amenity_total,
                    "tiles_ok": amenity_ok,
                    "cache_hits": amenity_cache_hits,
                    "amenity_nodes_added": amenity_nodes_after
                    - amenity_nodes_before,
                    # Ways and relations are pure amenity gains here: the road
                    # phase never fetches a way this phase could also claim.
                    "amenity_ways_added": amenity_elements_after["ways"]
                    - amenity_elements_before["ways"],
                    "amenity_relations_added": amenity_elements_after["relations"]
                    - amenity_elements_before["relations"],
                    "failure_count": amenity_failures,
                }

            store.write(
                target,
                min_lat=min_lat,
                min_lon=min_lon,
                max_lat=max_lat,
                max_lon=max_lon,
            )
        finally:
            store.close()

    return {
        "ok": True,
        "profile": resolved.name,
        "road_tiling": road_tiling,
        "amenity_tiling": amenity_tiling,
        "failures": [],
        "validation": validate_osm_extract(target),
    }

fetch_and_store_bbox_osm

fetch_and_store_bbox_osm(min_lat: float, min_lon: float, max_lat: float, max_lon: float, outpath: str | Path, *, include_amenities: bool | None = None, profile: FetchProfileName | str | None = None, poi_categories: Sequence[str] | None = None, progress: ProgressCallback | None = None, tile_cache_dir: str | Path | None = None, use_tile_cache: bool = True) -> dict[str, Any]

Fetch an arbitrary bbox, automatically tiling large or failed queries.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/osm/fetch.py

def fetch_and_store_bbox_osm(
    min_lat: float,
    min_lon: float,
    max_lat: float,
    max_lon: float,
    outpath: str | Path,
    *,
    include_amenities: bool | None = None,
    profile: FetchProfileName | str | None = None,
    poi_categories: Sequence[str] | None = None,
    progress: ProgressCallback | None = None,
    tile_cache_dir: str | Path | None = None,
    use_tile_cache: bool = True,
) -> dict[str, Any]:
    """Fetch an arbitrary bbox, automatically tiling large or failed queries."""
    if min_lat >= max_lat or min_lon >= max_lon:
        raise FetchError("OSM bbox must have positive latitude and longitude spans")

    target = Path(outpath)
    target.parent.mkdir(parents=True, exist_ok=True)
    resolved = _resolve_fetch_profile(
        profile,
        include_amenities=include_amenities,
        poi_categories=poi_categories,
    )
    bbox = {
        "minlat": min_lat,
        "minlon": min_lon,
        "maxlat": max_lat,
        "maxlon": max_lon,
    }
    bbox_text = f"({min_lat},{min_lon},{max_lat},{max_lon})"
    spans = _bbox_span_km(min_lat, min_lon, max_lat, max_lon)
    is_large = max(spans.values()) > LARGE_BBOX_SPAN_KM
    direct_failures: list[str] = []

    if not is_large:
        result, validation = _download_validated(
            build_road_overpass_query(
                bbox_text, road_classes=resolved.road_classes
            ),
            target,
            min_lat=min_lat,
            min_lon=min_lon,
            max_lat=max_lat,
            max_lon=max_lon,
        )
        direct_failures.extend(result.failures)
        if validation is not None:
            amenity_tiling = _empty_amenity_tiling()
            dataset_mode = "roads_only"
            if resolved.include_amenities:
                amenity_tiling = _fetch_direct_amenities(
                    min_lat=min_lat,
                    min_lon=min_lon,
                    max_lat=max_lat,
                    max_lon=max_lon,
                    outpath=target,
                    poi_categories=resolved.poi_categories,
                    progress=progress,
                    tile_cache_dir=tile_cache_dir,
                    use_tile_cache=use_tile_cache,
                )
                if not amenity_tiling["ok"]:
                    direct_amenity_failures = amenity_tiling["failure_count"]
                    amenity_tiling = fetch_tiled_amenities(
                        min_lat,
                        min_lon,
                        max_lat,
                        max_lon,
                        target,
                        poi_categories=resolved.poi_categories,
                        progress=progress,
                        tile_cache_dir=tile_cache_dir,
                        use_tile_cache=use_tile_cache,
                    )
                    amenity_tiling["failure_count"] += direct_amenity_failures
                if amenity_tiling["ok"]:
                    dataset_mode = "roads_and_amenities"
                elif amenity_tiling["tiles_ok"] > 0:
                    dataset_mode = "roads_plus_partial_amenities"
                else:
                    dataset_mode = "roads_only"
                validation = validate_osm_extract(target)
            warning = (
                "Some amenities could not be fetched; POI-based generation may "
                "have fewer candidates."
                if resolved.include_amenities and not amenity_tiling["ok"]
                else ""
            )
            return {
                "ok": True,
                "osm_path": str(target),
                "profile": resolved.name,
                "road_classes": (
                    list(resolved.road_classes)
                    if resolved.road_classes is not None
                    else None
                ),
                "poi_categories": (
                    list(resolved.poi_categories)
                    if resolved.poi_categories is not None
                    else None
                ),
                "dataset_mode": dataset_mode,
                "bbox": bbox,
                "bbox_span_km": spans,
                "road_tiling": _empty_road_tiling(),
                "amenity_tiling": amenity_tiling,
                "validation": validation,
                "warning": warning,
            }

    tiled = fetch_tiled_osm(
        min_lat,
        min_lon,
        max_lat,
        max_lon,
        target,
        profile=resolved.name,
        poi_categories=resolved.poi_categories,
        progress=progress,
        tile_cache_dir=tile_cache_dir,
        use_tile_cache=use_tile_cache,
    )
    if not tiled["ok"]:
        details = tiled.get("failures", []) + direct_failures
        raise FetchError(
            "Overpass could not provide complete road coverage: "
            + (" | ".join(details) or "no details")
        )

    amenity_tiling = tiled["amenity_tiling"]
    if not resolved.include_amenities:
        dataset_mode = "tiled_roads"
    elif amenity_tiling["ok"]:
        dataset_mode = "tiled_roads_and_amenities"
    elif amenity_tiling["amenity_nodes_added"] > 0:
        dataset_mode = "tiled_roads_plus_partial_amenities"
    else:
        dataset_mode = "tiled_roads_only"
    warning = (
        "Some amenities could not be fetched; POI-based generation may have "
        "fewer candidates."
        if resolved.include_amenities and not amenity_tiling["ok"]
        else ""
    )
    return {
        "ok": True,
        "osm_path": str(target),
        "profile": resolved.name,
        "road_classes": (
            list(resolved.road_classes)
            if resolved.road_classes is not None
            else None
        ),
        "poi_categories": (
            list(resolved.poi_categories)
            if resolved.poi_categories is not None
            else None
        ),
        "dataset_mode": dataset_mode,
        "bbox": bbox,
        "bbox_span_km": spans,
        "road_tiling": tiled["road_tiling"],
        "amenity_tiling": amenity_tiling,
        "validation": tiled["validation"],
        "warning": warning,
    }