mamut_routing_tools.td

Time-dependent traffic generation for the MAMUT road-graph model.

Ports the workbench's stage-3 traffic stage to Python on the tool's own road graph. :func:build_bridge assembles the TD-bridge records in memory (the streamlined path the family builder and the per-instance derivation consume); :func:export_bridge serializes the same records to the git-ignored JSON intermediate. The bridge record types and their JSON I/O live in :mod:mamut_routing_tools.td.bridge.

BridgeFormatError

Bases: ValueError

Raised when a TD bridge file violates the expected format.

BridgeBuild dataclass

BridgeBuild(graph: BridgeGraph, speeds: dict[tuple[str, str], BridgeSpeeds], nodes: dict[str, BridgeNodes])

The full TD bridge for one city, held in memory.

graph is shared by every combination; speeds is keyed by (model, intensity); nodes is keyed by instance base name (one entry per stage-1 meta). The consumer (mamut_routing_tools.family) takes these records directly, so a per-instance derivation skips the JSON round-trip entirely; :func:export_bridge serializes the same records to disk for the cached / inspectable path.

BridgeEdge dataclass

BridgeEdge(u: int, v: int, osm_u: int, osm_v: int, length_m: float, road_class: int)

One deduplicated directed edge of the bridge graph.

u / v are 0-based graph vertex indices; osm_u / osm_v are the OSM node ids that key the bridge.

TrafficModelError

Bases: ValueError

Raised on an unknown traffic model, intensity, or an empty edge set.

serialize_bridge_graph

serialize_bridge_graph(graph: BridgeGraph) -> dict

The graph.json payload for a :class:BridgeGraph.

Vertices ship as [osm_id, lon, lat] sorted by OSM id (the consumer reads them into a dict, so order is cosmetic but kept stable); edges ship as [osm_u, osm_v, length_m, class, free_speed_ms] in BridgeGraph.edges order (the speed profiles align with it).

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

def serialize_bridge_graph(graph: BridgeGraph) -> dict:
    """The ``graph.json`` payload for a :class:`BridgeGraph`.

    Vertices ship as ``[osm_id, lon, lat]`` sorted by OSM id (the consumer
    reads them into a dict, so order is cosmetic but kept stable); edges ship as
    ``[osm_u, osm_v, length_m, class, free_speed_ms]`` in ``BridgeGraph.edges``
    order (the speed profiles align with it).
    """
    vertices = sorted(
        ([int(osm_id), float(lon), float(lat)] for osm_id, (lon, lat) in graph.vertex_lonlat.items()),
        key=lambda row: row[0],
    )
    edges = [
        [int(osm_u), int(osm_v), float(length_m), int(road_class), float(free_speed)]
        for osm_u, osm_v, length_m, road_class, free_speed in graph.edges
    ]
    return {
        "schema_version": BRIDGE_SCHEMA_VERSION,
        "city": graph.city,
        "osm_file": graph.osm_file,
        "map_options": dict(graph.map_options),
        "num_bins": graph.num_bins,
        "bin_seconds": graph.bin_seconds,
        "speed_unit": "m/s",
        "length_unit": "m",
        "vertices": vertices,
        "edges": edges,
    }

serialize_bridge_nodes

serialize_bridge_nodes(nodes: BridgeNodes) -> dict

The nodes-<instance_base>.json payload for a :class:BridgeNodes.

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

def serialize_bridge_nodes(nodes: BridgeNodes) -> dict:
    """The ``nodes-<instance_base>.json`` payload for a :class:`BridgeNodes`."""
    return {
        "schema_version": BRIDGE_SCHEMA_VERSION,
        "city": nodes.city,
        "instance_base": nodes.instance_base,
        "depot_first": True,
        "node_osm_ids": [int(v) for v in nodes.node_osm_ids],
    }

serialize_bridge_speeds

serialize_bridge_speeds(speeds: BridgeSpeeds) -> dict

The speeds-<model>-<intensity>.json payload for a :class:BridgeSpeeds.

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

def serialize_bridge_speeds(speeds: BridgeSpeeds) -> dict:
    """The ``speeds-<model>-<intensity>.json`` payload for a :class:`BridgeSpeeds`."""
    return {
        "schema_version": BRIDGE_SCHEMA_VERSION,
        "city": speeds.city,
        "model": speeds.model,
        "intensity": speeds.intensity,
        "seed": speeds.seed,
        "num_trips": speeds.num_trips,
        "params": dict(speeds.params),
        "speeds": [list(row) for row in speeds.speeds],
    }

bpr_speeds

bpr_speeds(graph: RoadGraph, edges: list[BridgeEdge], osm_path: str | Path, intensity: str, seed: int) -> tuple[list[list[float]], int]

Per-edge 24-bin BPR speed profiles and the total number of trips routed.

Samples trips_per_vertex * |V| commuters (homes uniform on vertices, workplaces from the POI pool with uniform fallback), routes every trip on the free-flow fastest path, accumulates per-edge hourly flows at edge entry time, then applies the BPR volume-delay function with class-based capacities. Deterministic per seed within Python.

Routing is grouped by each trip's workplace endpoint (the small POI pool) rather than by home, so Dijkstra runs ~2*|work pool| times instead of once per home: work -> home trips keep a forward tree per workplace (bit- identical to a per-origin forward routing) and home -> work trips use a reverse tree per workplace over the transposed graph, which matches forward routing except on equal-cost shortest-path ties.

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

def bpr_speeds(
    graph: RoadGraph,
    edges: list[BridgeEdge],
    osm_path: str | Path,
    intensity: str,
    seed: int,
) -> tuple[list[list[float]], int]:
    """Per-edge 24-bin BPR speed profiles and the total number of trips routed.

    Samples ``trips_per_vertex * |V|`` commuters (homes uniform on vertices,
    workplaces from the POI pool with uniform fallback), routes every trip on
    the free-flow fastest path, accumulates per-edge hourly flows at edge
    entry time, then applies the BPR volume-delay function with class-based
    capacities. Deterministic per seed within Python.

    Routing is grouped by each trip's workplace endpoint (the small POI pool)
    rather than by home, so Dijkstra runs ~2*|work pool| times instead of once
    per home: ``work -> home`` trips keep a forward tree per workplace (bit-
    identical to a per-origin forward routing) and ``home -> work`` trips use a
    reverse tree per workplace over the transposed graph, which matches forward
    routing except on equal-cost shortest-path ties.
    """
    num_vertices = graph.vertex_count
    rng = np.random.Generator(np.random.PCG64(seed))
    commuters = round(TD_BPR_TRIPS_PER_VERTEX[intensity] * num_vertices)
    work_pool = bpr_work_pool(graph, osm_path)
    trips = _sample_trips(rng, commuters, num_vertices, work_pool)

    edge_index = {(edge.u, edge.v): index for index, edge in enumerate(edges)}
    times = np.array([edge.length_m / td_free_speed_ms(edge.road_class) for edge in edges], dtype=np.float64)
    rows = np.fromiter((edge.u for edge in edges), dtype=np.int64, count=len(edges))
    cols = np.fromiter((edge.v for edge in edges), dtype=np.int64, count=len(edges))
    csr = csr_matrix((times, (rows, cols)), shape=(num_vertices, num_vertices))

    flows = _accumulate_flows(csr, trips, edge_index, times, num_vertices)
    return _bpr_profiles(edges, flows), len(trips)

bpr_work_pool

bpr_work_pool(graph: RoadGraph, osm_path: str | Path) -> list[int]

Workplace vertex pool: amenity-POI-snapped graph vertices when the OSM file has enough of them (>= 50), otherwise all vertices (uniform fallback). Mirrors td_work_pool: each POI snaps to its nearest road node, which must itself be a graph vertex (the findnode + md.v membership rule); first-seen dedup; sorted.

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

def bpr_work_pool(graph: RoadGraph, osm_path: str | Path) -> list[int]:
    """Workplace vertex pool: amenity-POI-snapped graph vertices when the OSM
    file has enough of them (>= 50), otherwise all vertices (uniform
    fallback). Mirrors ``td_work_pool``: each POI snaps to its nearest road
    node, which must itself be a graph vertex (the ``findnode`` + ``md.v``
    membership rule); first-seen dedup; sorted."""
    pool: list[int] = []
    seen: set[int] = set()
    try:
        for poi in find_pois(osm_path, DEFAULT_CATEGORIES):
            osm_id = graph.nearest_node(poi.lat, poi.lon)
            if osm_id is None or osm_id not in graph.vertex_of:
                continue
            vertex = graph.vertex_of[osm_id]
            if vertex not in seen:
                seen.add(vertex)
                pool.append(vertex)
    except Exception:  # noqa: BLE001 - keep the partial pool and warn, as td_work_pool does
        warnings.warn("POI workplace pool failed; falling back to uniform workplaces", stacklevel=2)
    if len(pool) >= 50:
        return sorted(pool)
    return list(range(graph.vertex_count))

bridge_seed

bridge_seed(base_seed: int, model: str, intensity: str) -> int

Per-combination seed, matching the workbench's seed arithmetic so the recorded seed field is stable across regenerations.

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

def bridge_seed(base_seed: int, model: str, intensity: str) -> int:
    """Per-combination seed, matching the workbench's seed arithmetic so the
    recorded ``seed`` field is stable across regenerations."""
    model_index = TD_MODELS.index(model) + 1
    intensity_index = TD_INTENSITIES.index(intensity) + 1
    return base_seed + 101 * model_index + 10007 * intensity_index

build_bridge

build_bridge(*, osm_path: str | Path, city_slug: str, models: list[str] | tuple[str, ...] = TD_MODELS, intensities: list[str] | tuple[str, ...] = TD_INTENSITIES, seed: int = 42, only_intersections: bool = True, trim_to_connected: bool = True, metas: list[dict] | tuple[dict, ...] = ()) -> BridgeBuild

Build the full TD bridge for one city in memory (no disk round-trip).

Loads the city road graph once, collects its deduplicated edges, then assembles the shared :class:BridgeGraph, a :class:BridgeSpeeds for every model x intensity combination, and a :class:BridgeNodes for every stage-1 meta dict in metas (keyed by instance base name). This is the streamlined path the per-instance TD derivation uses; it produces exactly the records :func:export_bridge writes to and reads back from disk.

Each meta's graph_vertex_id values must index the same road graph the bridge is built on, so the metas' map_options must match only_intersections / trim_to_connected.

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

def build_bridge(
    *,
    osm_path: str | Path,
    city_slug: str,
    models: list[str] | tuple[str, ...] = TD_MODELS,
    intensities: list[str] | tuple[str, ...] = TD_INTENSITIES,
    seed: int = 42,
    only_intersections: bool = True,
    trim_to_connected: bool = True,
    metas: list[dict] | tuple[dict, ...] = (),
) -> BridgeBuild:
    """Build the full TD bridge for one city in memory (no disk round-trip).

    Loads the city road graph once, collects its deduplicated edges, then
    assembles the shared :class:`BridgeGraph`, a :class:`BridgeSpeeds` for every
    ``model x intensity`` combination, and a :class:`BridgeNodes` for every
    stage-1 ``meta`` dict in ``metas`` (keyed by instance base name). This is the
    streamlined path the per-instance TD derivation uses; it produces exactly the
    records :func:`export_bridge` writes to and reads back from disk.

    Each meta's ``graph_vertex_id`` values must index the same road graph the
    bridge is built on, so the metas' ``map_options`` must match
    ``only_intersections`` / ``trim_to_connected``.
    """
    _validate_combos(models, intensities)
    graph, edges, vertex_ll, center = _load_bridge_road_graph(
        osm_path, only_intersections, trim_to_connected
    )
    bridge_graph = build_bridge_graph(
        graph, edges, city_slug, osm_path, only_intersections, trim_to_connected
    )
    speeds: dict[tuple[str, str], BridgeSpeeds] = {}
    for model in models:
        for intensity in intensities:
            combo_seed = bridge_seed(seed, model, intensity)
            profiles, num_trips = _compute_speeds(
                graph, edges, vertex_ll, center, osm_path, model, intensity, combo_seed
            )
            speeds[(model, intensity)] = build_bridge_speeds(
                city_slug, model, intensity, combo_seed, num_trips, profiles
            )
    nodes: dict[str, BridgeNodes] = {}
    for meta in metas:
        record = build_bridge_nodes(
            graph,
            meta,
            city_slug,
            only_intersections=only_intersections,
            trim_to_connected=trim_to_connected,
        )
        nodes[record.instance_base] = record
    return BridgeBuild(graph=bridge_graph, speeds=speeds, nodes=nodes)

build_bridge_graph

build_bridge_graph(graph: RoadGraph, edges: list[BridgeEdge], city_slug: str, osm_path: str | Path, only_intersections: bool, trim_to_connected: bool) -> BridgeGraph

Assemble the shared :class:BridgeGraph for a road graph and its edges.

Edges carry the static free-flow limit (m/s, same rounding as the speed profiles) so the consumer never needs the class table; every vertex incident to an edge ships its WGS84 position for the consumer's geo cache. Serializing this record (serialize_bridge_graph) then loading it back (load_bridge_graph) returns an equal record, so the in-memory and disk paths never diverge.

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

def build_bridge_graph(
    graph: RoadGraph,
    edges: list[BridgeEdge],
    city_slug: str,
    osm_path: str | Path,
    only_intersections: bool,
    trim_to_connected: bool,
) -> BridgeGraph:
    """Assemble the shared :class:`BridgeGraph` for a road graph and its edges.

    Edges carry the static free-flow limit (m/s, same rounding as the speed
    profiles) so the consumer never needs the class table; every vertex
    incident to an edge ships its WGS84 position for the consumer's geo cache.
    Serializing this record (``serialize_bridge_graph``) then loading it back
    (``load_bridge_graph``) returns an equal record, so the in-memory and disk
    paths never diverge.
    """
    used = sorted({v for edge in edges for v in (edge.u, edge.v)})
    vertex_lonlat: dict[int, tuple[float, float]] = {}
    for graph_vertex in used:
        osm_id = int(graph.node_of[graph_vertex])
        lon, lat = graph.node_lonlat(osm_id)
        vertex_lonlat[osm_id] = (float(lon), float(lat))
    edge_tuples = [
        (
            int(edge.osm_u),
            int(edge.osm_v),
            float(edge.length_m),
            int(edge.road_class),
            td_round_speed(td_free_speed_ms(edge.road_class)),
        )
        for edge in edges
    ]
    return BridgeGraph(
        city=city_slug,
        osm_file=Path(osm_path).name,
        map_options={
            "only_intersections": only_intersections,
            "trim_to_connected_graph": trim_to_connected,
        },
        num_bins=TD_NUM_BINS,
        bin_seconds=TD_BIN_SECONDS,
        edges=edge_tuples,
        vertex_lonlat=vertex_lonlat,
    )

build_bridge_nodes

build_bridge_nodes(graph: RoadGraph, meta: dict, city_slug: str, *, only_intersections: bool, trim_to_connected: bool) -> BridgeNodes

Map one stage-1 meta's instance nodes (depot first) to a :class:BridgeNodes.

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

def build_bridge_nodes(
    graph: RoadGraph,
    meta: dict,
    city_slug: str,
    *,
    only_intersections: bool,
    trim_to_connected: bool,
) -> BridgeNodes:
    """Map one stage-1 ``meta``'s instance nodes (depot first) to a :class:`BridgeNodes`."""
    return BridgeNodes(
        city=city_slug,
        instance_base=str(meta["instance_name"]),
        node_osm_ids=node_osm_ids_from_meta(
            graph, meta, only_intersections=only_intersections, trim_to_connected=trim_to_connected
        ),
    )

build_bridge_speeds

build_bridge_speeds(city_slug: str, model: str, intensity: str, seed: int, num_trips: int, speeds: list[list[float]]) -> BridgeSpeeds

Wrap one combination's per-edge speed profiles in a :class:BridgeSpeeds.

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

def build_bridge_speeds(
    city_slug: str,
    model: str,
    intensity: str,
    seed: int,
    num_trips: int,
    speeds: list[list[float]],
) -> BridgeSpeeds:
    """Wrap one combination's per-edge speed profiles in a :class:`BridgeSpeeds`."""
    return BridgeSpeeds(
        city=city_slug,
        model=model,
        intensity=intensity,
        seed=seed,
        num_trips=num_trips,
        params=model_params(model, intensity),
        speeds=speeds,
    )

build_static_bridge

build_static_bridge(*, osm_path: str | Path, city_slug: str, metas: list[dict] | tuple[dict, ...] = (), only_intersections: bool = True, trim_to_connected: bool = True) -> BridgeBuild

The bridge a CVRP-only family needs: the graph and the nodes, no speeds.

:func:build_bridge simulates six traffic overlays per city, which is the dominant cost and is pure waste for a static family. This builds the same :class:BridgeGraph and :class:BridgeNodes records and leaves speeds empty, so family.build_base works unchanged while family.build_td stays unavailable (it would raise on the missing combinations, which is the correct failure for a collection that has no time-dependent layer).

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

def build_static_bridge(
    *,
    osm_path: str | Path,
    city_slug: str,
    metas: list[dict] | tuple[dict, ...] = (),
    only_intersections: bool = True,
    trim_to_connected: bool = True,
) -> BridgeBuild:
    """The bridge a CVRP-only family needs: the graph and the nodes, no speeds.

    :func:`build_bridge` simulates six traffic overlays per city, which is the
    dominant cost and is pure waste for a static family. This builds the same
    :class:`BridgeGraph` and :class:`BridgeNodes` records and leaves ``speeds``
    empty, so ``family.build_base`` works unchanged while ``family.build_td``
    stays unavailable (it would raise on the missing combinations, which is the
    correct failure for a collection that has no time-dependent layer).
    """
    graph, edges, _vertex_ll, _center = _load_bridge_road_graph(
        osm_path, only_intersections, trim_to_connected
    )
    bridge_graph = build_bridge_graph(
        graph, edges, city_slug, osm_path, only_intersections, trim_to_connected
    )
    nodes: dict[str, BridgeNodes] = {}
    for meta in metas:
        record = build_bridge_nodes(
            graph,
            meta,
            city_slug,
            only_intersections=only_intersections,
            trim_to_connected=trim_to_connected,
        )
        nodes[record.instance_base] = record
    return BridgeBuild(graph=bridge_graph, speeds={}, nodes=nodes)

collect_edges

collect_edges(graph: RoadGraph) -> list[BridgeEdge]

Deduplicated directed edge list, one entry per (u, v) graph-vertex pair, keeping the fastest free-flow representative (min free-flow time, then min length). Self-loops and non-positive lengths are dropped.

RoadGraph already stores a single edge per (u, v), so the dedup is a safety net; the filters mirror td_collect_edges exactly. Edges are returned sorted by (u, v) so speed profiles align by position.

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

def collect_edges(graph: RoadGraph) -> list[BridgeEdge]:
    """Deduplicated directed edge list, one entry per ``(u, v)`` graph-vertex
    pair, keeping the fastest free-flow representative (min free-flow time,
    then min length). Self-loops and non-positive lengths are dropped.

    ``RoadGraph`` already stores a single edge per ``(u, v)``, so the dedup is
    a safety net; the filters mirror ``td_collect_edges`` exactly. Edges are
    returned sorted by ``(u, v)`` so speed profiles align by position.
    """
    best: dict[tuple[int, int], BridgeEdge] = {}
    for index, (osm_u, osm_v) in enumerate(graph.edges):
        u = graph.vertex_of[osm_u]
        v = graph.vertex_of[osm_v]
        if u == v:
            continue
        length_m = float(graph.edge_weight[index])
        if length_m <= 0:
            continue
        candidate = BridgeEdge(u, v, int(osm_u), int(osm_v), length_m, int(graph.edge_class[index]))
        incumbent = best.get((u, v))
        if incumbent is None:
            best[(u, v)] = candidate
            continue
        new_time = candidate.length_m / SPEED_ROADS_URBAN[candidate.road_class]
        old_time = incumbent.length_m / SPEED_ROADS_URBAN[incumbent.road_class]
        if new_time < old_time or (new_time == old_time and candidate.length_m < incumbent.length_m):
            best[(u, v)] = candidate
    return sorted(best.values(), key=lambda e: (e.u, e.v))

export_bridge

export_bridge(*, osm_path: str | Path, city_slug: str, out_root: str | Path, models: list[str] | tuple[str, ...] = TD_MODELS, intensities: list[str] | tuple[str, ...] = TD_INTENSITIES, seed: int = 42, force: bool = False, only_intersections: bool = True, trim_to_connected: bool = True, meta_paths: list[str | Path] | tuple[str | Path, ...] = ()) -> Path

Serialize the TD bridge for one city under <out_root>/<city_slug>/.

Writes graph.json (deduplicated directed edges keyed by OSM node ids), speeds-<model>-<intensity>.json for every requested combination (speed profiles aligned with the graph edge order, m/s), one nodes-<instance_base>.json per stage-1 meta in meta_paths (instance node -> OSM node ids, depot first), and a bridge-manifest.json. The bytes written are exactly the serialized :func:build_bridge records, so load_bridge_* reads them back unchanged. Existing per-combination speed files are reused unless force=True (the disk path's only advantage over building in memory: skipping an expensive recompute). Returns the city output directory.

Each meta's graph_vertex_id values must index the same road graph the bridge is built on, so the metas' map_options must match only_intersections / trim_to_connected.

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

def export_bridge(
    *,
    osm_path: str | Path,
    city_slug: str,
    out_root: str | Path,
    models: list[str] | tuple[str, ...] = TD_MODELS,
    intensities: list[str] | tuple[str, ...] = TD_INTENSITIES,
    seed: int = 42,
    force: bool = False,
    only_intersections: bool = True,
    trim_to_connected: bool = True,
    meta_paths: list[str | Path] | tuple[str | Path, ...] = (),
) -> Path:
    """Serialize the TD bridge for one city under ``<out_root>/<city_slug>/``.

    Writes ``graph.json`` (deduplicated directed edges keyed by OSM node ids),
    ``speeds-<model>-<intensity>.json`` for every requested combination (speed
    profiles aligned with the graph edge order, m/s), one
    ``nodes-<instance_base>.json`` per stage-1 meta in ``meta_paths`` (instance
    node -> OSM node ids, depot first), and a ``bridge-manifest.json``. The bytes
    written are exactly the serialized :func:`build_bridge` records, so
    ``load_bridge_*`` reads them back unchanged. Existing per-combination speed
    files are reused unless ``force=True`` (the disk path's only advantage over
    building in memory: skipping an expensive recompute). Returns the city output
    directory.

    Each meta's ``graph_vertex_id`` values must index the same road graph the
    bridge is built on, so the metas' ``map_options`` must match
    ``only_intersections`` / ``trim_to_connected``.
    """
    _validate_combos(models, intensities)
    graph, edges, vertex_ll, center = _load_bridge_road_graph(
        osm_path, only_intersections, trim_to_connected
    )

    out_dir = Path(out_root) / city_slug
    out_dir.mkdir(parents=True, exist_ok=True)
    _write_json(
        out_dir / "graph.json",
        serialize_bridge_graph(
            build_bridge_graph(graph, edges, city_slug, osm_path, only_intersections, trim_to_connected)
        ),
    )

    written: list[str] = []
    for model in models:
        for intensity in intensities:
            speeds_path = out_dir / f"speeds-{model}-{intensity}.json"
            if speeds_path.is_file() and not force:
                written.append(f"{speeds_path.name} (kept)")
                continue
            combo_seed = bridge_seed(seed, model, intensity)
            profiles, num_trips = _compute_speeds(
                graph, edges, vertex_ll, center, osm_path, model, intensity, combo_seed
            )
            _write_json(
                speeds_path,
                serialize_bridge_speeds(
                    build_bridge_speeds(city_slug, model, intensity, combo_seed, num_trips, profiles)
                ),
            )
            written.append(speeds_path.name)

    node_files: list[str] = []
    for meta_path in meta_paths:
        meta = json.loads(Path(meta_path).read_text())
        record = build_bridge_nodes(
            graph, meta, city_slug, only_intersections=only_intersections, trim_to_connected=trim_to_connected
        )
        nodes_path = out_dir / f"nodes-{record.instance_base}.json"
        _write_json(nodes_path, serialize_bridge_nodes(record))
        node_files.append(nodes_path.name)

    _write_json(
        out_dir / "bridge-manifest.json",
        {
            "schema_version": BRIDGE_SCHEMA_VERSION,
            "city": city_slug,
            "num_vertices": graph.vertex_count,
            "num_edges": len(edges),
            "speed_files": written,
            "node_files": node_files,
        },
    )
    return out_dir

node_osm_ids_from_meta

node_osm_ids_from_meta(graph: RoadGraph, meta: dict, *, only_intersections: bool, trim_to_connected: bool) -> list[int]

Map a stage-1 meta's instance nodes (depot first) to OSM node ids on graph.

The meta's graph_vertex_id values are indices into the road graph the instance was sampled on, so the meta's map_options must match the graph the bridge is built on for the numbering to align (this is the stage-1a / stage-3 shared-numbering invariant).

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

def node_osm_ids_from_meta(
    graph: RoadGraph, meta: dict, *, only_intersections: bool, trim_to_connected: bool
) -> list[int]:
    """Map a stage-1 meta's instance nodes (depot first) to OSM node ids on
    ``graph``.

    The meta's ``graph_vertex_id`` values are indices into the road graph the
    instance was sampled on, so the meta's ``map_options`` must match the graph
    the bridge is built on for the numbering to align (this is the stage-1a /
    stage-3 shared-numbering invariant).
    """
    options = meta.get("map_options", {})
    if (
        bool(options.get("only_intersections", only_intersections)) != only_intersections
        or bool(options.get("trim_to_connected_graph", trim_to_connected)) != trim_to_connected
    ):
        raise TrafficModelError(
            f"meta {meta.get('instance_name')!r} map_options {options} do not match the bridge graph "
            f"options (only_intersections={only_intersections}, trim_to_connected_graph={trim_to_connected}); "
            "the graph_vertex_id numbering would not align"
        )
    node_osm_ids: list[int] = []
    for node in meta["nodes"]:
        gvid = int(node["graph_vertex_id"])
        if not 0 <= gvid < graph.vertex_count:
            raise TrafficModelError(
                f"meta {meta.get('instance_name')!r} graph_vertex_id {gvid} out of range [0, {graph.vertex_count})"
            )
        node_osm_ids.append(graph.node_of[gvid])
    return node_osm_ids

vertex_latlon

vertex_latlon(graph: RoadGraph) -> list[tuple[float, float]]

Per-vertex (lat, lon) indexed by 0-based graph vertex.

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

def vertex_latlon(graph: RoadGraph) -> list[tuple[float, float]]:
    """Per-vertex ``(lat, lon)`` indexed by 0-based graph vertex."""
    out: list[tuple[float, float]] = []
    for osm_id in graph.node_of:
        lla = graph.node_lla(osm_id)
        out.append((lla.lat, lla.lon))
    return out

wave_speeds

wave_speeds(edges: list[BridgeEdge], vertex_ll: list[tuple[float, float]], center_latlon: tuple[float, float], intensity: str, seed: int) -> list[list[float]]

Per-edge 24-bin wave speed profiles (m/s), aligned with edges.

Each edge draws one uniform jitter in edge order, then applies a bimodal rush dip scaled by amplitude, road-centre proximity and jitter, clamped to a free-flow floor.

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

def wave_speeds(
    edges: list[BridgeEdge],
    vertex_ll: list[tuple[float, float]],
    center_latlon: tuple[float, float],
    intensity: str,
    seed: int,
) -> list[list[float]]:
    """Per-edge 24-bin wave speed profiles (m/s), aligned with ``edges``.

    Each edge draws one uniform jitter in edge order, then applies a bimodal
    rush dip scaled by amplitude, road-centre proximity and jitter, clamped to
    a free-flow floor.
    """
    amplitude = TD_WAVE_AMPLITUDE[intensity]
    center_lat, center_lon = center_latlon
    rng = np.random.Generator(np.random.PCG64(seed))
    # One jitter draw per edge, in edge order (a block draw is the same PCG64
    # stream as consecutive scalar draws).
    jitter = (2.0 * rng.random(len(edges)) - 1.0) * TD_WAVE_JITTER
    rush = [td_rush_curve(b) for b in range(TD_NUM_BINS)]

    speeds: list[list[float]] = []
    for index, edge in enumerate(edges):
        u_lat, u_lon = vertex_ll[edge.u]
        v_lat, v_lon = vertex_ll[edge.v]
        mid_lat = (u_lat + v_lat) / 2.0
        mid_lon = (u_lon + v_lon) / 2.0
        centrality = math.exp(-haversine_m(mid_lat, mid_lon, center_lat, center_lon) / TD_WAVE_CENTER_DECAY_M)
        dip_share = TD_WAVE_FLOOR_SHARE + (1.0 - TD_WAVE_FLOOR_SHARE) * centrality
        free = td_free_speed_ms(edge.road_class)
        floor = free * TD_MIN_SPEED_FACTOR
        jitter_factor = 1.0 + float(jitter[index])
        profile = [
            td_round_speed(max(free * (1.0 - amplitude * rush[b] * dip_share) * jitter_factor, floor))
            for b in range(TD_NUM_BINS)
        ]
        speeds.append(profile)
    return speeds