Skip to content

mamut_routing_lib.td

Time-dependent routing (TDVRPTW / TDVRP) contract and canonical checker.

See the TD benchmark standard in the MAMUT-routing repository for the format specification: instance .vrp.json files reference an ATF sidecar (.atf.json / .atf.json.gz) holding one arrival-time NDCPWLF per arc.

ATF_GZIP_SUFFIX module-attribute 

ATF_GZIP_SUFFIX = '.atf.json.gz'

ATF_PLAIN_SUFFIX module-attribute 

ATF_PLAIN_SUFFIX = '.atf.json'

IGP_GZIP_SUFFIX module-attribute 

IGP_GZIP_SUFFIX = '.igp.json.gz'

IGP_MATERIALIZER_GENERATOR module-attribute 

IGP_MATERIALIZER_GENERATOR: dict[str, Any] = {
    "name": "igp-profile-materializer",
    "version": 1,
}

IGP_PLAIN_SUFFIX module-attribute 

IGP_PLAIN_SUFFIX = '.igp.json'

TD_OBJECTIVES module-attribute 

TD_OBJECTIVES = (
    ObjectiveFunction.DURATION,
    ObjectiveFunction.FLEET_COST_DURATION,
)

ATF_FORMAT module-attribute 

ATF_FORMAT = 'mamut-td-atf'

ATF_FORMAT_VERSION module-attribute 

ATF_FORMAT_VERSION = 1

IGP_CATEGORIES_FORMAT module-attribute 

IGP_CATEGORIES_FORMAT = 'mamut-td-igp-categories'

IGP_CATEGORIES_FORMAT_VERSION module-attribute 

IGP_CATEGORIES_FORMAT_VERSION = 1

ROAD_GRAPH_FORMAT module-attribute 

ROAD_GRAPH_FORMAT = 'mamut-road-graph'

ROAD_GRAPH_FORMAT_VERSION module-attribute 

ROAD_GRAPH_FORMAT_VERSION = 2

TRAFFIC_OVERLAY_FORMAT module-attribute 

TRAFFIC_OVERLAY_FORMAT = 'mamut-traffic-overlay'

TRAFFIC_OVERLAY_FORMAT_VERSION module-attribute 

TRAFFIC_OVERLAY_FORMAT_VERSION = 1

TD_ATF_MODEL module-attribute 

TD_ATF_MODEL = 'atf-ndcpwlf'

TD_IGP_MODEL module-attribute 

TD_IGP_MODEL = 'igp-profile'

TD_ROAD_MODEL module-attribute 

TD_ROAD_MODEL = 'road-graph'

AnyTDBenchmarkInstance module-attribute 

AnyTDBenchmarkInstance = (
    BenchmarkInstanceTDVRPTW | BenchmarkInstanceTDVRP
)

ROAD_GZIP_SUFFIX module-attribute 

ROAD_GZIP_SUFFIX = '.road.json.gz'

ROAD_MATERIALIZER_GENERATOR module-attribute 

ROAD_MATERIALIZER_GENERATOR: dict[str, Any] = {
    "name": "road-graph-materializer",
    "version": 2,
}

ROAD_PLAIN_SUFFIX module-attribute 

ROAD_PLAIN_SUFFIX = '.road.json'

TRAFFIC_GZIP_SUFFIX module-attribute 

TRAFFIC_GZIP_SUFFIX = '.json.gz'

TRAFFIC_INFIX module-attribute 

TRAFFIC_INFIX = '.traffic-'

TRAFFIC_PLAIN_SUFFIX module-attribute 

TRAFFIC_PLAIN_SUFFIX = '.json'

ATFFormatError

Bases: ValueError

Raised when an ATF sidecar file violates the canonical format.

InstanceATFs dataclass 

InstanceATFs(
    instance_name: str,
    benchmark_name: str,
    horizon: tuple[float, float],
    num_customers: int,
    arcs: dict[tuple[int, int], NDCPWLF],
    generator: dict[str, Any] = dict(),
    format_version: int = ATF_FORMAT_VERSION,
)

In-memory content of an ATF sidecar file.

instance_name instance-attribute 

instance_name: str

benchmark_name instance-attribute 

benchmark_name: str

horizon instance-attribute 

horizon: tuple[float, float]

num_customers instance-attribute 

num_customers: int

arcs instance-attribute 

arcs: dict[tuple[int, int], NDCPWLF]

generator class-attribute instance-attribute 

generator: dict[str, Any] = field(default_factory=dict)

format_version class-attribute instance-attribute 

format_version: int = ATF_FORMAT_VERSION

num_vertices 

num_vertices() -> int
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
53
54
def num_vertices(self) -> int:
    return self.num_customers + 1

LoadedTDInstance dataclass 

LoadedTDInstance(
    instance: AnyTDBenchmarkInstance,
    atfs: InstanceATFs,
    instance_path: Path,
    atf_path: Path | None,
    categories_path: Path | None = None,
    road_graph_path: Path | None = None,
    traffic_path: Path | None = None,
    collection_root: Path | None = None,
)

A TD instance paired with its arrival-time functions.

atf_path is the on-disk ATF sidecar for atf-ndcpwlf instances and None for the materialized models; categories_path (igp-profile) and road_graph_path + traffic_path (road-graph v2) point to the respective compact sidecars in the materialized cases. collection_root is set for instances living in a marker-rooted collection (road-graph sidecar paths resolve against it).

instance instance-attribute 

instance: AnyTDBenchmarkInstance

atfs instance-attribute 

atfs: InstanceATFs

instance_path instance-attribute 

instance_path: Path

atf_path instance-attribute 

atf_path: Path | None

categories_path class-attribute instance-attribute 

categories_path: Path | None = None

road_graph_path class-attribute instance-attribute 

road_graph_path: Path | None = None

traffic_path class-attribute instance-attribute 

traffic_path: Path | None = None

collection_root class-attribute instance-attribute 

collection_root: Path | None = None

IGPFormatError

Bases: ValueError

Raised when an IGP categories sidecar violates the canonical format.

InstanceCategories dataclass 

InstanceCategories(
    base_name: str,
    benchmark_name: str,
    num_customers: int,
    num_categories: int,
    categories: list[str],
    generator: dict[str, Any] = dict(),
    format_version: int = IGP_CATEGORIES_FORMAT_VERSION,
)

In-memory content of an IGP categories sidecar.

categories holds num_customers + 1 digit strings of the same length; categories[i][j] is the category of arc (i, j). The matrix is symmetric with a fixed '0' diagonal (self-arcs do not exist; the value is never read).

base_name instance-attribute 

base_name: str

benchmark_name instance-attribute 

benchmark_name: str

num_customers instance-attribute 

num_customers: int

num_categories instance-attribute 

num_categories: int

categories instance-attribute 

categories: list[str]

generator class-attribute instance-attribute 

generator: dict[str, Any] = field(default_factory=dict)

format_version class-attribute instance-attribute 

format_version: int = IGP_CATEGORIES_FORMAT_VERSION

category 

category(i: int, j: int) -> int
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
170
171
def category(self, i: int, j: int) -> int:
    return int(self.categories[i][j])

TDRouteEvaluation

Bases: BaseModel

route instance-attribute 

route: list[int]

feasible instance-attribute 

feasible: bool

duration class-attribute instance-attribute 

duration: float | None = None

departure_time class-attribute instance-attribute 

departure_time: float | None = None

TDSolutionCheckResult

Bases: BaseModel

status instance-attribute 

status: SolutionCheckStatus

routing_cost instance-attribute 

routing_cost: float | None

num_routes instance-attribute 

num_routes: int | None

error_message class-attribute instance-attribute 

error_message: str = ''

route_evaluations class-attribute instance-attribute 

route_evaluations: list[TDRouteEvaluation] = []

is_valid 

is_valid() -> bool
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/checker.py 
85
86
def is_valid(self) -> bool:
    return self.status == SolutionCheckStatus.VALID

make_invalid classmethod 

make_invalid(
    status: SolutionCheckStatus, error_message: str
) -> "TDSolutionCheckResult"
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/checker.py 
88
89
90
@classmethod
def make_invalid(cls, status: SolutionCheckStatus, error_message: str) -> "TDSolutionCheckResult":
    return cls(status=status, routing_cost=None, num_routes=None, error_message=error_message)

BenchmarkInstanceTDVRP

Bases: _TDInstanceValidationMixin

BenchmarkInstanceTDVRPTW

Bases: _TDInstanceValidationMixin

time_windows instance-attribute 

time_windows: list[tuple[int | float, int | float]]

validate_time_window_lengths classmethod 

validate_time_window_lengths(
    value: list[Any], info: Any
) -> list[Any]
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/models.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
@field_validator("time_windows")
@classmethod
def validate_time_window_lengths(cls, value: list[Any], info: Any) -> list[Any]:
    expected_length = info.data["num_customers"] + 1
    if len(value) != expected_length:
        raise ValueError(
            f"Length of time_windows must be {expected_length} "
            f"(based on num_customers={info.data['num_customers']} + 1 for depot)"
        )
    for index, (earliest, latest) in enumerate(value):
        if earliest > latest:
            raise ValueError(f"time window at index {index} has earliest > latest")
    return value

TDArrivalFunctionsRef

Bases: BaseModel

Reference to the ATF sidecar file of a TD instance.

atf_path is relative to the directory containing the instance file. atf_sha256 is computed over the uncompressed canonical JSON bytes of the sidecar, so it is stable across .json and .json.gz storage.

model class-attribute instance-attribute 

model: Literal['atf-ndcpwlf'] = TD_ATF_MODEL

atf_path instance-attribute 

atf_path: str

atf_sha256 class-attribute instance-attribute 

atf_sha256: str | None = None

validate_atf_path classmethod 

validate_atf_path(value: str) -> str
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/models.py 
55
56
57
58
@field_validator("atf_path")
@classmethod
def validate_atf_path(cls, value: str) -> str:
    return _validate_sidecar_path(value, "atf_path")

TDIGPProfileRef

Bases: BaseModel

Compact IGP (Ichoua-Gendreau-Potvin 2003) time-dependent travel model.

Instead of shipping a consolidated ATF sidecar, the instance stores the IGP data that fully determines it: contiguous time_periods (stored absolute floats are canonical — never recomputed), a strictly positive speeds matrix (num_categories x num_periods; FIFO by construction), and a reference to the arc-category sidecar. The canonical ATFs are materialized deterministically on load (mamut_routing_lib.td.igp); atf_sha256 pins the materialized canonical ATF bytes exactly as it would for a committed sidecar. See the TD benchmark standard and the Lera2026 family design note.

model class-attribute instance-attribute 

model: Literal['igp-profile'] = TD_IGP_MODEL

time_periods instance-attribute 

time_periods: list[tuple[float, float]]

speeds instance-attribute 

speeds: list[list[float]]

categories_path instance-attribute 

categories_path: str

categories_sha256 class-attribute instance-attribute 

categories_sha256: str | None = None

atf_sha256 class-attribute instance-attribute 

atf_sha256: str | None = None

validate_categories_path classmethod 

validate_categories_path(value: str) -> str
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/models.py 
84
85
86
87
@field_validator("categories_path")
@classmethod
def validate_categories_path(cls, value: str) -> str:
    return _validate_sidecar_path(value, "categories_path")

validate_time_periods classmethod 

validate_time_periods(
    value: list[tuple[float, float]],
) -> list[tuple[float, float]]
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/models.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@field_validator("time_periods")
@classmethod
def validate_time_periods(cls, value: list[tuple[float, float]]) -> list[tuple[float, float]]:
    if not value:
        raise ValueError("time_periods must be non-empty")
    for index, (start, end) in enumerate(value):
        if start >= end:
            raise ValueError(f"time period {index} is empty or reversed: [{start}, {end}]")
        if index > 0 and value[index - 1][1] != start:
            raise ValueError(
                f"time periods must be contiguous: period {index} starts at {start}, "
                f"previous ends at {value[index - 1][1]}"
            )
    return value

validate_speeds classmethod 

validate_speeds(
    value: list[list[float]], info: Any
) -> list[list[float]]
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/models.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@field_validator("speeds")
@classmethod
def validate_speeds(cls, value: list[list[float]], info: Any) -> list[list[float]]:
    if not value:
        raise ValueError("speeds must have at least one category row")
    num_periods = len(info.data.get("time_periods", []))
    for c, row in enumerate(value):
        if num_periods and len(row) != num_periods:
            raise ValueError(
                f"speeds row {c} has {len(row)} entries, expected one per time period ({num_periods})"
            )
        for p, speed in enumerate(row):
            if speed <= 0:
                raise ValueError(f"speeds[{c}][{p}] = {speed} must be strictly positive (FIFO)")
    return value

num_categories 

num_categories() -> int
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/models.py 
120
121
def num_categories(self) -> int:
    return len(self.speeds)

TDRoadGraphRef

Bases: BaseModel

Compact road-network time-dependent travel model (format v2).

The instance references two sidecars, both collection-root-relative and sha-pinned: the static road graph shared by every traffic subinstance of the base (<base>.road.json.gz: trimmed subgraph with per-edge free-flow speed_limit and vertex coordinates) and one traffic overlay (<base>.traffic-<model>-<intensity>.json.gz: per-edge 24-bin speeds aligned with the graph's edge order, clamped at free-flow). The canonical ATFs are materialized deterministically on load (mamut_routing_lib.td.roadgraph): pinned Dijkstra over the static free-flow times, exact per-edge arrival functions from the overlay speeds sampled exactly along the pinned paths on a fixed departure grid, then deterministic decimation. sample_step and simplify_tolerance are the authoritative materialization parameters and must equal the values recorded in the road sidecar. atf_sha256 pins the materialized canonical ATF bytes exactly as it would for a committed sidecar. See the TD benchmark standard (road-graph v2).

model class-attribute instance-attribute 

model: Literal['road-graph'] = TD_ROAD_MODEL

graph instance-attribute 

graph: SidecarRef

traffic instance-attribute 

traffic: SidecarRef

atf_sha256 class-attribute instance-attribute 

atf_sha256: str | None = None

sample_step instance-attribute 

sample_step: float

simplify_tolerance instance-attribute 

simplify_tolerance: float

validate_sample_step classmethod 

validate_sample_step(value: float) -> float
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/models.py 
153
154
155
156
157
158
@field_validator("sample_step")
@classmethod
def validate_sample_step(cls, value: float) -> float:
    if value <= 0:
        raise ValueError(f"sample_step must be strictly positive, got {value}")
    return value

validate_simplify_tolerance classmethod 

validate_simplify_tolerance(value: float) -> float
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/models.py 
160
161
162
163
164
165
@field_validator("simplify_tolerance")
@classmethod
def validate_simplify_tolerance(cls, value: float) -> float:
    if value < 0:
        raise ValueError(f"simplify_tolerance must be >= 0, got {value}")
    return value

NDCPWLF 

NDCPWLF(
    xs: list[float],
    ys: list[float],
    *,
    validate: bool = True,
)

A non-decreasing continuous piecewise linear function.

Immutable by convention: xs and ys must not be mutated after construction.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
49
50
51
52
53
54
55
56
57
58
59
def __init__(self, xs: list[float], ys: list[float], *, validate: bool = True) -> None:
    if validate:
        if len(xs) != len(ys):
            raise PWLFError(f"xs and ys must have the same length ({len(xs)} != {len(ys)})")
        for k in range(1, len(xs)):
            if xs[k] < xs[k - 1]:
                raise PWLFError(f"xs must be non-decreasing (violated at index {k})")
            if ys[k] < ys[k - 1]:
                raise PWLFError(f"ys must be non-decreasing (violated at index {k})")
    self.xs = xs
    self.ys = ys

xs instance-attribute 

xs = xs

ys instance-attribute 

ys = ys

min_domain property 

min_domain: float

max_domain property 

max_domain: float

min_image property 

min_image: float

max_image property 

max_image: float

identity classmethod 

identity(low: float, high: float) -> 'NDCPWLF'

The identity function over [low, high].

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
61
62
63
64
65
66
67
68
@classmethod
def identity(cls, low: float, high: float) -> "NDCPWLF":
    """The identity function over ``[low, high]``."""
    if low > high:
        raise PWLFError(f"identity domain is empty: [{low}, {high}]")
    if low == high:
        return cls([low], [low], validate=False)
    return cls([low, high], [low, high], validate=False)

empty classmethod 

empty() -> 'NDCPWLF'
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
70
71
72
@classmethod
def empty(cls) -> "NDCPWLF":
    return cls([], [], validate=False)

is_empty 

is_empty() -> bool
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
74
75
def is_empty(self) -> bool:
    return not self.xs

num_breakpoints 

num_breakpoints() -> int
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
77
78
def num_breakpoints(self) -> int:
    return len(self.xs)

evaluate 

evaluate(x: float) -> float

Evaluate the function at x (must lie within the domain).

At a vertical step the smallest value is returned.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def evaluate(self, x: float) -> float:
    """Evaluate the function at ``x`` (must lie within the domain).

    At a vertical step the smallest value is returned.
    """
    if self.is_empty() or x < self.xs[0] or x > self.xs[-1]:
        raise PWLFError(f"x={x!r} is outside the domain of the function")
    i = bisect_left(self.xs, x)
    if self.xs[i] == x:
        return self.ys[i]
    x_lo, x_hi = self.xs[i - 1], self.xs[i]
    y_lo, y_hi = self.ys[i - 1], self.ys[i]
    t = (x - x_lo) / (x_hi - x_lo)
    return y_lo + t * (y_hi - y_lo)

compose 

compose(g: 'NDCPWLF') -> 'NDCPWLF'

Return h = self ∘ g restricted to {x in dom(g) : g(x) in dom(self)}.

Two-pointer event merge over the common value axis dom(self) ∩ img(g); O(len(self.xs) + len(g.xs)).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def compose(self, g: "NDCPWLF") -> "NDCPWLF":
    """Return ``h = self ∘ g`` restricted to ``{x in dom(g) : g(x) in dom(self)}``.

    Two-pointer event merge over the common value axis
    ``dom(self) ∩ img(g)``; O(len(self.xs) + len(g.xs)).
    """
    f = self
    if f.is_empty() or g.is_empty():
        return NDCPWLF.empty()
    lo = max(f.xs[0], g.ys[0])
    hi = min(f.xs[-1], g.ys[-1])
    if lo > hi:
        return NDCPWLF.empty()

    fx, fy = f.xs, f.ys
    gx, gy = g.xs, g.ys
    nf, ng = len(fx), len(gx)

    i = 0
    while fx[i] < lo:
        i += 1
    j = 0
    while gy[j] < lo:
        j += 1

    hxs: list[float] = []
    hys: list[float] = []

    def emit(x: float, y: float) -> None:
        # Clamp monotone so rounding can never break the ND invariant,
        # then drop exact duplicates.
        if hxs:
            if x < hxs[-1]:
                x = hxs[-1]
            if y < hys[-1]:
                y = hys[-1]
            if x == hxs[-1] and y == hys[-1]:
                return
        hxs.append(x)
        hys.append(y)

    while True:
        has_f = i < nf and fx[i] <= hi
        has_g = j < ng and gy[j] <= hi
        if not has_f and not has_g:
            break
        if not has_g:
            u = fx[i]
        elif not has_f:
            u = gy[j]
        else:
            u = fx[i] if fx[i] <= gy[j] else gy[j]

        # Collect all f-breakpoints at value u (a step in f yields several ys).
        f_ys: list[float] = []
        while i < nf and fx[i] == u:
            f_ys.append(fy[i])
            i += 1
        # Collect all g-breakpoints whose image is u (a plateau in g yields several xs).
        g_xs: list[float] = []
        while j < ng and gy[j] == u:
            g_xs.append(gx[j])
            j += 1

        if not f_ys:
            # u lies strictly inside an f piece: fx[i-1] < u < fx[i].
            x_lo, x_hi = fx[i - 1], fx[i]
            t = (u - x_lo) / (x_hi - x_lo)
            f_ys = [fy[i - 1] + t * (fy[i] - fy[i - 1])]
        if not g_xs:
            # u lies strictly inside a g piece image: gy[j-1] < u < gy[j].
            y_lo, y_hi = gy[j - 1], gy[j]
            t = (u - y_lo) / (y_hi - y_lo)
            g_xs = [gx[j - 1] + t * (gx[j] - gx[j - 1])]

        for x_val in g_xs:
            emit(x_val, f_ys[0])
        for y_val in f_ys[1:]:
            emit(g_xs[-1], y_val)

    return NDCPWLF(hxs, hys, validate=False)

min_shifted_image 

min_shifted_image() -> tuple[float, float]

Return (min_k (ys[k] - xs[k]), xs[k*]) with k* the earliest argmin.

For a route ready-time function this is the optimal route duration and the associated optimal depot departure time. The minimum of y - x over the whole domain is attained at a breakpoint because the shifted function is linear between breakpoints.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def min_shifted_image(self) -> tuple[float, float]:
    """Return ``(min_k (ys[k] - xs[k]), xs[k*])`` with ``k*`` the earliest argmin.

    For a route ready-time function this is the optimal route duration and
    the associated optimal depot departure time. The minimum of ``y - x``
    over the whole domain is attained at a breakpoint because the shifted
    function is linear between breakpoints.
    """
    if self.is_empty():
        raise PWLFError("cannot minimize an empty function")
    best = self.ys[0] - self.xs[0]
    best_x = self.xs[0]
    for k in range(1, len(self.xs)):
        value = self.ys[k] - self.xs[k]
        if value < best:
            best = value
            best_x = self.xs[k]
    return best, best_x

PWLFError

Bases: ValueError

Raised when NDCPWLF invariants are violated.

InstanceRoadGraph dataclass 

InstanceRoadGraph(
    base_name: str,
    benchmark_name: str,
    num_customers: int,
    horizon: tuple[float, float],
    extension_end: float,
    bin_edges: list[float],
    sample_step: float,
    simplify_tolerance: float,
    num_vertices: int,
    vertex_osm_ids: list[int],
    vertex_lonlat: list[tuple[float, float]],
    node_vertices: list[int],
    edges: list[tuple[int, int, float, float]],
    generator: dict[str, Any] = dict(),
    format_version: int = ROAD_GRAPH_FORMAT_VERSION,
)

In-memory content of a road-graph sidecar (format v2).

Vertices are 0 .. num_vertices - 1; vertex_osm_ids[v] records the originating OSM node id of vertex v (strictly increasing — vertices are numbered in ascending OSM-id order, informative only); vertex_lonlat[v] is its WGS84 (lon, lat) position (informative only: rendering and heatmaps, never read by the materialization). edges holds [u, v, length_m, speed_limit] entries sorted strictly increasing by (u, v) (directed, no parallel edges); speed_limit is the static free-flow limit (m-per-time-unit in the instance's units, strictly positive) defining the pinned Dijkstra weight length / speed_limit. node_vertices[i] is the graph vertex of instance node i (depot first, all distinct). sample_step is the departure-grid spacing and simplify_tolerance the decimation tolerance of the pinned materialization; both are canonical (hashed) materialization parameters, repeated in the instance td block and cross-checked at load.

base_name instance-attribute 

base_name: str

benchmark_name instance-attribute 

benchmark_name: str

num_customers instance-attribute 

num_customers: int

horizon instance-attribute 

horizon: tuple[float, float]

extension_end instance-attribute 

extension_end: float

bin_edges instance-attribute 

bin_edges: list[float]

sample_step instance-attribute 

sample_step: float

simplify_tolerance instance-attribute 

simplify_tolerance: float

num_vertices instance-attribute 

num_vertices: int

vertex_osm_ids instance-attribute 

vertex_osm_ids: list[int]

vertex_lonlat instance-attribute 

vertex_lonlat: list[tuple[float, float]]

node_vertices instance-attribute 

node_vertices: list[int]

edges instance-attribute 

edges: list[tuple[int, int, float, float]]

generator class-attribute instance-attribute 

generator: dict[str, Any] = field(default_factory=dict)

format_version class-attribute instance-attribute 

format_version: int = ROAD_GRAPH_FORMAT_VERSION

RoadGraphFormatError

Bases: ValueError

Raised when a road-graph or traffic-overlay sidecar violates the canonical format.

TrafficOverlay dataclass 

TrafficOverlay(
    base_name: str,
    benchmark_name: str,
    traffic_model: str,
    intensity: str,
    bin_edges: list[float],
    edge_speeds: list[list[float]],
    generator: dict[str, Any] = dict(),
    format_version: int = TRAFFIC_OVERLAY_FORMAT_VERSION,
)

In-memory content of a traffic-overlay sidecar.

edge_speeds[k] is the per-bin speed row of edge k of the base's road-graph sidecar (same order, same count); every speed is strictly positive (FIFO by construction) and at most the edge's static speed_limit (validated against the road graph: overlays are slowdowns by contract). bin_edges must equal the road sidecar's bit-exactly.

base_name instance-attribute 

base_name: str

benchmark_name instance-attribute 

benchmark_name: str

traffic_model instance-attribute 

traffic_model: str

intensity instance-attribute 

intensity: str

bin_edges instance-attribute 

bin_edges: list[float]

edge_speeds instance-attribute 

edge_speeds: list[list[float]]

generator class-attribute instance-attribute 

generator: dict[str, Any] = field(default_factory=dict)

format_version class-attribute instance-attribute 

format_version: int = TRAFFIC_OVERLAY_FORMAT_VERSION

num_edges 

num_edges() -> int
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
180
181
def num_edges(self) -> int:
    return len(self.edge_speeds)

atfs_to_canonical_json_bytes 

atfs_to_canonical_json_bytes(atfs: InstanceATFs) -> bytes

Serialize to the canonical JSON bytes (the input of atf_sha256).

Fixed key order, one arc per line, floats via Python's shortest round-trip repr.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def atfs_to_canonical_json_bytes(atfs: InstanceATFs) -> bytes:
    """Serialize to the canonical JSON bytes (the input of ``atf_sha256``).

    Fixed key order, one arc per line, floats via Python's shortest
    round-trip repr.
    """
    header_lines = [
        "{",
        f'    "format": {json.dumps(ATF_FORMAT)},',
        f'    "format_version": {json.dumps(atfs.format_version)},',
        f'    "instance_name": {json.dumps(atfs.instance_name)},',
        f'    "benchmark_name": {json.dumps(atfs.benchmark_name)},',
        f'    "horizon": {json.dumps(list(atfs.horizon))},',
        f'    "num_customers": {json.dumps(atfs.num_customers)},',
        f'    "generator": {json.dumps(atfs.generator, sort_keys=True)},',
        '    "arcs": [',
    ]
    arc_lines = []
    for (i, j) in sorted(atfs.arcs):
        atf = atfs.arcs[(i, j)]
        arc_lines.append("        " + json.dumps([i, j, list(atf.xs), list(atf.ys)]))
    body = ",\n".join(arc_lines)
    text = "\n".join(header_lines) + "\n" + body + "\n    ]\n}\n"
    return text.encode("utf-8")

compute_atf_sha256 

compute_atf_sha256(atfs: InstanceATFs) -> str
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
151
152
def compute_atf_sha256(atfs: InstanceATFs) -> str:
    return hashlib.sha256(atfs_to_canonical_json_bytes(atfs)).hexdigest()

get_atf_path_for_instance 

get_atf_path_for_instance(
    instance_path: str | Path, atf_path: str | None = None
) -> Path

Resolve the sidecar path next to the instance file.

When atf_path (from the instance's td block) is given it is resolved relative to the instance directory; otherwise the conventional sibling names are probed (plain first, then gzip).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def get_atf_path_for_instance(instance_path: str | Path, atf_path: str | None = None) -> Path:
    """Resolve the sidecar path next to the instance file.

    When ``atf_path`` (from the instance's ``td`` block) is given it is
    resolved relative to the instance directory; otherwise the conventional
    sibling names are probed (plain first, then gzip).
    """
    base = Path(instance_path)
    if atf_path is not None:
        return base.parent / atf_path
    stem = base.name.removesuffix(".vrp.json")
    plain = base.parent / f"{stem}{ATF_PLAIN_SUFFIX}"
    if plain.exists():
        return plain
    return base.parent / f"{stem}{ATF_GZIP_SUFFIX}"

load_instance_atfs 

load_instance_atfs(
    path: str | Path, *, validate_complete: bool = True
) -> InstanceATFs
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
168
169
170
171
172
173
174
def load_instance_atfs(path: str | Path, *, validate_complete: bool = True) -> InstanceATFs:
    source = Path(path)
    raw = source.read_bytes()
    if source.name.endswith(ATF_GZIP_SUFFIX):
        raw = gzip.decompress(raw)
    payload = json.loads(raw.decode("utf-8"))
    return _atfs_from_payload(payload, validate_complete=validate_complete)

load_td_benchmark_instance 

load_td_benchmark_instance(
    instance_path: str | Path,
) -> AnyTDBenchmarkInstance

Load and validate the instance file only (no sidecar).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
194
195
196
197
def load_td_benchmark_instance(instance_path: str | Path) -> AnyTDBenchmarkInstance:
    """Load and validate the instance file only (no sidecar)."""
    payload = load_json_from_file(instance_path)
    return td_instance_from_payload(payload)

load_td_instance 

load_td_instance(
    instance_path: str | Path,
    *,
    verify_sha256: bool = True,
    validate_complete: bool = True,
    collection_root: str | Path | None = None,
) -> LoadedTDInstance

Load instance + sidecar, checking their mutual consistency.

For igp-profile and road-graph instances the ATFs are materialized deterministically from the td block and the compact sidecar(s); verify_sha256 then covers both the sidecar hashes (cheap) and atf_sha256 (a full canonical serialization of the materialized ATFs -- minutes at n=1000; pass verify_sha256=False in solver hot paths, materialization is deterministic either way). road-graph sidecar paths are collection-root-relative: the root is found by walking up to the mamut-collection.json marker, or passed explicitly via collection_root.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
def load_td_instance(
    instance_path: str | Path,
    *,
    verify_sha256: bool = True,
    validate_complete: bool = True,
    collection_root: str | Path | None = None,
) -> LoadedTDInstance:
    """Load instance + sidecar, checking their mutual consistency.

    For ``igp-profile`` and ``road-graph`` instances the ATFs are
    materialized deterministically from the td block and the compact
    sidecar(s); ``verify_sha256`` then covers both the sidecar hashes (cheap)
    and ``atf_sha256`` (a full canonical serialization of the materialized
    ATFs -- minutes at n=1000; pass ``verify_sha256=False`` in solver hot
    paths, materialization is deterministic either way). ``road-graph``
    sidecar paths are collection-root-relative: the root is found by walking
    up to the ``mamut-collection.json`` marker, or passed explicitly via
    ``collection_root``.
    """
    source = Path(instance_path)
    instance = load_td_benchmark_instance(source)
    if isinstance(instance.td, TDIGPProfileRef):
        return _load_td_instance_igp(source, instance, verify_sha256=verify_sha256)
    if isinstance(instance.td, TDRoadGraphRef):
        return _load_td_instance_roadgraph(
            source,
            instance,
            verify_sha256=verify_sha256,
            collection_root=Path(collection_root) if collection_root is not None else None,
        )
    atf_path = get_atf_path_for_instance(source, instance.td.atf_path)
    atfs = load_instance_atfs(atf_path, validate_complete=validate_complete)

    if atfs.instance_name != instance.instance_name:
        raise ATFFormatError(
            f"sidecar instance_name {atfs.instance_name!r} does not match {instance.instance_name!r}"
        )
    if atfs.num_customers != instance.num_customers:
        raise ATFFormatError(
            f"sidecar num_customers {atfs.num_customers} does not match {instance.num_customers}"
        )
    if (float(instance.horizon[0]), float(instance.horizon[1])) != atfs.horizon:
        raise ATFFormatError(
            f"sidecar horizon {atfs.horizon} does not match instance horizon {tuple(instance.horizon)}"
        )
    if verify_sha256 and instance.td.atf_sha256 is not None:
        digest = compute_atf_sha256(atfs)
        if digest != instance.td.atf_sha256:
            raise ATFFormatError(
                f"sidecar sha256 mismatch: computed {digest}, instance declares {instance.td.atf_sha256}"
            )
    return LoadedTDInstance(instance=instance, atfs=atfs, instance_path=source, atf_path=atf_path)

save_instance_atfs 

save_instance_atfs(
    atfs: InstanceATFs, path: str | Path
) -> None

Write the sidecar; gzip iff the path ends with .atf.json.gz.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
155
156
157
158
159
160
161
162
163
164
165
def save_instance_atfs(atfs: InstanceATFs, path: str | Path) -> None:
    """Write the sidecar; gzip iff the path ends with ``.atf.json.gz``."""
    target = Path(path)
    data = atfs_to_canonical_json_bytes(atfs)
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.name.endswith(ATF_GZIP_SUFFIX):
        target.write_bytes(gzip.compress(data, mtime=0))
    elif target.name.endswith(ATF_PLAIN_SUFFIX):
        target.write_bytes(data)
    else:
        raise ATFFormatError(f"ATF path must end with {ATF_PLAIN_SUFFIX} or {ATF_GZIP_SUFFIX}: {target.name}")

td_instance_from_payload 

td_instance_from_payload(
    payload: dict[str, Any],
) -> AnyTDBenchmarkInstance
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py 
200
201
202
203
def td_instance_from_payload(payload: dict[str, Any]) -> AnyTDBenchmarkInstance:
    if "time_windows" in payload:
        return BenchmarkInstanceTDVRPTW(**payload)
    return BenchmarkInstanceTDVRP(**payload)

annotate_td_bks_optimality 

annotate_td_bks_optimality(
    loaded: LoadedTDInstance | str | Path,
    optimality: OptimalityMetadata | dict[str, Any],
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> Path

Stamp the stored BKS of loaded with an optimality proof.

The stored BKS (of objective_function's store file) is re-validated by the TD checker before being rewritten with metadata["optimality"] set. When the stamp declares a proven_optimum, it must match the stored cost: an absolute difference beyond :data:OPTIMALITY_COST_TOLERANCE raises, and a dust-level difference is accepted only when the stamp's note explains it. Everything else in the stored file (routes, cost, authorship, other metadata) is left untouched — the proof stamp records who proved the solution optimal, which is independent of who found it.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/bks.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def annotate_td_bks_optimality(
    loaded: LoadedTDInstance | str | Path,
    optimality: OptimalityMetadata | dict[str, Any],
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> Path:
    """Stamp the stored BKS of ``loaded`` with an optimality proof.

    The stored BKS (of ``objective_function``'s store file) is re-validated by
    the TD checker before being rewritten with ``metadata["optimality"]`` set.
    When the stamp declares a ``proven_optimum``, it must match the stored
    cost: an absolute difference beyond :data:`OPTIMALITY_COST_TOLERANCE`
    raises, and a dust-level difference is accepted only when the stamp's
    ``note`` explains it. Everything else in the stored file (routes, cost,
    authorship, other metadata) is left untouched — the proof stamp records
    who *proved* the solution optimal, which is independent of who *found* it.
    """
    loaded = _as_loaded(loaded)
    _validate_td_objective(objective_function)
    if not isinstance(optimality, OptimalityMetadata):
        optimality = OptimalityMetadata.model_validate(optimality)

    bks_path = get_bks_path_for_instance(loaded.instance_path, objective_function)
    if not bks_path.exists():
        raise FileNotFoundError(f"No stored {objective_function.value} BKS to annotate at {bks_path}")

    existing_bks = load_bks(bks_path)
    existing_check = check_td_solution(loaded, existing_bks, objective_function)
    if not existing_check.is_valid():
        raise ValueError(f"Stored BKS is invalid at {bks_path}: {existing_check.error_message}")

    stored_cost = existing_bks.cost
    if optimality.proven_optimum is not None and stored_cost is not None:
        deviation = abs(optimality.proven_optimum - stored_cost)
        if deviation > OPTIMALITY_COST_TOLERANCE:
            raise ValueError(
                f"proven_optimum {optimality.proven_optimum!r} does not match the stored "
                f"cost {stored_cost!r} at {bks_path} (|diff| = {deviation})"
            )
        if deviation != 0.0 and not optimality.note:
            raise ValueError(
                f"proven_optimum {optimality.proven_optimum!r} differs from the stored "
                f"cost {stored_cost!r} at {bks_path} by dust ({deviation}); a "
                "self-contained `note` explaining the difference is required"
            )

    metadata = dict(existing_bks.metadata)
    metadata["optimality"] = optimality.model_dump(mode="json", exclude_none=True)
    # Reconstruct (rather than model_copy) so the metadata validator runs on the
    # stamped payload before it reaches the store.
    stamped = BenchmarkBKS(**{**existing_bks.model_dump(mode="json"), "metadata": metadata})
    save_bks(stamped, bks_path)
    return bks_path

create_td_bks_from_solution 

create_td_bks_from_solution(
    loaded: LoadedTDInstance | str | Path,
    solution: BenchmarkSolution,
    *,
    authors: str,
    metadata: dict[str, Any] | None = None,
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> BenchmarkBKS

Validate solution with the TD checker and wrap it as a BKS.

The BKS cost is the checker's canonical value under objective_function. If solution.cost is set, check_td_solution itself rejects any deviation from that value (exact doubles, no tolerance), so a solver that disagrees with the reference checker can never write to the BKS store.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/bks.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def create_td_bks_from_solution(
    loaded: LoadedTDInstance | str | Path,
    solution: BenchmarkSolution,
    *,
    authors: str,
    metadata: dict[str, Any] | None = None,
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> BenchmarkBKS:
    """Validate ``solution`` with the TD checker and wrap it as a BKS.

    The BKS cost is the checker's canonical value under
    ``objective_function``. If ``solution.cost`` is set, ``check_td_solution``
    itself rejects any deviation from that value (exact doubles, no
    tolerance), so a solver that disagrees with the reference checker can
    never write to the BKS store.
    """
    loaded = _as_loaded(loaded)
    _validate_td_objective(objective_function)
    check_result = check_td_solution(loaded, solution, objective_function)
    if not check_result.is_valid():
        raise ValueError(f"Cannot create BKS from invalid TD solution: {check_result.error_message}")

    if not authors.strip():
        raise ValueError("authors must be a non-empty string")

    merged_metadata = dict(metadata or {})
    merged_metadata["authors"] = authors
    merged_metadata.setdefault("validated_cost", check_result.routing_cost)
    merged_metadata.setdefault("validated_num_routes", check_result.num_routes)
    merged_metadata.setdefault("date", datetime.now(UTC).replace(tzinfo=None).isoformat(timespec="seconds"))
    return BenchmarkBKS(
        instance_name=loaded.instance.instance_name,
        objective_function=objective_function,
        routes=solution.routes,
        cost=check_result.routing_cost,
        metadata=merged_metadata,
    )

save_td_solution_as_bks_if_improved 

save_td_solution_as_bks_if_improved(
    loaded: LoadedTDInstance | str | Path,
    solution: BenchmarkSolution,
    *,
    authors: str,
    metadata: dict[str, Any] | None = None,
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> BKSUpdateResult

TD counterpart of save_solution_as_bks_if_improved.

The candidate and any stored BKS are both validated by the TD checker under objective_function; the stored file is replaced only on a strict improvement. Each objective has its own store file (<Name>.bks.<Objective>.json). Accepts a :class:LoadedTDInstance directly so callers that already hold the loaded instance (solvers, sweep runners) avoid re-reading the ATF sidecar.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/bks.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def save_td_solution_as_bks_if_improved(
    loaded: LoadedTDInstance | str | Path,
    solution: BenchmarkSolution,
    *,
    authors: str,
    metadata: dict[str, Any] | None = None,
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> BKSUpdateResult:
    """TD counterpart of ``save_solution_as_bks_if_improved``.

    The candidate and any stored BKS are both validated by the TD checker
    under ``objective_function``; the stored file is replaced only on a strict
    improvement. Each objective has its own store file
    (``<Name>.bks.<Objective>.json``). Accepts a :class:`LoadedTDInstance`
    directly so callers that already hold the loaded instance (solvers, sweep
    runners) avoid re-reading the ATF sidecar.
    """
    loaded = _as_loaded(loaded)
    _validate_td_objective(objective_function)
    bks = create_td_bks_from_solution(
        loaded, solution, authors=authors, metadata=metadata, objective_function=objective_function
    )
    bks_path = get_bks_path_for_instance(loaded.instance_path, objective_function)
    existing_path = bks_path if bks_path.exists() else None

    candidate_cost = bks.cost if bks.cost is not None else float("inf")

    if existing_path is None:
        save_bks(bks, bks_path)
        return BKSUpdateResult(
            action="created",
            path=bks_path,
            previous_path=None,
            candidate_cost=candidate_cost,
            candidate_num_routes=bks.num_routes,
        )

    existing_bks = load_bks(existing_path)
    existing_check = check_td_solution(loaded, existing_bks, objective_function)
    if not existing_check.is_valid():
        raise ValueError(f"Stored BKS is invalid at {existing_path}: {existing_check.error_message}")

    existing_cost = existing_bks.cost if existing_bks.cost is not None else float("inf")
    if is_better_solution(
        bks.routes,
        candidate_cost,
        existing_bks.routes,
        existing_cost,
        objective_function,
    ):
        save_bks(bks, bks_path)
        return BKSUpdateResult(
            action="replaced",
            path=bks_path,
            previous_path=existing_path,
            candidate_cost=candidate_cost,
            candidate_num_routes=bks.num_routes,
        )

    return BKSUpdateResult(
        action="kept_existing",
        path=bks_path,
        previous_path=existing_path,
        candidate_cost=candidate_cost,
        candidate_num_routes=bks.num_routes,
    )

build_arc_atf 

build_arc_atf(zones, speeds, distance, horizon)

Exact PWL arrival function: breakpoints at every zone-boundary crossing.

Zone bounds are coerced to floats: canonical ATF breakpoints must be serialized as JSON floats (1080.0), and integer zone bounds would otherwise leak python ints into xs and break the write/reload sha256 round-trip (1080 vs 1080.0).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def build_arc_atf(zones, speeds, distance, horizon):
    """Exact PWL arrival function: breakpoints at every zone-boundary crossing.

    Zone bounds are coerced to floats: canonical ATF breakpoints must be
    serialized as JSON floats (``1080.0``), and integer zone bounds would
    otherwise leak python ints into ``xs`` and break the write/reload sha256
    round-trip (``1080`` vs ``1080.0``).
    """
    zones = [(float(a), float(b)) for a, b in zones]
    t_end = float(horizon[1])
    if distance == 0.0:
        return NDCPWLF([float(horizon[0]), t_end], [float(horizon[0]), t_end])

    def zone_right(x):
        # Zone governing the speed immediately AFTER instant x (forward march).
        k = 0
        while k < len(zones) - 1 and x >= zones[k][1]:
            k += 1
        return k

    xs, ys = [], []
    t = horizon[0]
    while True:
        a = t + ichoua_travel_time(zones, speeds, distance, t)
        if ys and a < ys[-1]:
            a = ys[-1]  # monotone clamp (ulp-scale rounding only)
        xs.append(t)
        ys.append(a)
        if t >= t_end:
            break
        kd = zone_right(t)
        next_dep = zones[kd][1] if kd < len(zones) - 1 else t_end
        # First arrival-boundary crossing strictly after t. The arrival moves
        # at rate speeds[kd]/speeds[ka]; a candidate that fails to advance
        # (previous event one ulp below the boundary) is degenerate -- its
        # breakpoint is the one just emitted -- so target the next boundary.
        ka = zone_right(a)
        next_arr = t_end
        while ka < len(zones) - 1:
            candidate = t + (zones[ka][1] - a) * speeds[ka] / speeds[kd]
            if candidate > t:
                next_arr = candidate
                break
            ka += 1
        t_next = min(next_dep, next_arr, t_end)
        if t_next <= t:
            raise RuntimeError(f"marching stalled at t={t} (distance {distance})")
        t = t_next
    return NDCPWLF(xs, ys)

categories_to_canonical_json_bytes 

categories_to_canonical_json_bytes(
    cats: InstanceCategories,
) -> bytes

Serialize to the canonical JSON bytes (the input of categories_sha256).

Fixed key order, one category row per line, gzip-independent.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def categories_to_canonical_json_bytes(cats: InstanceCategories) -> bytes:
    """Serialize to the canonical JSON bytes (the input of ``categories_sha256``).

    Fixed key order, one category row per line, gzip-independent.
    """
    header_lines = [
        "{",
        f'    "format": {json.dumps(IGP_CATEGORIES_FORMAT)},',
        f'    "format_version": {json.dumps(cats.format_version)},',
        f'    "base_name": {json.dumps(cats.base_name)},',
        f'    "benchmark_name": {json.dumps(cats.benchmark_name)},',
        f'    "num_customers": {json.dumps(cats.num_customers)},',
        f'    "num_categories": {json.dumps(cats.num_categories)},',
        f'    "generator": {json.dumps(cats.generator, sort_keys=True)},',
        '    "categories": [',
    ]
    body = ",\n".join("        " + json.dumps(row) for row in cats.categories)
    text = "\n".join(header_lines) + "\n" + body + "\n    ]\n}\n"
    return text.encode("utf-8")

compute_categories_sha256 

compute_categories_sha256(cats: InstanceCategories) -> str
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
224
225
def compute_categories_sha256(cats: InstanceCategories) -> str:
    return hashlib.sha256(categories_to_canonical_json_bytes(cats)).hexdigest()

euclidean_distance 

euclidean_distance(a, b) -> float

Canonical distance of the igp-profile model: sqrt(dx*dx + dy*dy).

IEEE-754 binary64 with correctly rounded sqrt -- fully deterministic, and bit-identical to the stored Sintef2008 arc_costs on integer G&H coordinates.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
264
265
266
267
268
269
270
271
272
273
def euclidean_distance(a, b) -> float:
    """Canonical distance of the igp-profile model: ``sqrt(dx*dx + dy*dy)``.

    IEEE-754 binary64 with correctly rounded ``sqrt`` -- fully deterministic,
    and bit-identical to the stored Sintef2008 ``arc_costs`` on integer G&H
    coordinates.
    """
    dx = float(a[0]) - float(b[0])
    dy = float(a[1]) - float(b[1])
    return math.sqrt(dx * dx + dy * dy)

ichoua_travel_time 

ichoua_travel_time(zones, speeds, distance, t0)

Exact forward IGP loop; last-zone speed extended beyond the horizon.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def ichoua_travel_time(zones, speeds, distance, t0):
    """Exact forward IGP loop; last-zone speed extended beyond the horizon."""
    k = 0
    while k < len(zones) - 1 and t0 > zones[k][1]:
        k += 1
    t = t0
    d = distance
    tt = t + d / speeds[k]
    while k < len(zones) - 1 and tt > zones[k][1]:
        d = d - speeds[k] * (zones[k][1] - t)
        t = zones[k][1]
        if d <= 0.0:
            break
        k += 1
        tt = t + d / speeds[k]
    return tt - t0

load_instance_categories 

load_instance_categories(
    path: str | Path,
) -> InstanceCategories
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def load_instance_categories(path: str | Path) -> InstanceCategories:
    source = Path(path)
    raw = source.read_bytes()
    if source.name.endswith(IGP_GZIP_SUFFIX):
        raw = gzip.decompress(raw)
    payload = json.loads(raw.decode("utf-8"))
    if payload.get("format") != IGP_CATEGORIES_FORMAT:
        raise IGPFormatError(f"unexpected format marker: {payload.get('format')!r}")
    if payload.get("format_version") != IGP_CATEGORIES_FORMAT_VERSION:
        raise IGPFormatError(f"unsupported format_version: {payload.get('format_version')!r}")
    cats = InstanceCategories(
        base_name=str(payload["base_name"]),
        benchmark_name=str(payload["benchmark_name"]),
        num_customers=int(payload["num_customers"]),
        num_categories=int(payload["num_categories"]),
        categories=[str(row) for row in payload["categories"]],
        generator=dict(payload.get("generator", {})),
    )
    _validate_categories(cats)
    return cats

materialize_instance_atfs 

materialize_instance_atfs(
    instance: AnyTDBenchmarkInstance,
    categories: InstanceCategories,
) -> "InstanceATFs"

Build the canonical complete-graph InstanceATFs of an igp-profile instance.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def materialize_instance_atfs(
    instance: AnyTDBenchmarkInstance,
    categories: InstanceCategories,
) -> "InstanceATFs":
    """Build the canonical complete-graph ``InstanceATFs`` of an igp-profile instance."""
    from mamut_routing_lib.td.artifacts import InstanceATFs

    td = instance.td
    if not isinstance(td, TDIGPProfileRef):
        raise IGPFormatError(f"instance td model is {td.model!r}, expected igp-profile")
    if categories.num_customers != instance.num_customers:
        raise IGPFormatError(
            f"categories num_customers {categories.num_customers} does not match "
            f"instance {instance.num_customers}"
        )
    if categories.num_categories != td.num_categories():
        raise IGPFormatError(
            f"categories num_categories {categories.num_categories} does not match "
            f"the {td.num_categories()} speed rows of the td block"
        )

    horizon = (float(instance.horizon[0]), float(instance.horizon[1]))
    zones = [(float(a), float(b)) for a, b in td.time_periods]
    speeds = [[float(v) for v in row] for row in td.speeds]
    coordinates = instance.coordinates
    num_vertices = instance.num_customers + 1

    arcs: dict[tuple[int, int], NDCPWLF] = {}
    with _gc_paused():
        for i in range(num_vertices):
            for j in range(num_vertices):
                if i == j:
                    continue
                distance = euclidean_distance(coordinates[i], coordinates[j])
                arcs[(i, j)] = build_arc_atf(zones, speeds[categories.category(i, j)], distance, horizon)

    return InstanceATFs(
        instance_name=instance.instance_name,
        benchmark_name=instance.benchmark_name.value,
        horizon=horizon,
        num_customers=instance.num_customers,
        arcs=arcs,
        generator=dict(IGP_MATERIALIZER_GENERATOR),
    )

save_instance_categories 

save_instance_categories(
    cats: InstanceCategories, path: str | Path
) -> None

Write the sidecar; gzip iff the path ends with .igp.json.gz (mtime=0).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/igp.py 
228
229
230
231
232
233
234
235
236
237
238
239
def save_instance_categories(cats: InstanceCategories, path: str | Path) -> None:
    """Write the sidecar; gzip iff the path ends with ``.igp.json.gz`` (``mtime=0``)."""
    _validate_categories(cats)
    target = Path(path)
    data = categories_to_canonical_json_bytes(cats)
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.name.endswith(IGP_GZIP_SUFFIX):
        target.write_bytes(gzip.compress(data, mtime=0))
    elif target.name.endswith(IGP_PLAIN_SUFFIX):
        target.write_bytes(data)
    else:
        raise IGPFormatError(f"categories path must end with {IGP_PLAIN_SUFFIX} or {IGP_GZIP_SUFFIX}: {target.name}")

canonical_route_order 

canonical_route_order(
    routes: list[list[int]],
) -> list[list[int]]
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/checker.py 
164
165
def canonical_route_order(routes: list[list[int]]) -> list[list[int]]:
    return sorted(routes, key=lambda route: route[0])

check_td_solution 

check_td_solution(
    loaded: LoadedTDInstance,
    solution: BenchmarkSolution | BenchmarkBKS,
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> TDSolutionCheckResult
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/checker.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def check_td_solution(
    loaded: LoadedTDInstance,
    solution: BenchmarkSolution | BenchmarkBKS,
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> TDSolutionCheckResult:
    instance = loaded.instance
    atfs = loaded.atfs
    routes = solution.routes
    is_tdvrptw = isinstance(instance, BenchmarkInstanceTDVRPTW)

    # Contract misuse guards, not solution defects: raise instead of returning
    # an invalid result.
    fleet_fixed_cost = _fleet_fixed_cost_for(instance, objective_function)
    declared_objective = getattr(solution, "objective_function", None)
    if declared_objective is not None and declared_objective != objective_function:
        raise ValueError(
            f"Solution declares objective {declared_objective!r} but the check "
            f"was requested for {objective_function!r}"
        )

    served_customers: set[int] = set()
    for route in routes:
        current_load = 0
        for customer in route:
            if customer < 1 or customer > instance.num_customers:
                return TDSolutionCheckResult.make_invalid(
                    SolutionCheckStatus.INVALID_CUSTOMER_INDEX,
                    f"Invalid customer index: {customer}",
                )
            if customer in served_customers:
                return TDSolutionCheckResult.make_invalid(
                    SolutionCheckStatus.CUSTOMER_SERVED_MULTIPLE_TIMES,
                    f"Customer {customer} served more than once.",
                )
            served_customers.add(customer)
            current_load += instance.demands[customer]
            if current_load > instance.vehicle_capacity:
                return TDSolutionCheckResult.make_invalid(
                    SolutionCheckStatus.VEHICLE_CAPACITY_EXCEEDED,
                    f"Vehicle capacity exceeded on route: {route}",
                )

    missing_customers = set(range(1, instance.num_customers + 1)) - served_customers
    if missing_customers:
        return TDSolutionCheckResult.make_invalid(
            SolutionCheckStatus.NOT_ALL_CUSTOMERS_SERVED,
            f"Not all customers served. Missing: {sorted(missing_customers)}",
        )

    if instance.num_vehicles is not None and len(routes) > instance.num_vehicles:
        return TDSolutionCheckResult.make_invalid(
            SolutionCheckStatus.TOO_MANY_VEHICLES_USED,
            "Number of routes exceeds the declared number of vehicles.",
        )

    evaluations: list[TDRouteEvaluation] = []
    for route in routes:
        evaluation = compute_route_duration(instance, atfs, route)
        if not evaluation.feasible:
            status = (
                SolutionCheckStatus.TIME_WINDOW_VIOLATED
                if is_tdvrptw
                else SolutionCheckStatus.ROUTE_TIMING_INFEASIBLE
            )
            return TDSolutionCheckResult.make_invalid(
                status,
                f"Route has no feasible depot departure time: {route}",
            )
        evaluations.append(evaluation)

    total = 0.0
    for evaluation in sorted(evaluations, key=lambda item: item.route[0]):
        assert evaluation.duration is not None
        total += evaluation.duration
    if fleet_fixed_cost is not None:
        total += fleet_fixed_cost * len(routes)

    if solution.cost is not None and solution.cost != total:
        return TDSolutionCheckResult.make_invalid(
            SolutionCheckStatus.OBJECTIVE_VALUE_MISMATCH,
            f"Provided cost {solution.cost!r} does not match computed "
            f"{objective_function.value} {total!r}.",
        )

    return TDSolutionCheckResult(
        status=SolutionCheckStatus.VALID,
        routing_cost=total,
        num_routes=len(routes),
        route_evaluations=evaluations,
    )

compute_route_duration 

compute_route_duration(
    instance: AnyTDBenchmarkInstance,
    atfs: InstanceATFs,
    route: list[int],
) -> TDRouteEvaluation

Optimal Duration Δ*_r and earliest optimal depot departure t*_r.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/checker.py 
151
152
153
154
155
156
157
158
159
160
161
def compute_route_duration(
    instance: AnyTDBenchmarkInstance,
    atfs: InstanceATFs,
    route: list[int],
) -> TDRouteEvaluation:
    """Optimal Duration ``Δ*_r`` and earliest optimal depot departure ``t*_r``."""
    delta = compute_route_ready_time_function(instance, atfs, route)
    if delta.is_empty():
        return TDRouteEvaluation(route=route, feasible=False)
    duration, departure = delta.min_shifted_image()
    return TDRouteEvaluation(route=route, feasible=True, duration=duration, departure_time=departure)

compute_route_ready_time_function 

compute_route_ready_time_function(
    instance: AnyTDBenchmarkInstance,
    atfs: InstanceATFs,
    route: list[int],
) -> NDCPWLF

Route ready-time function δ_r over feasible depot departure times.

Returns an empty function when the route is time-infeasible.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/checker.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def compute_route_ready_time_function(
    instance: AnyTDBenchmarkInstance,
    atfs: InstanceATFs,
    route: list[int],
) -> NDCPWLF:
    """Route ready-time function ``δ_r`` over feasible depot departure times.

    Returns an empty function when the route is time-infeasible.
    """
    horizon_start, horizon_end = atfs.horizon
    depot = instance.depot

    depot_tw = _vertex_time_window(instance, depot)
    if depot_tw is not None:
        departure_low = max(horizon_start, depot_tw[0])
        departure_high = min(horizon_end, depot_tw[1])
    else:
        departure_low, departure_high = horizon_start, horizon_end
    if departure_low > departure_high:
        return NDCPWLF.empty()

    acc = NDCPWLF.identity(departure_low, departure_high)
    previous = depot
    for vertex in route:
        acc = atfs.arcs[(previous, vertex)].compose(acc)
        if acc.is_empty():
            return acc
        service_time = float(instance.service_times[vertex])
        time_window = _vertex_time_window(instance, vertex)
        if time_window is not None:
            theta = make_theta(time_window[0], time_window[1], service_time)
        else:
            theta = make_service_theta(acc.max_image, service_time)
        acc = theta.compose(acc)
        if acc.is_empty():
            return acc
        previous = vertex

    acc = atfs.arcs[(previous, depot)].compose(acc)
    if acc.is_empty():
        return acc
    if depot_tw is not None:
        # Restrict the return arrival to the depot due date, without any
        # waiting clamp: the route ends upon arrival.
        acc = NDCPWLF.identity(0.0, depot_tw[1]).compose(acc)
    return acc

compute_solution_cost 

compute_solution_cost(
    instance: AnyTDBenchmarkInstance,
    atfs: InstanceATFs,
    routes: list[list[int]],
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> float

Total cost under objective_function, canonical route order.

Duration: sum of per-route durations. FleetCostDuration: the same fold, then a single + fleet_fixed_cost * len(routes). Raises on infeasible routes and on a missing fleet_fixed_cost.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/checker.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def compute_solution_cost(
    instance: AnyTDBenchmarkInstance,
    atfs: InstanceATFs,
    routes: list[list[int]],
    objective_function: ObjectiveFunction = ObjectiveFunction.DURATION,
) -> float:
    """Total cost under ``objective_function``, canonical route order.

    Duration: sum of per-route durations. FleetCostDuration: the same fold,
    then a single ``+ fleet_fixed_cost * len(routes)``. Raises on infeasible
    routes and on a missing ``fleet_fixed_cost``.
    """
    fleet_fixed_cost = _fleet_fixed_cost_for(instance, objective_function)
    total = 0.0
    for route in canonical_route_order(routes):
        evaluation = compute_route_duration(instance, atfs, route)
        if not evaluation.feasible:
            raise ValueError(f"route {route} is time-infeasible")
        assert evaluation.duration is not None
        total += evaluation.duration
    if fleet_fixed_cost is not None:
        total += fleet_fixed_cost * len(routes)
    return total

make_service_theta 

make_service_theta(
    upper: float, service_time: float
) -> NDCPWLF

Vertex ready-time function θ without a time window, over [0, upper].

θ(t) = t + service_time, no waiting. upper is the largest arrival time the vertex can see. It shares the dedup of make_theta, so upper == 0 yields the single point ([0.0], [service_time]) instead of two equal consecutive points.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
254
255
256
257
258
259
260
261
262
263
264
def make_service_theta(upper: float, service_time: float) -> NDCPWLF:
    """Vertex ready-time function θ without a time window, over ``[0, upper]``.

    ``θ(t) = t + service_time``, no waiting. ``upper`` is the largest arrival
    time the vertex can see. It shares the dedup of ``make_theta``, so
    ``upper == 0`` yields the single point ``([0.0], [service_time])`` instead
    of two equal consecutive points.
    """
    xs = [0.0, upper]
    ys = [service_time, upper + service_time]
    return NDCPWLF(*_dedup_points(xs, ys))

make_theta 

make_theta(
    earliest: float, latest: float, service_time: float
) -> NDCPWLF

Vertex TW ready-time function θ over arrival times in [0, latest].

θ(t) = max(t, earliest) + service_time: arriving before earliest waits (plateau), arriving after latest is infeasible (out of domain).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/pwlf.py 
241
242
243
244
245
246
247
248
249
250
251
def make_theta(earliest: float, latest: float, service_time: float) -> NDCPWLF:
    """Vertex TW ready-time function θ over arrival times in ``[0, latest]``.

    ``θ(t) = max(t, earliest) + service_time``: arriving before ``earliest``
    waits (plateau), arriving after ``latest`` is infeasible (out of domain).
    """
    if earliest > latest:
        raise PWLFError(f"invalid time window [{earliest}, {latest}]")
    xs = [0.0, earliest, latest]
    ys = [earliest + service_time, earliest + service_time, latest + service_time]
    return NDCPWLF(*_dedup_points(xs, ys))

build_adjacency 

build_adjacency(road: InstanceRoadGraph) -> list[list[int]]

Outgoing edge indices per vertex, in edge-list (sorted (u, v)) order.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
603
604
605
606
607
608
def build_adjacency(road: InstanceRoadGraph) -> list[list[int]]:
    """Outgoing edge indices per vertex, in edge-list (sorted (u, v)) order."""
    adjacency: list[list[int]] = [[] for _ in range(road.num_vertices)]
    for index, (u, _, _, _) in enumerate(road.edges):
        adjacency[u].append(index)
    return adjacency

compute_fastest_path_tree 

compute_fastest_path_tree(
    road: InstanceRoadGraph,
    adjacency: list[list[int]],
    source_vertex: int,
) -> tuple[list[float], list[int]]

Pinned Dijkstra over static free-flow times from source_vertex.

Free-flow edge weight is length / speed_limit: a property of the base's road graph, independent of any traffic overlay. Determinism is pinned end to end: the heap orders (distance, vertex) tuples (total order, so equal-distance pops resolve by vertex id), outgoing edges relax in edge-list order, and labels update on strict < only — the first fastest path found is canonical. Returns (dist, pred_edge) where pred_edge[v] is the index into road.edges of the tree edge entering v (-1 at the source and at unreachable vertices).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
def compute_fastest_path_tree(
    road: InstanceRoadGraph,
    adjacency: list[list[int]],
    source_vertex: int,
) -> tuple[list[float], list[int]]:
    """Pinned Dijkstra over static free-flow times from ``source_vertex``.

    Free-flow edge weight is ``length / speed_limit``: a property of the
    base's road graph, independent of any traffic overlay. Determinism is
    pinned end to end: the heap orders ``(distance, vertex)`` tuples (total
    order, so equal-distance pops resolve by vertex id), outgoing edges relax
    in edge-list order, and labels update on strict ``<`` only — the first
    fastest path found is canonical. Returns ``(dist, pred_edge)`` where
    ``pred_edge[v]`` is the index into ``road.edges`` of the tree edge
    entering ``v`` (``-1`` at the source and at unreachable vertices).
    """
    edges = road.edges
    dist = [float("inf")] * road.num_vertices
    pred_edge = [-1] * road.num_vertices
    dist[source_vertex] = 0.0
    heap: list[tuple[float, int]] = [(0.0, source_vertex)]
    while heap:
        d, u = heapq.heappop(heap)
        if d > dist[u]:
            continue
        for edge_index in adjacency[u]:
            _, v, length, speed_limit = edges[edge_index]
            candidate = d + length / speed_limit
            if candidate < dist[v]:
                dist[v] = candidate
                pred_edge[v] = edge_index
                heapq.heappush(heap, (candidate, v))
    return dist, pred_edge

compute_road_graph_sha256 

compute_road_graph_sha256(road: InstanceRoadGraph) -> str
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
385
386
def compute_road_graph_sha256(road: InstanceRoadGraph) -> str:
    return hashlib.sha256(road_graph_to_canonical_json_bytes(road)).hexdigest()

compute_traffic_overlay_sha256 

compute_traffic_overlay_sha256(
    overlay: TrafficOverlay,
) -> str
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
389
390
def compute_traffic_overlay_sha256(overlay: TrafficOverlay) -> str:
    return hashlib.sha256(traffic_overlay_to_canonical_json_bytes(overlay)).hexdigest()

free_flow_node_times 

free_flow_node_times(
    road: InstanceRoadGraph,
) -> list[list[float]]

Free-flow fastest travel times between all instance nodes.

One pinned Dijkstra per node; entry [i][j] is the free-flow time from node i to node j (0.0 on the diagonal). This is the reference the published distances-fastest sidecar must match (after the family's rounding convention); this is the generation gate of the Poryos2026 collection.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
def free_flow_node_times(road: InstanceRoadGraph) -> list[list[float]]:
    """Free-flow fastest travel times between all instance nodes.

    One pinned Dijkstra per node; entry ``[i][j]`` is the free-flow time from
    node ``i`` to node ``j`` (0.0 on the diagonal). This is the reference the
    published ``distances-fastest`` sidecar must match (after the family's
    rounding convention); this is the generation gate of the Poryos2026 collection.
    """
    adjacency = build_adjacency(road)
    matrix: list[list[float]] = []
    for source_node in range(road.num_customers + 1):
        dist, _ = compute_fastest_path_tree(road, adjacency, road.node_vertices[source_node])
        row = [dist[road.node_vertices[target]] for target in range(road.num_customers + 1)]
        for target, value in enumerate(row):
            if value == float("inf"):
                raise RoadGraphFormatError(
                    f"node {target} unreachable from node {source_node} at free flow"
                )
        matrix.append(row)
    return matrix

load_instance_road_graph 

load_instance_road_graph(
    path: str | Path,
) -> InstanceRoadGraph
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
def load_instance_road_graph(path: str | Path) -> InstanceRoadGraph:
    source = Path(path)
    raw = source.read_bytes()
    if source.name.endswith(ROAD_GZIP_SUFFIX):
        raw = gzip.decompress(raw)
    payload = json.loads(raw.decode("utf-8"))
    if payload.get("format") != ROAD_GRAPH_FORMAT:
        raise RoadGraphFormatError(f"unexpected format marker: {payload.get('format')!r}")
    if payload.get("format_version") != ROAD_GRAPH_FORMAT_VERSION:
        raise RoadGraphFormatError(f"unsupported format_version: {payload.get('format_version')!r}")
    road = InstanceRoadGraph(
        base_name=str(payload["base_name"]),
        benchmark_name=str(payload["benchmark_name"]),
        num_customers=int(payload["num_customers"]),
        horizon=(payload["horizon"][0], payload["horizon"][1]),
        extension_end=payload["extension_end"],
        bin_edges=list(payload["bin_edges"]),
        sample_step=payload["sample_step"],
        simplify_tolerance=payload["simplify_tolerance"],
        num_vertices=int(payload["num_vertices"]),
        vertex_osm_ids=list(payload["vertex_osm_ids"]),
        vertex_lonlat=[(p[0], p[1]) for p in payload["vertex_lonlat"]],
        node_vertices=list(payload["node_vertices"]),
        edges=[(e[0], e[1], e[2], e[3]) for e in payload["edges"]],
        generator=dict(payload.get("generator", {})),
    )
    _validate_road_graph(road)
    return road

load_traffic_overlay 

load_traffic_overlay(path: str | Path) -> TrafficOverlay
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
def load_traffic_overlay(path: str | Path) -> TrafficOverlay:
    source = Path(path)
    raw = source.read_bytes()
    if source.name.endswith(TRAFFIC_GZIP_SUFFIX):
        raw = gzip.decompress(raw)
    payload = json.loads(raw.decode("utf-8"))
    if payload.get("format") != TRAFFIC_OVERLAY_FORMAT:
        raise RoadGraphFormatError(f"unexpected format marker: {payload.get('format')!r}")
    if payload.get("format_version") != TRAFFIC_OVERLAY_FORMAT_VERSION:
        raise RoadGraphFormatError(f"unsupported format_version: {payload.get('format_version')!r}")
    overlay = TrafficOverlay(
        base_name=str(payload["base_name"]),
        benchmark_name=str(payload["benchmark_name"]),
        traffic_model=str(payload["traffic_model"]),
        intensity=str(payload["intensity"]),
        bin_edges=list(payload["bin_edges"]),
        edge_speeds=[list(row) for row in payload["edge_speeds"]],
        generator=dict(payload.get("generator", {})),
    )
    _validate_traffic_overlay(overlay)
    return overlay

materialize_instance_atfs_roadgraph 

materialize_instance_atfs_roadgraph(
    instance: AnyTDBenchmarkInstance,
    road: InstanceRoadGraph,
    overlay: TrafficOverlay,
) -> "InstanceATFs"

Build the canonical complete-graph InstanceATFs of a road-graph instance.

Pinned fastest paths come from the road graph's static free-flow limits (shared by every subinstance of the base); per-edge arrival functions come from the overlay's bin speeds. One pinned Dijkstra per instance node, then per-arc ATFs by exact grid sampling: the departure grid is propagated down the fastest-path tree (arrival vectors memoized per tree vertex), each target's samples are decimated (exact-collinear drop, then Douglas-Peucker with the pinned tolerance), and the kept subset of exact samples becomes the arc ATF.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
def materialize_instance_atfs_roadgraph(
    instance: AnyTDBenchmarkInstance,
    road: InstanceRoadGraph,
    overlay: TrafficOverlay,
) -> "InstanceATFs":
    """Build the canonical complete-graph ``InstanceATFs`` of a road-graph instance.

    Pinned fastest paths come from the road graph's static free-flow limits
    (shared by every subinstance of the base); per-edge arrival functions come
    from the overlay's bin speeds. One pinned Dijkstra per instance node, then
    per-arc ATFs by exact grid sampling: the departure grid is propagated down
    the fastest-path tree (arrival vectors memoized per tree vertex), each
    target's samples are decimated (exact-collinear drop, then Douglas-Peucker
    with the pinned tolerance), and the kept subset of exact samples becomes
    the arc ATF.
    """
    from mamut_routing_lib.td.artifacts import InstanceATFs

    td = instance.td
    if not isinstance(td, TDRoadGraphRef):
        raise RoadGraphFormatError(f"instance td model is {td.model!r}, expected road-graph")
    _validate_road_graph(road)
    _validate_traffic_overlay(overlay)
    validate_overlay_against_road(overlay, road)
    if road.num_customers != instance.num_customers:
        raise RoadGraphFormatError(
            f"road-graph num_customers {road.num_customers} does not match "
            f"instance {instance.num_customers}"
        )
    instance_horizon = (float(instance.horizon[0]), float(instance.horizon[1]))
    if road.horizon != instance_horizon:
        raise RoadGraphFormatError(
            f"road-graph horizon {road.horizon} does not match instance horizon {instance_horizon}"
        )
    if float(td.sample_step) != road.sample_step:
        raise RoadGraphFormatError(
            f"td block sample_step {td.sample_step} does not match road sidecar {road.sample_step}"
        )
    if float(td.simplify_tolerance) != road.simplify_tolerance:
        raise RoadGraphFormatError(
            f"td block simplify_tolerance {td.simplify_tolerance} does not match "
            f"road sidecar {road.simplify_tolerance}"
        )

    zones = [(road.bin_edges[k], road.bin_edges[k + 1]) for k in range(len(road.bin_edges) - 1)]
    extended_horizon = (road.horizon[0], road.extension_end)
    tolerance = road.simplify_tolerance
    adjacency = build_adjacency(road)
    node_vertices = road.node_vertices
    num_nodes = road.num_customers + 1
    grid = departure_grid(road.horizon, road.sample_step)
    grid_root = _np.asarray(grid, dtype=_np.float64) if _np is not None else list(grid)

    evaluators: dict[int, _EdgeEvaluator] = {}

    def edge_evaluator(edge_index: int) -> _EdgeEvaluator:
        evaluator = evaluators.get(edge_index)
        if evaluator is None:
            _, _, length, _ = road.edges[edge_index]
            speeds = overlay.edge_speeds[edge_index]
            evaluator = _EdgeEvaluator(build_arc_atf(zones, speeds, length, extended_horizon))
            evaluators[edge_index] = evaluator
        return evaluator

    arcs: dict[tuple[int, int], NDCPWLF] = {}
    with _gc_paused():
        for source_node in range(num_nodes):
            source_vertex = node_vertices[source_node]
            _, pred_edge = compute_fastest_path_tree(road, adjacency, source_vertex)
            arrivals = {source_vertex: grid_root}
            for target_node in range(num_nodes):
                if target_node == source_node:
                    continue
                target_vertex = node_vertices[target_node]
                chain: list[int] = []
                vertex = target_vertex
                while vertex not in arrivals:
                    edge_index = pred_edge[vertex]
                    if edge_index < 0:
                        raise RoadGraphFormatError(
                            f"node {target_node} (OSM {road.vertex_osm_ids[target_vertex]}) "
                            f"unreachable from node {source_node} "
                            f"(OSM {road.vertex_osm_ids[source_vertex]})"
                        )
                    chain.append(vertex)
                    vertex = road.edges[edge_index][0]
                for vertex in reversed(chain):
                    edge_index = pred_edge[vertex]
                    parent = road.edges[edge_index][0]
                    values = edge_evaluator(edge_index).evaluate_many(arrivals[parent])
                    last = float(values[-1])
                    if last > road.extension_end:
                        raise RoadGraphFormatError(
                            f"arrival {last} exceeds extension_end {road.extension_end} "
                            f"— extension_end too small"
                        )
                    arrivals[vertex] = values
                samples = arrivals[target_vertex]
                ys = [float(v) for v in samples] if _np is not None else list(samples)
                xs, ys = _drop_exactly_collinear(grid, ys)
                xs, ys = _simplify_points(xs, ys, tolerance)
                arcs[(source_node, target_node)] = NDCPWLF(list(xs), list(ys), validate=False)
            # Free per-source memo before the next Dijkstra; keeps peak memory
            # to one shortest-path tree of arrival vectors at a time.
            del arrivals
    return InstanceATFs(
        instance_name=instance.instance_name,
        benchmark_name=instance.benchmark_name.value,
        horizon=road.horizon,
        num_customers=road.num_customers,
        arcs=arcs,
        generator=dict(ROAD_MATERIALIZER_GENERATOR),
    )

materialize_selected_atfs_roadgraph 

materialize_selected_atfs_roadgraph(
    instance: AnyTDBenchmarkInstance,
    road: InstanceRoadGraph,
    overlay: TrafficOverlay,
    selected_arcs: set[tuple[int, int]],
) -> dict[tuple[int, int], NDCPWLF]

Materialize only selected complete-graph arcs with canonical semantics.

This is the sparse counterpart of :func:materialize_instance_atfs_roadgraph. It uses the same pinned paths, departure grid, edge evaluation and simplification, but avoids constructing the quadratic matrix when a generation audit needs only the arcs of a route certificate.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
def materialize_selected_atfs_roadgraph(
    instance: AnyTDBenchmarkInstance,
    road: InstanceRoadGraph,
    overlay: TrafficOverlay,
    selected_arcs: set[tuple[int, int]],
) -> dict[tuple[int, int], NDCPWLF]:
    """Materialize only selected complete-graph arcs with canonical semantics.

    This is the sparse counterpart of :func:`materialize_instance_atfs_roadgraph`.
    It uses the same pinned paths, departure grid, edge evaluation and
    simplification, but avoids constructing the quadratic matrix when a
    generation audit needs only the arcs of a route certificate.
    """
    td = instance.td
    if not isinstance(td, TDRoadGraphRef):
        raise RoadGraphFormatError(f"instance td model is {td.model!r}, expected road-graph")
    _validate_road_graph(road)
    _validate_traffic_overlay(overlay)
    validate_overlay_against_road(overlay, road)
    if road.num_customers != instance.num_customers:
        raise RoadGraphFormatError("road-graph customer count does not match instance")
    num_nodes = road.num_customers + 1
    for source, target in selected_arcs:
        if not (0 <= source < num_nodes and 0 <= target < num_nodes) or source == target:
            raise RoadGraphFormatError(f"invalid selected arc {(source, target)}")

    zones = [(road.bin_edges[k], road.bin_edges[k + 1]) for k in range(len(road.bin_edges) - 1)]
    extended_horizon = (road.horizon[0], road.extension_end)
    adjacency = build_adjacency(road)
    grid = departure_grid(road.horizon, road.sample_step)
    grid_root = _np.asarray(grid, dtype=_np.float64) if _np is not None else list(grid)
    evaluators: dict[int, _EdgeEvaluator] = {}

    def edge_evaluator(edge_index: int) -> _EdgeEvaluator:
        evaluator = evaluators.get(edge_index)
        if evaluator is None:
            _, _, length, _ = road.edges[edge_index]
            evaluator = _EdgeEvaluator(
                build_arc_atf(zones, overlay.edge_speeds[edge_index], length, extended_horizon)
            )
            evaluators[edge_index] = evaluator
        return evaluator

    targets_by_source: dict[int, list[int]] = {}
    for source, target in sorted(selected_arcs):
        targets_by_source.setdefault(source, []).append(target)
    arcs: dict[tuple[int, int], NDCPWLF] = {}
    with _gc_paused():
        for source_node, targets in targets_by_source.items():
            source_vertex = road.node_vertices[source_node]
            _, pred_edge = compute_fastest_path_tree(road, adjacency, source_vertex)
            arrivals = {source_vertex: grid_root}
            for target_node in targets:
                target_vertex = road.node_vertices[target_node]
                chain: list[int] = []
                vertex = target_vertex
                while vertex not in arrivals:
                    edge_index = pred_edge[vertex]
                    if edge_index < 0:
                        raise RoadGraphFormatError(
                            f"node {target_node} unreachable from node {source_node}"
                        )
                    chain.append(vertex)
                    vertex = road.edges[edge_index][0]
                for vertex in reversed(chain):
                    edge_index = pred_edge[vertex]
                    parent = road.edges[edge_index][0]
                    values = edge_evaluator(edge_index).evaluate_many(arrivals[parent])
                    if float(values[-1]) > road.extension_end:
                        raise RoadGraphFormatError("selected-arc arrival exceeds extension_end")
                    arrivals[vertex] = values
                ys = [float(value) for value in arrivals[target_vertex]]
                xs, ys = _drop_exactly_collinear(grid, ys)
                xs, ys = _simplify_points(xs, ys, road.simplify_tolerance)
                arcs[(source_node, target_node)] = NDCPWLF(list(xs), list(ys), validate=False)
    return arcs

road_graph_to_canonical_json_bytes 

road_graph_to_canonical_json_bytes(
    road: InstanceRoadGraph,
) -> bytes

Serialize to the canonical JSON bytes (the input of the graph sha256).

Fixed key order, one edge / one vertex coordinate pair per line, floats via Python's shortest round-trip repr, gzip-independent.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def road_graph_to_canonical_json_bytes(road: InstanceRoadGraph) -> bytes:
    """Serialize to the canonical JSON bytes (the input of the graph sha256).

    Fixed key order, one edge / one vertex coordinate pair per line, floats
    via Python's shortest round-trip repr, gzip-independent.
    """
    header_lines = [
        "{",
        f'    "format": {json.dumps(ROAD_GRAPH_FORMAT)},',
        f'    "format_version": {json.dumps(road.format_version)},',
        f'    "base_name": {json.dumps(road.base_name)},',
        f'    "benchmark_name": {json.dumps(road.benchmark_name)},',
        f'    "num_customers": {json.dumps(road.num_customers)},',
        f'    "horizon": {json.dumps(list(road.horizon))},',
        f'    "extension_end": {json.dumps(road.extension_end)},',
        f'    "bin_edges": {json.dumps(road.bin_edges)},',
        f'    "sample_step": {json.dumps(road.sample_step)},',
        f'    "simplify_tolerance": {json.dumps(road.simplify_tolerance)},',
        f'    "num_vertices": {json.dumps(road.num_vertices)},',
        f'    "vertex_osm_ids": {json.dumps(road.vertex_osm_ids)},',
        f'    "node_vertices": {json.dumps(road.node_vertices)},',
        f'    "generator": {json.dumps(road.generator, sort_keys=True)},',
        '    "vertex_lonlat": [',
    ]
    lonlat_body = ",\n".join(
        "        " + json.dumps([lon, lat]) for lon, lat in road.vertex_lonlat
    )
    edge_header = '    ],\n    "edges": ['
    edges_body = ",\n".join(
        "        " + json.dumps([u, v, length, speed_limit])
        for u, v, length, speed_limit in road.edges
    )
    text = (
        "\n".join(header_lines)
        + "\n" + lonlat_body + "\n" + edge_header + "\n" + edges_body + "\n    ]\n}\n"
    )
    return text.encode("utf-8")

save_instance_road_graph 

save_instance_road_graph(
    road: InstanceRoadGraph, path: str | Path
) -> None

Write the sidecar; gzip iff the path ends with .road.json.gz (mtime=0).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
393
394
395
396
397
398
399
400
401
402
403
404
405
406
def save_instance_road_graph(road: InstanceRoadGraph, path: str | Path) -> None:
    """Write the sidecar; gzip iff the path ends with ``.road.json.gz`` (``mtime=0``)."""
    _validate_road_graph(road)
    target = Path(path)
    data = road_graph_to_canonical_json_bytes(road)
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.name.endswith(ROAD_GZIP_SUFFIX):
        target.write_bytes(gzip.compress(data, mtime=0))
    elif target.name.endswith(ROAD_PLAIN_SUFFIX):
        target.write_bytes(data)
    else:
        raise RoadGraphFormatError(
            f"road-graph path must end with {ROAD_PLAIN_SUFFIX} or {ROAD_GZIP_SUFFIX}: {target.name}"
        )

save_traffic_overlay 

save_traffic_overlay(
    overlay: TrafficOverlay, path: str | Path
) -> None

Write the overlay; gzip iff the path ends with .json.gz (mtime=0).

The conventional name is <base>.traffic-<model>-<intensity>.json[.gz]; the .traffic- infix is required so overlays are never confused with other sidecars.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
def save_traffic_overlay(overlay: TrafficOverlay, path: str | Path) -> None:
    """Write the overlay; gzip iff the path ends with ``.json.gz`` (``mtime=0``).

    The conventional name is ``<base>.traffic-<model>-<intensity>.json[.gz]``;
    the ``.traffic-`` infix is required so overlays are never confused with
    other sidecars.
    """
    _validate_traffic_overlay(overlay)
    target = Path(path)
    if TRAFFIC_INFIX not in target.name:
        raise RoadGraphFormatError(f"traffic-overlay path must contain {TRAFFIC_INFIX!r}: {target.name}")
    data = traffic_overlay_to_canonical_json_bytes(overlay)
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.name.endswith(TRAFFIC_GZIP_SUFFIX):
        target.write_bytes(gzip.compress(data, mtime=0))
    elif target.name.endswith(TRAFFIC_PLAIN_SUFFIX):
        target.write_bytes(data)
    else:
        raise RoadGraphFormatError(
            f"traffic-overlay path must end with {TRAFFIC_PLAIN_SUFFIX} or {TRAFFIC_GZIP_SUFFIX}: {target.name}"
        )

simplify_ndcpwlf 

simplify_ndcpwlf(f: NDCPWLF, tolerance: float) -> NDCPWLF

Deterministic Douglas-Peucker simplification of an NDCPWLF.

Keeps a subset of the original breakpoints (endpoints always kept), so the result is non-decreasing and continuous by construction and every kept breakpoint keeps ys >= xs. Interior breakpoints are dropped when their vertical deviation from the chord of the enclosing kept pair is <= tolerance; the split point is the deviation argmax, ties broken by the smallest index. A vertical step strictly inside a kept pair may be smoothed like any other breakpoint (its deviation is bounded the same way); only when the kept pair itself is a pure vertical run (equal x) is chord interpolation undefined and the run kept whole.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
def simplify_ndcpwlf(f: NDCPWLF, tolerance: float) -> NDCPWLF:
    """Deterministic Douglas-Peucker simplification of an NDCPWLF.

    Keeps a subset of the original breakpoints (endpoints always kept), so
    the result is non-decreasing and continuous by construction and every
    kept breakpoint keeps ``ys >= xs``. Interior breakpoints are dropped when
    their vertical deviation from the chord of the enclosing kept pair is
    ``<= tolerance``; the split point is the deviation argmax, ties broken by
    the smallest index. A vertical step strictly inside a kept pair may be
    smoothed like any other breakpoint (its deviation is bounded the same
    way); only when the kept pair itself is a pure vertical run (equal
    ``x``) is chord interpolation undefined and the run kept whole.
    """
    if tolerance < 0:
        raise RoadGraphFormatError(f"tolerance must be >= 0, got {tolerance}")
    if f.num_breakpoints() <= 2:
        return f
    xs, ys = _simplify_points(f.xs, f.ys, tolerance)
    if xs is f.xs:
        return f
    return NDCPWLF(xs, ys, validate=False)

traffic_overlay_to_canonical_json_bytes 

traffic_overlay_to_canonical_json_bytes(
    overlay: TrafficOverlay,
) -> bytes

Serialize to the canonical JSON bytes (the input of the traffic sha256).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def traffic_overlay_to_canonical_json_bytes(overlay: TrafficOverlay) -> bytes:
    """Serialize to the canonical JSON bytes (the input of the traffic sha256)."""
    header_lines = [
        "{",
        f'    "format": {json.dumps(TRAFFIC_OVERLAY_FORMAT)},',
        f'    "format_version": {json.dumps(overlay.format_version)},',
        f'    "base_name": {json.dumps(overlay.base_name)},',
        f'    "benchmark_name": {json.dumps(overlay.benchmark_name)},',
        f'    "traffic_model": {json.dumps(overlay.traffic_model)},',
        f'    "intensity": {json.dumps(overlay.intensity)},',
        f'    "bin_edges": {json.dumps(overlay.bin_edges)},',
        f'    "generator": {json.dumps(overlay.generator, sort_keys=True)},',
        '    "edge_speeds": [',
    ]
    body = ",\n".join("        " + json.dumps(row) for row in overlay.edge_speeds)
    text = "\n".join(header_lines) + "\n" + body + "\n    ]\n}\n"
    return text.encode("utf-8")

validate_overlay_against_road 

validate_overlay_against_road(
    overlay: TrafficOverlay, road: InstanceRoadGraph
) -> None

The alignment contract between a traffic overlay and its base road graph.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/roadgraph.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def validate_overlay_against_road(overlay: TrafficOverlay, road: InstanceRoadGraph) -> None:
    """The alignment contract between a traffic overlay and its base road graph."""
    if overlay.base_name != road.base_name:
        raise RoadGraphFormatError(
            f"overlay base_name {overlay.base_name!r} does not match road graph {road.base_name!r}"
        )
    if overlay.benchmark_name != road.benchmark_name:
        raise RoadGraphFormatError(
            f"overlay benchmark_name {overlay.benchmark_name!r} does not match "
            f"road graph {road.benchmark_name!r}"
        )
    if overlay.bin_edges != road.bin_edges:
        raise RoadGraphFormatError("overlay bin_edges do not equal the road graph's bit-exactly")
    if len(overlay.edge_speeds) != len(road.edges):
        raise RoadGraphFormatError(
            f"overlay has {len(overlay.edge_speeds)} edge rows, road graph has {len(road.edges)} edges"
        )
    for index, ((u, v, _, speed_limit), row) in enumerate(zip(road.edges, overlay.edge_speeds)):
        for b, speed in enumerate(row):
            if speed > speed_limit:
                raise RoadGraphFormatError(
                    f"edge {index} ({u}, {v}) speed[{b}] = {speed} exceeds the free-flow "
                    f"speed_limit {speed_limit} (overlays are slowdowns by contract)"
                )