mamut_routing_lib.geo

Geo sidecar of a collection base instance (format mamut-geo v3).

One geo sidecar per base instance (<base>.geo.json[.gz]), shared by every problem-type variant of the base. It carries everything the site/workbench needs to draw the instance on a map: per-node geodetic + local ENU positions, the geodetic reference frame, and (for n <= 100) a complete road-path cache over all ordered node pairs for the fastest and shortest metrics in an indexed encoding: one geo-local vertex_lonlat table (the union of all path vertices) plus per-metric maps from "i-j" arc keys to lists of indices into that table. Measured at n=100 this is ~5x smaller gzipped than repeating coordinate polylines per pair. The sidecar is static: it never changes when BKS routes change (v3 retires the on-demand road_cache growth of the v1/v2 meta.json).

Purely informative for solvers: the checker never reads it. Hashed over its uncompressed canonical JSON bytes like every other sidecar, gzip mtime=0.

GEO_FORMAT module-attribute

GEO_FORMAT = 'mamut-geo'

GEO_FORMAT_VERSION module-attribute

GEO_FORMAT_VERSION = 3

GEO_PLAIN_SUFFIX module-attribute

GEO_PLAIN_SUFFIX = '.geo.json'

GEO_GZIP_SUFFIX module-attribute

GEO_GZIP_SUFFIX = '.geo.json.gz'

GEO_ROAD_CACHE_METRICS module-attribute

GEO_ROAD_CACHE_METRICS = ('fastest', 'shortest')

GeoFormatError

Bases: ValueError

Raised when a geo sidecar violates the canonical format.

GeoNode dataclass

GeoNode(instance_node_id: int, poi_lon: float, poi_lat: float, enu_x: float, enu_y: float, demand: int, source_tag: str, graph_vertex_id: int | None = None)

One instance node on the map (depot included, instance_node_id 0-based).

instance_node_id instance-attribute

instance_node_id: int

poi_lon instance-attribute

poi_lon: float

poi_lat instance-attribute

poi_lat: float

enu_x instance-attribute

enu_x: float

enu_y instance-attribute

enu_y: float

demand instance-attribute

demand: int

source_tag instance-attribute

source_tag: str

graph_vertex_id class-attribute instance-attribute

graph_vertex_id: int | None = None

GeoRoadCache dataclass

GeoRoadCache(vertex_lonlat: list[tuple[float, float]], paths: dict[str, dict[str, list[int]]] = dict())

Indexed road-path cache: paths as index lists over a local vertex table.

paths[metric]["i-j"] is the vertex-index polyline of the cached road path from instance node i to instance node j (ordered pair, indices into vertex_lonlat, at least two points). A complete cache covers all ordered pairs of the instance for every metric it carries; completeness is a family policy (Poryos2026: n <= 100 only), not a format invariant.

vertex_lonlat instance-attribute

vertex_lonlat: list[tuple[float, float]]

paths class-attribute instance-attribute

paths: dict[str, dict[str, list[int]]] = field(default_factory=dict)

path_lonlat

path_lonlat(metric: str, from_node: int, to_node: int) -> list[tuple[float, float]]

Decode one cached path back to a lon/lat polyline.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/geo.py

def path_lonlat(self, metric: str, from_node: int, to_node: int) -> list[tuple[float, float]]:
    """Decode one cached path back to a lon/lat polyline."""
    entries = self.paths.get(metric)
    if entries is None:
        raise GeoFormatError(f"no cached paths for metric {metric!r}")
    key = f"{from_node}-{to_node}"
    path = entries.get(key)
    if path is None:
        raise GeoFormatError(f"no cached {metric} path for arc {key}")
    return [self.vertex_lonlat[v] for v in path]

InstanceGeo dataclass

InstanceGeo(base_name: str, benchmark_name: str, city: str, method: str, source_osm_file: str, reference_lla: dict[str, float], map_options: dict[str, Any], nodes: list[GeoNode], road_cache: GeoRoadCache | None = None, generator: dict[str, Any] = dict(), format_version: int = GEO_FORMAT_VERSION)

In-memory content of a geo sidecar.

base_name instance-attribute

base_name: str

benchmark_name instance-attribute

benchmark_name: str

city instance-attribute

city: str

method instance-attribute

method: str

source_osm_file instance-attribute

source_osm_file: str

reference_lla instance-attribute

reference_lla: dict[str, float]

map_options instance-attribute

map_options: dict[str, Any]

nodes instance-attribute

nodes: list[GeoNode]

road_cache class-attribute instance-attribute

road_cache: GeoRoadCache | None = None

generator class-attribute instance-attribute

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

format_version class-attribute instance-attribute

format_version: int = GEO_FORMAT_VERSION

num_customers

num_customers() -> int

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/geo.py

def num_customers(self) -> int:
    return len(self.nodes) - 1

geo_to_canonical_json_bytes

geo_to_canonical_json_bytes(geo: InstanceGeo) -> bytes

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

Fixed key order, one node / one cache vertex / one cached path per line, floats via Python's shortest round-trip repr, gzip-independent. Path maps are ordered by (i, j).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/geo.py

def geo_to_canonical_json_bytes(geo: InstanceGeo) -> bytes:
    """Serialize to the canonical JSON bytes (the input of the geo sha256).

    Fixed key order, one node / one cache vertex / one cached path per line,
    floats via Python's shortest round-trip repr, gzip-independent. Path maps
    are ordered by ``(i, j)``.
    """
    header_lines = [
        "{",
        f'    "format": {json.dumps(GEO_FORMAT)},',
        f'    "format_version": {json.dumps(geo.format_version)},',
        f'    "base_name": {json.dumps(geo.base_name)},',
        f'    "benchmark_name": {json.dumps(geo.benchmark_name)},',
        f'    "city": {json.dumps(geo.city)},',
        f'    "method": {json.dumps(geo.method)},',
        f'    "source_osm_file": {json.dumps(geo.source_osm_file)},',
        f'    "reference_lla": {json.dumps(geo.reference_lla, sort_keys=True)},',
        f'    "map_options": {json.dumps(geo.map_options, sort_keys=True)},',
        f'    "generator": {json.dumps(geo.generator, sort_keys=True)},',
        '    "nodes": [',
    ]
    node_lines = ",\n".join(
        "        "
        + json.dumps(
            {
                "instance_node_id": node.instance_node_id,
                "poi_lon": node.poi_lon,
                "poi_lat": node.poi_lat,
                "enu_x": node.enu_x,
                "enu_y": node.enu_y,
                "demand": node.demand,
                "source_tag": node.source_tag,
                "graph_vertex_id": node.graph_vertex_id,
            }
        )
        for node in geo.nodes
    )
    prefix = "\n".join(header_lines) + "\n" + node_lines + "\n"
    if geo.road_cache is None:
        return (prefix + '    ],\n    "road_cache": null\n}\n').encode("utf-8")
    cache = geo.road_cache
    cache_lines = ['    ],', '    "road_cache": {', '        "vertex_lonlat": [']
    cache_lines.append(
        ",\n".join("            " + json.dumps([lon, lat]) for lon, lat in cache.vertex_lonlat)
    )
    cache_lines.append('        ],')
    cache_lines.append('        "paths": {')
    metric_blocks = []
    for metric in sorted(cache.paths):
        entries = cache.paths[metric]
        entry_lines = ",\n".join(
            f"                {json.dumps(key)}: {json.dumps(entries[key])}"
            for key in sorted(entries, key=_path_key_order)
        )
        metric_blocks.append(
            f"            {json.dumps(metric)}: {{\n" + entry_lines + "\n            }"
        )
    cache_lines.append(",\n".join(metric_blocks))
    cache_lines.append("        }")
    cache_lines.append("    }")
    return (prefix + "\n".join(cache_lines) + "\n}\n").encode("utf-8")

compute_geo_sha256

compute_geo_sha256(geo: InstanceGeo) -> str

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/geo.py

def compute_geo_sha256(geo: InstanceGeo) -> str:
    return hashlib.sha256(geo_to_canonical_json_bytes(geo)).hexdigest()

save_instance_geo

save_instance_geo(geo: InstanceGeo, path: str | Path) -> None

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

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/geo.py

def save_instance_geo(geo: InstanceGeo, path: str | Path) -> None:
    """Write the sidecar; gzip iff the path ends with ``.geo.json.gz`` (``mtime=0``)."""
    _validate_geo(geo)
    target = Path(path)
    data = geo_to_canonical_json_bytes(geo)
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.name.endswith(GEO_GZIP_SUFFIX):
        target.write_bytes(gzip.compress(data, mtime=0))
    elif target.name.endswith(GEO_PLAIN_SUFFIX):
        target.write_bytes(data)
    else:
        raise GeoFormatError(f"geo path must end with {GEO_PLAIN_SUFFIX} or {GEO_GZIP_SUFFIX}: {target.name}")

load_instance_geo

load_instance_geo(path: str | Path) -> InstanceGeo

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/geo.py

def load_instance_geo(path: str | Path) -> InstanceGeo:
    source = Path(path)
    raw = source.read_bytes()
    if source.name.endswith(GEO_GZIP_SUFFIX):
        raw = gzip.decompress(raw)
    payload = json.loads(raw.decode("utf-8"))
    if payload.get("format") != GEO_FORMAT:
        raise GeoFormatError(f"unexpected format marker: {payload.get('format')!r}")
    if payload.get("format_version") != GEO_FORMAT_VERSION:
        raise GeoFormatError(f"unsupported format_version: {payload.get('format_version')!r}")
    cache_payload = payload.get("road_cache")
    road_cache = None
    if cache_payload is not None:
        road_cache = GeoRoadCache(
            vertex_lonlat=[(p[0], p[1]) for p in cache_payload["vertex_lonlat"]],
            paths={
                metric: dict(entries)
                for metric, entries in cache_payload.get("paths", {}).items()
            },
        )
    geo = InstanceGeo(
        base_name=str(payload["base_name"]),
        benchmark_name=str(payload["benchmark_name"]),
        city=str(payload["city"]),
        method=str(payload["method"]),
        source_osm_file=str(payload["source_osm_file"]),
        reference_lla=dict(payload["reference_lla"]),
        map_options=dict(payload["map_options"]),
        nodes=[GeoNode(**node) for node in payload["nodes"]],
        road_cache=road_cache,
        generator=dict(payload.get("generator", {})),
    )
    _validate_geo(geo)
    return geo