mamut_routing_tools.osm

audit_extract

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

What one extract holds: element counts, POI counts per type, and bounds.

Counted in a single streaming pass without building POI objects, because a workspace can hold dozens of extracts of 100+ MB each.

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

def audit_extract(osm_path: str | Path) -> dict[str, Any]:
    """What one extract holds: element counts, POI counts per type, and bounds.

    Counted in a single streaming pass without building POI objects, because a
    workspace can hold dozens of extracts of 100+ MB each.
    """
    path = Path(osm_path)
    stat = path.stat()
    elements: Counter[str] = Counter()
    pois: Counter[str] = Counter()
    bounds: dict[str, float] | None = None
    for _event, element in ET.iterparse(str(path), events=("end",)):
        if element.tag == "bounds" and bounds is None:
            try:
                bounds = {key: float(element.attrib[key]) for key in
                          ("minlat", "minlon", "maxlat", "maxlon")}
            except (KeyError, ValueError):
                bounds = None
            continue
        if element.tag not in _ELEMENT_TYPES:
            continue
        elements[element.tag] += 1
        for tag in element:
            if tag.tag == "tag" and tag.get("k") == "amenity":
                pois[element.tag] += 1
                break
        element.clear()

    poi_total = sum(pois.values())
    if poi_total == 0:
        status = STATUS_NO_AMENITIES
    elif pois["way"] or pois["relation"]:
        status = STATUS_COMPLETE
    else:
        status = STATUS_NODES_ONLY
    return {
        "city": path.stem,
        "path": str(path),
        "bytes": stat.st_size,
        "modified_at": stat.st_mtime,
        "nodes": elements["node"],
        "ways": elements["way"],
        "relations": elements["relation"],
        "poi_nodes": pois["node"],
        "poi_ways": pois["way"],
        "poi_relations": pois["relation"],
        "poi_total": poi_total,
        "status": status,
        # Nothing can be refilled without knowing the area to query.
        "can_refresh": bounds is not None,
        "bounds": bounds,
    }

audit_extracts

audit_extracts(osm_dir: str | Path) -> list[dict[str, Any]]

Audit every *.osm in a directory, by city name.

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

def audit_extracts(osm_dir: str | Path) -> list[dict[str, Any]]:
    """Audit every ``*.osm`` in a directory, by city name."""
    root = Path(osm_dir)
    if not root.is_dir():
        return []
    reports: list[dict[str, Any]] = []
    for path in sorted(root.glob("*.osm")):
        try:
            reports.append(audit_extract(path))
        except (OSError, ET.ParseError) as error:
            reports.append(
                {
                    "city": path.stem,
                    "path": str(path),
                    "status": "unreadable",
                    "error": str(error),
                    "can_refresh": False,
                }
            )
    return reports

refresh_extract_pois

refresh_extract_pois(
    osm_path: str | Path,
    *,
    poi_categories: tuple[str, ...] | None = tuple(
        POI_CATEGORIES
    ),
    progress: ProgressCallback | None = None,
    tile_cache_dir: str | Path | None = None,
    use_tile_cache: bool = True,
) -> dict[str, Any]

Backfill the amenities an extract is missing, in place.

Only the amenity query is re-run: the road network already in the file is left untouched, which is what makes this cheap enough to run over a whole workspace. The report pairs the audit before and after so a caller can say what each city gained.

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

def refresh_extract_pois(
    osm_path: str | Path,
    *,
    poi_categories: tuple[str, ...] | None = tuple(POI_CATEGORIES),
    progress: ProgressCallback | None = None,
    tile_cache_dir: str | Path | None = None,
    use_tile_cache: bool = True,
) -> dict[str, Any]:
    """Backfill the amenities an extract is missing, in place.

    Only the amenity query is re-run: the road network already in the file is
    left untouched, which is what makes this cheap enough to run over a whole
    workspace. The report pairs the audit before and after so a caller can say
    what each city gained.
    """
    path = Path(osm_path)
    before = audit_extract(path)
    if not before.get("can_refresh"):
        return {
            "city": before["city"],
            "path": str(path),
            "ok": False,
            "error": "extract has no <bounds>; re-fetch the city instead",
            "before": before,
        }
    bounds = before["bounds"]
    summary = fetch_tiled_amenities(
        bounds["minlat"],
        bounds["minlon"],
        bounds["maxlat"],
        bounds["maxlon"],
        path,
        poi_categories=poi_categories,
        progress=progress,
        tile_cache_dir=tile_cache_dir,
        use_tile_cache=use_tile_cache,
    )
    after = audit_extract(path)
    return {
        "city": before["city"],
        "path": str(path),
        "ok": bool(summary.get("ok")),
        "tiles_total": summary.get("tiles_total"),
        "tiles_ok": summary.get("tiles_ok"),
        "failure_count": summary.get("failure_count"),
        "poi_total_before": before["poi_total"],
        "poi_total_after": after["poi_total"],
        "gained": after["poi_total"] - before["poi_total"],
        "gained_ways": after["poi_ways"] - before["poi_ways"],
        "gained_relations": after["poi_relations"] - before["poi_relations"],
        "status_before": before["status"],
        "status_after": after["status"],
    }

refresh_extracts

refresh_extracts(
    paths: list[Path],
    *,
    poi_categories: tuple[str, ...] | None = tuple(
        POI_CATEGORIES
    ),
    on_city: Callable[[int, int, Path], None] | None = None,
    progress: ProgressCallback | None = None,
    tile_cache_dir: str | Path | None = None,
    use_tile_cache: bool = True,
) -> dict[str, Any]

Refresh several extracts, reporting each one rather than aborting.

One city failing -- a timeout, an Overpass outage -- must not cost the others: the run continues and the failure is reported on its own row.

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

def refresh_extracts(
    paths: list[Path],
    *,
    poi_categories: tuple[str, ...] | None = tuple(POI_CATEGORIES),
    on_city: Callable[[int, int, Path], None] | None = None,
    progress: ProgressCallback | None = None,
    tile_cache_dir: str | Path | None = None,
    use_tile_cache: bool = True,
) -> dict[str, Any]:
    """Refresh several extracts, reporting each one rather than aborting.

    One city failing -- a timeout, an Overpass outage -- must not cost the
    others: the run continues and the failure is reported on its own row.
    """
    results: list[dict[str, Any]] = []
    for index, path in enumerate(paths):
        if on_city is not None:
            on_city(index, len(paths), path)
        try:
            results.append(
                refresh_extract_pois(
                    path,
                    poi_categories=poi_categories,
                    progress=progress,
                    tile_cache_dir=tile_cache_dir,
                    use_tile_cache=use_tile_cache,
                )
            )
        except Exception as error:  # noqa: BLE001 - one bad city must not end the run
            results.append(
                {"city": Path(path).stem, "path": str(path), "ok": False, "error": str(error)}
            )
    return {
        "ok": all(entry.get("ok") for entry in results),
        "refreshed": sum(1 for entry in results if entry.get("ok")),
        "gained": sum(int(entry.get("gained") or 0) for entry in results),
        "results": results,
    }

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,
    }

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,
    }