mamut_routing_tools.campaign

Designed-diversity benchmark campaigns.

Generating a benchmark is not the same problem as generating an instance. An instance generator answers "make me one of these"; a campaign has to answer "which ones, so that the set as a whole is worth reporting results on".

The pieces, in the order a campaign uses them:

• :mod:~.city_profile measures how far each city's road network departs from the Euclidean plane, and strata the cities by it.
• :mod:~.design enumerates a balanced candidate pool and evaluates each candidate cheaply -- selection and demands only, no distance matrices.
• :mod:~.descriptors turns coordinates and demands into the features that say what an instance is like, and (after generation) how far its road metrics diverge from its Euclidean one.
• :mod:~.select keeps the subset that spans the feature space while honouring the coverage quotas in :mod:~.quotas.
• :mod:~.poi_capacity measures the ceiling on POI-only customers per city, which is what decides whether a large real-amenity instance is buildable there at all.

Generation itself stays where it was: generation.single for the artifacts and family.build_base for the published collection.

CityProfile dataclass

CityProfile(
    city: str,
    osm_file: str,
    num_vertices: int,
    num_edges: int,
    detour_mean: float,
    detour_p90: float,
    rank_tau_eucl_fast: float,
    num_pairs: int,
    distortion_stratum: str | None = None,
)

One city's road-network distortion, estimated over random node pairs.

InstanceDescriptors dataclass

InstanceDescriptors(
    n: int,
    extent_m: float,
    clark_evans_r: float,
    nnd_cv: float,
    radial_dispersion: float,
    depot_centrality: float,
    depot_eccentricity: float,
    demand_cv: float,
    demand_gini: float,
    demand_max_over_mean: float,
    demand_moran_i: float,
    lb_cap: int,
    route_size: float,
    capacity_slack: float,
)

What an instance looks like, from coordinates and demands alone.

MetricDivergence dataclass

MetricDivergence(
    detour_mean: float,
    detour_p90: float,
    asymmetry_shortest: float,
    asymmetry_fastest: float,
    rank_tau_eucl_short: float,
    rank_tau_eucl_fast: float,
    rank_tau_short_fast: float,
)

How far an instance's road metrics depart from its Euclidean one.

CandidateSpec dataclass

CandidateSpec(
    city: str,
    osm_path: Path,
    n: int,
    method: str,
    demand_type: int,
    avg_route_size: int,
    depot_mode: str,
    customer_mode: str,
    cluster_seeds: int,
    cluster_decay_meters: float,
    hybrid_poi_share: float,
    poi_attach_mode: str = CAMPAIGN_POI_ATTACH_MODE,
    categories: tuple[str, ...] = CAMPAIGN_POI_CATEGORIES,
    distortion_stratum: str | None = None,
    tier: str = "main",
    rung_key: str = "",
)

A complete, deterministic recipe for one base instance.

EvaluatedCandidate dataclass

EvaluatedCandidate(
    spec: CandidateSpec,
    descriptors: InstanceDescriptors,
    resolved_n: int,
    effective_method: str = "",
    poi_fraction: float = 0.0,
)

A candidate together with what it would look like if generated.

is_usable

is_usable() -> bool

Servable at the requested size, a genuine VRP, and fully described.

The size check is not pedantry. A city whose amenities are sparse serves a POI request short -- Nairobi's extract has 27 amenities in total -- and the generator only warns. Accepting the short instance would put a candidate selected for the 701-1000 bucket into the tree at n=198, which is exactly the kind of silent drift a designed benchmark must not have.

Neither is the route-count check. This gate read lb_cap >= 2 in v2 -- "not a TSP" -- and 22 of the 110 published bases cleared it while still being trivially partitioned, 19 of them into exactly two routes. Two is not a fleet. The band admissibility rule in :mod:~.ladder is what makes the guarantee hold by construction; this is the backstop that catches the one case it cannot, where Q = ceil(r * mean demand) rounds up far enough to cost a route at the very edge of a band.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/design.py

def is_usable(self) -> bool:
    """Servable at the requested size, a genuine VRP, and fully described.

    The size check is not pedantry. A city whose amenities are sparse serves
    a POI request short -- Nairobi's extract has 27 amenities in total -- and
    the generator only warns. Accepting the short instance would put a
    candidate selected for the 701-1000 bucket into the tree at n=198, which
    is exactly the kind of silent drift a designed benchmark must not have.

    Neither is the route-count check. This gate read ``lb_cap >= 2`` in v2 --
    "not a TSP" -- and 22 of the 110 published bases cleared it while still
    being trivially partitioned, 19 of them into exactly two routes. Two is
    not a fleet. The band admissibility rule in :mod:`~.ladder` is what makes
    the guarantee hold by construction; this is the backstop that catches the
    one case it cannot, where ``Q = ceil(r * mean demand)`` rounds up far
    enough to cost a route at the very edge of a band.
    """
    return (
        self.resolved_n >= self.spec.n
        and self.descriptors.lb_cap >= CAMPAIGN_MIN_ROUTES
        and all(isfinite(v) for v in self.feature_vector())
    )

LadderPlan dataclass

LadderPlan(
    assignments: list[RungAssignment] = list(),
    unfilled: list[UnfilledRung] = list(),
    method_weights: dict[str, float] = dict(),
    headroom: float = DEFAULT_HEADROOM,
    min_routes: int = CAMPAIGN_MIN_ROUTES,
)

The complete size/method/city assignment, plus what it could not fill.

band_size_corr property

band_size_corr: float

corr(log n, band) over the plan -- the decorrelation guard's result.

Reported rather than asserted: admissibility puts a hard floor under it (band 6 cannot appear below n = 300), so the honest thing is to publish what the assignment achieved, not to pretend the axes are independent.

guaranteed_min_routes property

guaranteed_min_routes: float

Fewest routes any assigned rung can produce, over every legal draw.

RungAssignment dataclass

RungAssignment(
    index: int,
    n: int,
    method: str,
    avg_route_size: int,
    city: str,
    osm_file: str,
    distortion_stratum: str | None,
    required: int,
    available: int,
)

One ladder position, and the city that will supply it.

rung_key property

rung_key: str

Stable identifier for quota'ing exactly one instance per rung.

The route-size band belongs in it. The key is what the candidate-pool cache is fingerprinted on, and a plan that differs only in its bands is a different design producing different demands and capacities -- but would hash identically without this, and silently reuse the other one's pool.

UnfilledRung dataclass

UnfilledRung(
    index: int,
    n: int,
    method: str,
    required: int,
    best_available: int,
    best_city: str | None,
)

A rung no remaining city could supply, and by how much it fell short.

PoiCapacity dataclass

PoiCapacity(
    city: str,
    osm_file: str,
    num_vertices: int,
    catalog_size: int,
    attached_pois: int,
    capacity: int,
    collapse_ratio: float,
    attach_mode: str,
    attach_radius_m: float,
    categories_digest: str = "",
)

One city's ceiling on POI-only customers.

supports

supports(n: int) -> bool

Whether a POI-only instance of n customers plus a depot fits.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/poi_capacity.py

def supports(self, n: int) -> bool:
    """Whether a POI-only instance of ``n`` customers plus a depot fits."""
    return self.capacity >= n + 1

Quota dataclass

Quota(
    name: str,
    key: Callable[[EvaluatedCandidate], Hashable],
    minimum: Mapping[Hashable, int] = (),
    maximum: Mapping[Hashable, int] | None = None,
    per_level_maximum: int | None = None,
)

A coverage requirement over one design axis.

key maps a candidate to the level it occupies (its demand type, its city, its size bucket). minimum is how many of each level the selection must end up with; maximum caps a level -- {city: 1} style caps are written with :meth:per_level_maximum.

QuotaInfeasibleError

Bases: ValueError

The pool cannot satisfy the quotas within the requested budget.

assign_strata

assign_strata(
    profiles: Sequence[CityProfile],
) -> list[CityProfile]

Label each profile low / mid / high by tercile of detour_mean.

Terciles of the observed set, not fixed thresholds: what counts as a high-distortion city is only meaningful relative to the cities on hand, and absolute cut-offs would silently empty a stratum on a different city list.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/city_profile.py

def assign_strata(profiles: Sequence[CityProfile]) -> list[CityProfile]:
    """Label each profile low / mid / high by tercile of ``detour_mean``.

    Terciles of the observed set, not fixed thresholds: what counts as a
    high-distortion city is only meaningful relative to the cities on hand, and
    absolute cut-offs would silently empty a stratum on a different city list.
    """
    if not profiles:
        return []
    means = np.asarray([p.detour_mean for p in profiles], dtype=float)
    low_cut, high_cut = np.quantile(means, [1 / 3, 2 / 3])

    labelled: list[CityProfile] = []
    for profile in profiles:
        if profile.detour_mean <= low_cut:
            stratum = "low"
        elif profile.detour_mean <= high_cut:
            stratum = "mid"
        else:
            stratum = "high"
        labelled.append(CityProfile(**{**profile.to_dict(), "distortion_stratum": stratum}))
    return labelled

profile_cities

profile_cities(
    cities: Iterable[tuple[str, Path]],
    *,
    num_sources: int = DEFAULT_SOURCES,
    num_targets: int = DEFAULT_TARGETS,
    seed: int = 0,
    jobs: int = 1,
    on_progress: Callable[[str], None] | None = None,
) -> list[CityProfile]

Profile every city, then assign the distortion strata across the set.

Each city is an independent OSM parse plus a handful of Dijkstras, so this parallelizes perfectly; a worker holds one city's road graph at a time. The strata are assigned afterwards over the whole set, which is why the results are gathered before anything is labelled.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/city_profile.py

def profile_cities(
    cities: Iterable[tuple[str, Path]],
    *,
    num_sources: int = DEFAULT_SOURCES,
    num_targets: int = DEFAULT_TARGETS,
    seed: int = 0,
    jobs: int = 1,
    on_progress: Callable[[str], None] | None = None,
) -> list[CityProfile]:
    """Profile every city, then assign the distortion strata across the set.

    Each city is an independent OSM parse plus a handful of Dijkstras, so this
    parallelizes perfectly; a worker holds one city's road graph at a time.
    The strata are assigned afterwards over the whole set, which is why the
    results are gathered before anything is labelled.
    """
    tasks = [
        (city, str(osm_path), num_sources, num_targets, seed) for city, osm_path in cities
    ]
    if not tasks:
        return []

    profiles: list[CityProfile] = []
    if jobs <= 1:
        for task in tasks:
            if on_progress is not None:
                on_progress(task[0])
            profiles.append(_profile_one(task))
    else:
        with ProcessPoolExecutor(max_workers=jobs) as pool:
            pending = {pool.submit(_profile_one, task): task[0] for task in tasks}
            for future in as_completed(pending):
                profile = future.result()
                if on_progress is not None:
                    on_progress(pending[future])
                profiles.append(profile)
        # Deterministic order regardless of which worker finished first.
        profiles.sort(key=lambda profile: profile.city)
    return assign_strata(profiles)

profile_city

profile_city(
    osm_path: str | Path,
    city: str,
    *,
    num_sources: int = DEFAULT_SOURCES,
    num_targets: int = DEFAULT_TARGETS,
    seed: int = 0,
) -> CityProfile

Profile one city from its OSM extract.

Runs num_sources single-source Dijkstras per metric and samples num_targets reachable targets from each, rather than an all-pairs pass: a few thousand pairs already pin the mean detour to well within the spread between cities, and an all-pairs pass on a 200k-vertex graph is not a "cheap pre-pass" by any reading.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/city_profile.py

def profile_city(
    osm_path: str | Path,
    city: str,
    *,
    num_sources: int = DEFAULT_SOURCES,
    num_targets: int = DEFAULT_TARGETS,
    seed: int = 0,
) -> CityProfile:
    """Profile one city from its OSM extract.

    Runs ``num_sources`` single-source Dijkstras per metric and samples
    ``num_targets`` reachable targets from each, rather than an all-pairs pass:
    a few thousand pairs already pin the mean detour to well within the spread
    between cities, and an all-pairs pass on a 200k-vertex graph is not a
    "cheap pre-pass" by any reading.
    """
    graph = load_road_graph(osm_path)
    rng = np.random.default_rng(seed)

    vertex_count = graph.vertex_count
    sources = rng.choice(vertex_count, size=min(num_sources, vertex_count), replace=False)
    targets = rng.choice(vertex_count, size=min(num_targets, vertex_count), replace=False)

    road_short = dijkstra(metric_csr(graph, "shortest"), directed=True, indices=sources)[:, targets]
    road_fast = dijkstra(metric_csr(graph, "fastest"), directed=True, indices=sources)[:, targets]

    enu = np.asarray([graph.node_enu[graph.node_of[v]][:2] for v in range(vertex_count)], dtype=float)
    straight = np.linalg.norm(enu[sources][:, None, :] - enu[targets][None, :, :], axis=2)

    # A pair is usable when both metrics reach it and the two nodes are not the
    # same point; unreachable pairs would otherwise poison every statistic.
    usable = np.isfinite(road_short) & np.isfinite(road_fast) & (straight > 0.0)
    if not usable.any():
        raise ValueError(f"{city}: no reachable sampled node pair on the road graph")

    detour = road_short[usable] / straight[usable]
    tau = kendalltau(straight[usable], road_fast[usable]).correlation

    return CityProfile(
        city=city,
        osm_file=Path(osm_path).name,
        num_vertices=vertex_count,
        num_edges=graph.edge_count,
        detour_mean=float(detour.mean()),
        detour_p90=float(np.percentile(detour, 90)),
        rank_tau_eucl_fast=float(tau),
        num_pairs=int(usable.sum()),
    )

compute_descriptors

compute_descriptors(
    coordinates: Sequence[Sequence[float]],
    demands: Sequence[int],
    capacity: int,
) -> InstanceDescriptors

Descriptors of one instance. Index 0 is the depot, in both sequences.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/descriptors.py

def compute_descriptors(
    coordinates: Sequence[Sequence[float]],
    demands: Sequence[int],
    capacity: int,
) -> InstanceDescriptors:
    """Descriptors of one instance. Index 0 is the depot, in both sequences."""
    points = np.asarray(coordinates, dtype=float)
    if points.ndim != 2 or points.shape[1] != 2:
        raise ValueError("coordinates must be a sequence of (x, y) pairs")
    if len(points) != len(demands):
        raise ValueError("coordinates and demands must have the same length")
    if capacity <= 0:
        raise ValueError("capacity must be positive")

    depot = points[0]
    customers = points[1:]
    n = len(customers)
    if n < 2:
        raise ValueError("at least 2 customers are required")

    quantities = np.asarray(demands[1:], dtype=float)
    if (quantities < 0).any():
        raise ValueError("customer demands must be non-negative")

    # --- spatial -----------------------------------------------------------
    area = _point_area(customers)
    distances, _ = cKDTree(customers).query(customers, k=2)
    nnd = distances[:, 1]
    mean_nnd = float(nnd.mean())
    if area > 0.0:
        expected_nnd = 0.5 * math.sqrt(area / n)
        clark_evans_r = mean_nnd / expected_nnd if expected_nnd > 0.0 else float("nan")
    else:
        clark_evans_r = float("nan")
    nnd_cv = float(nnd.std() / mean_nnd) if mean_nnd > 0.0 else float("nan")

    centroid = customers.mean(axis=0)
    radii = np.linalg.norm(customers - centroid, axis=1)
    mean_radius = float(radii.mean())
    radial_dispersion = float(radii.std() / mean_radius) if mean_radius > 0.0 else float("nan")
    depot_offset = float(np.linalg.norm(depot - centroid))
    depot_centrality = depot_offset / mean_radius if mean_radius > 0.0 else float("nan")

    # Where the depot sits in the instance's own "how remote is this point"
    # ordering. Scale-free and shape-free, unlike a raw distance.
    mean_to_customers = _mean_distance_to(points, customers)
    depot_eccentricity = float((mean_to_customers < mean_to_customers[0]).sum()) / n

    # --- demand ------------------------------------------------------------
    mean_demand = float(quantities.mean())
    if mean_demand > 0.0:
        demand_cv = float(quantities.std() / mean_demand)
        demand_max_over_mean = float(quantities.max() / mean_demand)
    else:
        demand_cv = float("nan")
        demand_max_over_mean = float("nan")
    demand_gini = _gini(quantities)
    demand_moran_i = _moran_i(customers, quantities, MORAN_NEIGHBOURS)

    # --- capacity ----------------------------------------------------------
    total_demand = float(quantities.sum())
    lb_cap = int(math.ceil(total_demand / capacity)) if total_demand > 0 else 0
    route_size = n / lb_cap if lb_cap > 0 else float("nan")
    capacity_slack = 1.0 - total_demand / (lb_cap * capacity) if lb_cap > 0 else float("nan")

    return InstanceDescriptors(
        n=n,
        extent_m=math.sqrt(area),
        clark_evans_r=clark_evans_r,
        nnd_cv=nnd_cv,
        radial_dispersion=radial_dispersion,
        depot_centrality=depot_centrality,
        depot_eccentricity=depot_eccentricity,
        demand_cv=demand_cv,
        demand_gini=demand_gini,
        demand_max_over_mean=demand_max_over_mean,
        demand_moran_i=demand_moran_i,
        lb_cap=lb_cap,
        route_size=route_size,
        capacity_slack=capacity_slack,
    )

compute_divergence

compute_divergence(
    euclidean: Sequence[Sequence[float]],
    shortest: Sequence[Sequence[float]],
    fastest: Sequence[Sequence[float]],
    *,
    seed: int = 0,
) -> MetricDivergence

How much the road metrics of one instance disagree with its Euclidean one.

seed fixes the pair subsample used for the rank correlations, so the reported values are reproducible for a given instance.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/descriptors.py

def compute_divergence(
    euclidean: Sequence[Sequence[float]],
    shortest: Sequence[Sequence[float]],
    fastest: Sequence[Sequence[float]],
    *,
    seed: int = 0,
) -> MetricDivergence:
    """How much the road metrics of one instance disagree with its Euclidean one.

    ``seed`` fixes the pair subsample used for the rank correlations, so the
    reported values are reproducible for a given instance.
    """
    d_eucl = np.asarray(euclidean, dtype=float)
    d_short = np.asarray(shortest, dtype=float)
    d_fast = np.asarray(fastest, dtype=float)
    if not (d_eucl.shape == d_short.shape == d_fast.shape):
        raise ValueError("the three matrices must have the same shape")
    if d_eucl.ndim != 2 or d_eucl.shape[0] != d_eucl.shape[1]:
        raise ValueError("matrices must be square")

    flat_eucl = _offdiagonal(d_eucl)
    flat_short = _offdiagonal(d_short)
    flat_fast = _offdiagonal(d_fast)

    positive = flat_eucl > 0.0
    if positive.any():
        detour = flat_short[positive] / flat_eucl[positive]
        detour_mean = float(detour.mean())
        detour_p90 = float(np.percentile(detour, 90))
    else:
        detour_mean = detour_p90 = float("nan")

    # One generator, drawn from in a fixed order: the three correlations use
    # different subsamples, and re-running with the same seed reproduces them.
    rng = np.random.default_rng(seed)
    return MetricDivergence(
        detour_mean=detour_mean,
        detour_p90=detour_p90,
        asymmetry_shortest=_asymmetry(d_short),
        asymmetry_fastest=_asymmetry(d_fast),
        rank_tau_eucl_short=_sampled_tau(flat_eucl, flat_short, rng),
        rank_tau_eucl_fast=_sampled_tau(flat_eucl, flat_fast, rng),
        rank_tau_short_fast=_sampled_tau(flat_short, flat_fast, rng),
    )

enumerate_candidates

enumerate_candidates(
    cities: Sequence[tuple[str, Path]],
    profiles: Sequence[CityProfile] = (),
    *,
    per_city: int = 8,
    size_grid: Sequence[int] = MAIN_SIZE_GRID,
    methods: Sequence[str] = METHODS,
    size_ceiling: Mapping[str, int] | None = None,
    tier: str = "main",
    base_seed: int = 0,
) -> list[CandidateSpec]

per_city candidates for every city, balanced on every design axis.

methods narrows the sourcing axis. The POI-only tier passes a single method because "every customer is a real amenity" is the point of that tier, not one option among three.

size_ceiling caps n per city. Sizes are drawn from a grid that knows nothing about how many customers a given city can actually serve, so without it a POI-only tier proposes n = 5000 in a city holding 1800 amenities and discovers the shortfall only after paying for the evaluation. The cap replaces an over-large draw with the largest grid size that still fits, and drops the candidate when even the smallest does not.

The route-size band is drawn as a balanced column and then repaired against the size, because which bands leave a real fleet depends on n: a target of 50-200 customers per route is an ordinary instance at n = 4000 and a two-route degenerate at n = 272. A draw the size cannot support falls back to the largest admissible band, which keeps the column's ordering -- longer draws still get longer routes -- without letting it off the end.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/design.py

def enumerate_candidates(
    cities: Sequence[tuple[str, Path]],
    profiles: Sequence[CityProfile] = (),
    *,
    per_city: int = 8,
    size_grid: Sequence[int] = MAIN_SIZE_GRID,
    methods: Sequence[str] = METHODS,
    size_ceiling: Mapping[str, int] | None = None,
    tier: str = "main",
    base_seed: int = 0,
) -> list[CandidateSpec]:
    """``per_city`` candidates for every city, balanced on every design axis.

    ``methods`` narrows the sourcing axis. The POI-only tier passes a single
    method because "every customer is a real amenity" is the point of that tier,
    not one option among three.

    ``size_ceiling`` caps ``n`` per city. Sizes are drawn from a grid that knows
    nothing about how many customers a given city can actually serve, so without
    it a POI-only tier proposes n = 5000 in a city holding 1800 amenities and
    discovers the shortfall only after paying for the evaluation. The cap
    replaces an over-large draw with the largest grid size that still fits, and
    drops the candidate when even the smallest does not.

    The route-size band is drawn as a balanced column and then *repaired against
    the size*, because which bands leave a real fleet depends on ``n``: a target
    of 50-200 customers per route is an ordinary instance at n = 4000 and a
    two-route degenerate at n = 272. A draw the size cannot support falls back to
    the largest admissible band, which keeps the column's ordering -- longer
    draws still get longer routes -- without letting it off the end.
    """
    if per_city < 1:
        raise ValueError("per_city must be >= 1")
    if not cities:
        return []
    if not methods:
        raise ValueError("methods must not be empty")

    stratum_of = {profile.city: profile.distortion_stratum for profile in profiles}
    total = per_city * len(cities)
    rng = np.random.default_rng(stable_seed("enumerate", tier, base_seed, len(cities), per_city))

    columns = {
        "n": _balanced_column(size_grid, total, rng),
        "method": _balanced_column(methods, total, rng),
        "demand_type": _balanced_column(DEMAND_TYPES, total, rng),
        "avg_route_size": _balanced_column(AVG_ROUTE_SIZES, total, rng),
        "depot_mode": _balanced_column(DEPOT_MODES, total, rng),
        "customer_mode": _balanced_column(CUSTOMER_MODES, total, rng),
        "cluster_seeds": _balanced_column(CLUSTER_SEEDS, total, rng),
        "cluster_decay_meters": _balanced_column(CLUSTER_DECAY_METERS, total, rng),
        "hybrid_poi_share": _balanced_column(HYBRID_POI_SHARES, total, rng),
    }

    ordered_sizes = sorted(size_grid)
    specs: list[CandidateSpec] = []
    for index in range(total):
        city, osm_path = cities[index // per_city]
        n = int(columns["n"][index])
        if size_ceiling is not None:
            ceiling = int(size_ceiling.get(city, 0))
            if n > ceiling:
                fitting = [size for size in ordered_sizes if size <= ceiling]
                if not fitting:
                    continue
                n = fitting[-1]
        specs.append(
            CandidateSpec(
                city=city,
                osm_path=Path(osm_path),
                n=n,
                method=str(columns["method"][index]),
                demand_type=int(columns["demand_type"][index]),
                avg_route_size=_admissible_band(int(columns["avg_route_size"][index]), n),
                depot_mode=str(columns["depot_mode"][index]),
                customer_mode=str(columns["customer_mode"][index]),
                cluster_seeds=int(columns["cluster_seeds"][index]),
                cluster_decay_meters=float(columns["cluster_decay_meters"][index]),
                hybrid_poi_share=float(columns["hybrid_poi_share"][index]),
                distortion_stratum=stratum_of.get(city),
                tier=tier,
            )
        )
    return specs

enumerate_for_rungs

enumerate_for_rungs(
    assignments: Sequence[Any],
    *,
    osm_dir: str | Path = "osmdata",
    per_rung: int = 6,
    base_seed: int = 0,
    tier: str = "main",
    categories: Sequence[str] = CAMPAIGN_POI_CATEGORIES,
    poi_attach_mode: str = CAMPAIGN_POI_ATTACH_MODE,
) -> list[CandidateSpec]

per_rung candidates for each ladder position.

The counterpart of :func:enumerate_candidates for a campaign that has already decided what each instance is -- city, size and sourcing method come from a :class:~.ladder.RungAssignment and are not up for negotiation here. What is still free is what the instance is like: how demand is distributed, how tightly the vehicles are loaded, where the depot sits and whether the customers cluster. Those are drawn as balanced columns across the whole pool exactly as before, so the selector still gets a spread to choose from on every axis it can still move.

Splitting it this way is what stops the two decisions from fighting. Drawn together, a size the city could not serve was silently downgraded; drawn in order, the size is guaranteed feasible before any descriptor is computed.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/design.py

def enumerate_for_rungs(
    assignments: Sequence[Any],
    *,
    osm_dir: str | Path = "osmdata",
    per_rung: int = 6,
    base_seed: int = 0,
    tier: str = "main",
    categories: Sequence[str] = CAMPAIGN_POI_CATEGORIES,
    poi_attach_mode: str = CAMPAIGN_POI_ATTACH_MODE,
) -> list[CandidateSpec]:
    """``per_rung`` candidates for each ladder position.

    The counterpart of :func:`enumerate_candidates` for a campaign that has
    already decided *what* each instance is -- city, size and sourcing method
    come from a :class:`~.ladder.RungAssignment` and are not up for negotiation
    here. What is still free is what the instance is *like*: how demand is
    distributed, how tightly the vehicles are loaded, where the depot sits and
    whether the customers cluster. Those are drawn as balanced columns across
    the whole pool exactly as before, so the selector still gets a spread to
    choose from on every axis it can still move.

    Splitting it this way is what stops the two decisions from fighting. Drawn
    together, a size the city could not serve was silently downgraded; drawn in
    order, the size is guaranteed feasible before any descriptor is computed.
    """
    if per_rung < 1:
        raise ValueError("per_rung must be >= 1")
    if not assignments:
        return []

    # A RungAssignment carries the extract's *file name*, not its path: capacity
    # measurements are shared across checkouts and an absolute path would not
    # survive the trip. The directory is supplied here instead.
    root = Path(osm_dir)
    total = per_rung * len(assignments)
    rng = np.random.default_rng(
        stable_seed("rungs", tier, base_seed, len(assignments), per_rung)
    )
    # ``avg_route_size`` is absent from this table on purpose. Which bands keep an
    # instance a genuine VRP depends on ``n``, so the band comes from the rung
    # alongside the city and the size; drawing it here as a balanced column over
    # all seven levels is precisely what put a 50-200 customer route target on a
    # hundred-customer instance in v2.
    columns = {
        "demand_type": _balanced_column(DEMAND_TYPES, total, rng),
        "depot_mode": _balanced_column(DEPOT_MODES, total, rng),
        "customer_mode": _balanced_column(CUSTOMER_MODES, total, rng),
        "cluster_seeds": _balanced_column(CLUSTER_SEEDS, total, rng),
        "cluster_decay_meters": _balanced_column(CLUSTER_DECAY_METERS, total, rng),
        "hybrid_poi_share": _balanced_column(HYBRID_POI_SHARES, total, rng),
    }

    specs: list[CandidateSpec] = []
    for index in range(total):
        rung = assignments[index // per_rung]
        specs.append(
            CandidateSpec(
                city=rung.city,
                osm_path=root / rung.osm_file,
                n=int(rung.n),
                method=str(rung.method),
                demand_type=int(columns["demand_type"][index]),
                avg_route_size=int(rung.avg_route_size),
                depot_mode=str(columns["depot_mode"][index]),
                customer_mode=str(columns["customer_mode"][index]),
                cluster_seeds=int(columns["cluster_seeds"][index]),
                cluster_decay_meters=float(columns["cluster_decay_meters"][index]),
                hybrid_poi_share=float(columns["hybrid_poi_share"][index]),
                poi_attach_mode=poi_attach_mode,
                categories=tuple(categories),
                distortion_stratum=rung.distortion_stratum,
                tier=tier,
                rung_key=rung.rung_key,
            )
        )
    return specs

evaluate_candidate

evaluate_candidate(
    spec: CandidateSpec,
) -> EvaluatedCandidate

What this candidate would be, short of its distance matrices.

Rebuilds the customer selection and draws the demands through exactly the calls generate_single_instance makes, from exactly the same seed, so the descriptors describe the instance that generation will actually produce.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/design.py

def evaluate_candidate(spec: CandidateSpec) -> EvaluatedCandidate:
    """What this candidate would be, short of its distance matrices.

    Rebuilds the customer selection and draws the demands through exactly the
    calls ``generate_single_instance`` makes, from exactly the same seed, so the
    descriptors describe the instance that generation will actually produce.
    """
    selection = build_generation_selection(spec.to_request())
    graph = selection.graph
    coordinates = [graph.node_enu[graph.node_of[v]][:2] for v in selection.vertices]

    customer_ll = list(zip(selection.poi_lats[1:], selection.poi_lons[1:]))
    demand_values, _total, _maximum, route_target = generate_demands(
        random.Random(spec.seed), customer_ll, spec.demand_type, spec.avg_route_size
    )
    capacity = capacity_from_avg_route_size(route_target, demand_values)

    customer_tags = selection.source_tags[1:]
    poi_count = sum(1 for tag in customer_tags if is_poi_source_tag(tag))
    return EvaluatedCandidate(
        spec=spec,
        descriptors=compute_descriptors(coordinates, [0, *demand_values], capacity),
        resolved_n=len(selection.vertices) - 1,
        effective_method=str(selection.params["method"]),
        poi_fraction=poi_count / len(customer_tags) if customer_tags else 0.0,
    )

evaluate_candidates

evaluate_candidates(
    specs: Sequence[CandidateSpec],
    *,
    jobs: int = 1,
    drop_unusable: bool = True,
    on_progress: Callable[[str, int, int], None]
    | None = None,
) -> list[EvaluatedCandidate]

Evaluate a whole pool, one process per city.

drop_unusable=False returns everything that evaluated, letting the caller apply :meth:EvaluatedCandidate.is_usable itself. That is what a campaign wants for its saved pool: usability depends on the feature space, and re-deciding it must not cost another pass over every OSM extract.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/design.py

def evaluate_candidates(
    specs: Sequence[CandidateSpec],
    *,
    jobs: int = 1,
    drop_unusable: bool = True,
    on_progress: Callable[[str, int, int], None] | None = None,
) -> list[EvaluatedCandidate]:
    """Evaluate a whole pool, one process per city.

    ``drop_unusable=False`` returns everything that evaluated, letting the
    caller apply :meth:`EvaluatedCandidate.is_usable` itself. That is what a
    campaign wants for its saved pool: usability depends on the feature space,
    and re-deciding it must not cost another pass over every OSM extract.
    """
    by_city: dict[str, list[CandidateSpec]] = {}
    for spec in specs:
        by_city.setdefault(spec.city, []).append(spec)
    groups = list(by_city.items())

    evaluated: list[EvaluatedCandidate] = []
    if jobs <= 1:
        for index, (city, city_specs) in enumerate(groups, start=1):
            if on_progress is not None:
                on_progress(city, index, len(groups))
            evaluated.extend(_evaluate_city(city_specs))
    else:
        with ProcessPoolExecutor(max_workers=jobs) as pool:
            futures = {
                pool.submit(_evaluate_city, city_specs): city for city, city_specs in groups
            }
            for index, (future, city) in enumerate(futures.items(), start=1):
                if on_progress is not None:
                    on_progress(city, index, len(groups))
                evaluated.extend(future.result())

    if not drop_unusable:
        return evaluated

    usable = [candidate for candidate in evaluated if candidate.is_usable()]
    dropped = len(evaluated) - len(usable)
    if dropped:
        short = sum(1 for c in evaluated if c.resolved_n < c.spec.n)
        warnings.warn(
            f"dropped {dropped} candidate(s): {short} could not be served at the requested "
            f"size, {dropped - short} had LB_cap < {CAMPAIGN_MIN_ROUTES} or an "
            "undefined descriptor",
            stacklevel=2,
        )
    return usable

resolve_cities

resolve_cities(
    osm_dir: str | Path, names: Iterable[str] | None = None
) -> list[tuple[str, Path]]

(city_slug, osm_path) for the extracts on disk, slugged like the tree.

Aix-en-Provence.osm becomes aix_en_provence, matching the place slugs the benchmark tree and the website already use.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/design.py

def resolve_cities(osm_dir: str | Path, names: Iterable[str] | None = None) -> list[tuple[str, Path]]:
    """``(city_slug, osm_path)`` for the extracts on disk, slugged like the tree.

    ``Aix-en-Provence.osm`` becomes ``aix_en_provence``, matching the place slugs
    the benchmark tree and the website already use.
    """
    directory = Path(osm_dir)
    wanted = {slugify(name) for name in names} if names else None
    cities: list[tuple[str, Path]] = []
    for path in sorted(directory.glob("*.osm")):
        slug = slugify(path.stem)
        if wanted is None or slug in wanted:
            cities.append((slug, path))
    if wanted is not None:
        missing = wanted - {slug for slug, _ in cities}
        if missing:
            raise FileNotFoundError(f"no OSM extract for: {', '.join(sorted(missing))}")
    return cities

stable_seed

stable_seed(*parts: Any) -> int

A seed derived from what the instance is, never from a counter.

Mirrors family.family._stable_seed: re-running the campaign, or regenerating one instance on its own, must land on the same draw.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/design.py

def stable_seed(*parts: Any) -> int:
    """A seed derived from what the instance *is*, never from a counter.

    Mirrors ``family.family._stable_seed``: re-running the campaign, or
    regenerating one instance on its own, must land on the same draw.
    """
    return zlib.crc32("|".join(str(part) for part in parts).encode("utf-8"))

assign_rungs

assign_rungs(
    ladder: Sequence[int],
    capacities: Iterable[PoiCapacity],
    profiles: Iterable[CityProfile] = (),
    *,
    method_weights: Mapping[str, float] | None = None,
    hybrid_share: float = 0.5,
    headroom: float = DEFAULT_HEADROOM,
    vertex_factor: float = DEFAULT_VERTEX_FACTOR,
    min_routes: int = CAMPAIGN_MIN_ROUTES,
    base_seed: int = 0,
) -> LadderPlan

Match every rung to a city that can demonstrably supply it.

Cheapest sufficient city, hardest rung first. Rungs are processed in decreasing order of what they demand, and each takes the unused city that satisfies it while costing the pool the least amenity capacity.

Amenity capacity is the scarce resource and road vertices are not: every city has plenty of road and only some have plenty of shops. So the cost of spending a city is its POI capacity, whatever the rung happens to need. For a POI or hybrid rung that is ordinary best fit -- the smallest capacity that still clears the bar, which keeps Tokyo available for the rungs that need Tokyo. For a parametric rung, which consumes no amenities at all, it means deliberately spending the amenity-poorest city that has enough road.

That second case is not a refinement, it is the difference between filling the ladder and not. Ranking parametric rungs by road vertices instead makes a city with 563 000 vertices and 57 amenities the last candidate for the only kind of rung it can serve, so it is never used -- and the ladder then runs out of cities while five amenity-dead extracts sit unspent.

Ties among equally sufficient cities break toward the distortion stratum that is currently most under-represented, so the road-network diversity the study depends on survives a matching that is otherwise driven entirely by amenity counts.

One instance per city: a city is removed from the pool once used.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/ladder.py

def assign_rungs(
    ladder: Sequence[int],
    capacities: Iterable[PoiCapacity],
    profiles: Iterable[CityProfile] = (),
    *,
    method_weights: Mapping[str, float] | None = None,
    hybrid_share: float = 0.5,
    headroom: float = DEFAULT_HEADROOM,
    vertex_factor: float = DEFAULT_VERTEX_FACTOR,
    min_routes: int = CAMPAIGN_MIN_ROUTES,
    base_seed: int = 0,
) -> LadderPlan:
    """Match every rung to a city that can demonstrably supply it.

    **Cheapest sufficient city, hardest rung first.** Rungs are processed in
    decreasing order of what they demand, and each takes the unused city that
    satisfies it while costing the pool the least *amenity capacity*.

    Amenity capacity is the scarce resource and road vertices are not: every
    city has plenty of road and only some have plenty of shops. So the cost of
    spending a city is its POI capacity, whatever the rung happens to need. For
    a POI or hybrid rung that is ordinary best fit -- the smallest capacity that
    still clears the bar, which keeps Tokyo available for the rungs that need
    Tokyo. For a parametric rung, which consumes no amenities at all, it means
    deliberately spending the amenity-*poorest* city that has enough road.

    That second case is not a refinement, it is the difference between filling
    the ladder and not. Ranking parametric rungs by road vertices instead makes
    a city with 563 000 vertices and 57 amenities the *last* candidate for the
    only kind of rung it can serve, so it is never used -- and the ladder then
    runs out of cities while five amenity-dead extracts sit unspent.

    Ties among equally sufficient cities break toward the distortion stratum
    that is currently most under-represented, so the road-network diversity the
    study depends on survives a matching that is otherwise driven entirely by
    amenity counts.

    One instance per city: a city is removed from the pool once used.
    """
    weights = dict(method_weights or DEFAULT_METHOD_WEIGHTS)
    rng = np.random.default_rng(stable_seed("ladder", base_seed, len(ladder), tuple(sorted(weights))))
    methods = weighted_column(weights, len(ladder), rng)
    bands = assign_route_size_bands(ladder, rng, min_routes=min_routes)

    stratum_of = {profile.city: profile.distortion_stratum for profile in profiles}
    pool = {item.city: item for item in capacities}

    demands: list[tuple[int, int, str, int, str, int]] = []
    for index, (n, method, band) in enumerate(zip(ladder, methods, bands)):
        resource, amount = required_supply(
            n,
            method,
            hybrid_share=hybrid_share,
            headroom=headroom,
            vertex_factor=vertex_factor,
        )
        demands.append((index, n, method, band, resource, amount))

    # Hardest first. Ordering across resources by raw amount is fine: it only
    # decides who picks first, and within a resource the order is exactly right.
    demands.sort(key=lambda item: (-item[5], item[0]))

    def supply(city: PoiCapacity, resource: str) -> int:
        return city.capacity if resource == "poi" else city.num_vertices

    used_strata: dict[str | None, int] = {stratum: 0 for stratum in STRATA}
    assignments: list[RungAssignment] = []
    unfilled: list[UnfilledRung] = []

    for index, n, method, band, resource, amount in demands:
        fitting = [city for city in pool.values() if supply(city, resource) >= amount]
        if not fitting:
            best = max(pool.values(), key=lambda c: supply(c, resource), default=None)
            unfilled.append(
                UnfilledRung(
                    index=index,
                    n=n,
                    method=method,
                    required=amount,
                    best_available=supply(best, resource) if best else 0,
                    best_city=best.city if best else None,
                )
            )
            continue

        chosen = min(
            fitting,
            key=lambda c: (
                c.capacity,                                  # cheapest in the scarce resource
                used_strata.get(stratum_of.get(c.city), 0),  # then rarest stratum
                c.city,                                      # then deterministic
            ),
        )
        del pool[chosen.city]
        stratum = stratum_of.get(chosen.city)
        used_strata[stratum] = used_strata.get(stratum, 0) + 1
        assignments.append(
            RungAssignment(
                index=index,
                n=n,
                method=method,
                avg_route_size=band,
                city=chosen.city,
                osm_file=chosen.osm_file,
                distortion_stratum=stratum,
                required=amount,
                available=supply(chosen, resource),
            )
        )

    assignments.sort(key=lambda item: item.index)
    unfilled.sort(key=lambda item: item.index)
    return LadderPlan(
        assignments=assignments,
        unfilled=unfilled,
        method_weights=weights,
        headroom=headroom,
        min_routes=min_routes,
    )

plan_summary

plan_summary(plan: LadderPlan) -> dict[str, Any]

Counts a human can scan to see whether the design came out as asked.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/ladder.py

def plan_summary(plan: LadderPlan) -> dict[str, Any]:
    """Counts a human can scan to see whether the design came out as asked."""
    methods: dict[str, int] = {}
    strata: dict[str, int] = {}
    bands: dict[int, int] = {}
    for item in plan.assignments:
        methods[item.method] = methods.get(item.method, 0) + 1
        key = item.distortion_stratum or "?"
        strata[key] = strata.get(key, 0) + 1
        bands[item.avg_route_size] = bands.get(item.avg_route_size, 0) + 1
    sizes = [item.n for item in plan.assignments]
    return {
        "assigned": len(plan.assignments),
        "unfilled": len(plan.unfilled),
        "methods": methods,
        "strata": strata,
        "route_size_bands": dict(sorted(bands.items())),
        # The two numbers that say whether the capacity design came out sound:
        # the worst case over every legal draw, and how far the route-size axis
        # ended up entangled with the size axis it is meant to be free of.
        "guaranteed_min_routes": round(plan.guaranteed_min_routes, 2),
        "band_size_corr": round(plan.band_size_corr, 3),
        "n_min": min(sizes) if sizes else None,
        "n_max": max(sizes) if sizes else None,
        "cities": len({item.city for item in plan.assignments}),
    }

size_ladder

size_ladder(
    k: int, n_min: int = 100, n_max: int = 1000
) -> list[int]

k distinct sizes rising smoothly from n_min to n_max.

Geometrically spaced, so the ladder spends its resolution where instances differ most: the step from 100 to 110 changes a CVRP far more than the step from 900 to 910. Rounding collides at the bottom, where consecutive rungs are less than a customer apart, so a repair pass walks the ladder and pushes each rung at least one past its predecessor. Both endpoints are exact.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/ladder.py

def size_ladder(k: int, n_min: int = 100, n_max: int = 1000) -> list[int]:
    """``k`` distinct sizes rising smoothly from ``n_min`` to ``n_max``.

    Geometrically spaced, so the ladder spends its resolution where instances
    differ most: the step from 100 to 110 changes a CVRP far more than the step
    from 900 to 910. Rounding collides at the bottom, where consecutive rungs
    are less than a customer apart, so a repair pass walks the ladder and pushes
    each rung at least one past its predecessor. Both endpoints are exact.
    """
    if k < 2:
        raise ValueError("a ladder needs at least two rungs")
    if n_min < 1 or n_max <= n_min:
        raise ValueError(f"need 1 <= n_min < n_max, got {n_min} and {n_max}")
    if n_max - n_min + 1 < k:
        raise ValueError(
            f"cannot fit {k} distinct sizes between {n_min} and {n_max}"
        )

    raw = np.geomspace(n_min, n_max, k)
    rungs: list[int] = []
    for index, value in enumerate(raw):
        rung = int(round(value))
        if rungs:
            rung = max(rung, rungs[-1] + 1)
        # Never overshoot so far that the tail cannot stay strictly increasing.
        rung = min(rung, n_max - (k - 1 - index))
        rungs.append(rung)
    rungs[-1] = n_max
    return rungs

weighted_column

weighted_column(
    weights: Mapping[str, float],
    length: int,
    rng: Generator,
) -> list[str]

length labels in the requested proportions, shuffled.

Exact rather than sampled: 50/25/25 of 100 is 50, 25 and 25, not "about that". Remainders from the largest-remainder apportionment go to the heaviest weights first, so a 50 % share is never the one left short.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/ladder.py

def weighted_column(
    weights: Mapping[str, float], length: int, rng: np.random.Generator
) -> list[str]:
    """``length`` labels in the requested proportions, shuffled.

    Exact rather than sampled: 50/25/25 of 100 is 50, 25 and 25, not "about
    that". Remainders from the largest-remainder apportionment go to the
    heaviest weights first, so a 50 % share is never the one left short.
    """
    if not weights:
        raise ValueError("weights must not be empty")
    total = sum(weights.values())
    if total <= 0:
        raise ValueError("weights must sum to something positive")

    exact = {name: length * weight / total for name, weight in weights.items()}
    counts = {name: int(value) for name, value in exact.items()}
    remainder = length - sum(counts.values())
    # Largest remainder, then largest weight, then name -- fully deterministic.
    order = sorted(
        weights,
        key=lambda name: (-(exact[name] - counts[name]), -weights[name], name),
    )
    for name in order[:remainder]:
        counts[name] += 1

    column = [name for name, count in counts.items() for _ in range(count)]
    rng.shuffle(column)
    return column

cities_supporting

cities_supporting(
    capacities: Iterable[PoiCapacity],
    n: int,
    *,
    headroom: float = 1.0,
) -> list[PoiCapacity]

Cities that can serve n POI customers, largest capacity first.

headroom above 1.0 demands slack: at exactly capacity == n + 1 the "selection" is the whole amenity pool, so the seed stops mattering and the instance is no longer a draw from anything.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/poi_capacity.py

def cities_supporting(
    capacities: Iterable[PoiCapacity], n: int, *, headroom: float = 1.0
) -> list[PoiCapacity]:
    """Cities that can serve ``n`` POI customers, largest capacity first.

    ``headroom`` above 1.0 demands slack: at exactly ``capacity == n + 1`` the
    "selection" is the whole amenity pool, so the seed stops mattering and the
    instance is no longer a draw from anything.
    """
    threshold = (n + 1) * headroom
    fitting = [item for item in capacities if item.capacity >= threshold]
    fitting.sort(key=lambda item: item.capacity, reverse=True)
    return fitting

categories_digest

categories_digest(categories: Sequence[str]) -> str

A short fingerprint of a category list, order-independent.

A capacity is only meaningful for the categories it was measured over, and curating the list changes every number in the file. Without a fingerprint a stale poi-capacity.json is indistinguishable from a fresh one, and the campaign would size instances against amenities it is no longer drawing from.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/poi_capacity.py

def categories_digest(categories: Sequence[str]) -> str:
    """A short fingerprint of a category list, order-independent.

    A capacity is only meaningful for the categories it was measured over, and
    curating the list changes every number in the file. Without a fingerprint a
    stale ``poi-capacity.json`` is indistinguishable from a fresh one, and the
    campaign would size instances against amenities it is no longer drawing
    from.
    """
    joined = ",".join(sorted(set(categories)))
    return hashlib.sha256(joined.encode("utf-8")).hexdigest()[:16]

load_capacities

load_capacities(
    path: str | Path,
    *,
    expect_categories: Sequence[str] | None = None,
) -> list[PoiCapacity]

Load a capacity file, refusing one measured over other categories.

expect_categories turns a silent mis-sizing into a loud failure: every number in the file is relative to a category list, so reading it against a different one would classify cities by amenities the campaign no longer draws from.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/poi_capacity.py

def load_capacities(
    path: str | Path, *, expect_categories: Sequence[str] | None = None
) -> list[PoiCapacity]:
    """Load a capacity file, refusing one measured over other categories.

    ``expect_categories`` turns a silent mis-sizing into a loud failure: every
    number in the file is relative to a category list, so reading it against a
    different one would classify cities by amenities the campaign no longer
    draws from.
    """
    payload = json.loads(Path(path).read_text(encoding="utf-8"))
    capacities = [PoiCapacity.from_dict(record) for record in payload["cities"]]
    if expect_categories is not None:
        wanted = categories_digest(expect_categories)
        stale = {item.categories_digest for item in capacities} - {wanted}
        if stale:
            raise ValueError(
                f"{path}: measured over categories {sorted(stale)}, "
                f"but this campaign uses {wanted} "
                "-- re-run the poi-capacity phase"
            )
    return capacities

measure_capacity

measure_capacity(
    osm_path: str | Path,
    city: str,
    *,
    categories: Sequence[str] = POI_CATEGORIES,
    attach_mode: str = POI_ATTACH_NEAREST_VERTEX,
    attach_radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M,
) -> PoiCapacity

Measure one city, reusing the same caches the generator uses.

Deliberately does not call select_customers_poi: that shuffles the pool and builds customer lists to arrive at the same attached count. The ceiling is a property of the attachment map alone, so the map is enough.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/poi_capacity.py

def measure_capacity(
    osm_path: str | Path,
    city: str,
    *,
    categories: Sequence[str] = POI_CATEGORIES,
    attach_mode: str = POI_ATTACH_NEAREST_VERTEX,
    attach_radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M,
) -> PoiCapacity:
    """Measure one city, reusing the same caches the generator uses.

    Deliberately does not call ``select_customers_poi``: that shuffles the pool
    and builds customer lists to arrive at the same ``attached`` count. The
    ceiling is a property of the attachment map alone, so the map is enough.
    """
    path = Path(osm_path)
    graph = load_road_graph(str(path), only_intersections=True, trim_to_connected=True)
    catalog = load_poi_catalog(str(path))
    attachment = catalog_attachment(graph, str(path), attach_mode, attach_radius_m)

    wanted = set(categories)
    vertices: set[int] = set()
    attached = 0
    catalog_size = 0
    for index, poi in enumerate(catalog):
        if poi.category not in wanted:
            continue
        catalog_size += 1
        hit = attachment[index]
        if hit is None:
            continue
        attached += 1
        vertices.add(hit[0])

    capacity = len(vertices)
    return PoiCapacity(
        city=city,
        osm_file=path.name,
        num_vertices=graph.vertex_count,
        catalog_size=catalog_size,
        attached_pois=attached,
        capacity=capacity,
        collapse_ratio=round(attached / capacity, 3) if capacity else 0.0,
        attach_mode=attach_mode,
        attach_radius_m=attach_radius_m,
        categories_digest=categories_digest(categories),
    )

measure_cities

measure_cities(
    cities: Iterable[tuple[str, Path]],
    *,
    categories: Sequence[str] = POI_CATEGORIES,
    attach_mode: str = POI_ATTACH_NEAREST_VERTEX,
    attach_radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M,
    jobs: int = 1,
    on_progress: Callable[[str], None] | None = None,
) -> list[PoiCapacity]

Measure every city; one worker holds one extract at a time.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/poi_capacity.py

def measure_cities(
    cities: Iterable[tuple[str, Path]],
    *,
    categories: Sequence[str] = POI_CATEGORIES,
    attach_mode: str = POI_ATTACH_NEAREST_VERTEX,
    attach_radius_m: float = DEFAULT_POI_ATTACH_RADIUS_M,
    jobs: int = 1,
    on_progress: Callable[[str], None] | None = None,
) -> list[PoiCapacity]:
    """Measure every city; one worker holds one extract at a time."""
    cats = tuple(categories)
    tasks = [
        (city, str(osm_path), cats, attach_mode, attach_radius_m)
        for city, osm_path in cities
    ]
    if not tasks:
        return []

    results: list[PoiCapacity] = []
    if jobs <= 1:
        for task in tasks:
            if on_progress is not None:
                on_progress(task[0])
            results.append(_measure_one(task))
    else:
        with ProcessPoolExecutor(max_workers=jobs) as pool:
            pending = {pool.submit(_measure_one, task): task[0] for task in tasks}
            for future in as_completed(pending):
                result = future.result()
                if on_progress is not None:
                    on_progress(pending[future])
                results.append(result)
    results.sort(key=lambda item: item.city)
    return results

ladder_tier_quotas

ladder_tier_quotas(rung_keys: Sequence[str]) -> list[Quota]

Coverage for a campaign whose sizes and sourcing are already assigned.

Four of the old quotas are gone because the ladder assignment decided them before selection began: method is fixed per rung, avg_route_size too, size_bucket is covered by construction (one instance per distinct n), and city is already one-per-instance. What replaces them is a single pinned quota -- exactly one instance per ladder position -- which is what turns "select 100 good instances" into "select the best instance for each of these 100 slots".

The axes that remain are the ones the assignment left free. They keep their floors so the set stays analysable on each of them, and demand_type gains a ceiling as well -- a floor alone does not stop a max-min selector from parking its spare budget on one level.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/quotas.py

def ladder_tier_quotas(rung_keys: Sequence[str]) -> list[Quota]:
    """Coverage for a campaign whose sizes and sourcing are already assigned.

    Four of the old quotas are gone because the ladder assignment decided them
    before selection began: ``method`` is fixed per rung, ``avg_route_size`` too,
    ``size_bucket`` is covered by construction (one instance per distinct ``n``),
    and ``city`` is already one-per-instance. What replaces them is a single pinned quota --
    exactly one instance per ladder position -- which is what turns "select 100
    good instances" into "select the best instance for each of these 100 slots".

    The axes that remain are the ones the assignment left free. They keep their
    floors so the set stays analysable on each of them, and ``demand_type`` gains
    a *ceiling* as well -- a floor alone does not stop a max-min selector from
    parking its spare budget on one level.
    """
    k = len(rung_keys)
    return [
        # min == max == 1: the selector must fill every rung and may not double
        # up. Quota.cap_for enforces the ceiling hard; _shortfall raises up
        # front if any rung has no usable candidate.
        Quota(
            name="rung",
            key=lambda c: c.spec.rung_key,
            minimum={key: 1 for key in rung_keys},
            maximum={key: 1 for key in rung_keys},
        ),
        Quota(name="city", key=lambda c: c.spec.city, per_level_maximum=1),
        Quota(
            name="distortion_stratum",
            key=lambda c: c.spec.distortion_stratum,
            minimum=_even_minimum(STRATA, k, 0.80),
        ),
        Quota(
            name="demand_type",
            key=lambda c: c.spec.demand_type,
            minimum=_even_minimum(DEMAND_TYPES, k, 0.70),
            maximum=_even_maximum(DEMAND_TYPES, k, 1.60),
        ),
        # No ``avg_route_size`` quota: the ladder assigns the band per rung, so
        # it is already exact rather than merely floored. See
        # ``ladder.assign_route_size_bands``.
        Quota(
            name="depot_mode",
            key=lambda c: c.spec.depot_mode,
            minimum=_even_minimum(DEPOT_MODES, k, 0.75),
        ),
        Quota(
            name="customer_mode",
            key=lambda c: c.spec.customer_mode,
            minimum=_even_minimum(CUSTOMER_MODES, k, 0.75),
        ),
    ]

large_tier_quotas

large_tier_quotas(k: int) -> list[Quota]

Coverage for the handful of large instances.

Far weaker, and deliberately so: at this size the budget is measured in hours per instance, so the only things worth insisting on are that no city appears twice and that the distortion strata are not all the same.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/quotas.py

def large_tier_quotas(k: int) -> list[Quota]:
    """Coverage for the handful of large instances.

    Far weaker, and deliberately so: at this size the budget is measured in
    hours per instance, so the only things worth insisting on are that no city
    appears twice and that the distortion strata are not all the same.
    """
    return [
        Quota(name="city", key=lambda c: c.spec.city, per_level_maximum=1),
        Quota(
            name="distortion_stratum",
            key=lambda c: c.spec.distortion_stratum,
            minimum={stratum: 1 for stratum in STRATA} if k >= len(STRATA) else {},
        ),
    ]

main_tier_quotas

main_tier_quotas(k: int) -> list[Quota]

Coverage the ~100-instance main tier must satisfy.

One instance per city, all three road-distortion strata represented, and every level of every design axis present in analysable numbers. The fractions are deliberately below an even share: they are a floor that makes the set analysable, not a straitjacket that turns selection back into a grid.

Superseded by :func:ladder_tier_quotas for the ladder campaign; kept for the grid-drawn tiers, which still draw their sizes and methods at random.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/quotas.py

def main_tier_quotas(k: int) -> list[Quota]:
    """Coverage the ~100-instance main tier must satisfy.

    One instance per city, all three road-distortion strata represented, and
    every level of every design axis present in analysable numbers. The
    fractions are deliberately below an even share: they are a floor that makes
    the set analysable, not a straitjacket that turns selection back into a grid.

    Superseded by :func:`ladder_tier_quotas` for the ladder campaign; kept for
    the grid-drawn tiers, which still draw their sizes and methods at random.
    """
    return [
        Quota(name="city", key=lambda c: c.spec.city, per_level_maximum=1),
        Quota(
            name="distortion_stratum",
            key=lambda c: c.spec.distortion_stratum,
            minimum=_even_minimum(STRATA, k, 0.80),
        ),
        Quota(
            name="demand_type",
            key=lambda c: c.spec.demand_type,
            minimum=_even_minimum(DEMAND_TYPES, k, 0.70),
        ),
        Quota(
            name="avg_route_size",
            key=lambda c: c.spec.avg_route_size,
            minimum=_even_minimum(AVG_ROUTE_SIZES, k, 0.55),
        ),
        Quota(
            name="depot_mode",
            key=lambda c: c.spec.depot_mode,
            minimum=_even_minimum(DEPOT_MODES, k, 0.75),
        ),
        Quota(
            name="customer_mode",
            key=lambda c: c.spec.customer_mode,
            minimum=_even_minimum(CUSTOMER_MODES, k, 0.75),
        ),
        Quota(
            name="method",
            # The realized sourcing, not the requested one: see
            # EvaluatedCandidate.effective_method.
            key=lambda c: c.effective_method or c.spec.method,
            minimum=_even_minimum(METHODS, k, 0.60),
        ),
        Quota(
            name="size_bucket",
            key=lambda c: size_bucket(c.spec.n),
            minimum=_even_minimum([name for name, _, _ in SIZE_BUCKETS], k, 0.60),
        ),
    ]

poi_tier_quotas

poi_tier_quotas(k: int) -> list[Quota]

Coverage for the POI-only large tier.

Three axes only. The sourcing axis is gone -- every instance in this tier is poi_categories by construction, which is the tier's whole point -- and the demand and depot axes are left to the max-min pass, because the pool is already small: only a handful of cities on Earth hold enough attachable amenities to serve thousands of customers, and quota'ing every axis over ~10 slots would fix the selection rather than guide it.

What is worth insisting on is that the tier spans its size range. An instance set that is nominally "1000 to 5000" but is really nine instances at 1100 and one at 5000 does not measure what happens as n grows.

Route size gets a ceiling and no floor. A floor here would be one of the quotas that fixes rather than guides -- seven levels over ten slots -- but a ceiling costs the selector almost nothing and stops the thing that actually happened in v2, where five of these ten instances landed in the longest band and mamut-athens-n4000-poi came out at 154 customers per route. This is the tier where the longest band is legitimate, so it is capped rather than excluded.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/campaign/quotas.py

def poi_tier_quotas(k: int) -> list[Quota]:
    """Coverage for the POI-only large tier.

    Three axes only. The sourcing axis is gone -- every instance in this tier is
    ``poi_categories`` by construction, which is the tier's whole point -- and
    the demand and depot axes are left to the max-min pass, because the pool is
    already small: only a handful of cities on Earth hold enough attachable
    amenities to serve thousands of customers, and quota'ing every axis over
    ~10 slots would fix the selection rather than guide it.

    What is worth insisting on is that the tier spans its *size* range. An
    instance set that is nominally "1000 to 5000" but is really nine instances
    at 1100 and one at 5000 does not measure what happens as n grows.

    Route size gets a *ceiling* and no floor. A floor here would be one of the
    quotas that fixes rather than guides -- seven levels over ten slots -- but a
    ceiling costs the selector almost nothing and stops the thing that actually
    happened in v2, where five of these ten instances landed in the longest band
    and ``mamut-athens-n4000-poi`` came out at 154 customers per route. This is
    the tier where the longest band is legitimate, so it is capped rather than
    excluded.
    """
    return [
        Quota(name="city", key=lambda c: c.spec.city, per_level_maximum=1),
        Quota(
            name="distortion_stratum",
            key=lambda c: c.spec.distortion_stratum,
            minimum={stratum: 1 for stratum in STRATA} if k >= len(STRATA) else {},
        ),
        Quota(
            name="poi_size_bucket",
            key=lambda c: poi_size_bucket(c.spec.n),
            minimum=_even_minimum([name for name, _, _ in POI_SIZE_BUCKETS], k, 0.60),
        ),
        Quota(
            name="avg_route_size",
            key=lambda c: c.spec.avg_route_size,
            per_level_maximum=max(2, int(k / len(AVG_ROUTE_SIZES) * 1.60)),
        ),
    ]

diversity_report

diversity_report(
    selected: Sequence[EvaluatedCandidate],
    pool: Sequence[EvaluatedCandidate],
    quotas: Sequence[Quota] = (),
) -> dict[str, Any]

Coverage, spread and confounding of a selected set, as publishable data.

feature_correlation is the part worth reading first. If two descriptors are strongly correlated across the selected set, the benchmark cannot separate their effects on a solver, however wide its spread looks.

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

def diversity_report(
    selected: Sequence[EvaluatedCandidate],
    pool: Sequence[EvaluatedCandidate],
    quotas: Sequence[Quota] = (),
) -> dict[str, Any]:
    """Coverage, spread and confounding of a selected set, as publishable data.

    ``feature_correlation`` is the part worth reading first. If two descriptors
    are strongly correlated *across the selected set*, the benchmark cannot
    separate their effects on a solver, however wide its spread looks.
    """
    if not selected:
        return {"selected": 0, "pool": len(pool)}

    # Descriptor units, not the selector's transformed space: this block is read
    # by people, and a route size reported as 1.097 when the instance holds three
    # customers per route is a misreport, not a rounding.
    selected_features = np.asarray(
        [descriptor_values(c.descriptors) for c in selected], dtype=float
    )
    pool_features = np.asarray(
        [descriptor_values(c.descriptors) for c in pool], dtype=float
    )

    features: dict[str, Any] = {}
    for position, name in enumerate(SELECTION_FEATURES):
        column, reference = selected_features[:, position], pool_features[:, position]
        pool_range = float(reference.max() - reference.min())
        selected_range = float(column.max() - column.min())
        features[name] = {
            "min": float(column.min()),
            "median": float(np.median(column)),
            "max": float(column.max()),
            "std": float(column.std()),
            # How much of what the pool offered on this axis the selection kept.
            "range_fraction_of_pool": selected_range / pool_range if pool_range > 0 else None,
        }

    # A feature that happens to be constant across the selection has no
    # correlation with anything; np.corrcoef would divide by its zero spread and
    # fill the row with NaN and a warning. Say "undefined" deliberately instead.
    varying = selected_features.std(axis=0) > 0.0
    correlation = np.full((len(SELECTION_FEATURES), len(SELECTION_FEATURES)), np.nan)
    if varying.sum() >= 2:
        block = np.corrcoef(selected_features[:, varying], rowvar=False)
        correlation[np.ix_(varying, varying)] = block
    worst_pair, worst_value = None, 0.0
    for i in range(len(SELECTION_FEATURES)):
        for j in range(i + 1, len(SELECTION_FEATURES)):
            value = correlation[i, j]
            if np.isfinite(value) and abs(value) > abs(worst_value):
                worst_pair = (SELECTION_FEATURES[i], SELECTION_FEATURES[j])
                worst_value = float(value)

    coverage = {
        quota.name: {
            str(level): count
            for level, count in sorted(
                Counter(quota.key(c) for c in selected).items(), key=lambda kv: str(kv[0])
            )
        }
        for quota in quotas
    }
    unmet = {
        quota.name: {str(level): short for level, short in quota.deficit(
            Counter(quota.key(c) for c in selected)
        ).items()}
        for quota in quotas
        if quota.deficit(Counter(quota.key(c) for c in selected))
    }

    standardized = _feature_matrix(list(pool))
    index_of = {id(c): i for i, c in enumerate(pool)}
    chosen = [index_of[id(c)] for c in selected if id(c) in index_of]
    spread = _spread_score(
        np.linalg.norm(standardized[:, None, :] - standardized[None, :, :], axis=2), chosen
    ) if len(chosen) == len(selected) else ()

    return {
        "selected": len(selected),
        "pool": len(pool),
        "coverage": coverage,
        "unmet_quotas": unmet,
        "features": features,
        "feature_correlation": {
            "names": list(SELECTION_FEATURES),
            # JSON has no NaN; an undefined correlation is null.
            "matrix": [
                [float(v) if np.isfinite(v) else None for v in row] for row in correlation
            ],
            "strongest_pair": list(worst_pair) if worst_pair else None,
            "strongest_value": worst_value,
        },
        "min_pairwise_distance": float(spread[0]) if spread else None,
    }

select_maxmin

select_maxmin(
    candidates: Sequence[EvaluatedCandidate],
    k: int,
    quotas: Sequence[Quota] = (),
    *,
    swap_passes: int = 2,
) -> list[EvaluatedCandidate]

k candidates spread over feature space and honouring every quota.

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

def select_maxmin(
    candidates: Sequence[EvaluatedCandidate],
    k: int,
    quotas: Sequence[Quota] = (),
    *,
    swap_passes: int = 2,
) -> list[EvaluatedCandidate]:
    """``k`` candidates spread over feature space and honouring every quota."""
    if k <= 0:
        raise ValueError("k must be positive")
    if len(candidates) < k:
        raise QuotaInfeasibleError(
            f"pool has {len(candidates)} usable candidates, fewer than the requested {k}"
        )

    features = _feature_matrix(candidates)
    distances = np.linalg.norm(features[:, None, :] - features[None, :, :], axis=2)
    order = sorted(range(len(candidates)), key=lambda i: _tie_break(candidates[i]))
    counts: dict[str, Counter] = {quota.name: Counter() for quota in quotas}
    chosen: list[int] = []
    chosen_set: set[int] = set()

    def level(quota: Quota, index: int) -> Hashable:
        return quota.key(candidates[index])

    def admissible(index: int) -> bool:
        return all(
            (cap := quota.cap_for(level(quota, index))) is None
            or counts[quota.name][level(quota, index)] < cap
            for quota in quotas
        )

    def deficits_reduced(index: int) -> int:
        """How many outstanding quota minimums taking ``index`` would advance."""
        return sum(
            1
            for quota in quotas
            if level(quota, index) in quota.deficit(counts[quota.name])
        )

    def outstanding() -> str:
        return ", ".join(
            f"{quota.name}={quota.deficit(counts[quota.name])}"
            for quota in quotas
            if quota.deficit(counts[quota.name])
        )

    def keeps_quotas_reachable(index: int) -> bool:
        """Would taking ``index`` leave every quota still satisfiable?

        A one-step lookahead rather than a "is this quota binding right now"
        test: it is what stops the greedy walking into a corner where the
        remaining budget cannot cover what the remaining candidates offer.
        """
        trial = {name: counter.copy() for name, counter in counts.items()}
        for quota in quotas:
            trial[quota.name][level(quota, index)] += 1
        return (
            _shortfall(quotas, candidates, trial, chosen_set | {index}, k - len(chosen) - 1)
            is None
        )

    def take(index: int) -> None:
        chosen.append(index)
        chosen_set.add(index)
        for quota in quotas:
            counts[quota.name][level(quota, index)] += 1

    # Fail before doing any work when the pool simply cannot meet the quotas.
    upfront = _shortfall(quotas, candidates, counts, chosen_set, k)
    if upfront is not None:
        raise QuotaInfeasibleError(upfront)

    # Seed with the most extreme admissible candidate: farthest from the pool's
    # centre of mass, so the first pick is an edge of the cloud rather than an
    # arbitrary point in the middle of it.
    from_centre = np.linalg.norm(features - features.mean(axis=0), axis=1)
    start = [i for i in order if admissible(i) and keeps_quotas_reachable(i)]
    if not start:
        raise QuotaInfeasibleError("no first pick leaves the quotas satisfiable")
    take(max(start, key=lambda i: (from_centre[i], _tie_break(candidates[i]))))

    while len(chosen) < k:
        open_picks = [i for i in order if i not in chosen_set and admissible(i)]
        if not open_picks:
            capped = ", ".join(
                quota.name for quota in quotas if quota.per_level_maximum or quota.maximum
            )
            raise QuotaInfeasibleError(
                f"only {len(chosen)} of {k} instances could be picked: every remaining "
                f"candidate is blocked by a per-level cap ({capped or 'none declared'})"
            )
        allowed = [i for i in open_picks if keeps_quotas_reachable(i)]
        if not allowed:
            # One-step lookahead cannot see joint tightness across several
            # quotas at once, so when it finds nothing safe, keep going with
            # whatever still advances a minimum.
            allowed = [i for i in open_picks if deficits_reduced(i) > 0]
        if not allowed:
            reason = _shortfall(quotas, candidates, counts, chosen_set, k - len(chosen))
            raise QuotaInfeasibleError(
                reason
                or f"no candidate advances the outstanding quotas after {len(chosen)} "
                f"pick(s): {outstanding()}"
            )

        # Coverage first, spread within it. While any minimum is outstanding the
        # primary key is how many deficits a pick advances; once they are all
        # met this is exactly pure max-min.
        #
        # The order matters and is not a detail. Chasing spread first and
        # repairing coverage at the end fails whenever the picks are already
        # nearly forced -- one instance per city over 102 cities with a budget of
        # 100 leaves the selector only *which* of a city's candidates to take,
        # and by the time the deficits are noticed the cities that could have
        # filled them are used up. Ties on deficit count are broken by max-min,
        # and early on most candidates advance the same number of quotas, so
        # spread still drives the choice.
        covering = any(quota.deficit(counts[quota.name]) for quota in quotas)
        nearest = distances[np.ix_(allowed, chosen)].min(axis=1)
        best = max(
            range(len(allowed)),
            key=lambda position: (
                deficits_reduced(allowed[position]) if covering else 0,
                nearest[position],
                _tie_break(candidates[allowed[position]]),
            ),
        )
        take(allowed[best])

    unmet = [quota for quota in quotas if quota.deficit(counts[quota.name])]
    if unmet:
        raise QuotaInfeasibleError(
            "the budget could not cover every minimum: "
            + ", ".join(f"{q.name}={q.deficit(counts[q.name])}" for q in unmet)
        )

    chosen = _improve_by_swapping(chosen, candidates, distances, quotas, swap_passes)
    return [candidates[index] for index in chosen]