Streaming OSM XML parsing for the road engine.
Reads only what the road graph needs: node coordinates, ways with their tags
and node lists, and the <bounds> element. City extracts run to ~150 MB,
so parsing is incremental with element recycling.
ensure_bounds(osm_path: Path) -> None
Inject a <bounds> element computed from node coordinates when the
extract ships without one (some Overpass responses omit it). Matches the
Julia pipeline's behavior so both read identical files afterwards.
Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/osmxml.py
| def ensure_bounds(osm_path: Path) -> None:
"""Inject a ``<bounds>`` element computed from node coordinates when the
extract ships without one (some Overpass responses omit it). Matches the
Julia pipeline's behavior so both read identical files afterwards."""
with osm_path.open("r", encoding="utf-8") as handle:
head = handle.read(64 * 1024)
if re.search(r"<bounds\b", head):
return
min_lat = min_lon = float("inf")
max_lat = max_lon = float("-inf")
count = 0
for _, elem in ET.iterparse(osm_path, events=("start",)):
if elem.tag == "node":
lat = float(elem.attrib["lat"])
lon = float(elem.attrib["lon"])
min_lat, max_lat = min(min_lat, lat), max(max_lat, lat)
min_lon, max_lon = min(min_lon, lon), max(max_lon, lon)
count += 1
elem.clear()
if count == 0:
raise ValueError(f"OSM file '{osm_path}' contains no nodes; cannot derive bounds")
bounds_line = (
f' <bounds minlat="{min_lat}" minlon="{min_lon}" maxlat="{max_lat}" maxlon="{max_lon}"/>\n'
)
text = osm_path.read_text(encoding="utf-8")
match = re.search(r"<osm\b[^>]*>\n?", text)
if match is None:
raise ValueError(f"OSM file '{osm_path}' has no <osm> root element")
insert_at = match.end()
osm_path.write_text(text[:insert_at] + bounds_line + text[insert_at:], encoding="utf-8")
|
crop_to_bounds(osm_data: OsmData) -> None
Crop ways and nodes to the <bounds> rectangle, as
get_map_data does before graph construction. Mutates in place.
Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/osmxml.py
| def crop_to_bounds(osm_data: OsmData) -> None:
"""Crop ways and nodes to the ``<bounds>`` rectangle, as
``get_map_data`` does before graph construction. Mutates in place."""
min_lat, min_lon, max_lat, max_lon = osm_data.bounds
if min_lon > max_lon:
raise ValueError("Antimeridian-crossing bounds are not supported")
next_id = max(osm_data.nodes.keys(), default=0) + 1
def allocate_node_id() -> int:
nonlocal next_id
value = next_id
next_id += 1
osm_data.synthetic_nodes.add(value)
return value
osm_data.ways = [
way for way in osm_data.ways if not _crop_way(osm_data.nodes, osm_data.bounds, way, allocate_node_id)
]
osm_data.nodes = {
node_id: point for node_id, point in osm_data.nodes.items() if _inbounds(point, osm_data.bounds)
}
|