guard high-extent tile reencoding

navsea_tile_reencode_guard.py

@@ -0,0 +1,77 @@
+#!/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']}"
+        )