#!/usr/bin/env python3 """Generate isolated NavSea Chart Domain prototype PBF packages. This builder is intentionally separate from the current production builders. It reads an existing engineering-style PBF root, re-groups layers by the new Chart Domain model, and writes prototype packages without touching the current engineering/delivery product line. Current output model: - safety package: safety_core + safety_extended - detail package: detail Layers marked as domain_pending are not emitted to final packages. They are instead recorded in the audit report so we can refine semantics before any production switch-over. """ from __future__ import annotations import argparse import json from collections import Counter, defaultdict from dataclasses import dataclass from pathlib import Path from typing import Any import mapbox_vector_tile from chart_domain_model import ( DETAIL, DOMAIN_PENDING, PHYSICAL_PACK_BY_DOMAIN, SAFETY_CORE, SAFETY_EXTENDED, get_layer_domain_spec, normalize_layer_std, ) @dataclass(frozen=True) class TileResult: safety_written: bool detail_written: bool feature_count: int pending_count: int def text_or_none(value: Any) -> str | None: if value is None: return None text = str(value).strip() return text or None class ChartDomainTileBuilder: def __init__( self, input_root: Path, output_root: Path, *, include_pending_in_safety: bool = False, ) -> None: self.input_root = input_root self.output_root = output_root self.include_pending_in_safety = include_pending_in_safety self.safety_root = output_root / "safety" self.detail_root = output_root / "detail" self.audit_json_path = output_root / "chart_domain_build_audit.json" self.audit_md_path = output_root / "chart_domain_build_audit.md" self.feature_count = 0 self.renderable_count = 0 self.pending_count = 0 self.unmapped_source_count = 0 self.domain_counter: Counter[str] = Counter() self.layer_counter: Counter[str] = Counter() self.pending_layer_counter: Counter[str] = Counter() self.pending_examples: list[dict[str, Any]] = [] def build(self) -> None: self.output_root.mkdir(parents=True, exist_ok=True) self.clear_existing_tiles() tile_paths = sorted(self.input_root.glob("*/*/*.pbf")) safety_written = 0 detail_written = 0 for tile_path in tile_paths: result = self.process_tile(tile_path) self.feature_count += result.feature_count self.pending_count += result.pending_count if result.safety_written: safety_written += 1 if result.detail_written: detail_written += 1 self.write_audit_report( tile_count=len(tile_paths), safety_written=safety_written, detail_written=detail_written, ) def clear_existing_tiles(self) -> None: for pack_root in (self.safety_root, self.detail_root): for path in pack_root.glob("*/*/*.pbf"): path.unlink() def process_tile(self, tile_path: Path) -> TileResult: rel = tile_path.relative_to(self.input_root) z = int(rel.parts[0]) x = int(rel.parts[1]) y = int(tile_path.stem) decoded = mapbox_vector_tile.decode(tile_path.read_bytes()) pack_layers: dict[str, dict[str, list[dict[str, Any]]]] = { "safety": defaultdict(list), "detail": defaultdict(list), } pack_extents: dict[str, dict[str, int]] = { "safety": {}, "detail": {}, } tile_feature_count = 0 tile_pending_count = 0 for input_layer_name, payload in decoded.items(): extent = int(payload.get("extent") or 4096) for feature in payload.get("features", []): tile_feature_count += 1 properties = dict(feature.get("properties") or {}) source_layer_jp = text_or_none(properties.get("source_layer_jp")) or input_layer_name raw_source_layer_std = text_or_none(properties.get("source_layer_std")) or input_layer_name source_layer_std = normalize_layer_std(raw_source_layer_std, source_layer_jp) spec = get_layer_domain_spec(source_layer_std) self.domain_counter[spec.chart_domain] += 1 self.layer_counter[source_layer_std] += 1 output_properties = dict(properties) output_properties["source_layer_jp"] = source_layer_jp output_properties["source_layer_std"] = source_layer_std output_properties["chart_domain"] = spec.chart_domain output_properties["offline_pack"] = spec.offline_pack output_properties["resolver_priority"] = spec.resolver_priority output_properties["domain_status"] = spec.domain_status output_properties["chart_domain_semantic_zh"] = spec.chinese_semantic pack_name = PHYSICAL_PACK_BY_DOMAIN.get(spec.chart_domain) if spec.chart_domain == DOMAIN_PENDING and self.include_pending_in_safety: pack_name = "safety" if pack_name is None: tile_pending_count += 1 self.pending_layer_counter[source_layer_std] += 1 if len(self.pending_examples) < 30: self.pending_examples.append( { "tile": f"{z}/{x}/{y}", "input_layer": input_layer_name, "source_layer_jp": source_layer_jp, "source_layer_std": source_layer_std, "canonical_object_type": output_properties.get("canonical_object_type"), "fid": output_properties.get("fid"), "reason": spec.notes or "domain_pending layer is excluded from final packages", } ) continue output_feature = { "geometry": feature["geometry"], "properties": output_properties, } if feature.get("id") is not None: output_feature["id"] = feature["id"] pack_layers[pack_name][source_layer_std].append(output_feature) pack_extents[pack_name].setdefault(source_layer_std, extent) self.renderable_count += 1 safety_written = self.write_pack_tile( pack_name="safety", rel=rel, layers=pack_layers["safety"], extents=pack_extents["safety"], ) detail_written = self.write_pack_tile( pack_name="detail", rel=rel, layers=pack_layers["detail"], extents=pack_extents["detail"], ) return TileResult( safety_written=safety_written, detail_written=detail_written, feature_count=tile_feature_count, pending_count=tile_pending_count, ) def write_pack_tile( self, *, pack_name: str, rel: Path, layers: dict[str, list[dict[str, Any]]], extents: dict[str, int], ) -> bool: if not layers: return False encoded_layers = [] per_layer_options: dict[str, dict[str, int]] = {} for layer_name, features in sorted(layers.items()): if not features: continue encoded_layers.append({"name": layer_name, "features": features}) per_layer_options[layer_name] = {"extents": extents[layer_name]} if not encoded_layers: return False pack_root = self.safety_root if pack_name == "safety" else self.detail_root output_path = pack_root / rel 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 True def write_audit_report(self, *, tile_count: int, safety_written: int, detail_written: int) -> None: payload = { "input_root": str(self.input_root), "output_root": str(self.output_root), "include_pending_in_safety": self.include_pending_in_safety, "tile_count": tile_count, "safety_written_tiles": safety_written, "detail_written_tiles": detail_written, "feature_count": self.feature_count, "renderable_feature_count": self.renderable_count, "pending_feature_count": self.pending_count, "domain_counts": dict(self.domain_counter), "layer_counts": dict(self.layer_counter.most_common()), "pending_layer_counts": dict(self.pending_layer_counter.most_common()), "pending_examples": self.pending_examples, } self.audit_json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") lines = [ "# NavSea Chart Domain Build Audit", "", f"- input_root: `{self.input_root}`", f"- output_root: `{self.output_root}`", f"- include_pending_in_safety: `{self.include_pending_in_safety}`", f"- tile_count: `{tile_count}`", f"- safety_written_tiles: `{safety_written}`", f"- detail_written_tiles: `{detail_written}`", f"- feature_count: `{self.feature_count}`", f"- renderable_feature_count: `{self.renderable_count}`", f"- pending_feature_count: `{self.pending_count}`", "", "## Domain Counts", "", ] for name, count in self.domain_counter.most_common(): lines.append(f"- `{name}`: `{count}`") lines.extend(["", "## Pending Layers", ""]) if not self.pending_layer_counter: lines.append("- none") else: for layer_name, count in self.pending_layer_counter.most_common(): lines.append(f"- `{layer_name}`: `{count}`") lines.extend(["", "## Pending Examples", ""]) if not self.pending_examples: lines.append("- none") else: for item in self.pending_examples[:20]: lines.append( f"- tile=`{item['tile']}` input=`{item['input_layer']}` std=`{item['source_layer_std']}` " f"object=`{item.get('canonical_object_type') or 'n/a'}` fid=`{item.get('fid')}`" ) lines.append(f" reason: {item['reason']}") self.audit_md_path.write_text("\n".join(lines) + "\n", encoding="utf-8") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Build isolated NavSea Chart Domain PBF packages.") parser.add_argument( "--input-root", type=Path, default=Path("/home/wwwroot/pbf-engineering-karatsu-10nm"), help="Input engineering PBF root to transform.", ) parser.add_argument( "--output-root", type=Path, default=Path("/home/wwwroot/pbf-domain-karatsu-10nm"), help="Output root for chart domain prototype packages.", ) parser.add_argument( "--include-pending-in-safety", action="store_true", help="Compatibility mode: emit domain_pending layers into the safety package instead of excluding them.", ) return parser.parse_args() def main() -> None: args = parse_args() builder = ChartDomainTileBuilder( input_root=args.input_root, output_root=args.output_root, include_pending_in_safety=args.include_pending_in_safety, ) builder.build() print( json.dumps( { "input_root": str(args.input_root), "output_root": str(args.output_root), "include_pending_in_safety": args.include_pending_in_safety, "audit_json": str(builder.audit_json_path), "audit_md": str(builder.audit_md_path), }, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()