mirror of
https://github.com/torlando-tech/pyxis.git
synced 2026-08-17 00:00:10 +00:00
979 lines
43 KiB
Python
979 lines
43 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (c) 2026 Pyxis contributors
|
|
# SPDX-License-Identifier: MIT
|
|
"""Build a Pyxis offline map pack exclusively from a local XYZ PNG tree.
|
|
|
|
Security-sensitive filesystem operations intentionally support Linux only. PNG
|
|
support is intentionally limited to non-interlaced images.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ctypes
|
|
from dataclasses import dataclass
|
|
import errno
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import struct
|
|
import sys
|
|
from typing import Any, cast, Iterable
|
|
import zlib
|
|
|
|
MAGIC = b"PMPK"
|
|
FORMAT_VERSION = 1
|
|
SPARSE_FORMAT_VERSION = 2
|
|
HEADER_SIZE = 16
|
|
EXTENT_SIZE = 26
|
|
ROW_SPAN_SIZE = 13
|
|
MAX_ROW_SPANS = 512
|
|
MAX_ZOOM = 22
|
|
DEFAULT_MAX_TILES = 100_000
|
|
DEFAULT_MAX_BYTES = 8 * 1024 * 1024 * 1024
|
|
MAX_VISITED_ENTRIES_BASE = 64
|
|
PACK_ID_RE = re.compile(r"[a-z0-9_-]{1,31}\Z")
|
|
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
|
|
_STRING_LIMITS = {"pack_id": 31, "name": 63, "attribution": 127, "source": 127, "license": 63}
|
|
MAX_MANIFEST_BYTES = 7100
|
|
_DIRECTORY_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
|
|
_FILE_FLAGS = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC
|
|
|
|
if sys.platform.startswith("linux"):
|
|
_libc = ctypes.CDLL(None, use_errno=True)
|
|
_renameat2 = getattr(_libc, "renameat2", None)
|
|
if _renameat2 is not None:
|
|
_renameat2.argtypes = [ctypes.c_int, ctypes.c_char_p,
|
|
ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
|
|
_renameat2.restype = ctypes.c_int
|
|
else:
|
|
_renameat2 = None
|
|
|
|
|
|
class PackError(ValueError):
|
|
"""Input cannot be represented as a safe, complete Pyxis map pack."""
|
|
|
|
|
|
class PublishedDurabilityError(PackError):
|
|
"""Publication committed, but syncing its parent directory failed."""
|
|
|
|
published = True
|
|
|
|
def __init__(self, target: Path, cause: OSError):
|
|
self.target = target
|
|
super().__init__(f"pack published at {target}, but durability is uncertain: {cause}")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ZoomExtent:
|
|
zoom: int
|
|
intervals: tuple[tuple[int, int], ...]
|
|
y_minimum: int
|
|
y_maximum: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RowSpan:
|
|
zoom: int
|
|
y: int
|
|
x_minimum: int
|
|
x_maximum: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Tile:
|
|
zoom: int
|
|
x: int
|
|
y: int
|
|
path: Path
|
|
size: int
|
|
zoom_identity: tuple[int, ...] = ()
|
|
x_identity: tuple[int, ...] = ()
|
|
file_identity: tuple[int, ...] = ()
|
|
|
|
|
|
def _identity(info: os.stat_result) -> tuple[int, ...]:
|
|
return (info.st_dev, info.st_ino, info.st_mode, info.st_size, info.st_mtime_ns, info.st_ctime_ns)
|
|
|
|
|
|
def _checked_text(field: str, value: str) -> bytes:
|
|
if not value:
|
|
raise PackError(f"{field} metadata is required")
|
|
try:
|
|
encoded = value.encode("ascii")
|
|
except UnicodeEncodeError as exc:
|
|
raise PackError(f"{field} must contain printable ASCII") from exc
|
|
if len(encoded) > _STRING_LIMITS[field] or any(byte < 0x20 or byte > 0x7E for byte in encoded):
|
|
raise PackError(f"{field} must be 1-{_STRING_LIMITS[field]} printable ASCII bytes")
|
|
return encoded
|
|
|
|
|
|
def _validate_metadata(pack_id: str, name: str, attribution: str, source: str, license: str) -> None:
|
|
if not PACK_ID_RE.fullmatch(pack_id):
|
|
raise PackError("pack ID must match [a-z0-9_-]{1,31}")
|
|
for field, value in (("name", name), ("attribution", attribution), ("source", source), ("license", license)):
|
|
_checked_text(field, value)
|
|
|
|
|
|
def _canonical_number(name: str, label: str) -> int:
|
|
if not name.isascii() or not name.isdecimal() or str(int(name)) != name:
|
|
raise PackError(f"noncanonical {label} path component: {name!r}")
|
|
return int(name)
|
|
|
|
|
|
def _fsync_directory_fd(descriptor: int) -> None:
|
|
os.fsync(descriptor)
|
|
|
|
|
|
def _open_path(directory: Path, *, create: bool = False) -> int:
|
|
path = Path(directory)
|
|
if path.is_absolute():
|
|
descriptor = os.open("/", _DIRECTORY_FLAGS)
|
|
parts = path.parts[1:]
|
|
else:
|
|
descriptor = os.open(".", _DIRECTORY_FLAGS)
|
|
parts = path.parts
|
|
try:
|
|
for part in parts:
|
|
if part in ("", "."):
|
|
continue
|
|
if part == "..":
|
|
raise PackError(f"parent traversal is not allowed: {path}")
|
|
try:
|
|
child = os.open(part, _DIRECTORY_FLAGS, dir_fd=descriptor)
|
|
except FileNotFoundError:
|
|
if not create:
|
|
raise PackError(f"directory does not exist: {path}")
|
|
os.mkdir(part, 0o755, dir_fd=descriptor)
|
|
_fsync_directory_fd(descriptor)
|
|
child = os.open(part, _DIRECTORY_FLAGS, dir_fd=descriptor)
|
|
except OSError as exc:
|
|
raise PackError(f"directory must be a real, nonsymlink directory: {path}: {exc}") from exc
|
|
os.close(descriptor)
|
|
descriptor = child
|
|
return descriptor
|
|
except BaseException:
|
|
os.close(descriptor)
|
|
raise
|
|
|
|
|
|
def _mkdir_open(parent_fd: int, name: str) -> tuple[int, tuple[int, ...]]:
|
|
try:
|
|
os.mkdir(name, 0o755, dir_fd=parent_fd)
|
|
_fsync_directory_fd(parent_fd)
|
|
except FileExistsError:
|
|
pass
|
|
try:
|
|
descriptor = os.open(name, _DIRECTORY_FLAGS, dir_fd=parent_fd)
|
|
except OSError as exc:
|
|
raise PackError(f"output component is not a real directory: {name}: {exc}") from exc
|
|
return descriptor, _identity(os.fstat(descriptor))
|
|
|
|
|
|
def _read_exact(descriptor: int, size: int, label: str) -> bytes:
|
|
result = bytearray()
|
|
while len(result) < size:
|
|
block = os.read(descriptor, size - len(result))
|
|
if not block:
|
|
raise PackError(f"invalid PNG structure: {label}")
|
|
result.extend(block)
|
|
return bytes(result)
|
|
|
|
|
|
def _write_all(descriptor: int, data: bytes | memoryview, label: str) -> None:
|
|
remaining = memoryview(data)
|
|
while remaining:
|
|
written = os.write(descriptor, remaining)
|
|
if written <= 0:
|
|
raise OSError(errno.EIO, f"short write while writing {label}")
|
|
remaining = remaining[written:]
|
|
|
|
|
|
def _inflate_block(inflater: Any, block: bytes, decoded: bytearray,
|
|
expected: int, label: str) -> None:
|
|
try:
|
|
output = inflater.decompress(block, expected + 1 - len(decoded))
|
|
except zlib.error as exc:
|
|
raise PackError(f"invalid PNG deflate stream: {label}") from exc
|
|
decoded.extend(output)
|
|
if len(decoded) > expected or inflater.unconsumed_tail or inflater.unused_data:
|
|
raise PackError(f"invalid PNG deflate length: {label}")
|
|
|
|
|
|
def _unfilter_png_rows(decoded: bytes, row_bytes: int, bytes_per_pixel: int,
|
|
label: str) -> bytes:
|
|
output = bytearray()
|
|
previous = bytearray(row_bytes)
|
|
stride = row_bytes + 1
|
|
for offset in range(0, len(decoded), stride):
|
|
filter_type = decoded[offset]
|
|
if filter_type > 4:
|
|
raise PackError(f"invalid PNG row filter: {label}")
|
|
encoded = decoded[offset + 1:offset + stride]
|
|
row = bytearray(row_bytes)
|
|
for index, value in enumerate(encoded):
|
|
left = row[index - bytes_per_pixel] if index >= bytes_per_pixel else 0
|
|
above = previous[index]
|
|
upper_left = previous[index - bytes_per_pixel] if index >= bytes_per_pixel else 0
|
|
if filter_type == 0:
|
|
predictor = 0
|
|
elif filter_type == 1:
|
|
predictor = left
|
|
elif filter_type == 2:
|
|
predictor = above
|
|
elif filter_type == 3:
|
|
predictor = (left + above) // 2
|
|
else:
|
|
estimate = left + above - upper_left
|
|
distances = (abs(estimate - left), abs(estimate - above), abs(estimate - upper_left))
|
|
predictor = (left, above, upper_left)[distances.index(min(distances))]
|
|
row[index] = (value + predictor) & 0xff
|
|
output.extend(row)
|
|
previous = row
|
|
return bytes(output)
|
|
|
|
|
|
def _validate_png_fd(descriptor: int, label: str, maximum_size: int) -> int:
|
|
before = os.fstat(descriptor)
|
|
if not stat.S_ISREG(before.st_mode):
|
|
raise PackError(f"PNG must be a regular file: {label}")
|
|
if before.st_size > maximum_size:
|
|
raise PackError(f"total bytes exceed quota of {maximum_size}: {label}")
|
|
os.lseek(descriptor, 0, os.SEEK_SET)
|
|
if _read_exact(descriptor, 8, label) != PNG_SIGNATURE:
|
|
raise PackError(f"invalid PNG signature: {label}")
|
|
remaining = before.st_size - 8
|
|
first = True
|
|
saw_plte = False
|
|
saw_idat = False
|
|
idat_closed = False
|
|
saw_iend = False
|
|
inflater = zlib.decompressobj()
|
|
decoded = bytearray()
|
|
expected = row_bytes = color_type = bit_depth = bytes_per_pixel = 0
|
|
palette_entries = 0
|
|
|
|
while remaining:
|
|
header = _read_exact(descriptor, 8, label)
|
|
remaining -= 8
|
|
length, kind = struct.unpack(">I4s", header)
|
|
if length > remaining - 4:
|
|
raise PackError(f"invalid PNG structure: {label}")
|
|
if first and (kind != b"IHDR" or length != 13):
|
|
raise PackError(f"invalid PNG IHDR: {label}")
|
|
if any(not (65 <= byte <= 90 or 97 <= byte <= 122) for byte in kind) or kind[2] & 0x20:
|
|
raise PackError(f"invalid PNG chunk type: {label}")
|
|
if kind == b"PLTE" and length > 768:
|
|
raise PackError(f"invalid PNG PLTE: {label}")
|
|
if not first and kind == b"IHDR":
|
|
raise PackError(f"invalid PNG critical chunk state: {label}")
|
|
if kind not in (b"IHDR", b"PLTE", b"IDAT", b"IEND") and kind[0] & 0x20 == 0:
|
|
raise PackError(f"unknown PNG critical chunk: {label}")
|
|
if saw_idat and kind != b"IDAT" and kind != b"IEND":
|
|
idat_closed = True
|
|
if kind == b"IDAT" and idat_closed:
|
|
raise PackError(f"nonconsecutive PNG IDAT chunks: {label}")
|
|
|
|
crc = zlib.crc32(kind)
|
|
payload = bytearray() if kind in (b"IHDR", b"PLTE") else None
|
|
unread = length
|
|
while unread:
|
|
block = os.read(descriptor, min(unread, 1024 * 1024))
|
|
if not block:
|
|
raise PackError(f"invalid PNG structure: {label}")
|
|
unread -= len(block)
|
|
crc = zlib.crc32(block, crc)
|
|
if payload is not None:
|
|
payload.extend(block)
|
|
if kind == b"IDAT":
|
|
_inflate_block(inflater, block, decoded, expected, label)
|
|
encoded_crc = _read_exact(descriptor, 4, label)
|
|
remaining -= length + 4
|
|
if crc != struct.unpack(">I", encoded_crc)[0]:
|
|
raise PackError(f"invalid PNG chunk CRC: {label}")
|
|
|
|
if kind == b"IHDR":
|
|
assert payload is not None
|
|
width, height, bit_depth, color_type, compression, filtering, interlace = struct.unpack(">IIBBBBB", payload)
|
|
valid_depths = {0: {1, 2, 4, 8, 16}, 2: {8, 16}, 3: {1, 2, 4, 8}, 4: {8, 16}, 6: {8, 16}}
|
|
if (width, height) != (256, 256):
|
|
raise PackError(f"PNG must be 256x256: {label}")
|
|
if (color_type not in valid_depths or bit_depth not in valid_depths[color_type]
|
|
or compression != 0 or filtering != 0 or interlace != 0):
|
|
raise PackError(f"invalid or unsupported non-interlaced PNG IHDR: {label}")
|
|
channels = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}[color_type]
|
|
row_bytes = (width * channels * bit_depth + 7) // 8
|
|
bytes_per_pixel = max(1, (channels * bit_depth + 7) // 8)
|
|
expected = (row_bytes + 1) * height
|
|
first = False
|
|
elif kind == b"PLTE":
|
|
if saw_plte or saw_idat or color_type in (0, 4) or payload is None:
|
|
raise PackError(f"invalid PNG PLTE state: {label}")
|
|
entries = len(payload) // 3
|
|
if len(payload) == 0 or len(payload) % 3 or entries > 256 or (color_type == 3 and entries > (1 << bit_depth)):
|
|
raise PackError(f"invalid PNG PLTE: {label}")
|
|
saw_plte = True
|
|
palette_entries = entries
|
|
elif kind == b"IDAT":
|
|
if color_type == 3 and not saw_plte:
|
|
raise PackError(f"indexed PNG requires PLTE before IDAT: {label}")
|
|
saw_idat = True
|
|
elif kind == b"IEND":
|
|
if length != 0 or not saw_idat or remaining != 0:
|
|
raise PackError(f"invalid PNG IEND: {label}")
|
|
saw_iend = True
|
|
|
|
if first or not saw_iend or not inflater.eof or inflater.unused_data or len(decoded) != expected:
|
|
raise PackError(f"invalid PNG structure, deflate, or terminal state: {label}")
|
|
pixels = _unfilter_png_rows(bytes(decoded), row_bytes, bytes_per_pixel, label)
|
|
if color_type == 3:
|
|
for row in range(256):
|
|
packed = pixels[row * row_bytes:(row + 1) * row_bytes]
|
|
for column in range(256):
|
|
bit_offset = column * bit_depth
|
|
byte = packed[bit_offset // 8]
|
|
shift = 8 - bit_depth - (bit_offset % 8)
|
|
if ((byte >> shift) & ((1 << bit_depth) - 1)) >= palette_entries:
|
|
raise PackError(f"indexed PNG palette index out of range: {label}")
|
|
after = os.fstat(descriptor)
|
|
if _identity(before) != _identity(after):
|
|
raise PackError(f"PNG changed while being validated: {label}")
|
|
return before.st_size
|
|
|
|
|
|
def _open_child_directory(parent_fd: int, name: str, label: str) -> tuple[int, tuple[int, ...]]:
|
|
try:
|
|
descriptor = os.open(name, _DIRECTORY_FLAGS, dir_fd=parent_fd)
|
|
except OSError as exc:
|
|
raise PackError(f"symlink or invalid source directory: {label}: {exc}") from exc
|
|
return descriptor, _identity(os.fstat(descriptor))
|
|
|
|
|
|
def _discover_tiles_fd(source_fd: int, source_label: Path, max_tiles: int, max_bytes: int) -> list[Tile]:
|
|
if max_tiles < 0 or max_bytes < 0:
|
|
raise PackError("quotas must not be negative")
|
|
if max_tiles > DEFAULT_MAX_TILES or max_bytes > DEFAULT_MAX_BYTES:
|
|
raise PackError("quota exceeds importer hard limit")
|
|
tiles: list[Tile] = []
|
|
keys: set[tuple[int, int, int]] = set()
|
|
total_bytes = 0
|
|
visited = 0
|
|
visit_limit = MAX_VISITED_ENTRIES_BASE + max_tiles * 3
|
|
|
|
def visit() -> None:
|
|
nonlocal visited
|
|
visited += 1
|
|
if visited > visit_limit:
|
|
raise PackError(f"visited-entry quota exceeded ({visit_limit})")
|
|
|
|
try:
|
|
zoom_names = (entry.name for entry in os.scandir(source_fd))
|
|
for zoom_name in zoom_names:
|
|
visit()
|
|
zoom = _canonical_number(zoom_name, "zoom")
|
|
if zoom > MAX_ZOOM:
|
|
raise PackError(f"zoom out of range: {zoom}")
|
|
zoom_fd, zoom_identity = _open_child_directory(source_fd, zoom_name, f"{source_label}/{zoom_name}")
|
|
try:
|
|
world_size = 1 << zoom
|
|
for x_entry in os.scandir(zoom_fd):
|
|
visit()
|
|
x = _canonical_number(x_entry.name, "x")
|
|
if x >= world_size:
|
|
raise PackError(f"x out of range at zoom {zoom}: {x}")
|
|
x_fd, x_identity = _open_child_directory(zoom_fd, x_entry.name, f"{source_label}/{zoom}/{x}")
|
|
try:
|
|
for y_entry in os.scandir(x_fd):
|
|
visit()
|
|
if not y_entry.name.endswith(".png"):
|
|
raise PackError(f"unexpected source entry: {source_label}/{zoom}/{x}/{y_entry.name}")
|
|
y = _canonical_number(y_entry.name[:-4], "y")
|
|
if y >= world_size:
|
|
raise PackError(f"y out of range at zoom {zoom}: {y}")
|
|
if len(tiles) >= max_tiles:
|
|
raise PackError(f"tile count exceeds quota of {max_tiles}")
|
|
key = (zoom, x, y)
|
|
if key in keys:
|
|
raise PackError(f"duplicate tile key: {zoom}/{x}/{y}")
|
|
try:
|
|
file_fd = os.open(y_entry.name, _FILE_FLAGS, dir_fd=x_fd)
|
|
except OSError as exc:
|
|
raise PackError(f"symlink or invalid PNG: {source_label}/{zoom}/{x}/{y}.png: {exc}") from exc
|
|
try:
|
|
info = os.fstat(file_fd)
|
|
if not stat.S_ISREG(info.st_mode):
|
|
raise PackError(f"PNG must be a regular file: {source_label}/{zoom}/{x}/{y}.png")
|
|
remaining_bytes = max_bytes - total_bytes
|
|
if info.st_size > remaining_bytes:
|
|
raise PackError(f"total bytes exceed quota of {max_bytes}")
|
|
size = _validate_png_fd(file_fd, f"{source_label}/{zoom}/{x}/{y}.png", remaining_bytes)
|
|
identity = _identity(os.fstat(file_fd))
|
|
finally:
|
|
os.close(file_fd)
|
|
keys.add(key)
|
|
total_bytes += size
|
|
tiles.append(Tile(zoom, x, y, source_label / str(zoom) / str(x) / f"{y}.png",
|
|
size, zoom_identity, x_identity, identity))
|
|
finally:
|
|
os.close(x_fd)
|
|
finally:
|
|
os.close(zoom_fd)
|
|
except OSError as exc:
|
|
raise PackError(f"cannot read source directory {source_label}: {exc}") from exc
|
|
if not tiles:
|
|
raise PackError("source contains no tiles")
|
|
return sorted(tiles, key=lambda tile: (tile.zoom, tile.x, tile.y))
|
|
|
|
|
|
def discover_tiles(source_directory: Path, max_tiles: int, max_bytes: int) -> list[Tile]:
|
|
descriptor = _open_path(Path(source_directory))
|
|
try:
|
|
return _discover_tiles_fd(descriptor, Path(source_directory), max_tiles, max_bytes)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def _runs(values: set[int]) -> tuple[tuple[int, int], ...]:
|
|
ordered = sorted(values)
|
|
runs: list[tuple[int, int]] = []
|
|
start = previous = ordered[0]
|
|
for value in ordered[1:]:
|
|
if value != previous + 1:
|
|
runs.append((start, previous))
|
|
start = value
|
|
previous = value
|
|
runs.append((start, previous))
|
|
return tuple(runs)
|
|
|
|
|
|
def calculate_extents(tiles: Iterable[Tile], antimeridian_zooms: set[int]) -> list[ZoomExtent]:
|
|
by_zoom: dict[int, set[tuple[int, int]]] = {}
|
|
for tile in tiles:
|
|
by_zoom.setdefault(tile.zoom, set()).add((tile.x, tile.y))
|
|
zooms = sorted(by_zoom)
|
|
if zooms != list(range(zooms[0], zooms[-1] + 1)):
|
|
raise PackError("zoom levels must be contiguous")
|
|
if not antimeridian_zooms.issubset(by_zoom):
|
|
raise PackError("antimeridian zoom option names a missing zoom")
|
|
extents: list[ZoomExtent] = []
|
|
for zoom in zooms:
|
|
keys = by_zoom[zoom]
|
|
xs = {key[0] for key in keys}
|
|
ys = {key[1] for key in keys}
|
|
x_runs = _runs(xs)
|
|
y_runs = _runs(ys)
|
|
if len(y_runs) != 1:
|
|
raise PackError(f"incomplete rectangle at zoom {zoom}: noncontiguous y range")
|
|
if zoom in antimeridian_zooms:
|
|
world_maximum = (1 << zoom) - 1
|
|
if len(x_runs) != 2 or x_runs[0][0] != 0 or x_runs[1][1] != world_maximum:
|
|
raise PackError(f"ambiguous antimeridian split at zoom {zoom}")
|
|
intervals = x_runs
|
|
else:
|
|
if len(x_runs) != 1:
|
|
raise PackError(f"incomplete rectangle at zoom {zoom}: noncontiguous x range")
|
|
intervals = x_runs
|
|
expected = sum(last - first + 1 for first, last in intervals) * (y_runs[0][1] - y_runs[0][0] + 1)
|
|
if len(keys) != expected:
|
|
raise PackError(f"incomplete rectangle at zoom {zoom}: missing tile")
|
|
extents.append(ZoomExtent(zoom, intervals, y_runs[0][0], y_runs[0][1]))
|
|
return extents
|
|
|
|
|
|
def calculate_row_spans(tiles: Iterable[Tile]) -> list[RowSpan]:
|
|
by_row: dict[tuple[int, int], set[int]] = {}
|
|
zooms: set[int] = set()
|
|
for tile in tiles:
|
|
by_row.setdefault((tile.zoom, tile.y), set()).add(tile.x)
|
|
zooms.add(tile.zoom)
|
|
ordered_zooms = sorted(zooms)
|
|
if not ordered_zooms or ordered_zooms != list(range(ordered_zooms[0], ordered_zooms[-1] + 1)):
|
|
raise PackError("zoom levels must be contiguous")
|
|
spans: list[RowSpan] = []
|
|
for (zoom, y), xs in sorted(by_row.items()):
|
|
for minimum, maximum in _runs(xs):
|
|
spans.append(RowSpan(zoom, y, minimum, maximum))
|
|
if len(spans) > MAX_ROW_SPANS:
|
|
raise PackError(f"sparse coverage exceeds row-span limit of {MAX_ROW_SPANS}")
|
|
return spans
|
|
|
|
|
|
def _validate_row_spans(row_spans: list[RowSpan], tile_count: int) -> None:
|
|
if not row_spans or len(row_spans) > MAX_ROW_SPANS:
|
|
raise PackError("manifest row-span count is invalid")
|
|
zooms: set[int] = set()
|
|
total = 0
|
|
previous: RowSpan | None = None
|
|
for span in row_spans:
|
|
if span.zoom < 0 or span.zoom > MAX_ZOOM:
|
|
raise PackError("invalid manifest row span")
|
|
world_maximum = (1 << span.zoom) - 1
|
|
if span.y < 0 or span.y > world_maximum or span.x_minimum < 0 \
|
|
or span.x_minimum > span.x_maximum or span.x_maximum > world_maximum:
|
|
raise PackError("invalid manifest row span")
|
|
if previous is not None and ((span.zoom, span.y) < (previous.zoom, previous.y) or
|
|
((span.zoom, span.y) == (previous.zoom, previous.y) and
|
|
span.x_minimum <= previous.x_maximum + 1)):
|
|
raise PackError("manifest row spans are not canonical")
|
|
total += span.x_maximum - span.x_minimum + 1
|
|
if total > 0xFFFFFFFF:
|
|
raise PackError("manifest tile count overflows uint32")
|
|
zooms.add(span.zoom)
|
|
previous = span
|
|
ordered_zooms = sorted(zooms)
|
|
if ordered_zooms != list(range(ordered_zooms[0], ordered_zooms[-1] + 1)):
|
|
raise PackError("manifest row spans must cover contiguous zooms")
|
|
if tile_count <= 0 or tile_count != total:
|
|
raise PackError("invalid manifest tile count")
|
|
|
|
|
|
def serialize_sparse_manifest(*, pack_id: str, name: str, attribution: str, source: str,
|
|
license: str, row_spans: list[RowSpan], tile_count: int) -> bytes:
|
|
_validate_metadata(pack_id, name, attribution, source, license)
|
|
_validate_row_spans(row_spans, tile_count)
|
|
payload = bytearray()
|
|
for field, value in (("pack_id", pack_id), ("name", name), ("attribution", attribution),
|
|
("source", source), ("license", license)):
|
|
encoded = _checked_text(field, value)
|
|
payload.append(len(encoded))
|
|
payload.extend(encoded)
|
|
payload.extend((row_spans[0].zoom, row_spans[-1].zoom))
|
|
payload.extend(struct.pack("<HI", len(row_spans), tile_count))
|
|
for span in row_spans:
|
|
payload.extend(struct.pack("<BIII", span.zoom, span.y, span.x_minimum, span.x_maximum))
|
|
length = HEADER_SIZE + len(payload) + 4
|
|
result = bytearray(struct.pack("<4sBBHII", MAGIC, SPARSE_FORMAT_VERSION, 0, HEADER_SIZE, length, 0))
|
|
result.extend(payload)
|
|
result.extend(struct.pack("<I", zlib.crc32(result)))
|
|
return bytes(result)
|
|
|
|
|
|
def _validate_manifest_values(extents: list[ZoomExtent], tile_count: int) -> None:
|
|
if not extents or len(extents) > MAX_ZOOM + 1:
|
|
raise PackError("manifest extent count is invalid")
|
|
if extents[0].zoom < 0 or extents[-1].zoom > MAX_ZOOM or [item.zoom for item in extents] != list(
|
|
range(extents[0].zoom, extents[-1].zoom + 1)):
|
|
raise PackError("manifest extents must cover contiguous valid zooms")
|
|
total = 0
|
|
for extent in extents:
|
|
world_maximum = (1 << extent.zoom) - 1
|
|
if len(extent.intervals) not in (1, 2) or extent.y_minimum < 0 or extent.y_minimum > extent.y_maximum \
|
|
or extent.y_maximum > world_maximum:
|
|
raise PackError("invalid manifest extent")
|
|
columns = 0
|
|
for minimum, maximum in extent.intervals:
|
|
if minimum < 0 or minimum > maximum or maximum > world_maximum:
|
|
raise PackError("invalid manifest extent")
|
|
columns += maximum - minimum + 1
|
|
if len(extent.intervals) == 2:
|
|
first, second = extent.intervals
|
|
if first[0] != 0 or second[1] != world_maximum or first[1] >= second[0] or first[1] + 1 == second[0]:
|
|
raise PackError("invalid manifest antimeridian intervals")
|
|
level = columns * (extent.y_maximum - extent.y_minimum + 1)
|
|
if level > 0xFFFFFFFF or total > 0xFFFFFFFF - level:
|
|
raise PackError("manifest tile count overflows uint32")
|
|
total += level
|
|
if tile_count <= 0 or tile_count > 0xFFFFFFFF or total != tile_count:
|
|
raise PackError("invalid manifest tile count")
|
|
|
|
|
|
def serialize_manifest(*, pack_id: str, name: str, attribution: str, source: str,
|
|
license: str, extents: list[ZoomExtent], tile_count: int) -> bytes:
|
|
_validate_metadata(pack_id, name, attribution, source, license)
|
|
_validate_manifest_values(extents, tile_count)
|
|
payload = bytearray()
|
|
for field, value in (("pack_id", pack_id), ("name", name), ("attribution", attribution),
|
|
("source", source), ("license", license)):
|
|
encoded = _checked_text(field, value)
|
|
payload.append(len(encoded))
|
|
payload.extend(encoded)
|
|
payload.extend((extents[0].zoom, extents[-1].zoom, len(extents)))
|
|
payload.extend(struct.pack("<I", tile_count))
|
|
for extent in extents:
|
|
intervals = extent.intervals + (((0, 0),) if len(extent.intervals) == 1 else ())
|
|
payload.extend((extent.zoom, len(extent.intervals)))
|
|
payload.extend(struct.pack("<II", extent.y_minimum, extent.y_maximum))
|
|
for minimum, maximum in intervals:
|
|
payload.extend(struct.pack("<II", minimum, maximum))
|
|
length = HEADER_SIZE + len(payload) + 4
|
|
result = bytearray(struct.pack("<4sBBHII", MAGIC, FORMAT_VERSION, 0, HEADER_SIZE, length, 0))
|
|
result.extend(payload)
|
|
result.extend(struct.pack("<I", zlib.crc32(result)))
|
|
return bytes(result)
|
|
|
|
|
|
def parse_manifest(data: bytes) -> dict[str, object]:
|
|
if len(data) < 20 or len(data) > MAX_MANIFEST_BYTES:
|
|
raise PackError("manifest is truncated or oversized")
|
|
magic, version, reserved, header_size, length, reserved_word = struct.unpack("<4sBBHII", data[:16])
|
|
if magic != MAGIC or version not in (FORMAT_VERSION, SPARSE_FORMAT_VERSION) or \
|
|
(reserved, header_size, length, reserved_word) != (0, 16, len(data), 0):
|
|
raise PackError("invalid manifest header")
|
|
if zlib.crc32(data[:-4]) != struct.unpack("<I", data[-4:])[0]:
|
|
raise PackError("invalid manifest CRC")
|
|
position = 16
|
|
values: dict[str, object] = {}
|
|
for field in ("pack_id", "name", "attribution", "source", "license"):
|
|
if position >= len(data) - 4:
|
|
raise PackError("manifest string is truncated")
|
|
size = data[position]
|
|
position += 1
|
|
if size > len(data) - 4 - position:
|
|
raise PackError("manifest string is truncated")
|
|
try:
|
|
value = data[position:position + size].decode("ascii")
|
|
except UnicodeDecodeError as exc:
|
|
raise PackError("invalid manifest string") from exc
|
|
position += size
|
|
_checked_text(field, value)
|
|
if field == "pack_id" and not PACK_ID_RE.fullmatch(value):
|
|
raise PackError("invalid manifest pack ID")
|
|
values[field] = value
|
|
if version == FORMAT_VERSION:
|
|
if position + 7 > len(data) - 4:
|
|
raise PackError("manifest fields are truncated")
|
|
minimum, maximum, count = data[position:position + 3]
|
|
position += 3
|
|
tile_count = struct.unpack("<I", data[position:position + 4])[0]
|
|
position += 4
|
|
if count > MAX_ZOOM + 1 or count > (len(data) - 4 - position) // EXTENT_SIZE:
|
|
raise PackError("invalid manifest extent count")
|
|
extents: list[ZoomExtent] = []
|
|
for _ in range(count):
|
|
zoom, interval_count = data[position:position + 2]
|
|
y_minimum, y_maximum, x0, x1, x2, x3 = struct.unpack("<IIIIII", data[position + 2:position + 26])
|
|
position += EXTENT_SIZE
|
|
if interval_count not in (1, 2) or (interval_count == 1 and (x2 != 0 or x3 != 0)):
|
|
raise PackError("invalid manifest extent")
|
|
intervals = ((x0, x1),) if interval_count == 1 else ((x0, x1), (x2, x3))
|
|
extents.append(ZoomExtent(zoom, intervals, y_minimum, y_maximum))
|
|
if position != len(data) - 4 or not extents or minimum != extents[0].zoom or maximum != extents[-1].zoom:
|
|
raise PackError("invalid manifest length or zoom range")
|
|
_validate_manifest_values(extents, tile_count)
|
|
values.update(format_version=version, min_zoom=minimum, max_zoom=maximum,
|
|
tile_count=tile_count, extents=extents, row_spans=[])
|
|
else:
|
|
if position + 8 > len(data) - 4:
|
|
raise PackError("manifest sparse fields are truncated")
|
|
minimum, maximum = data[position:position + 2]
|
|
position += 2
|
|
count, tile_count = struct.unpack("<HI", data[position:position + 6])
|
|
position += 6
|
|
if count == 0 or count > MAX_ROW_SPANS or count > (len(data) - 4 - position) // ROW_SPAN_SIZE:
|
|
raise PackError("invalid manifest row-span count")
|
|
row_spans: list[RowSpan] = []
|
|
for _ in range(count):
|
|
zoom, y, x_minimum, x_maximum = struct.unpack("<BIII", data[position:position + ROW_SPAN_SIZE])
|
|
position += ROW_SPAN_SIZE
|
|
row_spans.append(RowSpan(zoom, y, x_minimum, x_maximum))
|
|
if position != len(data) - 4 or minimum != row_spans[0].zoom or maximum != row_spans[-1].zoom:
|
|
raise PackError("invalid manifest length or zoom range")
|
|
_validate_row_spans(row_spans, tile_count)
|
|
values.update(format_version=version, min_zoom=minimum, max_zoom=maximum,
|
|
tile_count=tile_count, extents=[], row_spans=row_spans)
|
|
return values
|
|
|
|
|
|
def _open_verified_tile(source_fd: int, tile: Tile) -> int:
|
|
zoom_fd, zoom_identity = _open_child_directory(source_fd, str(tile.zoom), str(tile.path.parent.parent))
|
|
try:
|
|
if zoom_identity != tile.zoom_identity:
|
|
raise PackError(f"source directory changed after validation: {tile.path}")
|
|
x_fd, x_identity = _open_child_directory(zoom_fd, str(tile.x), str(tile.path.parent))
|
|
try:
|
|
if x_identity != tile.x_identity:
|
|
raise PackError(f"source directory changed after validation: {tile.path}")
|
|
try:
|
|
descriptor = os.open(f"{tile.y}.png", _FILE_FLAGS, dir_fd=x_fd)
|
|
except OSError as exc:
|
|
raise PackError(f"source PNG changed, became a symlink, or is invalid: {tile.path}: {exc}") from exc
|
|
if _identity(os.fstat(descriptor)) != tile.file_identity:
|
|
os.close(descriptor)
|
|
raise PackError(f"source PNG changed after validation: {tile.path}")
|
|
return descriptor
|
|
finally:
|
|
os.close(x_fd)
|
|
finally:
|
|
os.close(zoom_fd)
|
|
|
|
|
|
def _ensure_stage_path(stage_fd: int, tile: Tile) -> tuple[int, str]:
|
|
tiles_fd, _ = _mkdir_open(stage_fd, "tiles")
|
|
try:
|
|
zoom_fd, _ = _mkdir_open(tiles_fd, str(tile.zoom))
|
|
try:
|
|
x_fd, _ = _mkdir_open(zoom_fd, str(tile.x))
|
|
return x_fd, f"{tile.y}.png"
|
|
finally:
|
|
os.close(zoom_fd)
|
|
finally:
|
|
os.close(tiles_fd)
|
|
|
|
|
|
def copy_tile(source_fd: int, tile: Tile, stage_fd: int, maximum_bytes: int) -> None:
|
|
input_fd = _open_verified_tile(source_fd, tile)
|
|
output_parent_fd = -1
|
|
output_fd = -1
|
|
try:
|
|
output_parent_fd, output_name = _ensure_stage_path(stage_fd, tile)
|
|
output_fd = os.open(output_name, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC,
|
|
0o644, dir_fd=output_parent_fd)
|
|
copied = 0
|
|
while True:
|
|
block = os.read(input_fd, min(1024 * 1024, maximum_bytes + 1 - copied))
|
|
if not block:
|
|
break
|
|
copied += len(block)
|
|
if copied > maximum_bytes:
|
|
raise PackError(f"total bytes exceed quota of {maximum_bytes}")
|
|
_write_all(output_fd, block, f"staged {tile.zoom}/{tile.x}/{tile.y}.png")
|
|
if copied != tile.size or _identity(os.fstat(input_fd)) != tile.file_identity:
|
|
raise PackError(f"source PNG changed while copying: {tile.path}")
|
|
_validate_png_fd(output_fd, f"staged {tile.zoom}/{tile.x}/{tile.y}.png", maximum_bytes)
|
|
os.fsync(output_fd)
|
|
_fsync_directory_fd(output_parent_fd)
|
|
finally:
|
|
if output_fd >= 0:
|
|
os.close(output_fd)
|
|
if output_parent_fd >= 0:
|
|
os.close(output_parent_fd)
|
|
os.close(input_fd)
|
|
|
|
|
|
def _read_file_at(parent_fd: int, name: str, maximum: int) -> bytes:
|
|
descriptor = os.open(name, _FILE_FLAGS, dir_fd=parent_fd)
|
|
try:
|
|
info = os.fstat(descriptor)
|
|
if not stat.S_ISREG(info.st_mode) or info.st_size > maximum:
|
|
raise PackError(f"invalid or oversized file: {name}")
|
|
return _read_exact(descriptor, info.st_size, name)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def _validate_pack_fd(pack_fd: int, label: Path, max_bytes: int) -> dict[str, object]:
|
|
parsed = parse_manifest(_read_file_at(pack_fd, "manifest.pmp", MAX_MANIFEST_BYTES))
|
|
tiles_fd, _ = _open_child_directory(pack_fd, "tiles", f"{label}/tiles")
|
|
try:
|
|
tile_count = cast(int, parsed["tile_count"])
|
|
tiles = _discover_tiles_fd(tiles_fd, label / "tiles", tile_count, max_bytes)
|
|
finally:
|
|
os.close(tiles_fd)
|
|
if len(tiles) != tile_count:
|
|
raise PackError("emitted tile tree does not match manifest")
|
|
if parsed["format_version"] == FORMAT_VERSION:
|
|
expected_extents = cast(list[ZoomExtent], parsed["extents"])
|
|
extents = calculate_extents(tiles, {item.zoom for item in expected_extents if len(item.intervals) == 2})
|
|
if extents != expected_extents:
|
|
raise PackError("emitted tile tree does not match manifest")
|
|
else:
|
|
expected_spans = cast(list[RowSpan], parsed["row_spans"])
|
|
if calculate_row_spans(tiles) != expected_spans:
|
|
raise PackError("emitted tile tree does not match sparse manifest")
|
|
return parsed
|
|
|
|
|
|
def validate_pack(pack_directory: Path, max_bytes: int = DEFAULT_MAX_BYTES) -> dict[str, object]:
|
|
descriptor = _open_path(Path(pack_directory))
|
|
try:
|
|
return _validate_pack_fd(descriptor, Path(pack_directory), max_bytes)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def _remove_tree_at(parent_fd: int, name: str) -> None:
|
|
try:
|
|
descriptor = os.open(name, _DIRECTORY_FLAGS, dir_fd=parent_fd)
|
|
except FileNotFoundError:
|
|
return
|
|
try:
|
|
for entry in os.scandir(descriptor):
|
|
info = os.stat(entry.name, dir_fd=descriptor, follow_symlinks=False)
|
|
if stat.S_ISDIR(info.st_mode):
|
|
_remove_tree_at(descriptor, entry.name)
|
|
else:
|
|
os.unlink(entry.name, dir_fd=descriptor)
|
|
finally:
|
|
os.close(descriptor)
|
|
os.rmdir(name, dir_fd=parent_fd)
|
|
|
|
|
|
def _temporary_directory(packs_fd: int, pack_id: str) -> tuple[str, int]:
|
|
for attempt in range(256):
|
|
name = f".{pack_id}.tmp-{os.getpid():x}-{attempt:02x}"
|
|
try:
|
|
os.mkdir(name, 0o700, dir_fd=packs_fd)
|
|
descriptor = os.open(name, _DIRECTORY_FLAGS, dir_fd=packs_fd)
|
|
return name, descriptor
|
|
except FileExistsError:
|
|
continue
|
|
raise PackError("cannot allocate temporary output directory")
|
|
|
|
|
|
def _rename_noreplace(parent_fd: int, source: str, destination: str, target: Path) -> None:
|
|
if _renameat2 is None:
|
|
raise PackError("atomic no-replace unsupported on this host")
|
|
result = _renameat2(parent_fd, os.fsencode(source), parent_fd, os.fsencode(destination), 1)
|
|
if result == 0:
|
|
return
|
|
error = ctypes.get_errno()
|
|
if error == errno.EEXIST:
|
|
raise PackError(f"output already exists: {target}")
|
|
if error in (errno.ENOSYS, errno.EINVAL, errno.ENOTSUP, errno.EOPNOTSUPP):
|
|
raise PackError("atomic no-replace unsupported on this filesystem")
|
|
raise OSError(error, os.strerror(error), target)
|
|
|
|
|
|
def _verify_output_chain(output_root: Path, root_fd: int, pyxis_fd: int, pyxis_identity: tuple[int, ...],
|
|
packs_identity: tuple[int, ...]) -> None:
|
|
check_fd = _open_path(output_root)
|
|
try:
|
|
if _identity(os.fstat(check_fd)) != _identity(os.fstat(root_fd)):
|
|
raise PackError("output directory changed before publication")
|
|
finally:
|
|
os.close(check_fd)
|
|
try:
|
|
pyxis_info = os.stat("pyxis-map", dir_fd=root_fd, follow_symlinks=False)
|
|
packs_info = os.stat("packs", dir_fd=pyxis_fd, follow_symlinks=False)
|
|
except OSError as exc:
|
|
raise PackError(f"output directory changed before publication: {exc}") from exc
|
|
if _identity(pyxis_info)[:3] != pyxis_identity[:3] or _identity(packs_info)[:3] != packs_identity[:3]:
|
|
raise PackError("output directory changed before publication")
|
|
|
|
|
|
def build_map_pack(source_directory: Path, output_root: Path, *, pack_id: str, name: str,
|
|
attribution: str, source: str, license: str,
|
|
max_tiles: int = DEFAULT_MAX_TILES, max_bytes: int = DEFAULT_MAX_BYTES,
|
|
antimeridian_zooms: set[int] | None = None,
|
|
sparse: bool = False) -> Path:
|
|
if not sys.platform.startswith("linux"):
|
|
raise PackError("secure importer supports Linux hosts only")
|
|
_validate_metadata(pack_id, name, attribution, source, license)
|
|
antimeridian_zooms = set() if antimeridian_zooms is None else set(antimeridian_zooms)
|
|
if any(zoom < 0 or zoom > MAX_ZOOM for zoom in antimeridian_zooms):
|
|
raise PackError("antimeridian zoom is out of range")
|
|
source_directory = Path(source_directory)
|
|
output_root = Path(output_root)
|
|
target = output_root / "pyxis-map" / "packs" / pack_id
|
|
source_fd = _open_path(source_directory)
|
|
output_fd = pyxis_fd = packs_fd = stage_fd = -1
|
|
temporary = ""
|
|
committed = False
|
|
try:
|
|
tiles = _discover_tiles_fd(source_fd, source_directory, max_tiles, max_bytes)
|
|
try:
|
|
extents = calculate_extents(tiles, antimeridian_zooms)
|
|
except PackError as exc:
|
|
if not sparse or antimeridian_zooms or not str(exc).startswith("incomplete rectangle"):
|
|
raise
|
|
row_spans = calculate_row_spans(tiles)
|
|
manifest = serialize_sparse_manifest(pack_id=pack_id, name=name, attribution=attribution,
|
|
source=source, license=license,
|
|
row_spans=row_spans, tile_count=len(tiles))
|
|
else:
|
|
manifest = serialize_manifest(pack_id=pack_id, name=name, attribution=attribution, source=source,
|
|
license=license, extents=extents, tile_count=len(tiles))
|
|
output_fd = _open_path(output_root, create=True)
|
|
pyxis_fd, pyxis_identity = _mkdir_open(output_fd, "pyxis-map")
|
|
packs_fd, packs_identity = _mkdir_open(pyxis_fd, "packs")
|
|
temporary, stage_fd = _temporary_directory(packs_fd, pack_id)
|
|
copied = 0
|
|
for tile in tiles:
|
|
copy_tile(source_fd, tile, stage_fd, max_bytes - copied)
|
|
copied += tile.size
|
|
manifest_fd = os.open("manifest.pmp", os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC,
|
|
0o644, dir_fd=stage_fd)
|
|
try:
|
|
_write_all(manifest_fd, manifest, "manifest.pmp")
|
|
os.fsync(manifest_fd)
|
|
finally:
|
|
os.close(manifest_fd)
|
|
_fsync_directory_fd(stage_fd)
|
|
_validate_pack_fd(stage_fd, target, max_bytes)
|
|
_verify_output_chain(output_root, output_fd, pyxis_fd, pyxis_identity, packs_identity)
|
|
_rename_noreplace(packs_fd, temporary, pack_id, target)
|
|
committed = True
|
|
try:
|
|
_verify_output_chain(output_root, output_fd, pyxis_fd, pyxis_identity, packs_identity)
|
|
except BaseException as exc:
|
|
try:
|
|
_remove_tree_at(packs_fd, pack_id)
|
|
_fsync_directory_fd(packs_fd)
|
|
committed = False
|
|
except BaseException as cleanup_error:
|
|
raise PublishedDurabilityError(target, OSError(
|
|
f"publication location changed and rollback failed: {cleanup_error}")) from exc
|
|
raise PackError("output directory changed during publication; rolled back") from exc
|
|
try:
|
|
os.close(stage_fd)
|
|
except OSError as exc:
|
|
stage_fd = -1
|
|
raise PublishedDurabilityError(target, exc) from exc
|
|
stage_fd = -1
|
|
try:
|
|
_fsync_directory_fd(packs_fd)
|
|
except OSError as exc:
|
|
raise PublishedDurabilityError(target, exc) from exc
|
|
return target
|
|
finally:
|
|
active_exception = sys.exc_info()[0] is not None
|
|
close_error: OSError | None = None
|
|
if stage_fd >= 0:
|
|
try:
|
|
os.close(stage_fd)
|
|
except OSError as exc:
|
|
close_error = exc
|
|
if temporary and not committed and packs_fd >= 0:
|
|
_remove_tree_at(packs_fd, temporary)
|
|
for descriptor in (packs_fd, pyxis_fd, output_fd, source_fd):
|
|
if descriptor >= 0:
|
|
try:
|
|
os.close(descriptor)
|
|
except OSError as exc:
|
|
if close_error is None:
|
|
close_error = exc
|
|
if committed and close_error is not None and not active_exception:
|
|
raise PublishedDurabilityError(target, close_error)
|
|
|
|
|
|
def _parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("xyz_directory", type=Path)
|
|
parser.add_argument("output_root", type=Path, help="SD-card root under which pyxis-map/packs is created")
|
|
parser.add_argument("--pack-id", required=True)
|
|
parser.add_argument("--name", required=True)
|
|
parser.add_argument("--attribution", required=True)
|
|
parser.add_argument("--source", required=True)
|
|
parser.add_argument("--license", required=True)
|
|
parser.add_argument("--max-tiles", type=int, default=DEFAULT_MAX_TILES)
|
|
parser.add_argument("--max-bytes", type=int, default=DEFAULT_MAX_BYTES)
|
|
parser.add_argument("--antimeridian-zoom", action="append", type=int, default=[])
|
|
parser.add_argument("--sparse", action="store_true",
|
|
help="explicitly admit sparse state/polygon coverage and emit PMPK v2")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
arguments = _parser().parse_args(argv)
|
|
try:
|
|
target = build_map_pack(arguments.xyz_directory, arguments.output_root,
|
|
pack_id=arguments.pack_id, name=arguments.name,
|
|
attribution=arguments.attribution, source=arguments.source,
|
|
license=arguments.license, max_tiles=arguments.max_tiles,
|
|
max_bytes=arguments.max_bytes,
|
|
antimeridian_zooms=set(arguments.antimeridian_zoom),
|
|
sparse=arguments.sparse)
|
|
except PublishedDurabilityError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 3
|
|
except (PackError, OSError) as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 2
|
|
print(target)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|