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", } @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 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 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, engineering_mode: 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.engineering_mode = engineering_mode 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.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 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, COALESCE(NULLIF(r.render_layer, ''), r.source_layer) AS output_layer, 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): output_layer = str(row["output_layer"]) properties = dict(source_feature.get("properties") or {}) legacy_fid_raw = properties.get("fid", row["fid"]) 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 properties["fid_algo_id"] = "feistel32_aes_cyclewalk_v1" properties["fid_key_id"] = self.fid_key_id 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_legacy_raw"] = legacy_fid_raw properties["fid_navsea_int"] = navsea_fid_int properties["source_layer_jp"] = row["source_layer"] source_layer_std, source_layer_rule_id = self.mapping_registry.resolve_source_layer( str(row["source_layer"]) ) properties["source_layer_std"] = source_layer_std properties["normalization_bundle_id"] = self.bundle_id properties["source_layer_rule_id"] = source_layer_rule_id properties["feature_id"] = row["feature_id"] properties["canonical_object_type"] = row["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=str(row["source_layer"]), output_layer=output_layer, canonical_object_type=text_or_none(row["canonical_object_type"]) or "", 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) if self.engineering_mode: properties["trace_status"] = trace_status unresolved.extend(feature_unresolved) output_feature = { "geometry": source_feature["geometry"], "properties": properties, } if 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[str(row["source_layer"])]) 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 "", ) 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 @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 "灯" 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: 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: 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, ) -> str | None: if source_layer == "p航路標識群": 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 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 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("--engineering", action="store_true") 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") args = parser.parse_args() if not args.all_tiles: 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 args.engineering and not args.fid_key: parser.error("--fid-key is required when --engineering is used") 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, engineering_mode=args.engineering, fid_codec=fid_codec, fid_key_id=args.fid_key_id, bundle_id=args.bundle_id, ) builder.build() if __name__ == "__main__": main()