mamut_routing_tools.roadgraph.build

Road-graph construction: a direct Python port of OpenStreetMapX.jl's get_map_data as the MAMUT Julia pipeline used it.

The port preserves the exact graph semantics rather than approximating them with a generic OSM library: the same way filter (ROAD_CLASSES + visible + not services), the same oneway/reverse rules, the same intersection-splitting with ENU Euclidean segment lengths, the same largest-strongly-connected- component trim (computed on the full node-level graph, then rebuilt at intersection granularity), and the same option-fallback chain. Graph parity with the Julia engine is then a floating-point question, not a modeling one.

EmptyRoadGraphError

Bases: ValueError

The requested options produced no usable road graph (triggers the option-fallback chain, mirroring the Julia ArgumentError path).

RoadGraph dataclass

RoadGraph(osm_path: Path, only_intersections: bool, trim_to_connected: bool, ref_lla: LLA, node_enu: dict[int, tuple[float, float, float]], vertex_of: dict[int, int], node_of: list[int], edges: list[tuple[int, int]], edge_class: list[int], edge_weight: list[float], graph: PyDiGraph, loaded_with: tuple[bool, bool] | None = None, synthetic_nodes: frozenset[int] = frozenset(), _kdtree: cKDTree | None = None, _kdtree_nodes: list[int] = list(), _vertex_kdtree: cKDTree | None = None, _segments: tuple[Any, Any, Any] | None = None)

synthetic_vertices

synthetic_vertices() -> set[int]

Graph vertices that are synthetic crop nodes.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/build.py

def synthetic_vertices(self) -> set[int]:
    """Graph vertices that are synthetic crop nodes."""
    return {self.vertex_of[osm_id] for osm_id in self.synthetic_nodes if osm_id in self.vertex_of}

nearest_node

nearest_node(lat: float, lon: float, node_range_m: float = 100.0) -> int | None

Nearest road node (any node, not only graph vertices), matching OSMToolset's NodeSpatIndex/findnode contract exactly: candidate inclusion is an axis-aligned +-range box (L-infinity) around the query point, and the pick is the minimum EUCLIDEAN distance among the box hits. A node at 102 m Euclidean but inside the box still matches; one at 99 m outside the box does not exist geometrically.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/build.py

def nearest_node(self, lat: float, lon: float, node_range_m: float = 100.0) -> int | None:
    """Nearest road node (any node, not only graph vertices), matching
    OSMToolset's NodeSpatIndex/findnode contract exactly: candidate
    inclusion is an axis-aligned +-range box (L-infinity) around the
    query point, and the pick is the minimum EUCLIDEAN distance among
    the box hits. A node at 102 m Euclidean but inside the box still
    matches; one at 99 m outside the box does not exist geometrically."""
    self._ensure_node_kdtree()
    assert self._kdtree is not None
    enu = enu_from_lla(LLA(lat, lon), self.ref_lla)
    query = np.array([enu.east, enu.north])
    hits = self._kdtree.query_ball_point(query, r=node_range_m, p=np.inf)
    if not hits:
        return None
    best_index = min(hits, key=lambda i: float(np.sum((self._kdtree.data[i] - query) ** 2)))
    return self._kdtree_nodes[int(best_index)]

nearest_nodes

nearest_nodes(points: list[tuple[float, float]], node_range_m: float = 100.0) -> list[int | None]

nearest_node for many points at once, same contract per point.

One kd-tree call for the whole batch instead of one per point: the POI selection asks this for every amenity in a city, where the per-call overhead dominates the search itself. Results are identical to calling nearest_node in a loop, including the +-range box semantics.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/build.py

def nearest_nodes(
    self, points: list[tuple[float, float]], node_range_m: float = 100.0
) -> list[int | None]:
    """``nearest_node`` for many points at once, same contract per point.

    One kd-tree call for the whole batch instead of one per point: the POI
    selection asks this for every amenity in a city, where the per-call
    overhead dominates the search itself. Results are identical to calling
    ``nearest_node`` in a loop, including the ``+-range`` box semantics.
    """
    if not points:
        return []
    self._ensure_node_kdtree()
    assert self._kdtree is not None
    query = self._enu_east_north(points)
    hit_lists = self._kdtree.query_ball_point(query, r=node_range_m, p=np.inf)
    out: list[int | None] = []
    for index, hits in enumerate(hit_lists):
        if len(hits) == 0:
            out.append(None)
            continue
        candidates = np.asarray(hits, dtype=int)
        deltas = self._kdtree.data[candidates] - query[index]
        best = candidates[int(np.argmin(np.sum(deltas * deltas, axis=1)))]
        out.append(self._kdtree_nodes[int(best)])
    return out

nearest_vertex

nearest_vertex(lat: float, lon: float, max_radius_m: float = 2000.0) -> tuple[int, float] | None

Nearest GRAPH VERTEX to a point, as (vertex, distance_m).

Deliberately different from :meth:nearest_node: that one indexes every road node and honours the Julia findnode box contract, so a POI whose closest node sits mid-segment resolves to something absent from vertex_of. Manual picks must always land on a routable vertex, so this indexes node_of only and searches by true Euclidean distance, growing the radius until it finds one or gives up at max_radius_m.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/build.py

def nearest_vertex(
    self, lat: float, lon: float, max_radius_m: float = 2000.0
) -> tuple[int, float] | None:
    """Nearest GRAPH VERTEX to a point, as ``(vertex, distance_m)``.

    Deliberately different from :meth:`nearest_node`: that one indexes every
    road node and honours the Julia findnode box contract, so a POI whose
    closest node sits mid-segment resolves to something absent from
    ``vertex_of``. Manual picks must always land on a routable vertex, so
    this indexes ``node_of`` only and searches by true Euclidean distance,
    growing the radius until it finds one or gives up at ``max_radius_m``.
    """
    if not self.node_of:
        return None
    self._ensure_vertex_kdtree()
    assert self._vertex_kdtree is not None
    enu = enu_from_lla(LLA(lat, lon), self.ref_lla)
    query = np.array([enu.east, enu.north])
    distance, index = self._vertex_kdtree.query(query, k=1)
    if not np.isfinite(distance) or float(distance) > max_radius_m:
        return None
    return int(index), float(distance)

nearest_vertices

nearest_vertices(points: list[tuple[float, float]], max_radius_m: float = 2000.0) -> list[tuple[int, float] | None]

nearest_vertex for many points at once, same contract per point.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/build.py

def nearest_vertices(
    self, points: list[tuple[float, float]], max_radius_m: float = 2000.0
) -> list[tuple[int, float] | None]:
    """``nearest_vertex`` for many points at once, same contract per point."""
    if not points or not self.node_of:
        return [None] * len(points)
    self._ensure_vertex_kdtree()
    assert self._vertex_kdtree is not None
    query = self._enu_east_north(points)
    distances, indices = self._vertex_kdtree.query(query, k=1)
    return [
        None
        if not np.isfinite(distance) or float(distance) > max_radius_m
        else (int(index), float(distance))
        for distance, index in zip(distances, indices)
    ]

nearest_edge_projection

nearest_edge_projection(lat: float, lon: float, max_distance_m: float = 50.0) -> tuple[int, float, float] | None

Nearest EDGE to a point, as (edge_index, fraction, distance_m).

fraction is the position of the orthogonal projection along the directed edge (0 at its tail, 1 at its head). Only node-level graphs (only_intersections=False) qualify: their edges are the straight segments between consecutive OSM nodes, whereas an intersection-level edge stands for a whole polyline whose bends this graph no longer knows. Returns None on an intersection-level graph or when no edge passes within max_distance_m.

This is the escape hatch for a point that is on a road but far from every road node: a long straight segment, or a synthetic crop node of another extract of the same city (the generation-time bounds cut the road there; a re-fetched extract does not).

Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/build.py

def nearest_edge_projection(
    self, lat: float, lon: float, max_distance_m: float = 50.0
) -> tuple[int, float, float] | None:
    """Nearest EDGE to a point, as ``(edge_index, fraction, distance_m)``.

    ``fraction`` is the position of the orthogonal projection along the
    directed edge (0 at its tail, 1 at its head). Only node-level graphs
    (``only_intersections=False``) qualify: their edges are the straight
    segments between consecutive OSM nodes, whereas an intersection-level
    edge stands for a whole polyline whose bends this graph no longer
    knows. Returns ``None`` on an intersection-level graph or when no
    edge passes within ``max_distance_m``.

    This is the escape hatch for a point that is on a road but far from
    every road node: a long straight segment, or a synthetic crop node of
    another extract of the same city (the generation-time bounds cut the
    road there; a re-fetched extract does not).
    """
    if self.only_intersections or not self.edges:
        return None
    self._ensure_segments()
    assert self._segments is not None
    tails, deltas, lengths_sq = self._segments
    enu = enu_from_lla(LLA(lat, lon), self.ref_lla)
    query = np.array([enu.east, enu.north])
    fractions = np.clip(np.sum((query - tails) * deltas, axis=1) / lengths_sq, 0.0, 1.0)
    offsets = tails + fractions[:, None] * deltas - query
    distances = np.hypot(offsets[:, 0], offsets[:, 1])
    best = int(np.argmin(distances))
    if float(distances[best]) > max_distance_m:
        return None
    return best, float(fractions[best]), float(distances[best])

load_road_graph

load_road_graph(osm_path: str | Path, *, only_intersections: bool = True, trim_to_connected: bool = True) -> RoadGraph

Load with the same progressive option relaxation as the Julia servers: requested options first, then (oi, False), (False, trim), (False, False).

Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/build.py

def load_road_graph(
    osm_path: str | Path,
    *,
    only_intersections: bool = True,
    trim_to_connected: bool = True,
) -> RoadGraph:
    """Load with the same progressive option relaxation as the Julia servers:
    requested options first, then (oi, False), (False, trim), (False, False)."""
    resolved = Path(osm_path).resolve()
    key = (resolved, only_intersections, trim_to_connected)
    if key in _GRAPH_CACHE:
        return _GRAPH_CACHE[key]
    osm_data = _load_osm_data(resolved)

    attempts: list[tuple[bool, bool]] = []
    for option in ((only_intersections, trim_to_connected), *_FALLBACK_CHAIN):
        if option not in attempts:
            attempts.append(option)
    last_error: Exception | None = None
    for oi, trim in attempts:
        try:
            graph = build_road_graph(osm_data, resolved, only_intersections=oi, trim_to_connected=trim)
        except EmptyRoadGraphError as error:
            last_error = error
            continue
        graph.loaded_with = (oi, trim)
        _GRAPH_CACHE[(resolved, oi, trim)] = graph
        _GRAPH_CACHE[key] = graph
        return graph
    raise EmptyRoadGraphError(
        f"OSM file '{resolved}' produced an empty road graph for every option combination"
    ) from last_error

road_graph_candidates

road_graph_candidates(osm_path: str | Path, *, only_intersections: bool = True, trim_to_connected: bool = True) -> list[RoadGraph]

Candidate graphs in the cascade order the render path probes them.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/roadgraph/build.py

def road_graph_candidates(
    osm_path: str | Path,
    *,
    only_intersections: bool = True,
    trim_to_connected: bool = True,
) -> list[RoadGraph]:
    """Candidate graphs in the cascade order the render path probes them."""
    options: list[tuple[bool, bool]] = []
    for option in (
        (only_intersections, trim_to_connected),
        (False, trim_to_connected),
        (only_intersections, False),
        (False, False),
    ):
        if option not in options:
            options.append(option)
    candidates: list[RoadGraph] = []
    seen_ids: set[int] = set()
    for oi, trim in options:
        try:
            graph = load_road_graph(osm_path, only_intersections=oi, trim_to_connected=trim)
        except (EmptyRoadGraphError, OSError):
            continue
        if id(graph) not in seen_ids:
            seen_ids.add(id(graph))
            candidates.append(graph)
    return candidates