317 lines
11 KiB
Python
317 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
NavSea official port overlay export v1.0
|
|
|
|
从官方港湾 / 漁港原始包导出可视化叠加资产：
|
|
- C02-08.zip: 港湾
|
|
- C09-06.zip: 漁港
|
|
|
|
默认裁切到九州样区，便于人工核查。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import sqlite3 # kept for symmetry with other exporter scripts; not used directly
|
|
import zipfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from shapely.geometry import LineString, Point, mapping
|
|
|
|
|
|
RADIUS = 6378137.0
|
|
DEFAULT_BBOX = (123.0, 31.0, 132.7, 34.9) # Kyushu preview window
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SourceSpec:
|
|
path: Path
|
|
dataset: str
|
|
kind: str
|
|
|
|
|
|
def mercator_x(lon: float) -> float:
|
|
return RADIUS * math.radians(lon)
|
|
|
|
|
|
def mercator_y(lat: float) -> float:
|
|
lat = max(min(lat, 85.0511287798066), -85.0511287798066)
|
|
return RADIUS * math.log(math.tan(math.pi / 4.0 + math.radians(lat) / 2.0))
|
|
|
|
|
|
def mercator(lon: float, lat: float) -> tuple[float, float]:
|
|
return mercator_x(lon), mercator_y(lat)
|
|
|
|
|
|
def lonlat_from_text(text: str) -> tuple[float, float]:
|
|
parts = [p for p in text.replace(",", " ").split() if p]
|
|
if len(parts) < 2:
|
|
raise ValueError(f"invalid coordinate text: {text!r}")
|
|
lat = float(parts[0])
|
|
lon = float(parts[1])
|
|
return lon, lat
|
|
|
|
|
|
def lonlat_to_lonlat_text(text: str) -> tuple[float, float]:
|
|
# helper kept for readability when parsing GML/JPS coordinates
|
|
return lonlat_from_text(text)
|
|
|
|
|
|
def local_name(tag: str) -> str:
|
|
return tag.split("}", 1)[-1]
|
|
|
|
|
|
def load_xml(zip_path: Path) -> ET.Element:
|
|
with zipfile.ZipFile(zip_path) as zf:
|
|
xml_name = next(name for name in zf.namelist() if name.endswith(".xml") and "META" not in name)
|
|
return ET.fromstring(zf.read(xml_name))
|
|
|
|
|
|
def build_point_lookup(root: ET.Element) -> dict[str, tuple[float, float]]:
|
|
lookup: dict[str, tuple[float, float]] = {}
|
|
for point in root.iter():
|
|
if local_name(point.tag) != "GM_Point":
|
|
continue
|
|
point_id = point.attrib.get("id")
|
|
if not point_id:
|
|
continue
|
|
coord_text = None
|
|
for el in point.iter():
|
|
tag = local_name(el.tag)
|
|
if tag == "DirectPosition.coordinate" and (el.text or "").strip():
|
|
coord_text = el.text.strip()
|
|
break
|
|
if coord_text is None:
|
|
continue
|
|
lon, lat = lonlat_from_text(coord_text)
|
|
lookup[point_id] = (lon, lat)
|
|
return lookup
|
|
|
|
|
|
def extract_curve_coords(curve: ET.Element, point_lookup: dict[str, tuple[float, float]]) -> list[tuple[float, float]]:
|
|
coords: list[tuple[float, float]] = []
|
|
for pos in curve.iter():
|
|
tag = local_name(pos.tag)
|
|
if not tag.startswith("GM_Position"):
|
|
continue
|
|
direct = None
|
|
ref = None
|
|
for child in list(pos):
|
|
child_tag = local_name(child.tag)
|
|
if child_tag.endswith("DirectPosition.coordinate") and (child.text or "").strip():
|
|
direct = child.text.strip()
|
|
break
|
|
if child_tag.endswith("PointRef.point"):
|
|
ref = child.attrib.get("idref")
|
|
if direct is not None:
|
|
lon, lat = lonlat_from_text(direct)
|
|
coords.append((lon, lat))
|
|
elif ref and ref in point_lookup:
|
|
coords.append(point_lookup[ref])
|
|
return coords
|
|
|
|
|
|
def bbox_intersects(a: tuple[float, float, float, float], b: tuple[float, float, float, float]) -> bool:
|
|
return not (a[2] < b[0] or a[0] > b[2] or a[3] < b[1] or a[1] > b[3])
|
|
|
|
|
|
def point_in_bbox(pt: tuple[float, float], bbox: tuple[float, float, float, float]) -> bool:
|
|
lon, lat = pt
|
|
return bbox[0] <= lon <= bbox[2] and bbox[1] <= lat <= bbox[3]
|
|
|
|
|
|
def normalize_harbor_feature(tag: str, props: dict[str, str]) -> dict[str, str]:
|
|
if tag == "CK01":
|
|
props["feature_family"] = "harbor_point"
|
|
elif tag in ("CK11", "CK21"):
|
|
props["feature_family"] = "harbor_line"
|
|
elif tag == "CB01":
|
|
props["feature_family"] = "fish_port_point"
|
|
elif tag == "CB03":
|
|
props["feature_family"] = "fish_port_line"
|
|
else:
|
|
props["feature_family"] = "unknown"
|
|
return props
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Export official harbor/fishing-port preview assets for Kyushu")
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
default="src/pbf/port-overlay-official/kyushu",
|
|
help="输出目录",
|
|
)
|
|
parser.add_argument(
|
|
"--slug",
|
|
default="official_port_overlay_kyushu",
|
|
help="输出文件前缀",
|
|
)
|
|
parser.add_argument(
|
|
"--bbox",
|
|
nargs=4,
|
|
type=float,
|
|
default=list(DEFAULT_BBOX),
|
|
metavar=("MIN_LON", "MIN_LAT", "MAX_LON", "MAX_LAT"),
|
|
help="预览裁切范围，默认九州",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
project_root = Path(__file__).resolve().parent.parent
|
|
out_dir = Path(args.output_dir)
|
|
if not out_dir.is_absolute():
|
|
out_dir = project_root / out_dir
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
bbox = tuple(args.bbox) # type: ignore[assignment]
|
|
|
|
sources = [
|
|
SourceSpec(project_root / "coastline" / "C02-08.zip", "C02-08", "harbor"),
|
|
SourceSpec(project_root / "coastline" / "C09-06.zip", "C09-06", "fish_port"),
|
|
]
|
|
|
|
line_features: list[dict] = []
|
|
point_features: list[dict] = []
|
|
stats = []
|
|
|
|
for source in sources:
|
|
if not source.path.exists():
|
|
raise SystemExit(f"missing input: {source.path}")
|
|
root = load_xml(source.path)
|
|
point_lookup = build_point_lookup(root)
|
|
obj = root.find(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}OBJ")
|
|
if obj is None:
|
|
raise SystemExit(f"OBJ block missing in {source.path}")
|
|
|
|
line_count = 0
|
|
point_count = 0
|
|
bbox_minx = bbox_miny = float("inf")
|
|
bbox_maxx = bbox_maxy = float("-inf")
|
|
|
|
tags = ["CK11", "CK21"] if source.dataset == "C02-08" else ["CB03"]
|
|
point_tags = ["CK01"] if source.dataset == "C02-08" else ["CB01"]
|
|
|
|
for tag in tags:
|
|
for feature in obj.findall(f".//{{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}}{tag}"):
|
|
loc = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}LOC")
|
|
if loc is None:
|
|
continue
|
|
ref = loc.attrib.get("idref")
|
|
if not ref:
|
|
continue
|
|
curve = obj.find(f".//*[@id='{ref}']")
|
|
if curve is None:
|
|
continue
|
|
coords = extract_curve_coords(curve, point_lookup)
|
|
if len(coords) < 2:
|
|
continue
|
|
xs = [p[0] for p in coords]
|
|
ys = [p[1] for p in coords]
|
|
geom_bbox = (min(xs), min(ys), max(xs), max(ys))
|
|
if not bbox_intersects(geom_bbox, bbox):
|
|
continue
|
|
props = {"source_dataset": source.dataset, "source_kind": source.kind, "tag": tag}
|
|
for key in ("PRC", "HBC", "FPC"):
|
|
el = feature.find(f"{{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}}{key}")
|
|
if el is not None and (el.text or "").strip():
|
|
props[key.lower()] = el.text.strip()
|
|
props = normalize_harbor_feature(tag, props)
|
|
if len(coords) == 1:
|
|
lon, lat = coords[0]
|
|
geometry = mapping(Point(lon, lat))
|
|
else:
|
|
geometry = mapping(LineString(coords))
|
|
line_features.append({"type": "Feature", "properties": props, "geometry": geometry})
|
|
line_count += 1
|
|
bbox_minx = min(bbox_minx, geom_bbox[0])
|
|
bbox_miny = min(bbox_miny, geom_bbox[1])
|
|
bbox_maxx = max(bbox_maxx, geom_bbox[2])
|
|
bbox_maxy = max(bbox_maxy, geom_bbox[3])
|
|
|
|
for tag in point_tags:
|
|
for feature in obj.findall(f".//{{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}}{tag}"):
|
|
pos = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}POS")
|
|
if pos is None:
|
|
continue
|
|
ref = pos.attrib.get("idref")
|
|
if not ref or ref not in point_lookup:
|
|
continue
|
|
lon, lat = point_lookup[ref]
|
|
if not point_in_bbox((lon, lat), bbox):
|
|
continue
|
|
props = {"source_dataset": source.dataset, "source_kind": source.kind, "tag": tag}
|
|
for key in ("NA4", "AD7", "NA2", "FCF", "AAC", "CH1", "CH2", "CFP", "FPA", "PRC", "HBC", "FPC"):
|
|
el = feature.find(f"{{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}}{key}")
|
|
if el is not None and (el.text or "").strip():
|
|
props[key.lower()] = el.text.strip()
|
|
props = normalize_harbor_feature(tag, props)
|
|
point_features.append(
|
|
{
|
|
"type": "Feature",
|
|
"properties": props,
|
|
"geometry": mapping(Point(lon, lat)),
|
|
}
|
|
)
|
|
point_count += 1
|
|
bbox_minx = min(bbox_minx, lon)
|
|
bbox_miny = min(bbox_miny, lat)
|
|
bbox_maxx = max(bbox_maxx, lon)
|
|
bbox_maxy = max(bbox_maxy, lat)
|
|
|
|
stats.append(
|
|
{
|
|
"dataset": source.dataset,
|
|
"kind": source.kind,
|
|
"path": str(source.path),
|
|
"line_count": line_count,
|
|
"point_count": point_count,
|
|
"bbox_lonlat": None if line_count + point_count == 0 else [bbox_minx, bbox_miny, bbox_maxx, bbox_maxy],
|
|
}
|
|
)
|
|
|
|
line_geojson = {
|
|
"type": "FeatureCollection",
|
|
"name": "official_port_overlay_lines",
|
|
"features": line_features,
|
|
}
|
|
point_geojson = {
|
|
"type": "FeatureCollection",
|
|
"name": "official_port_overlay_points",
|
|
"features": point_features,
|
|
}
|
|
manifest = {
|
|
"version": "v1.0",
|
|
"source_packages": [str(s.path) for s in sources],
|
|
"bbox_lonlat": list(bbox),
|
|
"line_count": len(line_features),
|
|
"point_count": len(point_features),
|
|
"stats": stats,
|
|
"line_file": f"{args.slug}_lines.geojson",
|
|
"point_file": f"{args.slug}_points.geojson",
|
|
}
|
|
|
|
(out_dir / f"{args.slug}_lines.geojson").write_text(
|
|
json.dumps(line_geojson, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
(out_dir / f"{args.slug}_points.geojson").write_text(
|
|
json.dumps(point_geojson, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
(out_dir / f"{args.slug}_manifest.json").write_text(
|
|
json.dumps(manifest, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
print("已导出:")
|
|
print(" -", out_dir / f"{args.slug}_lines.geojson")
|
|
print(" -", out_dir / f"{args.slug}_points.geojson")
|
|
print(" -", out_dir / f"{args.slug}_manifest.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|