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

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],
    )