1691 lines
67 KiB
Python
1691 lines
67 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import re
|
|
import struct
|
|
from collections import defaultdict
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import mapbox_vector_tile
|
|
import mercantile
|
|
import pymysql
|
|
from navsea_mapping_registry import NavSeaMappingRegistry
|
|
|
|
try:
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
except ImportError: # pragma: no cover - depends on host Python environment
|
|
Cipher = None
|
|
algorithms = None
|
|
modes = None
|
|
default_backend = None
|
|
|
|
|
|
SOURCE_TILE_ROOT = Path(
|
|
"/home/wwwroot/newpec/exported_auto/"
|
|
"tile.mapple-on.jp__newpec-mvt-20260106__z___x___y_.pbf/tiles"
|
|
)
|
|
DEFAULT_OUTPUT_ROOT = Path("/home/wwwroot/pbf")
|
|
|
|
# Use the full signed int32 domain for stable, dataset-wide reversible fid mapping.
|
|
# Real legacy fid values already exceed the narrower example range in fiddecode.md.
|
|
MIN_VAL = -(1 << 31)
|
|
MAX_VAL = (1 << 31) - 1
|
|
N = MAX_VAL - MIN_VAL + 1
|
|
ROUNDS = 10
|
|
DOMAIN_BITS = 32
|
|
DOMAIN_SIZE = 1 << DOMAIN_BITS
|
|
FLOAT_RE = re.compile(r"-?\d+(?:\.\d+)?")
|
|
LIGHT_COLOR_REMARK_MAP = {
|
|
"G": "green",
|
|
"R": "red",
|
|
"Y": "yellow",
|
|
"W": "white",
|
|
"B": "blue",
|
|
"V": "violet",
|
|
"O": "orange",
|
|
"A": "amber",
|
|
}
|
|
|
|
NAVIGATION_MARK_DISPLAY_ICON_MAP = {
|
|
"30300000": "symbol-daytime-303",
|
|
"30300002": "symbol-daytime-303",
|
|
"30300004": "symbol-daytime-303",
|
|
"30500001": "symbol-daytime-30500001",
|
|
"30500002": "symbol-daytime-30500002",
|
|
"30500003": "symbol-daytime-30500003",
|
|
"30500004": "symbol-daytime-30500003",
|
|
"30500010": "symbol-daytime-30500003",
|
|
"30600000": "symbol-daytime-303",
|
|
"30600001": "symbol-daytime-303",
|
|
"30600002": "symbol-daytime-303",
|
|
"30600003": "symbol-daytime-303",
|
|
"30600004": "symbol-daytime-303",
|
|
"30600006": "symbol-daytime-303",
|
|
"30700000": "symbol-daytime-30700003",
|
|
"30700001": "symbol-daytime-30700001",
|
|
"30700002": "symbol-daytime-30700002",
|
|
"30700003": "symbol-daytime-30700003",
|
|
"30700004": "symbol-daytime-30700003",
|
|
"30700007": "symbol-daytime-30700003",
|
|
"30800000": "symbol-daytime-308",
|
|
"30900000": "symbol-daytime-30500003",
|
|
"30900002": "symbol-daytime-30500002",
|
|
"30900004": "symbol-daytime-30500003",
|
|
"30900009": "symbol-daytime-30500003",
|
|
"30900011": "symbol-daytime-30500003",
|
|
}
|
|
|
|
FINAL_RELEASE_PROPERTY_ALLOWLIST = frozenset(
|
|
{
|
|
"canonical_object_type",
|
|
"class_code",
|
|
"chart_fill_pattern",
|
|
"chart_fill_style",
|
|
"chart_icon_image",
|
|
"chart_label_position_code",
|
|
"chart_label_subtext",
|
|
"chart_label_text",
|
|
"chart_line_color",
|
|
"chart_line_width",
|
|
"chart_symbol_code",
|
|
"chart_text_color",
|
|
"chart_text_style",
|
|
"clearance_height_m",
|
|
"depth_value_m",
|
|
"display_code",
|
|
"least_depth_m",
|
|
"light_color_code",
|
|
"light_sector_mode",
|
|
"name_ja",
|
|
"place_name_en",
|
|
"shape_class_code",
|
|
}
|
|
)
|
|
|
|
RELEASE_CANONICAL_OBJECT_TYPE_MAP = {
|
|
"港湾灯台": "harbor_lighthouse",
|
|
"防波堤灯台": "breakwater_lighthouse",
|
|
"沿岸灯台 (15M over)": "coastal_lighthouse_over_15m",
|
|
"灯 (Lt)": "minor_light",
|
|
"灯標": "light_beacon",
|
|
"浮標 (やぐら型)": "lattice_buoy",
|
|
"円柱型浮標": "pillar_buoy",
|
|
"円筒型浮標": "can_buoy",
|
|
"船体露出沈船": "wreck_hull_exposed",
|
|
"危険全沈没船": "dangerous_fully_submerged_wreck",
|
|
"測定済みの沈船": "surveyed_wreck",
|
|
"魚礁": "fish_reef",
|
|
"錨泊地": "anchorage",
|
|
"海上地名": "sea_place_name",
|
|
"陆上地名": "land_place_name",
|
|
"底質": "seabed_material",
|
|
"等深线": "depth_contour",
|
|
"未分类对象": "unclassified_object",
|
|
"P基本線ククリ": "baseline_outline",
|
|
"P穴": "seabed_hole",
|
|
"p地名": "sea_place_name",
|
|
"p地名陸": "land_place_name",
|
|
"潜提": "submerged_reef",
|
|
"税関": "customs_office",
|
|
"河川域": "river_area",
|
|
"湖沼域": "lake_area",
|
|
"陸上水域": "inland_water_area",
|
|
"干潮帯": "tidal_flat",
|
|
"未測海域": "unsurveyed_area",
|
|
"浅所危険界": "shoal_danger_area",
|
|
"防波堤": "breakwater",
|
|
"浮施設・桟橋": "floating_facility_pier",
|
|
"撤去跡": "removed_structure_remains",
|
|
"煙突": "chimney",
|
|
"塔、やぐら、風車": "tower_yagura_windmill",
|
|
"海事関係署": "maritime_office",
|
|
"漁業協同組合": "fishing_cooperative",
|
|
"山頂": "mountain_top",
|
|
"その他 (記念碑等)": "other_landmark_monument",
|
|
"9000m以深": "over_9000m",
|
|
"等深線": "depth_contour_line",
|
|
"L海底地形": "bathymetry_line",
|
|
"P陸域": "land_area",
|
|
"道路": "road",
|
|
"P危険界ククリ": "hazard_boundary_outline",
|
|
"区画漁業": "demarcated_fishery",
|
|
"橋 (水門)": "bridge_water_gate",
|
|
"一本線で表わす桟橋": "pier_single_line",
|
|
"暗岩": "sunken_rock",
|
|
"概略等深線": "overview_depth_contour",
|
|
"険悪物": "dangerous_object",
|
|
"ビル": "building",
|
|
"漁港": "fishing_port",
|
|
"海底線 (電信電話)": "subsea_cable_telecom",
|
|
"一本線で表わす海上バース": "offshore_berth_single_line",
|
|
"P投錨注意障害物ククリ": "anchor_caution_hazard_outline",
|
|
"p投錨注意障害物": "anchor_caution_hazard_point",
|
|
"タンク": "tank",
|
|
"立標": "beacon",
|
|
"洗岩": "awash_rock",
|
|
"海底線 (電力)": "subsea_power_cable",
|
|
"灯浮標 (やぐら型)": "light_buoy_lattice",
|
|
"橋梁灯": "bridge_light",
|
|
"円錐型浮標": "conical_buoy",
|
|
"ポンツーン、パイル、杭": "pontoon_pile_posts",
|
|
"一本線で表わす潜提": "submerged_reef_single_line",
|
|
"一本線で表わす撤去跡": "removed_structure_single_line",
|
|
"海底輸送管 (水)": "subsea_water_pipeline",
|
|
"架空線、送電線": "overhead_transmission_line",
|
|
"海草": "seagrass",
|
|
"シーバース灯": "sea_berth_light",
|
|
"平水境界": "shoreline_boundary",
|
|
"港則法による境界": "port_regulation_boundary",
|
|
"p高さ制限": "clearance_limit_point",
|
|
"ドルフィン": "mooring_dolphin",
|
|
"P施設・境界線等ククリ": "facility_boundary_outline",
|
|
"全沈没船 (危険なし)": "fully_submerged_wreck_non_dangerous",
|
|
"マリーナ": "marina",
|
|
"ケーソン仮置き場": "caisson_storage_yard",
|
|
"サンドウェーブ": "sand_wave",
|
|
"一般港湾 (港則法区域、重要港湾)": "general_port_regulated_major",
|
|
"海底設置物、放水口、取水口": "seabed_installation_outfall_intake",
|
|
"定置漁業": "fixed_fishery",
|
|
"孤立危険物": "isolated_danger",
|
|
"P錨泊地等ククリ": "anchorage_outline",
|
|
"p錨泊地等": "anchorage_point",
|
|
"障害物": "obstruction",
|
|
"導灯": "leading_light",
|
|
"指向灯/照射灯": "directional_light",
|
|
"検疫錨地": "quarantine_anchorage",
|
|
"P754ククリ": "special_outline_754",
|
|
"養殖場": "aquaculture_area",
|
|
"海の駅": "sea_station",
|
|
"塔、やぐら、測台": "tower_yagura_survey_platform",
|
|
"P航行危険障害物ククリ": "navigation_hazard_outline",
|
|
"土砂捨て場": "spoil_ground",
|
|
"錨泊禁止区域": "anchorage_prohibited_area",
|
|
"漁網": "fishing_net",
|
|
}
|
|
|
|
|
|
def normalize_release_canonical_object_type(source_layer: str, canonical_object_type: str) -> str:
|
|
object_type = text_or_none(canonical_object_type) or ""
|
|
if source_layer == "p高さ制限":
|
|
return "clearance_limit_point"
|
|
if source_layer == "L高さ制限":
|
|
return "clearance_limit_line"
|
|
return RELEASE_CANONICAL_OBJECT_TYPE_MAP.get(object_type, object_type)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DbConfig:
|
|
host: str = "localhost"
|
|
port: int = 3306
|
|
user: str = "root"
|
|
password: str = "2chi9ks2"
|
|
database: str = "pbf_analysis"
|
|
unix_socket: str | None = "/tmp/mysql.sock"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TileJob:
|
|
z: int
|
|
x: int
|
|
y: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TileProcessResult:
|
|
written: bool
|
|
unresolved: tuple[dict, ...] = ()
|
|
|
|
|
|
class NavSeaFidCodec:
|
|
def __init__(self, key: bytes) -> None:
|
|
if Cipher is None or algorithms is None or modes is None or default_backend is None:
|
|
raise RuntimeError(
|
|
"cryptography is required for engineering fid encryption. "
|
|
"Run with a Python environment that has cryptography installed."
|
|
)
|
|
if len(key) not in (16, 24, 32):
|
|
raise ValueError("AES key must be 16, 24, or 32 bytes")
|
|
self.key = key
|
|
self._cipher = Cipher(algorithms.AES(self.key), modes.ECB(), backend=default_backend())
|
|
|
|
def _prf(self, round_no: int, value: int) -> int:
|
|
encryptor = self._cipher.encryptor()
|
|
data = struct.pack(">IQ", round_no, value)
|
|
block = hashlib.sha256(data).digest()[:16]
|
|
out = encryptor.update(block) + encryptor.finalize()
|
|
return int.from_bytes(out[-2:], "big")
|
|
|
|
def _feistel_permute(self, x: int, encrypt: bool) -> int:
|
|
if not (0 <= x < DOMAIN_SIZE):
|
|
raise ValueError("x out of 32-bit domain")
|
|
|
|
left = (x >> 16) & 0xFFFF
|
|
right = x & 0xFFFF
|
|
rounds = range(ROUNDS) if encrypt else reversed(range(ROUNDS))
|
|
|
|
for rnd in rounds:
|
|
if encrypt:
|
|
fval = self._prf(rnd, right)
|
|
left, right = right, left ^ fval
|
|
else:
|
|
fval = self._prf(rnd, left)
|
|
left, right = right ^ fval, left
|
|
|
|
return ((left << 16) | right) & 0xFFFFFFFF
|
|
|
|
def encrypt_number(self, x: int) -> int:
|
|
if not (MIN_VAL <= x <= MAX_VAL):
|
|
raise ValueError(f"fid {x} out of supported range [{MIN_VAL}, {MAX_VAL}]")
|
|
|
|
result = x - MIN_VAL
|
|
while True:
|
|
result = self._feistel_permute(result, True)
|
|
if result < N:
|
|
return result
|
|
|
|
def decrypt_number(self, y: int) -> int:
|
|
if not (0 <= y < N):
|
|
raise ValueError(f"encrypted fid {y} out of range [0, {N - 1}]")
|
|
|
|
result = y
|
|
while True:
|
|
result = self._feistel_permute(result, False)
|
|
if result < N:
|
|
return result + MIN_VAL
|
|
|
|
@staticmethod
|
|
def to_hex(value: int) -> str:
|
|
# Fixed-width 32-bit uppercase hex string for stable external IDs.
|
|
return f"{value:08X}"
|
|
|
|
|
|
def text_or_none(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None
|
|
|
|
|
|
def contains_non_ascii(text: str) -> bool:
|
|
return any(ord(ch) > 127 for ch in text)
|
|
|
|
|
|
def parse_number(value: object) -> float | None:
|
|
text = text_or_none(value)
|
|
if not text:
|
|
return None
|
|
match = FLOAT_RE.search(text.replace(",", "."))
|
|
if not match:
|
|
return None
|
|
try:
|
|
return float(match.group(0))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def feature_id_from_current_fid(fid_value: object) -> int | None:
|
|
text = text_or_none(fid_value)
|
|
if not text:
|
|
return None
|
|
try:
|
|
if re.fullmatch(r"[0-9A-Fa-f]{8}", text):
|
|
return int(text, 16)
|
|
return int(text)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def infer_light_color_code_from_remark(light_remark: object) -> str | None:
|
|
remark_text = text_or_none(light_remark)
|
|
if not remark_text:
|
|
return None
|
|
|
|
for token, color in LIGHT_COLOR_REMARK_MAP.items():
|
|
if remark_text == token or f" {token} " in f" {remark_text} ":
|
|
return color
|
|
return None
|
|
|
|
|
|
def infer_light_character_code(light_remark: object) -> str | None:
|
|
remark_text = text_or_none(light_remark)
|
|
if not remark_text:
|
|
return None
|
|
|
|
patterns = (
|
|
"V-AIS",
|
|
"Q(6)+L Fl",
|
|
"Q(3)",
|
|
"LFl",
|
|
"Fl",
|
|
"Iso",
|
|
"Oc",
|
|
"Mo(A)",
|
|
"Mo(U)",
|
|
"F",
|
|
"Q",
|
|
)
|
|
for pattern in patterns:
|
|
if pattern in remark_text:
|
|
return pattern
|
|
if remark_text in LIGHT_COLOR_REMARK_MAP:
|
|
return "F"
|
|
return None
|
|
|
|
|
|
class NavSeaTileBuilder:
|
|
def __init__(
|
|
self,
|
|
db_config: DbConfig,
|
|
source_root: Path,
|
|
output_root: Path,
|
|
center_lat: float,
|
|
center_lon: float,
|
|
radius_nm: float,
|
|
zmin: int,
|
|
zmax: int,
|
|
workers: int,
|
|
all_tiles: bool,
|
|
reference_tile_root: Path | None,
|
|
engineering_mode: bool,
|
|
strip_legacy_japanese_delivery: bool,
|
|
release_minimal: bool,
|
|
fid_codec: NavSeaFidCodec | None,
|
|
fid_key_id: str | None,
|
|
bundle_id: str | None,
|
|
) -> None:
|
|
self.db_config = db_config
|
|
self.source_root = source_root
|
|
self.output_root = output_root
|
|
self.center_lat = center_lat
|
|
self.center_lon = center_lon
|
|
self.radius_nm = radius_nm
|
|
self.zmin = zmin
|
|
self.zmax = zmax
|
|
self.workers = workers
|
|
self.all_tiles = all_tiles
|
|
self.reference_tile_root = reference_tile_root
|
|
self.engineering_mode = engineering_mode
|
|
self.strip_legacy_japanese_delivery = strip_legacy_japanese_delivery
|
|
self.release_minimal = release_minimal
|
|
self.fid_codec = fid_codec
|
|
self.fid_key_id = fid_key_id
|
|
self.bundle_id = bundle_id
|
|
with self.connect() as conn:
|
|
self.mapping_registry = NavSeaMappingRegistry.load(conn, bundle_id=bundle_id)
|
|
self.bundle_id = self.mapping_registry.bundle_id
|
|
|
|
def connect(self):
|
|
kwargs = {
|
|
"host": self.db_config.host,
|
|
"port": self.db_config.port,
|
|
"user": self.db_config.user,
|
|
"password": self.db_config.password,
|
|
"database": self.db_config.database,
|
|
"charset": "utf8mb4",
|
|
"autocommit": True,
|
|
"cursorclass": pymysql.cursors.DictCursor,
|
|
}
|
|
if self.db_config.unix_socket and self.db_config.host in {"localhost", "127.0.0.1"}:
|
|
kwargs["unix_socket"] = self.db_config.unix_socket
|
|
return pymysql.connect(**kwargs)
|
|
|
|
def build(self) -> None:
|
|
self.output_root.mkdir(parents=True, exist_ok=True)
|
|
self.clear_existing_tiles()
|
|
supported_zooms = self.fetch_supported_zooms()
|
|
jobs = self.build_jobs(supported_zooms)
|
|
|
|
for z in range(self.zmin, min(self.zmax, max(supported_zooms, default=self.zmin)) + 1):
|
|
(self.output_root / str(z)).mkdir(parents=True, exist_ok=True)
|
|
|
|
print(
|
|
f"AOI center=({self.center_lat},{self.center_lon}) radius_nm={self.radius_nm} "
|
|
f"supported_zooms={supported_zooms} tile_jobs={len(jobs)}"
|
|
)
|
|
|
|
written = 0
|
|
skipped = 0
|
|
unresolved_events: list[dict] = []
|
|
with ThreadPoolExecutor(max_workers=self.workers) as executor:
|
|
futures = {executor.submit(self.process_tile, job): job for job in jobs}
|
|
for future in as_completed(futures):
|
|
job = futures[future]
|
|
result = future.result()
|
|
unresolved_events.extend(result.unresolved)
|
|
if result.written:
|
|
written += 1
|
|
print(f"wrote z={job.z} x={job.x} y={job.y}")
|
|
else:
|
|
skipped += 1
|
|
print(f"skipped z={job.z} x={job.x} y={job.y}")
|
|
|
|
self.write_audit_report(unresolved_events, jobs, written, skipped)
|
|
print(f"build complete: wrote={written}, skipped={skipped}, output={self.output_root}")
|
|
|
|
def clear_existing_tiles(self) -> None:
|
|
for path in self.output_root.glob("*/*/*.pbf"):
|
|
path.unlink()
|
|
|
|
def fetch_supported_zooms(self) -> list[int]:
|
|
with self.connect() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT DISTINCT z
|
|
FROM features
|
|
WHERE z BETWEEN %s AND %s
|
|
ORDER BY z
|
|
""",
|
|
(self.zmin, self.zmax),
|
|
)
|
|
return [int(row["z"]) for row in cur.fetchall()]
|
|
|
|
def build_jobs(self, supported_zooms: list[int]) -> list[TileJob]:
|
|
if self.reference_tile_root is not None:
|
|
return self.build_reference_jobs(supported_zooms)
|
|
if self.all_tiles:
|
|
return self.build_all_jobs(supported_zooms)
|
|
|
|
west, south, east, north = self.aoi_bbox()
|
|
jobs: list[TileJob] = []
|
|
|
|
for z in supported_zooms:
|
|
for tile in mercantile.tiles(west, south, east, north, [z]):
|
|
if (self.source_root / str(z) / str(tile.x) / f"{tile.y}.pbf").exists():
|
|
jobs.append(TileJob(z=tile.z, x=tile.x, y=tile.y))
|
|
|
|
return jobs
|
|
|
|
def build_all_jobs(self, supported_zooms: list[int]) -> list[TileJob]:
|
|
jobs: list[TileJob] = []
|
|
zoom_set = set(supported_zooms)
|
|
|
|
for path in sorted(self.source_root.glob("*/*/*.pbf")):
|
|
try:
|
|
z = int(path.parent.parent.name)
|
|
x = int(path.parent.name)
|
|
y = int(path.stem)
|
|
except ValueError:
|
|
continue
|
|
if z not in zoom_set or z < self.zmin or z > self.zmax:
|
|
continue
|
|
jobs.append(TileJob(z=z, x=x, y=y))
|
|
|
|
return jobs
|
|
|
|
def build_reference_jobs(self, supported_zooms: list[int]) -> list[TileJob]:
|
|
jobs: list[TileJob] = []
|
|
zoom_set = set(supported_zooms)
|
|
assert self.reference_tile_root is not None
|
|
|
|
for path in sorted(self.reference_tile_root.glob("*/*/*.pbf")):
|
|
try:
|
|
z = int(path.parent.parent.name)
|
|
x = int(path.parent.name)
|
|
y = int(path.stem)
|
|
except ValueError:
|
|
continue
|
|
if z not in zoom_set or z < self.zmin or z > self.zmax:
|
|
continue
|
|
if not (self.source_root / str(z) / str(x) / f"{y}.pbf").exists():
|
|
continue
|
|
jobs.append(TileJob(z=z, x=x, y=y))
|
|
|
|
return jobs
|
|
|
|
def aoi_bbox(self) -> tuple[float, float, float, float]:
|
|
radius_km = self.radius_nm * 1.852
|
|
lat_delta = radius_km / 111.32
|
|
lon_delta = radius_km / (111.32 * math.cos(math.radians(self.center_lat)))
|
|
return (
|
|
self.center_lon - lon_delta,
|
|
self.center_lat - lat_delta,
|
|
self.center_lon + lon_delta,
|
|
self.center_lat + lat_delta,
|
|
)
|
|
|
|
def process_tile(self, job: TileJob) -> TileProcessResult:
|
|
rows = self.fetch_tile_rows(job)
|
|
if not rows:
|
|
return TileProcessResult(written=False)
|
|
|
|
source_path = self.source_root / str(job.z) / str(job.x) / f"{job.y}.pbf"
|
|
if not source_path.exists():
|
|
return TileProcessResult(written=False)
|
|
|
|
decoded = mapbox_vector_tile.decode(source_path.read_bytes())
|
|
encoded_layers, per_layer_options, unresolved = self.build_output_layers(decoded, rows, job)
|
|
if not encoded_layers:
|
|
return TileProcessResult(written=False, unresolved=tuple(unresolved))
|
|
|
|
output_path = self.output_root / str(job.z) / str(job.x) / f"{job.y}.pbf"
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_bytes(
|
|
mapbox_vector_tile.encode(
|
|
encoded_layers,
|
|
per_layer_options=per_layer_options,
|
|
)
|
|
)
|
|
return TileProcessResult(written=True, unresolved=tuple(unresolved))
|
|
|
|
def fetch_tile_rows(self, job: TileJob) -> list[dict]:
|
|
with self.connect() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
r.feature_id,
|
|
COALESCE(fid.v, '') AS fid,
|
|
r.source_layer,
|
|
r.geom_type,
|
|
r.canonical_object_type,
|
|
r.canonical_family,
|
|
r.semantic_key,
|
|
r.detection_key,
|
|
r.render_layer
|
|
FROM pbf_relayer_candidates r
|
|
LEFT JOIN properties fid
|
|
ON fid.feature_id = r.feature_id
|
|
AND fid.k = 'fid'
|
|
WHERE r.z = %s
|
|
AND r.x = %s
|
|
AND r.y = %s
|
|
ORDER BY r.feature_id
|
|
""",
|
|
(job.z, job.x, job.y),
|
|
)
|
|
return list(cur.fetchall())
|
|
|
|
def build_output_layers(
|
|
self, decoded_tile: dict, rows: list[dict], job: TileJob
|
|
) -> tuple[list[dict], dict[str, dict[str, int]], list[dict]]:
|
|
source_groups: dict[tuple[str, str, str], list[dict]] = defaultdict(list)
|
|
source_extents: dict[str, int] = {}
|
|
for source_layer, payload in decoded_tile.items():
|
|
source_extents[source_layer] = int(payload.get("extent") or 4096)
|
|
for feature in payload.get("features", []):
|
|
properties = feature.get("properties") or {}
|
|
key = (
|
|
str(properties.get("fid", "")),
|
|
source_layer,
|
|
str(feature.get("geometry", {}).get("type", "")),
|
|
)
|
|
source_groups[key].append(feature)
|
|
|
|
db_groups: dict[tuple[str, str, str], list[dict]] = defaultdict(list)
|
|
for row in rows:
|
|
key = (str(row["fid"]), str(row["source_layer"]), str(row["geom_type"]))
|
|
db_groups[key].append(row)
|
|
|
|
missing_in_source = sorted(set(db_groups) - set(source_groups))
|
|
if missing_in_source:
|
|
preview = missing_in_source[:5]
|
|
raise RuntimeError(f"source feature mismatch for tile {job}: {preview}")
|
|
|
|
layer_features: dict[str, list[dict]] = defaultdict(list)
|
|
layer_extents: dict[str, int] = {}
|
|
unresolved: list[dict] = []
|
|
for key in sorted(db_groups):
|
|
source_features = source_groups[key]
|
|
db_rows = db_groups[key]
|
|
if len(source_features) != len(db_rows):
|
|
raise RuntimeError(
|
|
"feature multiplicity mismatch for tile "
|
|
f"{job}: key={key} source={len(source_features)} db={len(db_rows)}"
|
|
)
|
|
|
|
for source_feature, row in zip(source_features, db_rows):
|
|
source_layer_jp = str(row["source_layer"])
|
|
output_layer, source_layer_rule_id = self.mapping_registry.resolve_source_layer(source_layer_jp)
|
|
properties = dict(source_feature.get("properties") or {})
|
|
legacy_fid_raw = properties.get("fid", row["fid"])
|
|
current_feature_id = feature_id_from_current_fid(legacy_fid_raw)
|
|
navsea_fid_pair = None
|
|
if self.fid_codec is not None:
|
|
navsea_fid_pair = self.build_engineering_fid(legacy_fid_raw, row, job)
|
|
navsea_fid_int, navsea_fid_hex = navsea_fid_pair
|
|
properties["fid"] = navsea_fid_hex
|
|
current_feature_id = navsea_fid_int
|
|
if self.engineering_mode:
|
|
if navsea_fid_pair is None:
|
|
raise RuntimeError("engineering mode requires encrypted fid generation")
|
|
navsea_fid_int, _ = navsea_fid_pair
|
|
properties["fid_algo_id"] = "feistel32_aes_cyclewalk_v1"
|
|
properties["fid_key_id"] = self.fid_key_id
|
|
properties["fid_legacy_raw"] = legacy_fid_raw
|
|
properties["fid_navsea_int"] = navsea_fid_int
|
|
properties["source_layer_jp"] = source_layer_jp
|
|
properties["source_layer_std"] = output_layer
|
|
properties["normalization_bundle_id"] = self.bundle_id
|
|
properties["source_layer_rule_id"] = source_layer_rule_id
|
|
properties["feature_id"] = row["feature_id"]
|
|
|
|
raw_canonical_object_type = text_or_none(row["canonical_object_type"]) or ""
|
|
release_canonical_object_type = normalize_release_canonical_object_type(
|
|
source_layer_jp,
|
|
raw_canonical_object_type,
|
|
)
|
|
properties["canonical_object_type"] = raw_canonical_object_type
|
|
properties["canonical_family"] = row["canonical_family"]
|
|
properties["semantic_key"] = row["semantic_key"]
|
|
properties["detection_key"] = row["detection_key"]
|
|
properties["render_layer"] = output_layer
|
|
chart_properties, feature_unresolved, trace_status = self.build_chart_properties(
|
|
properties=properties,
|
|
source_layer=source_layer_jp,
|
|
output_layer=output_layer,
|
|
canonical_object_type=raw_canonical_object_type,
|
|
canonical_family=text_or_none(row["canonical_family"]) or "",
|
|
geom_type=str(source_feature.get("geometry", {}).get("type", "")),
|
|
feature_id=int(row["feature_id"]),
|
|
tile=job,
|
|
)
|
|
properties.update(chart_properties)
|
|
properties["canonical_object_type"] = release_canonical_object_type
|
|
if self.engineering_mode:
|
|
properties["trace_status"] = trace_status
|
|
properties, field_name_unresolved = self.standardize_output_properties(
|
|
properties=properties,
|
|
source_layer=source_layer_jp,
|
|
canonical_object_type=release_canonical_object_type,
|
|
canonical_family=text_or_none(row["canonical_family"]) or "",
|
|
geom_type=str(source_feature.get("geometry", {}).get("type", "")),
|
|
feature_id=int(row["feature_id"]),
|
|
tile=job,
|
|
)
|
|
if self.release_minimal and not self.engineering_mode:
|
|
properties = self.minimize_delivery_properties(properties)
|
|
unresolved.extend(feature_unresolved)
|
|
unresolved.extend(field_name_unresolved)
|
|
|
|
output_feature = {
|
|
"geometry": source_feature["geometry"],
|
|
"properties": properties,
|
|
}
|
|
if current_feature_id is not None:
|
|
output_feature["id"] = current_feature_id
|
|
elif source_feature.get("id") is not None:
|
|
output_feature["id"] = source_feature["id"]
|
|
|
|
layer_features[output_layer].append(output_feature)
|
|
layer_extents.setdefault(output_layer, source_extents[source_layer_jp])
|
|
|
|
encoded_layers = []
|
|
per_layer_options: dict[str, dict[str, int]] = {}
|
|
for layer_name, features in sorted(layer_features.items()):
|
|
if not features:
|
|
continue
|
|
encoded_layers.append({"name": layer_name, "features": features})
|
|
per_layer_options[layer_name] = {"extents": layer_extents[layer_name]}
|
|
|
|
return encoded_layers, per_layer_options, unresolved
|
|
|
|
def build_engineering_fid(self, legacy_fid_raw: object, row: dict, job: TileJob) -> tuple[int, str]:
|
|
if self.fid_codec is None:
|
|
raise RuntimeError("engineering mode requires a configured fid codec")
|
|
try:
|
|
legacy_int = int(str(legacy_fid_raw))
|
|
except (TypeError, ValueError) as exc:
|
|
raise RuntimeError(
|
|
f"invalid legacy fid for engineering tile {job}: "
|
|
f"feature_id={row['feature_id']} fid={legacy_fid_raw!r}"
|
|
) from exc
|
|
navsea_int = self.fid_codec.encrypt_number(legacy_int)
|
|
return navsea_int, self.fid_codec.to_hex(navsea_int)
|
|
|
|
def build_chart_properties(
|
|
self,
|
|
properties: dict,
|
|
source_layer: str,
|
|
output_layer: str,
|
|
canonical_object_type: str,
|
|
canonical_family: str,
|
|
geom_type: str,
|
|
feature_id: int,
|
|
tile: TileJob,
|
|
) -> tuple[dict[str, object], list[dict], str]:
|
|
unresolved: list[dict] = []
|
|
trace_status = "mapped"
|
|
if canonical_family == "unknown" or canonical_object_type == "未分类对象":
|
|
unresolved.append(
|
|
self.make_unresolved_event(
|
|
issue_type="taxonomy_unresolved",
|
|
reason="canonical taxonomy is unknown and requires explicit rule coverage",
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
canonical_family=canonical_family,
|
|
geom_type=geom_type,
|
|
feature_id=feature_id,
|
|
tile=tile,
|
|
fid=properties.get("fid"),
|
|
)
|
|
)
|
|
trace_status = "needs_review"
|
|
render_context = {
|
|
"source_layer": source_layer,
|
|
"output_layer": output_layer,
|
|
"canonical_object_type": canonical_object_type,
|
|
"canonical_family": canonical_family,
|
|
"geom_type": geom_type,
|
|
"class_name": text_or_none(properties.get("名称")) or canonical_object_type,
|
|
}
|
|
chart, render_rule_id = self.mapping_registry.resolve_render_rule(render_context)
|
|
if not render_rule_id or "FALLBACK" in render_rule_id:
|
|
chart = {}
|
|
unresolved.append(
|
|
self.make_unresolved_event(
|
|
issue_type="render_rule_unresolved",
|
|
reason="no specific render rule matched; heuristic fill-in was used",
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
canonical_family=canonical_family,
|
|
geom_type=geom_type,
|
|
feature_id=feature_id,
|
|
tile=tile,
|
|
fid=properties.get("fid"),
|
|
)
|
|
)
|
|
trace_status = "needs_review"
|
|
|
|
chart_render_type = text_or_none(chart.get("chart_render_type")) or self.infer_chart_render_type(
|
|
source_layer,
|
|
geom_type,
|
|
)
|
|
chart["chart_render_type"] = chart_render_type
|
|
chart.setdefault(
|
|
"chart_priority",
|
|
self.infer_chart_priority(
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
chart_render_type=chart_render_type,
|
|
),
|
|
)
|
|
if self.engineering_mode and render_rule_id:
|
|
chart["render_rule_id"] = render_rule_id
|
|
|
|
minzoom, maxzoom = self.infer_chart_visibility(source_layer, chart_render_type)
|
|
if minzoom is not None:
|
|
chart.setdefault("chart_visibility_min", minzoom)
|
|
if maxzoom is not None:
|
|
chart.setdefault("chart_visibility_max", maxzoom)
|
|
|
|
chart_collision_group = self.infer_collision_group(source_layer, chart_render_type)
|
|
if chart_collision_group:
|
|
chart.setdefault("chart_collision_group", chart_collision_group)
|
|
|
|
symbol_family, symbol_code = self.infer_symbol_semantics(
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
)
|
|
if symbol_family:
|
|
chart.setdefault("chart_symbol_family", symbol_family)
|
|
if symbol_code:
|
|
chart.setdefault("chart_symbol_code", symbol_code)
|
|
|
|
line_style = self.infer_line_style(source_layer, canonical_object_type)
|
|
if line_style:
|
|
chart.setdefault("chart_line_style", line_style)
|
|
|
|
fill_style = self.infer_fill_style(source_layer, canonical_object_type, properties)
|
|
if fill_style:
|
|
chart.setdefault("chart_fill_style", fill_style)
|
|
|
|
text_style = self.infer_text_style(source_layer, canonical_object_type, properties)
|
|
if text_style:
|
|
chart.setdefault("chart_text_style", text_style)
|
|
|
|
label_text, label_subtext = self.infer_label_text(
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
properties=properties,
|
|
)
|
|
if label_text:
|
|
chart.setdefault("chart_label_text", label_text)
|
|
if label_subtext:
|
|
chart.setdefault("chart_label_subtext", label_subtext)
|
|
|
|
label_anchor = self.infer_label_anchor(source_layer)
|
|
if label_anchor:
|
|
chart.setdefault("chart_label_anchor", label_anchor)
|
|
|
|
label_position_code = self.infer_label_position_code(
|
|
properties.get("表示位置"),
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
geom_type=geom_type,
|
|
)
|
|
if label_position_code:
|
|
chart["chart_label_position_code"] = label_position_code
|
|
|
|
light_color_code = self.mapping_registry.standardize_field_value(
|
|
"灯色",
|
|
properties.get("灯色"),
|
|
"light_color_code",
|
|
context={
|
|
"source_layer": source_layer,
|
|
"canonical_object_type": canonical_object_type,
|
|
"geom_type": geom_type,
|
|
},
|
|
)
|
|
if not light_color_code:
|
|
light_color_code = infer_light_color_code_from_remark(properties.get("灯略記"))
|
|
if light_color_code:
|
|
unresolved.append(
|
|
self.make_unresolved_event(
|
|
issue_type="field_value_rule_unresolved",
|
|
reason="light color was inferred from remark because no explicit field-value rule matched",
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
canonical_family=canonical_family,
|
|
geom_type=geom_type,
|
|
feature_id=feature_id,
|
|
tile=tile,
|
|
fid=properties.get("fid"),
|
|
field_name="灯色",
|
|
legacy_value=properties.get("灯色"),
|
|
target_field="light_color_code",
|
|
)
|
|
)
|
|
trace_status = "needs_review"
|
|
if light_color_code:
|
|
chart["light_color_code"] = light_color_code
|
|
|
|
light_character_code = infer_light_character_code(properties.get("灯略記"))
|
|
if light_character_code:
|
|
chart["light_character_code"] = light_character_code
|
|
|
|
light_sector_mode = self.infer_light_sector_mode(properties.get("明弧/分孤"), source_layer, canonical_object_type)
|
|
if light_sector_mode:
|
|
chart["light_sector_mode"] = light_sector_mode
|
|
|
|
hazard_class, hazard_severity = self.infer_hazard_semantics(source_layer, canonical_object_type)
|
|
if hazard_class:
|
|
chart.setdefault("hazard_class", hazard_class)
|
|
if hazard_severity:
|
|
chart.setdefault("hazard_severity", hazard_severity)
|
|
|
|
area_usage_class = self.infer_area_usage_class(source_layer, canonical_object_type)
|
|
if area_usage_class:
|
|
chart.setdefault("area_usage_class", area_usage_class)
|
|
|
|
chart_icon_image = self.infer_chart_icon_image(
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
chart_symbol_code=text_or_none(chart.get("chart_symbol_code")) or "",
|
|
light_color_code=text_or_none(chart.get("light_color_code")) or "",
|
|
hazard_class=text_or_none(chart.get("hazard_class")) or "",
|
|
display_code=text_or_none(properties.get("表示用番号")) or text_or_none(properties.get("display_code")) or "",
|
|
)
|
|
if chart_icon_image:
|
|
chart.setdefault("chart_icon_image", chart_icon_image)
|
|
|
|
chart_fill_pattern = self.infer_chart_fill_pattern(
|
|
source_layer=source_layer,
|
|
chart_fill_style=text_or_none(chart.get("chart_fill_style")) or "",
|
|
hazard_class=text_or_none(chart.get("hazard_class")) or "",
|
|
)
|
|
if chart_fill_pattern:
|
|
chart.setdefault("chart_fill_pattern", chart_fill_pattern)
|
|
|
|
chart_line_color = self.infer_chart_line_color(
|
|
source_layer=source_layer,
|
|
chart_line_style=text_or_none(chart.get("chart_line_style")) or "",
|
|
hazard_class=text_or_none(chart.get("hazard_class")) or "",
|
|
)
|
|
if chart_line_color:
|
|
chart.setdefault("chart_line_color", chart_line_color)
|
|
|
|
chart_line_width = self.infer_chart_line_width(
|
|
source_layer=source_layer,
|
|
chart_line_style=text_or_none(chart.get("chart_line_style")) or "",
|
|
)
|
|
if chart_line_width is not None:
|
|
chart.setdefault("chart_line_width", chart_line_width)
|
|
|
|
chart_text_color = self.infer_chart_text_color(
|
|
source_layer=source_layer,
|
|
chart_symbol_code=text_or_none(chart.get("chart_symbol_code")) or "",
|
|
chart_text_style=text_or_none(chart.get("chart_text_style")) or "",
|
|
)
|
|
if chart_text_color:
|
|
chart.setdefault("chart_text_color", chart_text_color)
|
|
|
|
depth_value_m = parse_number(properties.get("水深値(m)"))
|
|
if depth_value_m is not None:
|
|
chart["depth_value_m"] = depth_value_m
|
|
|
|
clearance_height_m = parse_number(properties.get("高さ(m)"))
|
|
if clearance_height_m is not None:
|
|
chart["clearance_height_m"] = clearance_height_m
|
|
|
|
least_depth_m = parse_number(properties.get("高さ/深度(m)"))
|
|
if least_depth_m is not None:
|
|
chart["least_depth_m"] = least_depth_m
|
|
|
|
bearing_deg = parse_number(properties.get("角度"))
|
|
if bearing_deg is not None:
|
|
chart["bearing_deg"] = bearing_deg
|
|
|
|
if not unresolved and render_rule_id and "FALLBACK" not in render_rule_id:
|
|
trace_status = "db_rule_matched"
|
|
|
|
return chart, unresolved, trace_status
|
|
|
|
def standardize_output_properties(
|
|
self,
|
|
*,
|
|
properties: dict[str, object],
|
|
source_layer: str,
|
|
canonical_object_type: str,
|
|
canonical_family: str,
|
|
geom_type: str,
|
|
feature_id: int,
|
|
tile: TileJob,
|
|
) -> tuple[dict[str, object], list[dict]]:
|
|
normalized: dict[str, object] = {}
|
|
unresolved: list[dict] = []
|
|
|
|
for key, value in properties.items():
|
|
if key == "fid":
|
|
if self.engineering_mode:
|
|
# Engineering output keeps the public NavSea fid in properties
|
|
# so it can be inspected alongside trace metadata.
|
|
normalized[key] = value
|
|
continue
|
|
rule = self.mapping_registry.get_field_name_rule(str(key))
|
|
if rule is None:
|
|
normalized[key] = value
|
|
if (
|
|
not self.engineering_mode
|
|
and self.strip_legacy_japanese_delivery
|
|
and contains_non_ascii(str(key))
|
|
):
|
|
unresolved.append(
|
|
self.make_unresolved_event(
|
|
issue_type="field_name_unmapped",
|
|
reason="delivery output still contains an unmapped non-ASCII field name",
|
|
source_layer=source_layer,
|
|
canonical_object_type=canonical_object_type,
|
|
canonical_family=canonical_family,
|
|
geom_type=geom_type,
|
|
feature_id=feature_id,
|
|
tile=tile,
|
|
fid=properties.get("fid"),
|
|
field_name=str(key),
|
|
)
|
|
)
|
|
continue
|
|
|
|
if self.engineering_mode:
|
|
if rule.keep_in_engineering:
|
|
normalized[key] = value
|
|
normalized[rule.field_name_std] = value
|
|
continue
|
|
|
|
if not self.strip_legacy_japanese_delivery:
|
|
normalized[key] = value
|
|
if self.release_minimal and rule.field_name_std in {"class_code", "display_code", "shape_class_code"}:
|
|
normalized[rule.field_name_std] = value
|
|
if rule.keep_in_delivery:
|
|
normalized[rule.field_name_std] = value
|
|
|
|
return normalized, unresolved
|
|
|
|
@staticmethod
|
|
def minimize_delivery_properties(properties: dict[str, object]) -> dict[str, object]:
|
|
return {
|
|
key: value
|
|
for key, value in properties.items()
|
|
if key in FINAL_RELEASE_PROPERTY_ALLOWLIST
|
|
}
|
|
|
|
@staticmethod
|
|
def make_unresolved_event(
|
|
*,
|
|
issue_type: str,
|
|
reason: str,
|
|
source_layer: str,
|
|
canonical_object_type: str,
|
|
canonical_family: str,
|
|
geom_type: str,
|
|
feature_id: int,
|
|
tile: TileJob,
|
|
fid: object,
|
|
field_name: str | None = None,
|
|
legacy_value: object | None = None,
|
|
target_field: str | None = None,
|
|
) -> dict:
|
|
event = {
|
|
"issue_type": issue_type,
|
|
"reason": reason,
|
|
"source_layer": source_layer,
|
|
"canonical_object_type": canonical_object_type,
|
|
"canonical_family": canonical_family,
|
|
"geom_type": geom_type,
|
|
"feature_id": feature_id,
|
|
"fid": None if fid is None else str(fid),
|
|
"z": tile.z,
|
|
"x": tile.x,
|
|
"y": tile.y,
|
|
}
|
|
if field_name:
|
|
event["field_name"] = field_name
|
|
if legacy_value is not None:
|
|
event["legacy_value"] = str(legacy_value)
|
|
if target_field:
|
|
event["target_field"] = target_field
|
|
return event
|
|
|
|
def write_audit_report(
|
|
self,
|
|
unresolved_events: list[dict],
|
|
jobs: list[TileJob],
|
|
written: int,
|
|
skipped: int,
|
|
) -> None:
|
|
report_json = self.output_root.parent / f"{self.output_root.name}.mapping_audit.json"
|
|
report_md = self.output_root.parent / f"{self.output_root.name}.mapping_audit.md"
|
|
|
|
buckets: dict[tuple[str, str, str, str], list[dict]] = defaultdict(list)
|
|
for event in unresolved_events:
|
|
key = (
|
|
event["issue_type"],
|
|
event["source_layer"],
|
|
event["canonical_object_type"],
|
|
event["geom_type"],
|
|
)
|
|
buckets[key].append(event)
|
|
|
|
summary = []
|
|
for key, events in sorted(buckets.items(), key=lambda item: (-len(item[1]), item[0])):
|
|
issue_type, source_layer, canonical_object_type, geom_type = key
|
|
summary.append(
|
|
{
|
|
"issue_type": issue_type,
|
|
"source_layer": source_layer,
|
|
"canonical_object_type": canonical_object_type,
|
|
"geom_type": geom_type,
|
|
"count": len(events),
|
|
"sample_features": events[:5],
|
|
}
|
|
)
|
|
|
|
payload = {
|
|
"bundle_id": self.bundle_id,
|
|
"output_root": str(self.output_root),
|
|
"tile_jobs": len(jobs),
|
|
"written_tiles": written,
|
|
"skipped_tiles": skipped,
|
|
"unresolved_event_count": len(unresolved_events),
|
|
"unresolved_summary": summary,
|
|
}
|
|
report_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
lines = [
|
|
"# NavSea Mapping Audit Report",
|
|
"",
|
|
f"- bundle_id: `{self.bundle_id}`",
|
|
f"- output_root: `{self.output_root}`",
|
|
f"- tile_jobs: `{len(jobs)}`",
|
|
f"- written_tiles: `{written}`",
|
|
f"- skipped_tiles: `{skipped}`",
|
|
f"- unresolved_event_count: `{len(unresolved_events)}`",
|
|
"",
|
|
"## Unresolved Summary",
|
|
"",
|
|
]
|
|
if not summary:
|
|
lines.append("- No unresolved mapping events were detected.")
|
|
else:
|
|
for item in summary:
|
|
lines.append(
|
|
f"- `{item['issue_type']}` | `{item['source_layer']}` | "
|
|
f"`{item['canonical_object_type']}` | `{item['geom_type']}` | count=`{item['count']}`"
|
|
)
|
|
for sample in item["sample_features"][:3]:
|
|
lines.append(
|
|
f" sample: fid=`{sample.get('fid')}` feature_id=`{sample['feature_id']}` "
|
|
f"tile=`{sample['z']}/{sample['x']}/{sample['y']}` reason=`{sample['reason']}`"
|
|
)
|
|
report_md.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
@staticmethod
|
|
def infer_chart_render_type(source_layer: str, geom_type: str) -> str:
|
|
if source_layer.startswith("p"):
|
|
if source_layer in {"p地名", "p地名陸", "p底質", "p高さ制限"}:
|
|
return "label"
|
|
return "symbol"
|
|
if source_layer.startswith("L") or "ククリ" in source_layer:
|
|
return "line"
|
|
if geom_type.endswith("Polygon"):
|
|
return "fill"
|
|
if geom_type.endswith("Point"):
|
|
return "symbol"
|
|
return "line"
|
|
|
|
@staticmethod
|
|
def infer_chart_priority(source_layer: str, canonical_object_type: str, chart_render_type: str) -> int:
|
|
object_type = canonical_object_type or ""
|
|
if source_layer == "p航路標識群":
|
|
return 900
|
|
if "危険" in object_type or "暗岩" in object_type or "沈船" in object_type:
|
|
return 890
|
|
if "魚礁" in object_type or "障害物" in object_type or "洗岩" in object_type:
|
|
return 870
|
|
if source_layer in {"p航行危険障害物", "p投錨注意障害物", "P航行危険障害物", "P投錨注意障害物"}:
|
|
return 860
|
|
if source_layer in {"P錨泊地等", "P航路", "P漁具定置箇所"}:
|
|
return 760
|
|
if source_layer in {"p地名", "p地名陸"}:
|
|
return 450
|
|
if source_layer == "p底質":
|
|
return 500
|
|
if source_layer in {"L海底地形", "L等深線", "L概略等深線"}:
|
|
return 640
|
|
if chart_render_type == "fill":
|
|
return 650
|
|
if chart_render_type == "line":
|
|
return 700
|
|
if chart_render_type == "label":
|
|
return 520
|
|
return 600
|
|
|
|
@staticmethod
|
|
def infer_chart_visibility(source_layer: str, chart_render_type: str) -> tuple[int | None, int | None]:
|
|
if source_layer == "L概略等深線":
|
|
return 5, 24
|
|
if source_layer in {"L等深線", "L海底地形"}:
|
|
return 9, 24
|
|
if source_layer == "p底質":
|
|
return 11, 24
|
|
if source_layer in {"p地名", "p地名陸"}:
|
|
return 10, 24
|
|
if source_layer == "p航路標識群":
|
|
return 10, 24
|
|
if source_layer == "p高さ制限":
|
|
return 11, 24
|
|
if chart_render_type in {"fill", "line"}:
|
|
return 5, 24
|
|
if chart_render_type == "symbol":
|
|
return 9, 24
|
|
if chart_render_type == "label":
|
|
return 10, 24
|
|
return None, None
|
|
|
|
@staticmethod
|
|
def infer_collision_group(source_layer: str, chart_render_type: str) -> str | None:
|
|
if source_layer in {"p地名", "p地名陸"}:
|
|
return "place_label"
|
|
if source_layer == "p航路標識群":
|
|
return "light_label"
|
|
if source_layer == "p底質":
|
|
return "seabed_label"
|
|
if source_layer == "p高さ制限":
|
|
return "clearance_label"
|
|
if source_layer == "L海底地形":
|
|
return "depth_label"
|
|
if chart_render_type == "symbol":
|
|
return "symbol"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_symbol_semantics(source_layer: str, canonical_object_type: str) -> tuple[str | None, str | None]:
|
|
object_type = canonical_object_type or ""
|
|
|
|
if source_layer == "p航路標識群":
|
|
if "AIS" in object_type:
|
|
return "navigation_light", "vais"
|
|
if "灯浮標" in object_type:
|
|
return "navigation_light", "light_buoy"
|
|
if "灯台" in object_type:
|
|
return "navigation_light", "lighthouse"
|
|
if "導灯" in object_type:
|
|
return "navigation_light", "leading_light"
|
|
if object_type == "灯 (Lt)":
|
|
return "navigation_light", "minor_light"
|
|
if "灯" in object_type or "灯標" in object_type or "灯柱" in object_type:
|
|
return "navigation_light", "light_beacon"
|
|
if "浮標" in object_type:
|
|
return "navigation_mark", "buoy"
|
|
if "立標" in object_type:
|
|
return "navigation_mark", "beacon"
|
|
return "navigation_mark", "nav_mark"
|
|
|
|
if source_layer in {"p投錨注意障害物", "p航行危険障害物"}:
|
|
if "魚礁" in object_type:
|
|
return "hazard", "fish_reef"
|
|
if "沈船" in object_type or "沈没船" in object_type:
|
|
return "hazard", "wreck"
|
|
if any(token in object_type for token in ("暗岩", "洗岩", "干出岩", "水上岩", "露出岩")):
|
|
return "hazard", "rock_awash"
|
|
if any(token in object_type for token in ("岩礁", "サンゴ礁")):
|
|
return "hazard", "reef"
|
|
if any(token in object_type for token in ("障害物", "危険物", "海底設置物", "沈木")):
|
|
return "hazard", "obstruction"
|
|
return "hazard", "hazard_mark"
|
|
|
|
if source_layer == "pパイロットステーション":
|
|
return "usage_area", "pilot_station"
|
|
|
|
if source_layer == "p施設・境界線等":
|
|
return "facility", "facility_mark"
|
|
|
|
if source_layer == "p陸上構造物":
|
|
return "landmark", "landmark"
|
|
|
|
if source_layer == "p錨泊地等":
|
|
return "usage_area", "anchorage_mark"
|
|
|
|
if source_layer == "p航路境界等":
|
|
return "boundary", "route_boundary_mark"
|
|
|
|
return None, None
|
|
|
|
@staticmethod
|
|
def infer_line_style(source_layer: str, canonical_object_type: str) -> str | None:
|
|
object_type = canonical_object_type or ""
|
|
if source_layer == "L概略等深線":
|
|
return "contour_overview"
|
|
if source_layer == "L等深線":
|
|
return "contour_minor"
|
|
if source_layer == "L海底地形":
|
|
return "bathymetry_support"
|
|
if source_layer == "L海底線":
|
|
return "subsea_cable"
|
|
if source_layer == "L基本線":
|
|
return "baseline"
|
|
if source_layer == "L高さ制限":
|
|
return "clearance_limit"
|
|
if source_layer in {"L739", "L741", "L危険界"}:
|
|
return "regulatory_boundary"
|
|
if source_layer in {"P危険界ククリ", "P航行危険障害物ククリ", "P投錨注意障害物ククリ"}:
|
|
return "hazard_boundary"
|
|
if source_layer in {"P航路ククリ", "L航路", "P誘導線ククリ"}:
|
|
return "route_boundary"
|
|
if source_layer == "P錨泊地等ククリ":
|
|
return "anchorage_boundary"
|
|
if source_layer == "P施設・境界線等ククリ":
|
|
return "facility_boundary"
|
|
if source_layer == "P754ククリ":
|
|
return "special_outline_754"
|
|
if source_layer.endswith("ククリ"):
|
|
return "outline"
|
|
if "基本線" in object_type:
|
|
return "baseline"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_fill_style(source_layer: str, canonical_object_type: str, properties: dict) -> str | None:
|
|
object_type = canonical_object_type or ""
|
|
if source_layer == "P陸域":
|
|
return "land_area"
|
|
if source_layer == "P穴":
|
|
return "seabed_hole"
|
|
if source_layer == "P潜堤":
|
|
return "submerged_reef_area"
|
|
if source_layer == "P漁具定置箇所":
|
|
return "fishery_area"
|
|
if source_layer == "P錨泊地等":
|
|
return "anchorage_area"
|
|
if source_layer == "P航路":
|
|
return "route_area"
|
|
if source_layer in {"P投錨注意障害物", "P航行危険障害物"}:
|
|
if "魚礁" in object_type:
|
|
return "fish_reef_area"
|
|
return "hazard_area"
|
|
if source_layer == "P施設・境界線等":
|
|
return "facility_area"
|
|
if source_layer == "P橋りょう等構造物":
|
|
return "bridge_area"
|
|
if source_layer == "P陸上構造物陸":
|
|
return "land_structure_area"
|
|
if source_layer == "P施設・境界線等透明":
|
|
return "facility_transparent_area"
|
|
if source_layer == "P基本線":
|
|
if object_type in {"河川域", "湖沼域", "陸上水域"}:
|
|
return "river_water"
|
|
if object_type in {"防波堤", "浮施設・桟橋", "撤去跡"}:
|
|
return "coast_structure_area"
|
|
if "未測" in object_type:
|
|
return "unsurveyed_area"
|
|
if "干出" in object_type or "干潮" in object_type:
|
|
return "tidal_flat"
|
|
if "危険" in object_type or "浅" in object_type:
|
|
return "shoal_danger_area"
|
|
if "-" in object_type and "m" in object_type:
|
|
return f"depth_zone_{object_type.lower().replace(' ', '').replace('m', 'm').replace('/', '_')}"
|
|
depth_value = parse_number(properties.get("水深値(m)"))
|
|
if depth_value is not None:
|
|
return f"depth_zone_{int(depth_value)}m"
|
|
return "water_area"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_text_style(source_layer: str, canonical_object_type: str, properties: dict) -> str | None:
|
|
if source_layer == "p地名":
|
|
return "place_name_sea"
|
|
if source_layer == "p地名陸":
|
|
return "place_name_land"
|
|
if source_layer == "p底質":
|
|
return "seabed_text"
|
|
if source_layer == "p高さ制限":
|
|
return "clearance_label"
|
|
if source_layer == "p航路標識群":
|
|
return "light_label"
|
|
if source_layer in {"L海底地形", "L等深線"} and (
|
|
parse_number(properties.get("水深値(m)")) is not None
|
|
or parse_number(properties.get("高さ/深度(m)")) is not None
|
|
):
|
|
return "depth_text"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_label_text(source_layer: str, canonical_object_type: str, properties: dict) -> tuple[str | None, str | None]:
|
|
shape_code = text_or_none(properties.get("形状分類番号"))
|
|
if source_layer == "p地名":
|
|
return text_or_none(properties.get("日本語地名")) or text_or_none(properties.get("英文字地名")), None
|
|
if source_layer == "p地名陸":
|
|
return text_or_none(properties.get("日本語地名")) or text_or_none(properties.get("英文字地名")), None
|
|
if source_layer == "p底質":
|
|
return text_or_none(properties.get("名称")), None
|
|
if source_layer == "p高さ制限":
|
|
height_text = text_or_none(properties.get("高さ(m)")) or text_or_none(properties.get("高さ/深度(m)"))
|
|
return height_text, None
|
|
if source_layer == "p航路標識群":
|
|
name_text = text_or_none(properties.get("名称")) or text_or_none(properties.get("名称補助"))
|
|
if shape_code in {"308", "335"}:
|
|
name_text = None
|
|
subtext = "V-AIS" if shape_code == "335" else text_or_none(properties.get("灯略記"))
|
|
return name_text, subtext
|
|
if source_layer == "L海底地形":
|
|
depth_value = parse_number(properties.get("水深値(m)"))
|
|
if depth_value is not None:
|
|
return str(int(depth_value) if depth_value.is_integer() else depth_value), None
|
|
if source_layer == "L等深線":
|
|
depth_value = parse_number(properties.get("高さ/深度(m)"))
|
|
if depth_value is not None:
|
|
return str(int(depth_value) if depth_value.is_integer() else depth_value), None
|
|
return text_or_none(properties.get("名称")), text_or_none(properties.get("名称補助"))
|
|
|
|
@staticmethod
|
|
def infer_label_anchor(source_layer: str) -> str | None:
|
|
if source_layer in {"p地名", "p地名陸", "p底質", "p高さ制限"}:
|
|
return "center"
|
|
if source_layer == "p航路標識群":
|
|
return "top"
|
|
return None
|
|
|
|
def infer_label_position_code(
|
|
self,
|
|
position_value: object,
|
|
*,
|
|
source_layer: str,
|
|
canonical_object_type: str,
|
|
geom_type: str,
|
|
) -> str | None:
|
|
mapped = self.mapping_registry.standardize_field_value(
|
|
"表示位置",
|
|
position_value,
|
|
"chart_label_position_code",
|
|
context={
|
|
"source_layer": source_layer,
|
|
"canonical_object_type": canonical_object_type,
|
|
"geom_type": geom_type,
|
|
},
|
|
)
|
|
if mapped:
|
|
return mapped
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_light_sector_mode(sector_value: object, source_layer: str, canonical_object_type: str) -> str | None:
|
|
if source_layer != "p航路標識群":
|
|
return None
|
|
sector_text = text_or_none(sector_value)
|
|
if sector_text and sector_text not in {"0", "0.0"}:
|
|
return "sector"
|
|
if "灯" in canonical_object_type or "AIS" in canonical_object_type:
|
|
return "omni"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_hazard_semantics(source_layer: str, canonical_object_type: str) -> tuple[str | None, str | None]:
|
|
object_type = canonical_object_type or ""
|
|
if source_layer not in {
|
|
"p投錨注意障害物",
|
|
"p航行危険障害物",
|
|
"P投錨注意障害物",
|
|
"P航行危険障害物",
|
|
"P潜堤",
|
|
"P漁具定置箇所",
|
|
}:
|
|
return None, None
|
|
|
|
if "魚礁" in object_type:
|
|
return "reef", "major"
|
|
if any(token in object_type for token in ("暗岩", "干出岩", "洗岩", "露出岩", "水上岩")):
|
|
return "rock", "critical"
|
|
if any(token in object_type for token in ("岩礁", "サンゴ礁")):
|
|
return "reef", "major"
|
|
if "沈船" in object_type or "沈没船" in object_type:
|
|
if "危険" in object_type or "露出" in object_type:
|
|
return "wreck", "critical"
|
|
return "wreck", "major"
|
|
if any(token in object_type for token in ("障害物", "危険物", "海底設置物", "沈木")):
|
|
return "obstruction", "major"
|
|
if "潜堤" in object_type:
|
|
return "shoal", "major"
|
|
return "obstruction", "context"
|
|
|
|
@staticmethod
|
|
def infer_area_usage_class(source_layer: str, canonical_object_type: str) -> str | None:
|
|
object_type = canonical_object_type or ""
|
|
if source_layer in {
|
|
"P投錨注意障害物",
|
|
"P航行危険障害物",
|
|
"p投錨注意障害物",
|
|
"p航行危険障害物",
|
|
}:
|
|
return "restricted"
|
|
if source_layer == "P錨泊地等" or "錨" in object_type:
|
|
return "anchorage"
|
|
if source_layer in {"P航路", "P航路ククリ", "L航路", "p航路境界等"} or "航路" in object_type:
|
|
return "route"
|
|
if source_layer == "P漁具定置箇所" or "漁" in object_type or "魚礁" in object_type:
|
|
return "fishery"
|
|
if source_layer in {"P投錨注意障害物", "P航行危険障害物"}:
|
|
return "restricted"
|
|
if source_layer == "P基本線":
|
|
return "water"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_chart_icon_image(
|
|
source_layer: str,
|
|
canonical_object_type: str,
|
|
chart_symbol_code: str,
|
|
light_color_code: str,
|
|
hazard_class: str,
|
|
display_code: str,
|
|
) -> str | None:
|
|
if source_layer == "p航路標識群":
|
|
if display_code:
|
|
mapped = NAVIGATION_MARK_DISPLAY_ICON_MAP.get(display_code)
|
|
if mapped:
|
|
return mapped
|
|
if chart_symbol_code == "leading_light":
|
|
return "symbol-daytime-300"
|
|
if chart_symbol_code == "lighthouse":
|
|
return "symbol-daytime-301"
|
|
if chart_symbol_code in {"beacon", "light_beacon"}:
|
|
return "symbol-daytime-310"
|
|
if chart_symbol_code == "light_buoy":
|
|
return "symbol-daytime-320"
|
|
if "浮標 (やぐら型)" in canonical_object_type:
|
|
return "symbol-daytime-327"
|
|
if "円柱型浮標" in canonical_object_type:
|
|
return "symbol-daytime-323"
|
|
if "円筒型浮標" in canonical_object_type:
|
|
return "symbol-daytime-325"
|
|
if chart_symbol_code in {"buoy", "nav_mark"}:
|
|
return "symbol-daytime-321"
|
|
if chart_symbol_code == "vais":
|
|
return "symbol-daytime-335359"
|
|
if light_color_code:
|
|
return "light-daytime-3"
|
|
if source_layer in {"p航行危険障害物", "p投錨注意障害物"}:
|
|
if "船体露出沈船" in canonical_object_type:
|
|
return "symbol-daytime-410"
|
|
if "危険全沈没船" in canonical_object_type:
|
|
return "symbol-daytime-412"
|
|
if "測定済みの沈船" in canonical_object_type:
|
|
return "symbol-daytime-413"
|
|
if chart_symbol_code == "fish_reef":
|
|
return "symbol-daytime-428"
|
|
if hazard_class == "rock":
|
|
return "symbol-daytime-405"
|
|
if hazard_class == "reef":
|
|
return "symbol-daytime-421"
|
|
if hazard_class == "wreck":
|
|
return "symbol-daytime-429"
|
|
if hazard_class == "obstruction":
|
|
return "symbol-daytime-428"
|
|
if source_layer == "p錨泊地等":
|
|
return "symbol-daytime-719"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_chart_fill_pattern(source_layer: str, chart_fill_style: str, hazard_class: str) -> str | None:
|
|
if source_layer in {"P投錨注意障害物", "P航行危険障害物"}:
|
|
if chart_fill_style == "fish_reef_area":
|
|
return "fill-daytime-428"
|
|
if chart_fill_style == "reef_area" or hazard_class == "reef":
|
|
return "fill-daytime-421"
|
|
if chart_fill_style == "submerged_reef_area":
|
|
return "fill-daytime-421"
|
|
if chart_fill_style == "hazard_area":
|
|
return "fill-daytime-405"
|
|
if source_layer == "P施設・境界線等" and chart_fill_style == "facility_area":
|
|
return "fill-daytime-754"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_chart_line_color(source_layer: str, chart_line_style: str, hazard_class: str) -> str | None:
|
|
if source_layer in {"P航行危険障害物ククリ", "P投錨注意障害物ククリ"}:
|
|
if hazard_class == "reef":
|
|
return "rgba(212,177,221,1)"
|
|
if hazard_class in {"rock", "wreck"}:
|
|
return "rgba(198,77,187,1)"
|
|
if hazard_class == "obstruction":
|
|
return "rgba(136,152,139,1)"
|
|
if chart_line_style == "anchorage_boundary":
|
|
return "rgba(198,77,187,1)"
|
|
if chart_line_style == "facility_boundary":
|
|
return "rgba(136,152,139,1)"
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_chart_line_width(source_layer: str, chart_line_style: str) -> int | None:
|
|
if source_layer in {"P航行危険障害物ククリ", "P投錨注意障害物ククリ"}:
|
|
return 1
|
|
if chart_line_style == "anchorage_boundary":
|
|
return 2
|
|
if chart_line_style == "facility_boundary":
|
|
return 1
|
|
return None
|
|
|
|
@staticmethod
|
|
def infer_chart_text_color(source_layer: str, chart_symbol_code: str, chart_text_style: str) -> str | None:
|
|
if source_layer == "p航路標識群" and chart_symbol_code == "vais":
|
|
return "#c64dbb"
|
|
if chart_text_style in {"light_label", "place_name_sea", "place_name_land", "seabed_text", "depth_text"}:
|
|
return "#000000"
|
|
return None
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Build NavSea AOI vector tiles from semantic overlays.")
|
|
parser.add_argument("--center-lat", type=float)
|
|
parser.add_argument("--center-lon", type=float)
|
|
parser.add_argument("--radius-nm", type=float)
|
|
parser.add_argument("--zmin", type=int, default=0)
|
|
parser.add_argument("--zmax", type=int, default=14)
|
|
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT_ROOT)
|
|
parser.add_argument("--source-root", type=Path, default=SOURCE_TILE_ROOT)
|
|
parser.add_argument("--workers", type=int, default=4)
|
|
parser.add_argument("--all-tiles", action="store_true")
|
|
parser.add_argument("--reference-tile-root", type=Path)
|
|
parser.add_argument("--engineering", action="store_true")
|
|
parser.add_argument(
|
|
"--strip-legacy-japanese-delivery",
|
|
action="store_true",
|
|
dest="strip_legacy_japanese_delivery",
|
|
help="drop mapped Japanese structural keys from delivery tiles after writing standardized aliases",
|
|
)
|
|
parser.add_argument(
|
|
"--keep-legacy-japanese-delivery",
|
|
action="store_false",
|
|
dest="strip_legacy_japanese_delivery",
|
|
help="keep mapped Japanese structural keys in delivery tiles for legacy compatibility",
|
|
)
|
|
parser.add_argument(
|
|
"--release-minimal",
|
|
action="store_true",
|
|
help="emit a final release tile payload with only the minimal style-driven delivery fields",
|
|
)
|
|
parser.add_argument("--fid-key")
|
|
parser.add_argument("--fid-key-id", default="navsea-fid-key-v1")
|
|
parser.add_argument("--bundle-id", default="navsea-reversible-v1")
|
|
parser.set_defaults(strip_legacy_japanese_delivery=True)
|
|
args = parser.parse_args()
|
|
|
|
if not args.all_tiles and args.reference_tile_root is None:
|
|
missing = [
|
|
name
|
|
for name in ("center_lat", "center_lon", "radius_nm")
|
|
if getattr(args, name) is None
|
|
]
|
|
if missing:
|
|
parser.error(
|
|
"--center-lat, --center-lon, and --radius-nm are required unless --all-tiles is used"
|
|
)
|
|
if not args.fid_key:
|
|
parser.error("--fid-key is required for NavSea delivery and engineering builds")
|
|
|
|
return args
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
fid_codec = NavSeaFidCodec(args.fid_key.encode("utf-8")) if args.fid_key else None
|
|
builder = NavSeaTileBuilder(
|
|
db_config=DbConfig(),
|
|
source_root=args.source_root,
|
|
output_root=args.output,
|
|
center_lat=args.center_lat,
|
|
center_lon=args.center_lon,
|
|
radius_nm=args.radius_nm,
|
|
zmin=args.zmin,
|
|
zmax=args.zmax,
|
|
workers=max(1, args.workers),
|
|
all_tiles=args.all_tiles,
|
|
reference_tile_root=args.reference_tile_root,
|
|
engineering_mode=args.engineering,
|
|
strip_legacy_japanese_delivery=args.strip_legacy_japanese_delivery,
|
|
release_minimal=args.release_minimal,
|
|
fid_codec=fid_codec,
|
|
fid_key_id=args.fid_key_id,
|
|
bundle_id=args.bundle_id,
|
|
)
|
|
builder.build()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|