mamut_routing_tools.generation.single

Single-instance generation driver (selection, demands, matrices, artifacts), the port of the workbench's build_generation_selection + generate_single_instance.

GenerationRequest dataclass

GenerationRequest(city: str, osm_path: Path, method: str = 'poi_categories', n_customers: int = 50, seed: int = 0, demand_type: int = 7, avg_route_size: int = 4, depot_mode: str = 'center', customer_mode: str = 'random_clustered', cluster_seeds: int | None = None, cluster_decay_meters: float = 800.0, categories: list[str] = (lambda: list(DEFAULT_CATEGORIES))(), hybrid_poi_share: float = 0.5, only_intersections: bool = True, trim_to_connected_graph: bool = True, poi_attach_mode: str = POI_ATTACH_NEAREST_NODE, poi_attach_radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M, manual_poi_ids: list[int] = list(), manual_depot_poi_id: int | None = None, manual_poi_refs: list[tuple[str, int]] = list(), manual_depot_poi_ref: tuple[str, int] | None = None)

picked_refs

picked_refs() -> list[tuple[str, int]]

Every hand-picked POI as a (osm_type, osm_id) pair, in pick order.

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

def picked_refs(self) -> list[tuple[str, int]]:
    """Every hand-picked POI as a ``(osm_type, osm_id)`` pair, in pick order."""
    refs = [(str(kind), int(value)) for kind, value in self.manual_poi_refs]
    seen = set(refs)
    for osm_id in self.manual_poi_ids:
        ref = ("node", int(osm_id))
        if ref not in seen:
            seen.add(ref)
            refs.append(ref)
    return refs

effective_method

effective_method(requested_method: str, customer_tags: list[str]) -> str

The method that describes what was actually selected.

A POI request the extract cannot serve is completed with parametric road points, which makes the instance a hybrid one -- or a parametric one when no amenity could be used at all. Naming it after the request would put _poi on a file whose customers sit on plain intersections, so the composition has the last word. Only the fallback direction is reclassified: a hybrid that reaches its POI target stays hybrid, because that mix was asked for.

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

def effective_method(requested_method: str, customer_tags: list[str]) -> str:
    """The method that describes what was actually selected.

    A POI request the extract cannot serve is completed with parametric road
    points, which makes the instance a hybrid one -- or a parametric one when no
    amenity could be used at all. Naming it after the request would put ``_poi``
    on a file whose customers sit on plain intersections, so the composition has
    the last word. Only the fallback direction is reclassified: a hybrid that
    reaches its POI target stays hybrid, because that mix was asked for.
    """
    if requested_method not in ("poi_categories", "hybrid"):
        return requested_method
    has_poi = any(is_poi_source_tag(tag) for tag in customer_tags)
    has_parametric = any(not is_poi_source_tag(tag) for tag in customer_tags)
    if not has_poi:
        return "parametric_attach"
    if requested_method == "poi_categories" and has_parametric:
        return "hybrid"
    return requested_method

graph_option_params

graph_option_params(request: GenerationRequest, graph: RoadGraph) -> dict[str, Any]

The graph options actually in force, plus the ones that were asked for.

load_road_graph relaxes only_intersections / trim_to_connected when the requested combination yields an empty graph, so recording the request would make an instance claim a graph it does not have. The effective values are the ones written to the metadata; the request is kept beside them and the difference is what the caller warns about.

graph.loaded_with is the loader's own record. A graph built without it went through no fallback chain, so the request stands.

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

def graph_option_params(request: GenerationRequest, graph: RoadGraph) -> dict[str, Any]:
    """The graph options actually in force, plus the ones that were asked for.

    ``load_road_graph`` relaxes ``only_intersections`` / ``trim_to_connected``
    when the requested combination yields an empty graph, so recording the
    request would make an instance claim a graph it does not have. The effective
    values are the ones written to the metadata; the request is kept beside them
    and the difference is what the caller warns about.

    ``graph.loaded_with`` is the loader's own record. A graph built without it
    went through no fallback chain, so the request stands.
    """
    requested = {
        "only_intersections": request.only_intersections,
        "trim_to_connected_graph": request.trim_to_connected_graph,
    }
    loaded = graph.loaded_with
    effective = (
        {"only_intersections": loaded[0], "trim_to_connected_graph": loaded[1]}
        if loaded is not None
        else dict(requested)
    )
    return {
        **effective,
        "requested_graph_options": requested,
        "graph_options_relaxed": requested if requested != effective else None,
    }

selection_composition

selection_composition(*, method: str, requested: int, customer_tags: list[str], poi_stats: dict[str, Any], poi_share: float, categories: list[str], graph_options_relaxed: dict[str, Any] | None = None, effective: str | None = None, attach_mode: str = POI_ATTACH_NEAREST_NODE) -> dict[str, Any]

What the selection really is made of, versus what was asked for.

A POI-backed method that runs out of amenities keeps going with parametric road points, which is the right fallback but changes the nature of the instance. This record is what lets the caller say so before generating: poi_shortfall is how many customers meant to sit on a real amenity did not, and poi_pool_attachable is the ceiling that caused it (None when the pool was never exhausted, so no ceiling was observed).

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

def selection_composition(
    *,
    method: str,
    requested: int,
    customer_tags: list[str],
    poi_stats: dict[str, Any],
    poi_share: float,
    categories: list[str],
    graph_options_relaxed: dict[str, Any] | None = None,
    effective: str | None = None,
    attach_mode: str = POI_ATTACH_NEAREST_NODE,
) -> dict[str, Any]:
    """What the selection really is made of, versus what was asked for.

    A POI-backed method that runs out of amenities keeps going with parametric
    road points, which is the right fallback but changes the nature of the
    instance. This record is what lets the caller say so before generating:
    ``poi_shortfall`` is how many customers meant to sit on a real amenity did
    not, and ``poi_pool_attachable`` is the ceiling that caused it (``None``
    when the pool was never exhausted, so no ceiling was observed).
    """
    poi_customers = sum(1 for tag in customer_tags if is_poi_source_tag(tag))
    delivered = len(customer_tags)
    if method == "poi_categories":
        poi_target = requested
    elif method == "hybrid":
        # Mirrors select_customers_hybrid, so the target matches what it aimed for.
        poi_target = min(requested, max(0, round(requested * poi_share)))
    else:
        poi_target = 0
    return {
        # The requested method: the advice below is about the controls the user
        # actually has. ``effective_method`` is what the instance is recorded as.
        "method": method,
        "effective_method": effective if effective is not None else method,
        "requested": requested,
        "delivered": delivered,
        "poi_customers": poi_customers,
        "parametric_customers": delivered - poi_customers,
        "poi_target": poi_target,
        "poi_shortfall": max(0, poi_target - poi_customers),
        "poi_share": poi_share if method == "hybrid" else None,
        "poi_pool_matching": int(poi_stats.get("pool_matching", 0)),
        # The whole pool is walked, so this is the real ceiling on POI customers
        # for these categories, not just where the search happened to stop.
        "poi_pool_attachable": (
            int(poi_stats["attached"]) if "attached" in poi_stats else None
        ),
        # A POI whose vertex is the depot cannot also be a customer; without this
        # the resulting shortfall looks like a missing amenity.
        "poi_used_as_depot": int(poi_stats.get("used_as_depot", 0)),
        # POIs the attachment rule discarded outright. Under the strict rule
        # this is most of a city, and it is the first thing to explain when a
        # user cannot find an amenity they know is in the extract.
        "poi_pool_unattached": max(
            0, int(poi_stats.get("pool_matching", 0)) - int(poi_stats.get("attached", 0))
        ),
        "poi_attach_mode": attach_mode,
        "category_count": len(categories) if poi_target else 0,
        "category_total": len(POI_CATEGORIES),
        "graph_options_relaxed": graph_options_relaxed,
    }

composition_notice

composition_notice(composition: dict[str, Any] | None) -> str | None

Everything about this selection the user did not ask for, or None.

Names the final split, the pool that capped it, and the ways out (fewer customers, more categories to draw from), plus any option the loader had to relax to build the graph at all.

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

def composition_notice(composition: dict[str, Any] | None) -> str | None:
    """Everything about this selection the user did not ask for, or ``None``.

    Names the final split, the pool that capped it, and the ways out (fewer
    customers, more categories to draw from), plus any option the loader had to
    relax to build the graph at all.
    """
    if not composition:
        return None
    relaxed = composition.get("graph_options_relaxed")
    relaxed_note = (
        "The requested road-graph options (only intersections="
        f"{relaxed['only_intersections']}, trim to connected graph="
        f"{relaxed['trim_to_connected_graph']}) yield an empty graph for this extract, so it is "
        "loaded with relaxed options; the instance records the ones actually used."
        if relaxed
        else None
    )
    if composition.get("method") == "manual":
        notes: list[str | None] = []
        missing = int(composition["requested"]) - int(composition["delivered"])
        if missing > 0:
            dropped = composition.get("dropped") or {}
            reasons = [
                f"{dropped[key]} {label}"
                for key, label in (
                    ("unknown", "not found in this extract"),
                    ("unreachable", "too far from any road"),
                    ("collapsed", "snapped onto an already-used intersection"),
                )
                if dropped.get(key)
            ]
            notes.append(
                f"{missing} of your {composition['requested']} picked POIs cannot become customers"
                + (f" ({', '.join(reasons)})" if reasons else "")
                + f". The instance will hold {composition['delivered']} customers."
            )
        fallback_mode = composition.get("depot_fell_back_to_mode")
        if fallback_mode:
            notes.append(
                "The POI marked as depot could not be resolved, so the depot falls back to the "
                f"'{fallback_mode}' rule."
            )
        notes.append(relaxed_note)
        return " ".join(note for note in notes if note) or None
    shortfall = int(composition.get("poi_shortfall") or 0)
    requested = int(composition["requested"])
    delivered = int(composition["delivered"])
    # A road graph too small for the request is fatal at generation time; saying
    # so here beats letting the job fail after the user waited for it.
    too_small = (
        f"Only {delivered} of the {requested} customers requested can be placed: the road graph "
        f"of this city does not offer enough distinct vertices. Generation will fail unless the "
        f"customer count drops to {delivered}."
        if delivered < requested
        else None
    )
    if shortfall <= 0:
        return " ".join(note for note in (too_small, relaxed_note) if note) or None
    poi = int(composition["poi_customers"])
    target = int(composition["poi_target"])
    pool = int(composition["poi_pool_matching"])
    ticked = int(composition["category_count"])
    total = int(composition["category_total"])
    ceiling = composition.get("poi_pool_attachable")
    effective = composition.get("effective_method")
    became = (
        f"The instance will be recorded and named as '{effective}' rather than "
        f"'{composition['method']}', since that is what its customers are."
        if effective and effective != composition["method"]
        else None
    )
    if pool == 0:
        # Nothing to draw from at all: the POI share and the customer count are
        # both beside the point, so neither is offered as a remedy.
        return " ".join(
            note
            for note in (
                too_small,
                f"None of the {ticked} selected "
                + ("category has" if ticked == 1 else "categories have")
                + f" a POI in this extract, so all {delivered} customers are parametric road "
                "points.",
                f"Select other POI categories ({ticked} of {total} selected), or re-fetch the "
                "city with its amenities.",
                became,
                relaxed_note,
            )
            if note
        )
    parts = [
        *([too_small] if too_small else []),
        f"Only {poi} of the {target} POI customers this method asked for sit on a real amenity; "
        f"the remaining {shortfall} are parametric road points. "
        f"Final instance: {delivered} customers = {poi} POI + "
        f"{int(composition['parametric_customers'])} parametric."
    ]
    ceiling_text = (
        f", of which {ceiling} attach to a distinct road intersection"
        if ceiling is not None
        else ""
    )
    as_depot = int(composition.get("poi_used_as_depot") or 0)
    depot_text = (
        f" {as_depot} matching POI(s) became the depot rather than a customer."
        if as_depot
        else ""
    )
    parts.append(
        f"The {ticked} selected "
        + (f"category holds {pool}" if ticked == 1 else f"categories hold {pool}")
        + f" POI(s) in this extract{ceiling_text}.{depot_text}"
    )
    options = []
    if ceiling:
        share = composition.get("poi_share")
        max_customers = (
            int(ceiling / share) if composition["method"] == "hybrid" and share else int(ceiling)
        )
        if max_customers < requested:
            options.append(f"reduce the customer count to {max_customers}")
    if composition["method"] == "hybrid" and composition.get("poi_share"):
        options.append("lower the POI share")
    unattached = int(composition.get("poi_pool_unattached") or 0)
    if unattached and composition.get("poi_attach_mode") == POI_ATTACH_NEAREST_NODE:
        # By far the largest source of "but that amenity is in the extract".
        options.append(
            f"switch POI attachment to 'nearest vertex' ({unattached} POI(s) here sit closest "
            "to a mid-segment road node and are discarded by the strict rule)"
        )
    options.append(f"select more POI categories ({ticked} of {total} selected)")
    joined = options[0] if len(options) == 1 else ", ".join(options[:-1]) + f" or {options[-1]}"
    parts.append(f"To keep every customer on a POI, {joined}.")
    parts.append(became)
    parts.append(relaxed_note)
    return " ".join(part for part in parts if part)

poi_attributes

poi_attributes(selection: Selection, index: int) -> tuple[str | None, int | None, str | None, str | None]

(category, osm_id, name, osm_type) of the POI behind a selected vertex.

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

def poi_attributes(
    selection: Selection, index: int
) -> tuple[str | None, int | None, str | None, str | None]:
    """``(category, osm_id, name, osm_type)`` of the POI behind a selected vertex."""
    poi = selection.poi_meta[index] if index < len(selection.poi_meta) else None
    if poi is None:
        return None, None, None, None
    return poi.category, poi.osm_id, poi.name, poi.osm_type

merged_poi_records

merged_poi_records(selection: Selection, index: int) -> list[dict[str, Any]]

The other amenities standing on this node, as plain serialisable rows.

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

def merged_poi_records(selection: Selection, index: int) -> list[dict[str, Any]]:
    """The other amenities standing on this node, as plain serialisable rows."""
    merged = selection.poi_merged[index] if index < len(selection.poi_merged) else []
    return [
        {
            "name": poi.name,
            "category": poi.category,
            "osm_id": poi.osm_id,
            "osm_type": poi.osm_type,
        }
        for poi in merged
    ]

poi_attribute_columns

poi_attribute_columns(selection: Selection, count: int) -> tuple[list[str | None], list[int | None], list[str | None], list[str | None], list[list[dict[str, Any]]], list[float]]

The six per-node POI columns in one pass over selection.

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

def poi_attribute_columns(
    selection: Selection, count: int
) -> tuple[
    list[str | None],
    list[int | None],
    list[str | None],
    list[str | None],
    list[list[dict[str, Any]]],
    list[float],
]:
    """The six per-node POI columns in one pass over ``selection``."""
    categories: list[str | None] = []
    osm_ids: list[int | None] = []
    names: list[str | None] = []
    osm_types: list[str | None] = []
    merged: list[list[dict[str, Any]]] = []
    snaps: list[float] = []
    for index in range(count):
        category, osm_id, name, osm_type = poi_attributes(selection, index)
        categories.append(category)
        osm_ids.append(osm_id)
        names.append(name)
        osm_types.append(osm_type)
        merged.append(merged_poi_records(selection, index))
        snaps.append(
            selection.snap_distances_m[index]
            if index < len(selection.snap_distances_m)
            else 0.0
        )
    return categories, osm_ids, names, osm_types, merged, snaps

materialize_instance

materialize_instance(selection: Selection, output_root: str | Path, *, demand_type: int, avg_route_size: int, rng: Random, precomputed: dict[str, Any] | None = None, name_suffix: str = '', reserved_names: set[str] | None = None) -> dict[str, Any]

Write the 3-metric .vrp set, _meta.json, _manifest.json and .vrp.json files for a selection; precomputed lets the bulk driver pass sliced matrices/geometry instead of recomputing them, and name_suffix keeps same-parameter-different-seed instances from overwriting one another.

reserved_names is the set of instance paths a caller has already written in this run. The final name is only known here -- it encodes route_count, which follows from the drawn demands rather than from any request field -- so this is the only place that can guarantee two instances of one run never land on the same files. Names claimed by earlier runs are untouched, keeping regeneration of the same configuration idempotent.

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

def materialize_instance(
    selection: Selection,
    output_root: str | Path,
    *,
    demand_type: int,
    avg_route_size: int,
    rng: random.Random,
    precomputed: dict[str, Any] | None = None,
    name_suffix: str = "",
    reserved_names: set[str] | None = None,
) -> dict[str, Any]:
    """Write the 3-metric .vrp set, _meta.json, _manifest.json and .vrp.json
    files for a selection; ``precomputed`` lets the bulk driver pass sliced
    matrices/geometry instead of recomputing them, and ``name_suffix`` keeps
    same-parameter-different-seed instances from overwriting one another.

    ``reserved_names`` is the set of instance paths a caller has already written
    in this run. The final name is only known here -- it encodes ``route_count``,
    which follows from the drawn demands rather than from any request field -- so
    this is the only place that can guarantee two instances of one run never land
    on the same files. Names claimed by *earlier* runs are untouched, keeping
    regeneration of the same configuration idempotent."""
    graph = selection.graph
    params = selection.params
    vertices = selection.vertices
    ref = graph.ref_lla

    customer_ll = list(zip(selection.poi_lats[1:], selection.poi_lons[1:]))
    demand_values, sum_demands, _max_demand, r = generate_demands(rng, customer_ll, demand_type, avg_route_size)
    demands = [0, *demand_values]
    capacity = capacity_from_avg_route_size(r, demand_values)
    # The exact bin-packing minimum, the way CVRPLIB derives the ``k`` in
    # ``X-n101-k25`` -- not ``ceil(sum / capacity)``, which is only the
    # continuous relaxation of it and can understate the fleet by a wide margin
    # when the demands do not divide the capacity. It is a *lower bound* on a
    # solution's route count, not a cap: ``num_vehicles`` stays unset.
    fleet = minimum_bins(demand_values, capacity)
    route_count = fleet.lower

    if precomputed is None:
        d_short, d_fast, geom_short, geom_fast = compute_matrices(graph, vertices)
        d_eucl, coords = euclidean_matrix_from_vertices(graph, vertices)
    else:
        d_short = precomputed["d_short"]
        d_fast = precomputed["d_fast"]
        geom_short = precomputed["geom_short"]
        geom_fast = precomputed["geom_fast"]
        d_eucl = precomputed["d_eucl"]
        coords = precomputed["coords"]

    n_customers = len(vertices) - 1
    folder, base = instance_path_plan(
        str(params["city"]),
        str(params["method"]),
        n_customers,
        route_count,
        output_root,
        name_suffix,
    )
    if reserved_names is not None:
        base = _claim_unique_name(folder, base, reserved_names)
    folder.mkdir(parents=True, exist_ok=True)

    files = {
        "shortest": f"{base}_shortest.vrp",
        "fastest": f"{base}_fastest.vrp",
        "euclidean": f"{base}_euclidean.vrp",
        "meta": f"{base}_meta.json",
    }
    manifest_name = f"{base}_manifest.json"
    ref_str = f"LLA({ref.lat}, {ref.lon}, {ref.alt})"
    write_cvrplib(folder / files["shortest"], f"{base}_shortest", f"Shortest distances; ENU ref: {ref_str}", coords, demands, d_short, capacity)
    write_cvrplib(folder / files["fastest"], f"{base}_fastest", f"Fastest distances; ENU ref: {ref_str}", coords, demands, d_fast, capacity)
    write_cvrplib(folder / files["euclidean"], f"{base}_euclidean", f"Euclidean distances; ENU ref: {ref_str}", coords, demands, d_eucl, capacity)

    generation_params = dict(params)
    generation_params["demand_type"] = demand_type
    generation_params["avg_route_size"] = avg_route_size
    (
        poi_categories,
        poi_osm_ids,
        poi_names,
        poi_osm_types,
        poi_merged,
        poi_snaps,
    ) = poi_attribute_columns(selection, len(vertices))
    write_instance_metadata(
        folder / files["meta"],
        city=str(params["city"]),
        osm_file=str(params["osm_path"]),
        instance_name=base,
        metric_files=[files["shortest"], files["fastest"], files["euclidean"]],
        ref_lla=ref,
        vertices=vertices,
        poi_lats=selection.poi_lats,
        poi_lons=selection.poi_lons,
        coords=coords,
        demands=demands,
        method=str(params["method"]),
        source_tags=selection.source_tags,
        only_intersections=bool(params["only_intersections"]),
        trim_to_connected_graph=bool(params["trim_to_connected_graph"]),
        generation_params=generation_params,
        poi_categories=poi_categories,
        poi_osm_ids=poi_osm_ids,
        poi_names=poi_names,
        poi_osm_types=poi_osm_types,
        poi_merged=poi_merged,
        snap_distances_m=poi_snaps,
        road_cache={"shortest": geom_short, "fastest": geom_fast},
    )

    generated_at = datetime.now().isoformat()
    place_slug = slugify(str(params["city"]))
    vrp_json_files: dict[str, str] = {}
    for metric in ("shortest", "fastest", "euclidean"):
        parsed = parse_cvrp_vrp(folder / files[metric])
        json_name = files[metric].replace(".vrp", ".vrp.json")
        payload = build_vrp_json_payload(
            parsed,
            instance_name=f"{base}_{metric}",
            metric_variant=metric,
            place_slug=place_slug,
            source_base_name=base,
            source_city=str(params["city"]),
            source_seed=int(params["seed"]),
            source_folder=str(folder),
            num_vehicles_lb=route_count,
            artifact_paths={m: files[m] for m in files},
            sibling_variant_paths={
                m: files[m].replace(".vrp", ".vrp.json") for m in ("shortest", "fastest", "euclidean")
            },
            reference_lla={"lat": ref.lat, "lon": ref.lon, "alt": ref.alt},
            generated_at=generated_at,
        )
        write_json(folder / json_name, payload)
        vrp_json_files[metric] = json_name

    manifest = {
        "generated_at": generated_at,
        "base_name": base,
        "folder": str(folder),
        "files": {**files, "vrp_json": vrp_json_files},
        "params": generation_params,
        "generator": "mamut-routing-tools",
        "demand_type": demand_type,
        "avg_route_size": avg_route_size,
        "route_count": route_count,
        "route_count_proven": fleet.proven,
        "route_count_method": fleet.method,
        "capacity": capacity,
        "total_demand": sum_demands,
    }
    write_json(folder / manifest_name, manifest)

    poi_count = sum(1 for tag in selection.source_tags[1:] if is_poi_source_tag(tag))
    composition = params.get("composition")
    return {
        "ok": True,
        "base_name": base,
        "folder": str(folder),
        "files": manifest["files"],
        "manifest": manifest_name,
        # Not None only when the selection could not be served as asked; the
        # caller shows it so a parametric top-up never passes unnoticed.
        "notice": composition_notice(composition),
        "summary": {
            "customers": n_customers,
            "composition": composition,
            "capacity": capacity,
            "total_demand": sum_demands,
            "method": str(params["method"]),
            "demand_type": demand_type,
            "avg_route_size": avg_route_size,
            "route_count": route_count,
            "route_count_proven": fleet.proven,
            "route_count_method": fleet.method,
            "poi_customers": poi_count,
            "parametric_customers": n_customers - poi_count,
        },
    }