def derive_td_from_vrptw(
folder: str | Path,
base: str,
*,
model: str = "bpr",
intensity: str = "moderate",
all_combos: bool = False,
seed: int = 42,
force: bool = False,
) -> dict[str, Any]:
"""Derive the TDVRP + TDVRPTW twins of a generated instance in place.
Reuses the PROVIDED VRPTW windows (lifting deadlines to time-dependent
feasibility, never regenerating them) and produces canonical road-graph
v2 twins that self-verify through ``load_td_instance(path,
verify_sha256=True)``.
"""
if model not in TD_MODELS:
raise TDDerivationError(f"unknown traffic model {model!r}; known: {TD_MODELS}")
if intensity not in TD_INTENSITIES:
raise TDDerivationError(f"unknown intensity {intensity!r}; known: {TD_INTENSITIES}")
folder = Path(folder)
meta_path = folder / f"{base}_meta.json"
cvrptw_path = folder / f"{base}_fastest.cvrptw.vrp"
manifest_path = folder / f"{base}_vrptw_manifest.json"
for required in (meta_path, cvrptw_path, manifest_path):
if not required.is_file():
raise FileNotFoundError(f"missing input for {base}: {required}")
meta = _load_json(meta_path)
manifest = _load_json(manifest_path)
cvrptw = _parse_cvrptw_vrp(cvrptw_path)
nodes_meta = meta["nodes"]
num_customers = len(nodes_meta) - 1
if cvrptw["dimension"] != num_customers + 1:
raise TDDerivationError(
f"{base}: CVRPTW dimension {cvrptw['dimension']} does not match "
f"meta node count {num_customers + 1}"
)
options = meta.get("map_options", {})
only_intersections = bool(options.get("only_intersections", True))
trim_to_connected = bool(options.get("trim_to_connected_graph", True))
city_slug = slugify(str(meta.get("city", base.split("_", 1)[0])))
method_tag = str(meta.get("method", meta.get("generation_params", {}).get("method", "")))
combos = (
[(m, i) for m in TD_MODELS for i in TD_INTENSITIES]
if all_combos
else [(model, intensity)]
)
subs = [subinstance_name(m, i) for m, i in combos]
twin_paths: dict[tuple[str, str], dict[str, Path]] = {}
for (m, i), sub in zip(combos, subs):
name = td_instance_name(base, m, i)
twin_paths[(m, i)] = {
"TDVRPTW": folder / f"{name}.vrp.json",
"TDVRP": folder / f"{name}.tdvrp.vrp.json",
}
all_targets = [p for paths in twin_paths.values() for p in paths.values()]
if not force and all(p.exists() for p in all_targets):
return {
"ok": True,
"base": base,
"folder": str(folder),
"action": "kept",
"num_customers": num_customers,
"combos": [
{
"model": m,
"intensity": i,
"sub": sub,
"tdvrptw_twin": twin_paths[(m, i)]["TDVRPTW"].name,
"tdvrp_twin": twin_paths[(m, i)]["TDVRP"].name,
}
for (m, i), sub in zip(combos, subs)
],
}
osm_path = _resolve_osm_path(str(meta["source_osm_file"]))
# 1. TD bridge for this city, built in memory (shared graph + one speed
# profile per requested combination + this instance's node -> OSM
# mapping); the family records are consumed directly, no JSON round-trip.
bridge_models = sorted({m for m, _ in combos})
bridge_intensities = sorted({i for _, i in combos})
bridge = build_bridge(
osm_path=osm_path,
city_slug=city_slug,
models=bridge_models,
intensities=bridge_intensities,
metas=[meta],
seed=seed,
only_intersections=only_intersections,
trim_to_connected=trim_to_connected,
)
bridge_graph = bridge.graph
bridge_nodes = bridge.nodes[base]
# 2. Instance road graph: full city graph from the bridge, trimmed to the
# union of pinned free-flow fastest paths, saved next to the instance.
ensure_collection_root(folder) # collection marker: sidecar refs resolve here
full = _full_city_road_graph(
bridge_graph,
bridge_nodes,
base=base,
sample_step=DEFAULT_SAMPLE_STEP,
simplify_tolerance=simplify_tolerance_for(num_customers),
extension_end=DEFAULT_EXTENSION_END,
generator=_generator("derive-td", {"city": city_slug, "method": method_tag}),
)
road, _ = _trim_road_graph(full)
road_file = f"{base}.road.json.gz"
save_instance_road_graph(road, folder / road_file)
road_sha = compute_road_graph_sha256(road)
# 3. Per-combo overlays + ATF materialization. Anchor arcs certify the TW
# lift; the full ATF set pins atf_sha256.
anchor_custs, anchor_source = _anchor_customers(manifest, cvrptw["matrix"])
anchor_arc_keys = {
(previous, customer)
for previous, customer in zip([0, *anchor_custs], [*anchor_custs, 0])
}
provided_time_windows = [tuple(w) for w in cvrptw["time_windows"]]
service_times = [int(s) for s in cvrptw["service_times"]]
coordinates = [[float(node["enu_x"]), float(node["enu_y"])] for node in nodes_meta]
demands = [int(node["demand"]) for node in nodes_meta]
reference_lla = meta.get("reference_lla")
overlay_refs: dict[str, dict[str, Any]] = {}
overlay_sha: dict[str, str] = {}
atf_sha: dict[str, str] = {}
traffic_info: dict[str, dict[str, Any]] = {}
anchor_arcs_by_sub: dict[str, dict[tuple[int, int], Any]] = {}
for m, i in combos:
sub = subinstance_name(m, i)
speeds = bridge.speeds[(m, i)]
overlay = _align_overlay(road, bridge_graph, speeds)
overlay_file = f"{base}.traffic-{sub}.json.gz"
save_traffic_overlay(overlay, folder / overlay_file)
overlay_sha[sub] = compute_traffic_overlay_sha256(overlay)
overlay_refs[sub] = {"path": overlay_file, "sha256": overlay_sha[sub]}
traffic_info[sub] = {
"model": speeds.model,
"intensity": speeds.intensity,
"seed": speeds.seed,
"num_trips": speeds.num_trips,
"params": speeds.params,
}
probe_payload = {
"instance_name": td_instance_name(base, m, i),
"instance_origin": "OsmCvrpGen",
"benchmark_name": FAMILY,
"num_customers": num_customers,
"num_vehicles": None,
"vehicle_capacity": int(cvrptw["capacity"]),
"coordinates": coordinates,
"demands": demands,
"service_times": service_times,
"depot": 0,
"horizon": list(TD_HORIZON),
"td": {
"model": "road-graph",
"graph": {"path": road_file, "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
)
missing = anchor_arc_keys - anchor_arcs_by_sub[sub].keys()
if missing:
raise TDDerivationError(f"missing anchor arcs under {sub}: {sorted(missing)}")
atfs = materialize_instance_atfs_roadgraph(instance, road, overlay)
atf_sha[sub] = compute_atf_sha256(atfs)
del atfs
# 4. Shared deadline lift certifying the anchor route under every derived
# overlay. Earliest bounds are never reduced; zero-width windows raise.
lifted, repairs = _audit_and_lift(
provided_time_windows,
service_times,
[anchor_custs],
anchor_arcs_by_sub,
TD_HORIZON[1],
)
for customer in range(1, len(lifted)):
earliest_before = provided_time_windows[customer][0]
earliest_after, latest_after = lifted[customer]
if earliest_after < earliest_before:
raise TDDerivationError(
f"customer {customer}: lifted earliest {earliest_after} below provided {earliest_before}"
)
if earliest_after >= latest_after:
raise TDDerivationError(f"customer {customer}: non-positive lifted window width")
lift_entries = [e for e in repairs.values() if "deadline_after" in e]
tw_repair = {
"policy": "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
),
"anchor_route_source": anchor_source,
"repairs": repairs,
}
# 5. Emit the twins (TDVRP without windows, TDVRPTW with the lifted set).
generated_at = meta.get("generated_at") or manifest.get("generated_at", "")
combos_out: list[dict[str, Any]] = []
for m, i in combos:
sub = subinstance_name(m, i)
name = td_instance_name(base, m, i)
td_metadata = {
"authors": AUTHORS,
"generated_at": generated_at,
"city": city_slug,
"method": method_tag,
"base_instance_name": base,
"subinstance": sub,
"generator": _generator("derive-td", {"city": city_slug, "method": method_tag}),
"traffic": traffic_info[sub],
"notes": (
"Time dependence derives from the city's OSM road network: the shared "
"road-graph sidecar pins free-flow fastest paths for this 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. Windows reuse the derive-vrptw set, lifted to time-dependent "
"feasibility along the anchor route."
),
}
common = {
"instance_name": name,
"instance_origin": "OsmCvrpGen",
"benchmark_name": FAMILY,
"num_customers": num_customers,
"num_vehicles": None,
"vehicle_capacity": int(cvrptw["capacity"]),
"coordinates": coordinates,
"demands": demands,
"service_times": service_times,
"depot": 0,
"reference_lla": reference_lla,
"horizon": list(TD_HORIZON),
"td": {
"model": "road-graph",
"graph": {"path": road_file, "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": tw_repair,
}
save_json_to_file(tdvrp_payload, twin_paths[(m, i)]["TDVRP"])
save_json_to_file(tdvrptw_payload, twin_paths[(m, i)]["TDVRPTW"])
combos_out.append(
{
"model": m,
"intensity": i,
"sub": sub,
"traffic_sidecar": overlay_refs[sub]["path"],
"traffic_sha256": overlay_sha[sub],
"atf_sha256": atf_sha[sub],
"tdvrptw_twin": twin_paths[(m, i)]["TDVRPTW"].name,
"tdvrp_twin": twin_paths[(m, i)]["TDVRP"].name,
}
)
return {
"ok": True,
"base": base,
"folder": str(folder),
"action": "derived",
"num_customers": num_customers,
"collection_marker": "mamut-collection.json",
"road_sidecar": road_file,
"road_sha256": road_sha,
"anchor_route_source": anchor_source,
"lifted_customers": len(lift_entries),
"max_lift_seconds": tw_repair["max_lift_seconds"],
"combos": combos_out,
}