mamut_routing_tools.generation.select

Depot and customer selection on a road graph (port of the workbench selection methods: parametric attach, POI categories, hybrid).

vertex_latlon

vertex_latlon(graph: RoadGraph) -> list[tuple[float, float]]

Per-vertex (lat, lon), indexed by 0-based graph vertex.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/generation/select.py

def vertex_latlon(graph: RoadGraph) -> list[tuple[float, float]]:
    """Per-vertex (lat, lon), indexed by 0-based graph vertex."""
    out: list[tuple[float, float]] = []
    for osm_id in graph.node_of:
        lla = graph.node_lla(osm_id)
        out.append((lla.lat, lla.lon))
    return out

pick_depot_vertex

pick_depot_vertex(mode: str, vertex_ll: list[tuple[float, float]], rng: Random, exclude: set[int] | frozenset[int] = frozenset()) -> int

The depot vertex for mode, never one of exclude.

exclude is meant for the graph's synthetic crop nodes: corner takes the vertex nearest the south-west corner of the extract, which is exactly where a road leaving the bounds gets its invented boundary node, so 16 of the 100 Mamut2026 depots landed on a point that exists in no other extract of their city. center is guarded the same way; random redraws only when it hits an excluded vertex, so seeded draws that never touched one are unchanged. When every vertex is excluded the guard is dropped rather than failing the generation.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/generation/select.py

def pick_depot_vertex(
    mode: str,
    vertex_ll: list[tuple[float, float]],
    rng: random.Random,
    exclude: set[int] | frozenset[int] = frozenset(),
) -> int:
    """The depot vertex for ``mode``, never one of ``exclude``.

    ``exclude`` is meant for the graph's synthetic crop nodes: ``corner``
    takes the vertex nearest the south-west corner of the extract, which is
    exactly where a road leaving the bounds gets its invented boundary node,
    so 16 of the 100 Mamut2026 depots landed on a point that exists in no
    other extract of their city. ``center`` is guarded the same way; ``random``
    redraws only when it hits an excluded vertex, so seeded draws that never
    touched one are unchanged. When every vertex is excluded the guard is
    dropped rather than failing the generation.
    """
    n = len(vertex_ll)
    if n < 1:
        raise ValueError("Cannot pick depot from empty vertex list")
    candidates = [v for v in range(n) if v not in exclude] or list(range(n))
    if mode == "random":
        pick = rng.randrange(n)
        while pick not in candidates:
            pick = rng.randrange(n)
        return pick
    if mode == "center":
        c_lat = sum(t[0] for t in vertex_ll) / n
        c_lon = sum(t[1] for t in vertex_ll) / n
        return min(candidates, key=lambda v: haversine_m(vertex_ll[v][0], vertex_ll[v][1], c_lat, c_lon))
    if mode == "corner":
        min_lat = min(t[0] for t in vertex_ll)
        min_lon = min(t[1] for t in vertex_ll)
        return min(candidates, key=lambda v: haversine_m(vertex_ll[v][0], vertex_ll[v][1], min_lat, min_lon))
    raise ValueError(f"Unsupported depot mode '{mode}'")

attach_pois

attach_pois(graph: RoadGraph, pois: list[Poi], mode: str = POI_ATTACH_NEAREST_NODE, radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M) -> list[tuple[int, float] | None]

Per POI, the (vertex, snap_distance_m) it attaches to, or None.

nearest_node is the Julia findnode + md.v membership semantic: the nearest ROAD NODE must itself be a graph vertex, so a POI whose closest node sits mid-segment is dropped even when an intersection stands a metre further. It keeps generated instances identical to the published pipeline, at the cost of discarding most of a city's amenities.

nearest_vertex instead snaps each POI to the closest routable vertex within radius_m, which is what manual picking has always done. Far more of the pool becomes usable; the distance is returned so every customer can record how far its amenity was moved.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/generation/select.py

def attach_pois(
    graph: RoadGraph,
    pois: list[Poi],
    mode: str = POI_ATTACH_NEAREST_NODE,
    radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M,
) -> list[tuple[int, float] | None]:
    """Per POI, the ``(vertex, snap_distance_m)`` it attaches to, or ``None``.

    ``nearest_node`` is the Julia ``findnode`` + ``md.v`` membership semantic:
    the nearest ROAD NODE must itself be a graph vertex, so a POI whose closest
    node sits mid-segment is dropped even when an intersection stands a metre
    further. It keeps generated instances identical to the published pipeline,
    at the cost of discarding most of a city's amenities.

    ``nearest_vertex`` instead snaps each POI to the closest routable vertex
    within ``radius_m``, which is what manual picking has always done. Far more
    of the pool becomes usable; the distance is returned so every customer can
    record how far its amenity was moved.
    """
    if mode == POI_ATTACH_NEAREST_VERTEX:
        return graph.nearest_vertices([(poi.lat, poi.lon) for poi in pois], radius_m)
    if mode != POI_ATTACH_NEAREST_NODE:
        raise ValueError(f"Unsupported POI attach mode '{mode}'")
    nodes = graph.nearest_nodes([(poi.lat, poi.lon) for poi in pois])
    out: list[tuple[int, float] | None] = []
    for osm_id in nodes:
        vertex = graph.vertex_of.get(osm_id) if osm_id is not None else None
        # Zero by construction: the POI attached to the node it is nearest to.
        out.append(None if vertex is None else (vertex, 0.0))
    return out

catalog_attachment

catalog_attachment(graph: RoadGraph, osm_path: str | Path, mode: str = POI_ATTACH_NEAREST_NODE, radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M) -> list[tuple[int, float] | None]

attach_pois over the whole cached catalog, computed at most once.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/generation/select.py

def catalog_attachment(
    graph: RoadGraph,
    osm_path: str | Path,
    mode: str = POI_ATTACH_NEAREST_NODE,
    radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M,
) -> list[tuple[int, float] | None]:
    """``attach_pois`` over the whole cached catalog, computed at most once."""
    path = Path(osm_path)
    stat = path.stat()
    key = (
        str(path.resolve()),
        int(stat.st_mtime_ns),
        int(stat.st_size),
        _graph_identity(graph),
        mode,
        # Only the snapping mode reads the radius; folding it in unconditionally
        # would split the cache for a value the strict mode ignores.
        radius_m if mode == POI_ATTACH_NEAREST_VERTEX else None,
    )
    cached = _ATTACHMENT_CACHE.get(key)
    if cached is not None:
        _ATTACHMENT_CACHE.move_to_end(key)
        return cached
    attachment = attach_pois(graph, load_poi_catalog(path), mode, radius_m)
    _ATTACHMENT_CACHE[key] = attachment
    while len(_ATTACHMENT_CACHE) > _ATTACHMENT_CACHE_MAX:
        _ATTACHMENT_CACHE.popitem(last=False)
    return attachment

select_customers_poi

select_customers_poi(graph: RoadGraph, osm_path: str | Path, n_customers: int, categories: list[str], rng: Random, stats: dict[str, Any] | None = None, attach_mode: str = POI_ATTACH_NEAREST_NODE, attach_radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M, exclude: Iterable[int] | None = None) -> tuple[list[int], list[float], list[float], list[str], list[Poi | None]]

Chosen vertices plus, per vertex, the POI it came from.

The trailing list[Poi | None] carries the amenity value, OSM node id and display name so callers can persist them; None marks a non-POI customer.

exclude names vertices that must not become customers -- in practice the depot. Without it the caller has to drop a POI that landed on the depot after the fact, which silently costs one customer: the selection returns the n it was asked for, one is discarded, and the shortfall is made up with a sampled road point. A POI-only instance then quietly stops being one, even in a city with four times the amenities it needs.

stats, when given, is filled with how big the pool actually was: pool_matching (POIs of these categories in the extract), attached (the number of distinct graph vertices the whole pool can reach, which is the hard ceiling on POI customers) and co_located (per chosen vertex, the other POIs that snap to it and are therefore not customers of their own). Callers need the first two to explain a shortfall and the last to say which amenities a single customer stands for.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/generation/select.py

def select_customers_poi(
    graph: RoadGraph,
    osm_path: str | Path,
    n_customers: int,
    categories: list[str],
    rng: random.Random,
    stats: dict[str, Any] | None = None,
    attach_mode: str = POI_ATTACH_NEAREST_NODE,
    attach_radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M,
    exclude: Iterable[int] | None = None,
) -> tuple[list[int], list[float], list[float], list[str], list[Poi | None]]:
    """Chosen vertices plus, per vertex, the POI it came from.

    The trailing ``list[Poi | None]`` carries the amenity value, OSM node id and
    display name so callers can persist them; ``None`` marks a non-POI customer.

    ``exclude`` names vertices that must not become customers -- in practice the
    depot. Without it the caller has to drop a POI that landed on the depot
    *after* the fact, which silently costs one customer: the selection returns
    the ``n`` it was asked for, one is discarded, and the shortfall is made up
    with a sampled road point. A POI-only instance then quietly stops being one,
    even in a city with four times the amenities it needs.

    ``stats``, when given, is filled with how big the pool actually was:
    ``pool_matching`` (POIs of these categories in the extract), ``attached``
    (the number of distinct graph vertices the whole pool can reach, which is
    the hard ceiling on POI customers) and ``co_located`` (per chosen vertex,
    the other POIs that snap to it and are therefore *not* customers of their
    own). Callers need the first two to explain a shortfall and the last to say
    which amenities a single customer stands for.
    """
    cats = categories or DEFAULT_CATEGORIES
    # The cached catalog holds every known category in file order, so filtering
    # it gives exactly what find_pois would return without re-parsing an extract
    # that preview, preflight and generate each walk in turn -- and the cached
    # attachment map spares them the spatial search too. A category outside the
    # catalog's own list is not in there, so that case still parses and searches.
    if set(cats) <= set(POI_CATEGORIES):
        wanted = set(cats)
        catalog = load_poi_catalog(osm_path)
        attachment = catalog_attachment(graph, osm_path, attach_mode, attach_radius_m)
        selected = [index for index, poi in enumerate(catalog) if poi.category in wanted]
        pois = [catalog[index] for index in selected]
        poi_hits = [attachment[index] for index in selected]
    else:
        pois = find_pois(osm_path, list(cats))
        poi_hits = attach_pois(graph, pois, attach_mode, attach_radius_m)
    if stats is not None:
        stats.update(
            {"pool_matching": len(pois), "attached": 0, "co_located": {}, "snap_m": {}}
        )
    if not pois:
        raise NoPoiFoundError(f"No POI found for selected categories: {', '.join(cats)}")

    rows = list(range(len(pois)))
    rng.shuffle(rows)

    # Seeded rather than filtered afterwards: a vertex that must not be a
    # customer is simply never taken, so the walk goes on to the next POI and
    # still returns the full count.
    taken: set[int] = set(exclude or ())
    verts: list[int] = []
    poi_lats: list[float] = []
    poi_lons: list[float] = []
    sources: list[Poi | None] = []
    # Two amenities on the same street corner collapse into one customer. The
    # ones that lose are kept here, per winning vertex, so the instance can say
    # which places that single customer actually stands for.
    co_located: dict[int, list[Poi]] = {}
    attachable: set[int] = set()
    # How far each winning POI had to move to reach its vertex; zero under the
    # strict rule, which never moves a POI at all.
    snap_m: dict[int, float] = {}
    for i in rows:
        hit = poi_hits[i]
        if hit is None:
            continue
        v, distance = hit
        attachable.add(v)
        if v in taken:
            # Amenities standing on an excluded vertex are still worth naming --
            # the depot is usually a real place too -- so they are recorded the
            # same way as ones that lost a corner to another customer.
            co_located.setdefault(v, []).append(pois[i])
            continue
        # The pool keeps being walked past the requested count, which costs one
        # spatial lookup per remaining POI and buys two things the selection
        # cannot report otherwise: the true attachable ceiling, and the full set
        # of amenities standing behind each chosen point.
        if len(verts) >= n_customers:
            continue
        taken.add(v)
        verts.append(v)
        poi_lats.append(pois[i].lat)
        poi_lons.append(pois[i].lon)
        sources.append(pois[i])
        snap_m[v] = distance

    if stats is not None:
        stats.update(
            {"attached": len(attachable), "co_located": co_located, "snap_m": snap_m}
        )
    if len(verts) < n_customers:
        warnings.warn(
            f"Only found {len(verts)} POI-attached unique graph vertices; requested {n_customers}",
            stacklevel=2,
        )
    n_actual = min(n_customers, len(verts))
    return (
        verts[:n_actual],
        poi_lats[:n_actual],
        poi_lons[:n_actual],
        ["poi"] * n_actual,
        sources[:n_actual],
    )