mamut_routing_tools.family

Benchmark-family generation: publish the whole family into a marker-rooted collection tree (slim CVRP / VRPTW instances, shared geo / road / traffic / distances sidecars, and the TDVRP/TDVRPTW twins of the road-graph v2 td model).

Relocated from mamut_routing_publish.td_generation (session 34): generation now lives in the generation tool. Consumes the TD bridge produced by mamut_routing_tools.td.traffic plus the stage-1 sampling intermediates from mamut_routing_tools.generation.

build_base

build_base(
    *,
    graph: BridgeGraph,
    nodes: BridgeNodes,
    meta: dict[str, Any],
    manifest: dict[str, Any],
    city: str,
    method_tag: str,
    collection_root: str | Path,
    sample_step: float = DEFAULT_SAMPLE_STEP,
    extension_end: float = DEFAULT_EXTENSION_END,
    generated_at: str | None = None,
    force: bool = False,
    family: str = FAMILY,
    commit_distances: bool = True,
) -> BuiltBase | None

Publish the base: road + geo + distances sidecars and the 3 CVRP instances.

commit_distances=False computes and hashes the distance matrices exactly as usual, records the pins in every instance's arc_costs_source, and then does not write the two .distances-*.json.gz files. The instance is a complete descriptor that a consumer regenerates its matrices for, verified against the pins -- the treatment the Blauth2024 family already uses for its oversized sidecars. It implies pin_shortest_paths, since a regenerated shortest matrix has to come from the published road graph.

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

def build_base(
    *,
    graph: BridgeGraph,
    nodes: BridgeNodes,
    meta: dict[str, Any],
    manifest: dict[str, Any],
    city: str,
    method_tag: str,
    collection_root: str | Path,
    sample_step: float = DEFAULT_SAMPLE_STEP,
    extension_end: float = DEFAULT_EXTENSION_END,
    generated_at: str | None = None,
    force: bool = False,
    family: str = FAMILY,
    commit_distances: bool = True,
) -> BuiltBase | None:
    """Publish the base: road + geo + distances sidecars and the 3 CVRP instances.

    ``commit_distances=False`` computes and hashes the distance matrices exactly
    as usual, records the pins in every instance's ``arc_costs_source``, and then
    does not write the two ``.distances-*.json.gz`` files. The instance is a
    complete descriptor that a consumer regenerates its matrices for, verified
    against the pins -- the treatment the Blauth2024 family already uses for its
    oversized sidecars. It implies ``pin_shortest_paths``, since a regenerated
    shortest matrix has to come from the published road graph.
    """
    started = time.perf_counter()
    num_customers = len(nodes.node_osm_ids) - 1
    # Generation already solved the bin-packing minimum and recorded it; families
    # that put the fleet in the name read it from there rather than recomputing.
    base = base_instance_name(
        city,
        num_customers,
        method_tag,
        family,
        route_count=(
            int(manifest["route_count"])
            if family in FAMILIES_WITH_FLEET_IN_NAME
            else None
        ),
    )
    root = ensure_collection_root(collection_root, family)
    generated_at = generated_at or time.strftime("%Y-%m-%d")

    side_dir = sidecar_dir(root, city, num_customers, base)
    road_path = side_dir / f"{base}.road.json.gz"
    geo_path = side_dir / f"{base}.geo.json.gz"
    cvrp_json_paths = {
        metric: cvrp_dir(root, metric, city, num_customers, base) / f"{base}.vrp.json"
        for metric in ("euclidean", "shortest", "fastest")
    }
    if not force and road_path.exists() and all(p.exists() for p in cvrp_json_paths.values()):
        return None

    tolerance = simplify_tolerance_for(num_customers)
    generator = _base_generator(
        "generate-base",
        {"city": city, "method": method_tag, "sampling_seed": sampling_seed(base)},
    )
    full = _full_city_road_graph(
        graph,
        nodes,
        base=base,
        sample_step=sample_step,
        simplify_tolerance=tolerance,
        extension_end=extension_end,
        generator=generator,
        family=family,
    )
    # An uncommitted distance sidecar has to be reproducible from published
    # bytes alone, which means the road graph must pin the shortest-distance
    # paths as well as the fastest-time ones.
    road, _ = _trim_road_graph(full, pin_shortest_paths=not commit_distances)
    save_instance_road_graph(road, road_path)
    road_sha = compute_road_graph_sha256(road)

    # Distance matrices: fastest on the trimmed graph (identical to the full
    # graph by the trim gate), shortest on the full graph (its paths may use
    # edges outside the fastest trim).
    fastest_matrix = _round_matrix(free_flow_node_times(road))
    shortest_full = _length_weighted(full)
    shortest_values, shortest_paths = _node_paths(
        shortest_full, want_paths=num_customers <= ROAD_CACHE_MAX_N
    )
    shortest_matrix = _round_matrix(shortest_values)

    if not commit_distances:
        # The pin is a promise that these bytes can be rebuilt from what is
        # published. Check it here rather than discovering at load time that the
        # trim lost an edge the shortest metric needed.
        rebuilt, _ = _node_paths(_length_weighted(road), want_paths=False)
        if _round_matrix(rebuilt) != shortest_matrix:
            raise AssertionError(
                f"{base}: the published road graph does not reproduce "
                "distances-shortest; the trim is missing shortest-path edges"
            )

    distances_sha: dict[str, str] = {}
    for metric, values in (("fastest", fastest_matrix), ("shortest", shortest_matrix)):
        distances = InstanceDistances(
            base_name=base,
            benchmark_name=family,
            metric=metric,
            num_customers=num_customers,
            values=values,
            generator=_base_generator("generate-base"),
        )
        distances_sha[metric] = compute_distances_sha256(distances)
        if commit_distances:
            save_instance_distances(distances, side_dir / f"{base}.distances-{metric}.json.gz")

    # Geo sidecar: nodes + (n <= 100) the complete indexed road cache with
    # fastest paths pinned on the trimmed graph and shortest paths on the
    # full graph, over one shared local vertex table keyed by OSM id.
    road_cache = None
    if num_customers <= ROAD_CACHE_MAX_N:
        _, fastest_paths = _node_paths(road)
        used_osm: set[int] = set()
        for path in fastest_paths.values():
            used_osm.update(road.vertex_osm_ids[v] for v in path)
        for path in shortest_paths.values():
            used_osm.update(shortest_full.vertex_osm_ids[v] for v in path)
        local_osm = sorted(used_osm)
        local_of = {osm: index for index, osm in enumerate(local_osm)}
        cache_paths = {
            "fastest": {
                f"{i}-{j}": [local_of[road.vertex_osm_ids[v]] for v in path]
                for (i, j), path in fastest_paths.items()
            },
            "shortest": {
                f"{i}-{j}": [local_of[shortest_full.vertex_osm_ids[v]] for v in path]
                for (i, j), path in shortest_paths.items()
            },
        }
        road_cache = GeoRoadCache(
            vertex_lonlat=[graph.vertex_lonlat[osm] for osm in local_osm],
            paths=cache_paths,
        )
    meta_nodes = meta["nodes"]
    geo = InstanceGeo(
        base_name=base,
        benchmark_name=family,
        city=city,
        method=method_tag,
        source_osm_file=str(meta.get("source_osm_file", graph.osm_file)),
        reference_lla={
            "lat": float(meta["reference_lla"]["lat"]),
            "lon": float(meta["reference_lla"]["lon"]),
            "alt": float(meta["reference_lla"].get("alt", 0.0)),
        },
        map_options=dict(meta.get("map_options", graph.map_options)),
        nodes=[
            GeoNode(
                instance_node_id=index,
                poi_lon=float(node["poi_lon"]),
                poi_lat=float(node["poi_lat"]),
                enu_x=float(node["enu_x"]),
                enu_y=float(node["enu_y"]),
                demand=int(node["demand"]),
                source_tag=str(node["source_tag"]),
                graph_vertex_id=road.node_vertices[index],
            )
            for index, node in enumerate(meta_nodes)
        ],
        road_cache=road_cache,
        generator=_base_generator("generate-base", {"city": city, "method": method_tag}),
    )
    save_instance_geo(geo, geo_path)
    geo_sha = compute_geo_sha256(geo)
    geo_ref = {
        "path": sidecar_relpath(city, num_customers, base, f"{base}.geo.json.gz"),
        "sha256": geo_sha,
    }

    # The 3 slim CVRP metric instances (+ CVRPLIB .vrp exports for n <= 100).
    matrices = {"shortest": shortest_matrix, "fastest": fastest_matrix}
    coordinates = [[float(node["enu_x"]), float(node["enu_y"])] for node in meta_nodes]
    demands = [int(node["demand"]) for node in meta_nodes]
    for metric in ("euclidean", "shortest", "fastest"):
        if metric == "euclidean":
            source = {"model": "euclidean", "decimals": DISTANCE_DECIMALS}
        else:
            source = {
                "model": "distances-sidecar",
                "distances": {
                    "path": sidecar_relpath(
                        city, num_customers, base, f"{base}.distances-{metric}.json.gz"
                    ),
                    "sha256": distances_sha[metric],
                },
            }
        payload = _static_instance_payload(
            base=base,
            meta=meta,
            manifest=manifest,
            metric=metric,
            arc_costs_source=source,
            city=city,
            method_tag=method_tag,
            geo_ref=geo_ref,
            generated_at=generated_at,
            family=family,
        )
        target = cvrp_json_paths[metric]
        target.parent.mkdir(parents=True, exist_ok=True)
        save_json_to_file(payload, target)
        if num_customers <= VRP_EXPORT_MAX_N:
            if metric == "euclidean":
                import math

                matrix = [
                    [
                        0.0 if i == j else round(math.hypot(bx - ax, by - ay), DISTANCE_DECIMALS)
                        for j, (bx, by) in enumerate(coordinates)
                    ]
                    for i, (ax, ay) in enumerate(coordinates)
                ]
            else:
                matrix = matrices[metric]
            _write_cvrplib(
                target.parent / f"{base}.vrp",
                name=base,
                comment=_cvrplib_comment(family, metric, city, base, manifest),
                coordinates=coordinates,
                demands=demands,
                matrix=matrix,
                capacity=int(manifest["capacity"]),
            )

    return BuiltBase(
        base=base,
        city=city,
        num_customers=num_customers,
        method_tag=method_tag,
        road_sha256=road_sha,
        geo_sha256=geo_sha,
        distances_sha256=distances_sha,
        num_road_vertices=road.num_vertices,
        num_road_edges=len(road.edges),
        build_seconds=time.perf_counter() - started,
        cvrp_paths=cvrp_json_paths,
        distances_committed=commit_distances,
    )

build_td

build_td(
    *,
    collection_root: str | Path,
    graph: BridgeGraph,
    speeds_by_combo: dict[tuple[str, str], BridgeSpeeds],
    city: str,
    num_customers: int,
    method_tag: str,
    generated_at: str | None = None,
    force: bool = False,
    verify: bool = True,
    tdvrptw_only: bool = False,
    reuse_traffic: bool = False,
    family: str = FAMILY,
    route_count: int | None = None,
) -> BuiltTDBase | None

Publish the TD layer of a base: 6 overlays, TW lift, 12 slim twins.

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

def build_td(
    *,
    collection_root: str | Path,
    graph: BridgeGraph,
    speeds_by_combo: dict[tuple[str, str], BridgeSpeeds],
    city: str,
    num_customers: int,
    method_tag: str,
    generated_at: str | None = None,
    force: bool = False,
    verify: bool = True,
    tdvrptw_only: bool = False,
    reuse_traffic: bool = False,
    family: str = FAMILY,
    #: Bin-packing minimum fleet, for families whose base names carry it.
    #: Required for those and rejected for the others -- see
    #: ``naming.base_instance_name``.
    route_count: int | None = None,
) -> BuiltTDBase | None:
    """Publish the TD layer of a base: 6 overlays, TW lift, 12 slim twins."""
    reuse_traffic = reuse_traffic or tdvrptw_only
    started = time.perf_counter()
    root = Path(collection_root)
    base = base_instance_name(
        city, num_customers, method_tag, family, route_count=route_count
    )
    side_dir = sidecar_dir(root, city, num_customers, base)
    road_path = side_dir / f"{base}.road.json.gz"
    if not road_path.exists():
        raise FileNotFoundError(f"road sidecar missing for {base}: run generate-base first")
    vrptw_path = vrptw_dir(root, city, num_customers, base) / f"{base}.vrp.json"
    if not vrptw_path.exists():
        raise FileNotFoundError(f"VRPTW candidate missing for {base}: run derive-vrptw first")

    combos = [
        (model, intensity)
        for model in TD_MODELS
        for intensity in TD_INTENSITIES
    ]
    missing = [combo for combo in combos if combo not in speeds_by_combo] if not reuse_traffic else []
    if missing:
        raise ValueError(f"missing traffic speeds for {base}: {missing}")

    subs = [subinstance_name(model, intensity) for model, intensity in combos]
    twin_paths = {
        (pt, sub): td_instance_dir(root, pt, city, num_customers, base, sub)
        / f"{td_instance_name(base, *sub.split('-', 1))}.vrp.json"
        for pt in ("TDVRP", "TDVRPTW")
        for sub in subs
    }
    target_problem_types = ("TDVRPTW",) if tdvrptw_only else ("TDVRP", "TDVRPTW")
    if not force and all(
        twin_paths[(problem_type, sub)].exists()
        for problem_type in target_problem_types
        for sub in subs
    ):
        return None

    road = load_instance_road_graph(road_path)
    road_sha = compute_road_graph_sha256(road)

    # Gate: the published distances-fastest equals the road graph's free-flow
    # node times after the family rounding.
    stored = load_instance_distances(side_dir / f"{base}.distances-fastest.json.gz")
    recomputed = _round_matrix(free_flow_node_times(road))
    if stored.values != recomputed:
        raise AssertionError(f"distances-fastest gate failed for {base}")

    import json as _json

    vrptw_payload = _json.loads(vrptw_path.read_text())
    service_times = [int(v) for v in vrptw_payload["service_times"]]
    time_windows = [tuple(int(v) for v in window) for window in vrptw_payload["time_windows"]]
    anchor_routes = [
        [int(customer) for customer in route]
        for route in vrptw_payload.get("metadata", {}).get("tw_anchor", {}).get("routes", [])
    ]
    if not anchor_routes:
        raise ValueError(f"route-centered TW anchor missing for {base}: re-run derive-vrptw")
    anchor_arc_keys = {
        (previous, customer)
        for route in anchor_routes
        for previous, customer in zip([0, *route], [*route, 0])
    }

    # Overlays + per-subinstance materialization (ATFs are TW-independent).
    # Only the anchor arcs are retained for the global TW certificate: full
    # n=1000 ATF sets are heavy and all other arcs are irrelevant here.
    overlay_sha: dict[str, str] = {}
    atf_sha: dict[str, str] = {}
    anchor_arcs_by_sub: dict[str, dict[tuple[int, int], Any]] = {}
    overlay_refs: dict[str, dict[str, str]] = {}
    existing_tdvrp_payloads: dict[str, dict[str, Any]] = {}
    for model, intensity in combos:
        sub = subinstance_name(model, intensity)
        overlay_file = f"{base}.traffic-{sub}.json.gz"
        overlay_path = side_dir / overlay_file
        if reuse_traffic and overlay_path.exists():
            overlay = load_traffic_overlay(overlay_path)
        else:
            if (model, intensity) not in speeds_by_combo:
                raise ValueError(f"missing traffic speeds for {base}: {(model, intensity)}")
            overlay = _align_overlay(road, graph, speeds_by_combo[(model, intensity)], family)
            save_traffic_overlay(overlay, overlay_path)
        overlay_sha[sub] = compute_traffic_overlay_sha256(overlay)
        overlay_refs[sub] = {
            "path": sidecar_relpath(city, num_customers, base, overlay_file),
            "sha256": overlay_sha[sub],
        }
        name = td_instance_name(base, model, intensity)
        probe_payload = {
            "instance_name": name,
            "instance_origin": "OsmCvrpGen",
            "benchmark_name": family,
            "num_customers": num_customers,
            "num_vehicles": None,
            "vehicle_capacity": int(vrptw_payload["vehicle_capacity"]),
            "coordinates": vrptw_payload["coordinates"],
            "demands": vrptw_payload["demands"],
            "service_times": service_times,
            "depot": 0,
            "horizon": list(TD_HORIZON),
            "td": {
                "model": "road-graph",
                "graph": {
                    "path": sidecar_relpath(city, num_customers, base, f"{base}.road.json.gz"),
                    "sha256": road_sha,
                },
                "traffic": dict(overlay_refs[sub]),
                "sample_step": road.sample_step,
                "simplify_tolerance": road.simplify_tolerance,
            },
            "metadata": {},
        }
        instance = td_instance_from_payload(probe_payload)
        anchor_arcs_by_sub[sub] = materialize_selected_atfs_roadgraph(
            instance, road, overlay, anchor_arc_keys
        )
        existing_tdvrp_path = twin_paths[("TDVRP", sub)]
        if existing_tdvrp_path.exists():
            existing_tdvrp = _json.loads(existing_tdvrp_path.read_text())
            existing_tdvrp_payloads[sub] = existing_tdvrp
            atf_sha[sub] = str(existing_tdvrp["td"]["atf_sha256"])
        else:
            atfs = materialize_instance_atfs_roadgraph(instance, road, overlay)
            atf_sha[sub] = compute_atf_sha256(atfs)
            del atfs
        missing_anchor_arcs = anchor_arc_keys - anchor_arcs_by_sub[sub].keys()
        if missing_anchor_arcs:
            raise AssertionError(f"missing anchor arcs under {sub}: {sorted(missing_anchor_arcs)}")

    # Global anchor audit + minimal shared deadline repair; finalize VRPTW.
    lifted, repairs = _audit_and_lift(
        time_windows,
        service_times,
        anchor_routes,
        anchor_arcs_by_sub,
        TD_HORIZON[1],
    )
    vrptw_payload["time_windows"] = [list(window) for window in lifted]
    vrptw_metadata = dict(vrptw_payload["metadata"])
    lift_entries = [e for e in repairs.values() if "deadline_after" in e]
    cut_entries = [e for e in repairs.values() if "earliest_after" in e]
    vrptw_metadata["tw_repair"] = {
        "policy": "global-anchor-minimal-shared-deadline-lift",
        "overlays_audited": subs,
        "lifted_customers": len(lift_entries),
        "max_lift_seconds": max(
            (e["deadline_after"] - e["deadline_before"] for e in lift_entries), default=0
        ),
        "reduced_customers": 0,
        "max_reduction_seconds": 0,
        "anchor_routes": len(anchor_routes),
        "repairs": repairs,
    }
    if generated_at:
        vrptw_metadata["generated_at"] = generated_at
    vrptw_payload["metadata"] = vrptw_metadata
    save_json_to_file(vrptw_payload, vrptw_path)

    # Emit the 12 slim TD twins.
    reference_lla = vrptw_payload.get("reference_lla")
    geo_ref = vrptw_metadata.get("sidecars", {}).get("geo")
    for model, intensity in combos:
        sub = subinstance_name(model, intensity)
        name = td_instance_name(base, model, intensity)
        if reuse_traffic:
            td_metadata = dict(existing_tdvrp_payloads[sub]["metadata"])
            td_metadata.pop("problem_type", None)
            td_metadata["generated_at"] = generated_at or td_metadata.get("generated_at", "")
            td_metadata["generator"] = _base_generator(
                "build-td", {"city": city, "method": method_tag}
            )
            td_metadata["sidecars"] = {"geo": dict(geo_ref)} if geo_ref else {}
        else:
            speeds = speeds_by_combo[(model, intensity)]
            td_metadata = {
                "authors": AUTHORS,
                "generated_at": generated_at or vrptw_metadata.get("generated_at", ""),
                "city": city,
                "method": method_tag,
                "base_instance_name": base,
                "subinstance": sub,
                "generator": _base_generator(
                    "build-td", {"city": city, "method": method_tag}
                ),
                "traffic": {
                    "model": speeds.model,
                    "intensity": speeds.intensity,
                    "seed": speeds.seed,
                    "num_trips": speeds.num_trips,
                    "params": speeds.params,
                },
                "sidecars": {"geo": dict(geo_ref)} if geo_ref else {},
                "notes": (
                    "Time dependence derives from the city's OSM road network: the shared "
                    "road-graph sidecar pins free-flow fastest paths per base; this "
                    "subinstance's traffic overlay carries per-edge hourly speeds; "
                    "arrival-time functions are materialized deterministically on load "
                    "and pinned by td.atf_sha256."
                ),
            }
        common = {
            "instance_name": name,
            "instance_origin": "OsmCvrpGen",
            "benchmark_name": family,
            "num_customers": num_customers,
            "num_vehicles": None,
            "vehicle_capacity": int(vrptw_payload["vehicle_capacity"]),
            "coordinates": vrptw_payload["coordinates"],
            "demands": vrptw_payload["demands"],
            "service_times": service_times,
            "depot": 0,
            "reference_lla": reference_lla,
            "horizon": list(TD_HORIZON),
            "td": {
                "model": "road-graph",
                "graph": {
                    "path": sidecar_relpath(city, num_customers, base, f"{base}.road.json.gz"),
                    "sha256": road_sha,
                },
                "traffic": dict(overlay_refs[sub]),
                "atf_sha256": atf_sha[sub],
                "sample_step": road.sample_step,
                "simplify_tolerance": road.simplify_tolerance,
            },
        }
        tdvrp_payload = dict(common)
        tdvrp_payload["metadata"] = {**td_metadata, "problem_type": "TDVRP"}
        tdvrptw_payload = dict(common)
        tdvrptw_payload["time_windows"] = [list(window) for window in lifted]
        tdvrptw_payload["metadata"] = {
            **td_metadata,
            "problem_type": "TDVRPTW",
            "tw_repair": vrptw_metadata["tw_repair"],
        }
        payloads = (("TDVRPTW", tdvrptw_payload),) if tdvrptw_only else (
            ("TDVRP", tdvrp_payload),
            ("TDVRPTW", tdvrptw_payload),
        )
        for pt, payload in payloads:
            target = twin_paths[(pt, sub)]
            target.parent.mkdir(parents=True, exist_ok=True)
            save_json_to_file(payload, target)
            if verify:
                load_td_instance(target, verify_sha256=True)

    return BuiltTDBase(
        base=base,
        subinstances=subs,
        atf_sha256=atf_sha,
        traffic_sha256=overlay_sha,
        lifted_customers=len(lift_entries),
        max_lift_seconds=max(
            (e["deadline_after"] - e["deadline_before"] for e in lift_entries), default=0
        ),
        reduced_customers=0,
        max_reduction_seconds=0,
        build_seconds=time.perf_counter() - started,
    )

capacity_lower_bound

capacity_lower_bound(
    demands: list[int], capacity: int
) -> int

Return the standard capacity lower bound, excluding the depot demand.

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

def capacity_lower_bound(demands: list[int], capacity: int) -> int:
    """Return the standard capacity lower bound, excluding the depot demand."""
    if capacity <= 0:
        raise ValueError("vehicle capacity must be positive")
    return math.ceil(sum(demands[1:]) / capacity)

derive_vrptw

derive_vrptw(
    *,
    collection_root: str | Path,
    city: str,
    num_customers: int,
    method_tag: str,
    tw_set: str = TW_SET_TD_SHARED,
    generated_at: str | None = None,
    force: bool = False,
    family: str = FAMILY,
    route_count: int | None = None,
) -> tuple[str, Path]

Emit one VRPTW TW-set instance of a base.

td-shared emits the candidate finalized by build_td (audit + minimal shared TW repair); the static-only sets (tight, spread) are final as written and never touched by build_td. Returns (action, path) with action in derived / kept / annotated (the latter: an existing td-shared instance received the tw_set metadata block in place, windows untouched).

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

def derive_vrptw(
    *,
    collection_root: str | Path,
    city: str,
    num_customers: int,
    method_tag: str,
    tw_set: str = TW_SET_TD_SHARED,
    generated_at: str | None = None,
    force: bool = False,
    family: str = FAMILY,
    #: Bin-packing minimum fleet, for families whose base names carry it.
    #: Required for those and rejected for the others -- see
    #: ``naming.base_instance_name``.
    route_count: int | None = None,
) -> tuple[str, Path]:
    """Emit one VRPTW TW-set instance of a base.

    ``td-shared`` emits the candidate finalized by ``build_td`` (audit +
    minimal shared TW repair); the static-only sets (``tight``, ``spread``)
    are final as written and never touched by ``build_td``. Returns
    ``(action, path)`` with action in ``derived`` / ``kept`` / ``annotated``
    (the latter: an existing td-shared instance received the ``tw_set``
    metadata block in place, windows untouched).
    """
    if tw_set not in ALL_TW_SETS:
        raise ValueError(f"unknown TW set {tw_set!r} (expected one of {ALL_TW_SETS})")
    root = Path(collection_root)
    base = base_instance_name(
        city, num_customers, method_tag, family, route_count=route_count
    )
    name = vrptw_instance_name(base, tw_set)
    target_dir = vrptw_dir(root, city, num_customers, base)
    target = target_dir / f"{name}.vrp.json"
    if not force and target.exists():
        if tw_set == TW_SET_TD_SHARED and _annotate_tw_set(target):
            return "annotated", target
        return "kept", target

    cvrp_fastest = cvrp_dir(root, "fastest", city, num_customers, base) / f"{base}.vrp.json"
    if not cvrp_fastest.exists():
        raise FileNotFoundError(f"CVRP fastest instance missing for {base}: run generate-base first")
    import json as _json

    cvrp_payload = _json.loads(cvrp_fastest.read_text())
    fastest = load_instance_distances(
        root / cvrp_payload["arc_costs_source"]["distances"]["path"]
    ).values

    service_times = synthesize_service_times(Random(_stable_seed(base, "service")), num_customers)
    tw_seed_label = "tw" if tw_set == TW_SET_TD_SHARED else f"tw:{tw_set}"
    tw_rng = Random(_stable_seed(base, tw_seed_label))
    if tw_set == TW_SET_SPREAD:
        tw_anchor = "uniform-feasible"
        time_windows = synthesize_time_windows_spread(tw_rng, fastest, service_times)
    else:
        tw_anchor = "free-flow-fastest"
        demands = [int(value) for value in cvrp_payload["demands"]]
        capacity = int(cvrp_payload["vehicle_capacity"])
        anchor_routes, visit_times = construct_anchor_routes(
            fastest, demands, capacity, service_times
        )
        if tw_set == TW_SET_TIGHT:
            time_windows = synthesize_time_windows(
                tw_rng,
                fastest,
                service_times,
                visit_times,
                width_ratio_mean=TIGHT_TW_WIDTH_RATIO_MEAN,
                width_ratio_std=TIGHT_TW_WIDTH_RATIO_STD,
                width_ratio_min=TIGHT_TW_WIDTH_RATIO_MIN,
                width_ratio_max=TIGHT_TW_WIDTH_RATIO_MAX,
            )
        else:
            time_windows = synthesize_time_windows(tw_rng, fastest, service_times, visit_times)
        validate_static_anchor(
            anchor_routes,
            fastest,
            demands,
            capacity,
            service_times,
            time_windows,
        )

    generator_extra: dict[str, Any] = {
        "city": city,
        "method": method_tag,
        "service_seed": _stable_seed(base, "service"),
        "tw_seed": _stable_seed(base, tw_seed_label),
        "tw_anchor": tw_anchor,
    }
    if tw_set != TW_SET_TD_SHARED:
        generator_extra["tw_set"] = tw_set

    payload = dict(cvrp_payload)
    payload["instance_name"] = name
    payload["service_times"] = service_times
    payload["time_windows"] = [list(window) for window in time_windows]
    metadata = dict(cvrp_payload["metadata"])
    metadata["problem_type"] = "VRPTW"
    metadata["generator"] = _base_generator("derive-vrptw", generator_extra)
    if generated_at:
        metadata["generated_at"] = generated_at
    metadata["tw_set"] = TW_SET_METADATA[tw_set]
    if tw_set != TW_SET_SPREAD:
        metadata["tw_anchor"] = {
            "policy": "deterministic-capacity-horizon-nearest-neighbour",
            "routes": anchor_routes,
            "num_routes": len(anchor_routes),
            "min_customers_per_route": min(map(len, anchor_routes)),
            "max_customers_per_route": max(map(len, anchor_routes)),
        }
    payload["metadata"] = metadata
    target_dir.mkdir(parents=True, exist_ok=True)
    save_json_to_file(payload, target)
    return "derived", target

materialize_distances

materialize_distances(
    instance_path: str | Path,
    *,
    collection_root: str | Path | None = None,
    force: bool = False,
) -> dict[str, Any]

Rebuild an instance's distance sidecar from its published road graph.

The counterpart of build_base(commit_distances=False). Large instances ship as descriptors: the .vrp.json records the sidecar's path and its sha256, and the bytes themselves are regenerated locally rather than stored. This recomputes them from the committed .road.json.gz and refuses to write anything whose digest does not match the recorded pin, so a materialized sidecar is exactly the one the instance was published against.

Returns a record describing what happened; status is written, kept (already present) or euclidean (nothing to do -- that variant derives its costs from the coordinates).

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

def materialize_distances(
    instance_path: str | Path,
    *,
    collection_root: str | Path | None = None,
    force: bool = False,
) -> dict[str, Any]:
    """Rebuild an instance's distance sidecar from its published road graph.

    The counterpart of ``build_base(commit_distances=False)``. Large instances
    ship as descriptors: the ``.vrp.json`` records the sidecar's path and its
    sha256, and the bytes themselves are regenerated locally rather than stored.
    This recomputes them from the committed ``.road.json.gz`` and refuses to
    write anything whose digest does not match the recorded pin, so a
    materialized sidecar is exactly the one the instance was published against.

    Returns a record describing what happened; ``status`` is ``written``,
    ``kept`` (already present) or ``euclidean`` (nothing to do -- that variant
    derives its costs from the coordinates).
    """
    path = Path(instance_path)
    payload = load_json_from_file(path)
    base = str(payload["instance_name"])
    source = payload.get("arc_costs_source") or {}
    if source.get("model") != "distances-sidecar":
        return {"instance": base, "status": "euclidean", "path": None}

    reference = source["distances"]
    if collection_root is not None:
        root = Path(collection_root)
    else:
        discovered = find_collection_root(path)
        if discovered is None:
            raise FileNotFoundError(
                f"{path}: no {COLLECTION_MARKER_FILENAME} above it; "
                "pass collection_root explicitly"
            )
        root = Path(discovered)
    target = root / reference["path"]
    if target.exists() and not force:
        return {"instance": base, "status": "kept", "path": target}

    metric = str(payload["metric_variant"])
    road_path = target.parent / f"{base}.road.json.gz"
    if not road_path.is_file():
        raise FileNotFoundError(
            f"{base}: road sidecar {road_path} is required to rebuild {metric} distances"
        )
    road = load_instance_road_graph(road_path)

    if metric == "fastest":
        values = free_flow_node_times(road)
    elif metric == "shortest":
        values, _ = _node_paths(_length_weighted(road), want_paths=False)
    else:
        raise ValueError(f"{base}: cannot rebuild distances for metric {metric!r}")

    distances = InstanceDistances(
        base_name=base,
        benchmark_name=str(payload["benchmark_name"]),
        metric=metric,
        num_customers=int(payload["num_customers"]),
        values=_round_matrix(values),
        generator=_base_generator("generate-base"),
    )
    digest = compute_distances_sha256(distances)
    if digest != reference["sha256"]:
        raise AssertionError(
            f"{base}: rebuilt {metric} distances hash {digest}, "
            f"but the instance pins {reference['sha256']}"
        )
    save_instance_distances(distances, target)
    return {"instance": base, "status": "written", "path": target, "sha256": digest}

sampling_seed

sampling_seed(base: str) -> int

Stage-1 sampling seed, derived from the base name (recorded provenance; reproducible under the pinned Julia toolchain only).

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

def sampling_seed(base: str) -> int:
    """Stage-1 sampling seed, derived from the base name (recorded provenance;
    reproducible under the pinned Julia toolchain only)."""
    return _stable_seed(base, "sample")

simplify_tolerance_for

simplify_tolerance_for(num_customers: int) -> float

Frozen family policy: 1.0 s for n <= 100, 2.0 s for n >= 500.

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

def simplify_tolerance_for(num_customers: int) -> float:
    """Frozen family policy: 1.0 s for n <= 100, 2.0 s for n >= 500."""
    return 1.0 if num_customers <= 100 else 2.0

base_instance_name

base_instance_name(
    city: str,
    n: int,
    method_tag: str,
    family: str = FAMILY,
    *,
    route_count: int | None = None,
) -> str

<family>-<city>-n<N>[-k<K>]-<method>.

route_count is the bin-packing minimum fleet. It is only written into the name for families in :data:FAMILIES_WITH_FLEET_IN_NAME, and passing it for a family that does not want it is a caller error rather than a silent no-op: a name that sometimes carries k and sometimes does not is unparseable.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/family/naming.py

def base_instance_name(
    city: str,
    n: int,
    method_tag: str,
    family: str = FAMILY,
    *,
    route_count: int | None = None,
) -> str:
    """``<family>-<city>-n<N>[-k<K>]-<method>``.

    ``route_count`` is the bin-packing minimum fleet. It is only written into the
    name for families in :data:`FAMILIES_WITH_FLEET_IN_NAME`, and passing it for
    a family that does not want it is a caller error rather than a silent no-op:
    a name that sometimes carries ``k`` and sometimes does not is unparseable.
    """
    prefix = f"{family_prefix(family)}-{city}-n{n}"
    if family in FAMILIES_WITH_FLEET_IN_NAME:
        if route_count is None:
            raise ValueError(
                f"{family} base names carry the fleet lower bound; pass route_count"
            )
        prefix = f"{prefix}-k{int(route_count)}"
    elif route_count is not None:
        raise ValueError(
            f"{family} base names do not carry a fleet lower bound; drop route_count"
        )
    return f"{prefix}-{method_tag}".lower()

family_prefix

family_prefix(family: str = FAMILY) -> str

A family's base-name prefix: its name minus the trailing release year.

Poryos2026 -> poryos, Mamut2026 -> mamut. A rule rather than a lookup table, so adding a family means picking a name and nothing else.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/family/naming.py

def family_prefix(family: str = FAMILY) -> str:
    """A family's base-name prefix: its name minus the trailing release year.

    ``Poryos2026 -> poryos``, ``Mamut2026 -> mamut``. A rule rather than a
    lookup table, so adding a family means picking a name and nothing else.
    """
    stem = _RELEASE_YEAR_SUFFIX.sub("", family)
    if not stem:
        raise ValueError(f"family {family!r} has no name part before its release year")
    return stem.lower()

vrptw_instance_name

vrptw_instance_name(base: str, tw_set: str) -> str

TD-paired set: the bare base name; static-only sets: <base>-tw-<set>.

Source code in MAMUT-routing-tools/src/mamut_routing_tools/family/naming.py

def vrptw_instance_name(base: str, tw_set: str) -> str:
    """TD-paired set: the bare base name; static-only sets: ``<base>-tw-<set>``."""
    if tw_set == TW_SET_TD_SHARED:
        return base
    if tw_set not in EXTRA_TW_SETS:
        raise ValueError(f"unknown TW set {tw_set!r} (expected one of {ALL_TW_SETS})")
    return f"{base}-tw-{tw_set}"