IO for TD benchmark artifacts: ATF sidecar files and TD instance loading.
The ATF sidecar (<Instance>.atf.json or <Instance>.atf.json.gz) is the
canonical ground truth of a TD instance: one arrival-time NDCPWLF per arc of
the complete customer-based graph. Plain JSON is used for small instances
(debuggable at a glance), gzip for large ones; the loader accepts both. The
gzip form is written with mtime=0 so bytes are deterministic, and the
atf_sha256 recorded in instance files always hashes the uncompressed
canonical JSON bytes, making it storage-form independent.
module-attribute
ATF_PLAIN_SUFFIX = '.atf.json'
module-attribute
ATF_GZIP_SUFFIX = '.atf.json.gz'
Bases: ValueError
Raised when an ATF sidecar file violates the canonical format.
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-attribute
instance-attribute
instance-attribute
horizon: tuple[float, float]
instance-attribute
instance-attribute
arcs: dict[tuple[int, int], NDCPWLF]
class-attribute
instance-attribute
generator: dict[str, Any] = field(default_factory=dict)
format_version: int = ATF_FORMAT_VERSION
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py
| def num_vertices(self) -> int:
return self.num_customers + 1
|
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-attribute
instance: AnyTDBenchmarkInstance
instance-attribute
instance-attribute
class-attribute
instance-attribute
categories_path: Path | None = None
class-attribute
instance-attribute
road_graph_path: Path | None = None
class-attribute
instance-attribute
traffic_path: Path | None = None
class-attribute
instance-attribute
collection_root: Path | None = None
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
| 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(atfs: InstanceATFs) -> str
Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/td/artifacts.py
| def compute_atf_sha256(atfs: InstanceATFs) -> str:
return hashlib.sha256(atfs_to_canonical_json_bytes(atfs)).hexdigest()
|
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
| 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}")
|
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
| 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)
|
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
| 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_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
| 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)
|
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
| def td_instance_from_payload(payload: dict[str, Any]) -> AnyTDBenchmarkInstance:
if "time_windows" in payload:
return BenchmarkInstanceTDVRPTW(**payload)
return BenchmarkInstanceTDVRP(**payload)
|
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
| 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)
|