78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import mapbox_vector_tile
|
|
|
|
from navsea_geometry_native_audit import analyze_tile
|
|
|
|
|
|
def decode_tile_with_extents(raw: bytes) -> tuple[dict[str, Any], dict[str, int]]:
|
|
decoded = mapbox_vector_tile.decode(raw)
|
|
extents = {
|
|
layer_name: int(layer.get("extent") or 4096)
|
|
for layer_name, layer in decoded.items()
|
|
}
|
|
return decoded, extents
|
|
|
|
|
|
def encode_layers_preserving_extents(
|
|
encoded_layers: list[dict[str, Any]],
|
|
layer_extents: dict[str, int],
|
|
) -> bytes:
|
|
per_layer_options = {
|
|
layer["name"]: {"extents": int(layer_extents[layer["name"]])}
|
|
for layer in encoded_layers
|
|
}
|
|
return mapbox_vector_tile.encode(encoded_layers, per_layer_options=per_layer_options)
|
|
|
|
|
|
def validate_reencoded_tile(
|
|
source_tile_path: Path,
|
|
output_tile_path: Path,
|
|
*,
|
|
tolerance: int = 4096,
|
|
) -> dict[str, Any]:
|
|
source_analysis = analyze_tile(source_tile_path)
|
|
output_analysis = analyze_tile(output_tile_path)
|
|
|
|
source_extent = max(source_analysis["extent_values"]) if source_analysis["extent_values"] else 4096
|
|
output_extent = max(output_analysis["extent_values"]) if output_analysis["extent_values"] else 4096
|
|
source_overflow = int(source_analysis["max_overflow"])
|
|
output_overflow = int(output_analysis["max_overflow"])
|
|
suspicious_extent_shift = (
|
|
output_extent > 4096
|
|
and output_overflow >= source_overflow + (output_extent - 4096) - tolerance
|
|
)
|
|
suspicious = output_overflow > source_overflow + tolerance or suspicious_extent_shift
|
|
|
|
return {
|
|
"source_extent_values": source_analysis["extent_values"],
|
|
"output_extent_values": output_analysis["extent_values"],
|
|
"source_max_overflow": source_overflow,
|
|
"output_max_overflow": output_overflow,
|
|
"suspicious_extent_shift": suspicious_extent_shift,
|
|
"suspicious": suspicious,
|
|
}
|
|
|
|
|
|
def assert_reencoded_tile_safe(
|
|
source_tile_path: Path,
|
|
output_tile_path: Path,
|
|
*,
|
|
tolerance: int = 4096,
|
|
) -> None:
|
|
verdict = validate_reencoded_tile(source_tile_path, output_tile_path, tolerance=tolerance)
|
|
if verdict["suspicious"]:
|
|
raise RuntimeError(
|
|
"reencoded tile geometry drift detected: "
|
|
f"source={source_tile_path} output={output_tile_path} "
|
|
f"source_extent={verdict['source_extent_values']} "
|
|
f"output_extent={verdict['output_extent_values']} "
|
|
f"source_overflow={verdict['source_max_overflow']} "
|
|
f"output_overflow={verdict['output_max_overflow']} "
|
|
f"suspicious_extent_shift={verdict['suspicious_extent_shift']}"
|
|
)
|