119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
|
|
import mapbox_vector_tile
|
|
|
|
|
|
INPUT_ROOT = Path("/home/wwwroot/newpec/exported_auto/tile.mapple-on.jp__newpec-mvt-20260106__z___x___y_.pbf/tiles")
|
|
OUTPUT_ROOT = Path("/home/wwwroot/newpec/exported_auto/geojson_linux")
|
|
|
|
|
|
def tile_coord_to_lonlat(z: int, x: int, y: int, tx: float, ty: float, extent: int) -> list[float]:
|
|
lon = ((x + (tx / extent)) / (2**z)) * 360.0 - 180.0
|
|
lat = math.degrees(
|
|
math.atan(
|
|
math.sinh(
|
|
math.pi * (1.0 - (2.0 * (y + (ty / extent)) / (2**z)))
|
|
)
|
|
)
|
|
)
|
|
return [lon, lat]
|
|
|
|
|
|
def convert_geometry(geometry: dict, z: int, x: int, y: int, extent: int) -> dict:
|
|
geom_type = geometry["type"]
|
|
coords = geometry["coordinates"]
|
|
|
|
if geom_type == "Point":
|
|
return {"type": "Point", "coordinates": tile_coord_to_lonlat(z, x, y, coords[0], coords[1], extent)}
|
|
if geom_type == "MultiPoint":
|
|
return {
|
|
"type": "MultiPoint",
|
|
"coordinates": [tile_coord_to_lonlat(z, x, y, px, py, extent) for px, py in coords],
|
|
}
|
|
if geom_type == "LineString":
|
|
return {
|
|
"type": "LineString",
|
|
"coordinates": [tile_coord_to_lonlat(z, x, y, px, py, extent) for px, py in coords],
|
|
}
|
|
if geom_type == "MultiLineString":
|
|
return {
|
|
"type": "MultiLineString",
|
|
"coordinates": [
|
|
[tile_coord_to_lonlat(z, x, y, px, py, extent) for px, py in line]
|
|
for line in coords
|
|
],
|
|
}
|
|
if geom_type == "Polygon":
|
|
return {
|
|
"type": "Polygon",
|
|
"coordinates": [
|
|
[tile_coord_to_lonlat(z, x, y, px, py, extent) for px, py in ring]
|
|
for ring in coords
|
|
],
|
|
}
|
|
if geom_type == "MultiPolygon":
|
|
return {
|
|
"type": "MultiPolygon",
|
|
"coordinates": [
|
|
[
|
|
[tile_coord_to_lonlat(z, x, y, px, py, extent) for px, py in ring]
|
|
for ring in polygon
|
|
]
|
|
for polygon in coords
|
|
],
|
|
}
|
|
raise RuntimeError(f"unsupported geometry type: {geom_type}")
|
|
|
|
|
|
def convert_tile(path: Path) -> tuple[dict, int]:
|
|
z = int(path.parent.parent.name)
|
|
x = int(path.parent.name)
|
|
y = int(path.stem)
|
|
|
|
with path.open("rb") as fh:
|
|
decoded = mapbox_vector_tile.decode(fh.read())
|
|
|
|
features = []
|
|
for vt_layer, payload in decoded.items():
|
|
extent = int(payload.get("extent") or 4096)
|
|
for feature in payload.get("features", []):
|
|
properties = dict(feature.get("properties") or {})
|
|
properties["vt_layer"] = vt_layer
|
|
out_feature = {
|
|
"type": "Feature",
|
|
"geometry": convert_geometry(feature["geometry"], z, x, y, extent),
|
|
"properties": properties,
|
|
}
|
|
if feature.get("id") is not None:
|
|
out_feature["id"] = feature["id"]
|
|
features.append(out_feature)
|
|
|
|
return {"type": "FeatureCollection", "features": features}, len(features)
|
|
|
|
|
|
def main() -> None:
|
|
OUTPUT_ROOT.mkdir(parents=True, exist_ok=True)
|
|
tile_paths = sorted(INPUT_ROOT.glob("*/*/*.pbf"))
|
|
total_features = 0
|
|
|
|
for index, path in enumerate(tile_paths, start=1):
|
|
collection, count = convert_tile(path)
|
|
total_features += count
|
|
output_path = OUTPUT_ROOT / f"{path.parent.parent.name}_{path.parent.name}_{path.stem}.json"
|
|
output_path.write_text(
|
|
json.dumps(collection, ensure_ascii=False, separators=(",", ":")),
|
|
encoding="utf-8",
|
|
)
|
|
if index % 5000 == 0:
|
|
print(f"processed {index}/{len(tile_paths)} tiles, wrote {total_features} features")
|
|
|
|
print(f"Converted {len(tile_paths)} tiles into {OUTPUT_ROOT} with {total_features} features")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|