pbf/navsea_build_kyushu_semantic_icon_test.py

#!/usr/bin/env python3
"""从当前全国 Full 基线生成九州语义图标替换测试包。"""

from __future__ import annotations

import json
import os
import shutil
from collections import Counter
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from typing import Any

import mercantile

from navsea_tile_reencode_guard import (
    assert_reencoded_tile_safe,
    decode_tile_with_extents,
    encode_layers_preserving_extents,
)


REPO_ROOT = Path(__file__).resolve().parent
CONFIG_PATH = REPO_ROOT / "tasks/pbf/navsea_semantic_icon_map_v1.json"
SOURCE_ROOT = Path("/home/wwwroot/pbf-delivery-full-20260418-rebuild")
TARGET_ROOT = Path("/home/wwwroot/pbf-delivery-kyushu-semantic-icon-v1-20260806")
TARGET_TILE_URL = "http://192.168.200.184/pbf-delivery-kyushu-semantic-icon-v1-20260806/{z}/{x}/{y}.pbf"
KYUSHU_BBOX = (128.0, 30.0, 132.5, 34.9)
ZOOMS = tuple(range(5, 13))
ASSET_VERSION = "kyushu-semantic-icon-v1-20260806-r2"

SPRITE_SOURCE = Path("/mnt/sda1/www/newpec/sprite-semantic")
SPRITE_TARGET = Path(f"/mnt/sda1/www/newpec/sprite-{ASSET_VERSION}")
SPRITE_URL = f"http://192.168.200.184/newpec/sprite-{ASSET_VERSION}/sprite"

STYLE_SOURCE = REPO_ROOT / "src/pbf/style.navsea-delivery-full-semantic-extentfix.json"
STYLE_OUTPUT = REPO_ROOT / "src/pbf/style.navsea-delivery-kyushu-semantic-icon-v1.json"
STYLE_DEPLOY = Path("/mnt/sda1/www/newpec/domain/style.navsea-delivery-kyushu-semantic-icon-v1.json")

NEWPEC_STYLE_SOURCE = REPO_ROOT / "src/pbf/style.json"
NEWPEC_STYLE_OUTPUT = REPO_ROOT / "src/pbf/style.navsea-newpec-kyushu-icon-test-v1.json"
NEWPEC_STYLE_DEPLOY = Path("/mnt/sda1/www/newpec/domain/style.navsea-newpec-kyushu-icon-test-v1.json")

REPORT_DIR = REPO_ROOT / "report/kyushu_semantic_icon_v1_2026-08-06"

SIMPLE_ICON_LAYERS = {
    "hazard-points",
    "anchorage-symbols",
    "anchor-hazard-points",
    "anchor-hazard-points-428",
    "facility-points",
    "landmark-points",
}

STYLE_EXTRA_MAP = {
    "light_flare_green": "fl_G",
    "light_flare_red": "fl_R",
    "light_flare_yellow_white": "fl_YW",
}

CANONICAL_ICON_MAP = {
    "coastal_lighthouse_over_15m": "lt_cst",
    "harbor_lighthouse": "lt_hbr",
    "breakwater_lighthouse": "lt_bkw",
    "minor_light": "lt_min",
    "light_beacon": "lt_bcn",
    "light_buoy": "bu_lit",
    "lattice_buoy": "bu_lat",
    "pillar_buoy": "bu_pil",
    "can_buoy": "bu_can",
    "buoy_generic": "bu_gen",
    "nav_mark_vais": "mk_vais",
}


def load_config() -> dict[str, Any]:
    return json.loads(CONFIG_PATH.read_text(encoding="utf-8"))


def collect_tiles() -> list[tuple[int, int, int, Path]]:
    tiles: list[tuple[int, int, int, Path]] = []
    west, south, east, north = KYUSHU_BBOX
    for z in ZOOMS:
        for tile in mercantile.tiles(west, south, east, north, [z]):
            rel = Path(str(tile.z)) / str(tile.x) / f"{tile.y}.pbf"
            source = SOURCE_ROOT / rel
            if source.exists():
                tiles.append((tile.z, tile.x, tile.y, rel))
    return sorted(tiles)


def copy_link(source: Path, target: Path) -> None:
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.exists():
        target.unlink()
    try:
        target.hardlink_to(source)
    except OSError:
        shutil.copy2(source, target)


def class_icon(layer_name: str, class_code: object, config: dict[str, Any]) -> str | None:
    code = str(class_code)
    if layer_name in {"navigation_hazard_point", "anchor_caution_hazard_point"}:
        return config["hazard_class_map"].get(code)
    if layer_name == "onshore_structure_point":
        return config["landmark_class_map"].get(code)
    if layer_name == "anchorage_point":
        return config["anchorage_class_map"].get(code)
    if layer_name == "facility_boundary_point":
        return config["facility_class_map"].get(code)
    return None


def navigation_icon(properties: dict[str, Any], config: dict[str, Any]) -> str | None:
    legacy = properties.get("chart_icon_image")
    if isinstance(legacy, str):
        mapped = config["sprite_key_map"].get(legacy)
        if mapped:
            return mapped
    display_code = properties.get("display_code")
    if display_code is not None:
        mapped = config["display_code_map"].get(str(display_code))
        if mapped:
            return mapped
    canonical = properties.get("canonical_object_type")
    if isinstance(canonical, str):
        return CANONICAL_ICON_MAP.get(canonical)
    return None


def arc_icon(properties: dict[str, Any], config: dict[str, Any]) -> str | None:
    if properties.get("chart_symbol_code") != "lighthouse":
        return None
    if not properties.get("light_color_code"):
        return None
    if (
        properties.get("canonical_object_type") == "coastal_lighthouse_over_15m"
        and properties.get("light_sector_mode") == "sector"
    ):
        return config["arc_rules"]["sector"]
    color = str(properties.get("light_color_code"))
    return config["arc_rules"].get(color, config["arc_rules"]["other"])


def normalize_pattern_properties(layer_name: str, properties: dict[str, Any]) -> None:
    if (
        layer_name == "anchor_caution_hazard_area"
        and str(properties.get("class_code")) == "420"
        and properties.get("chart_fill_pattern") == "fill-daytime-405"
    ):
        properties["chart_fill_pattern"] = "fill-daytime-420"
    if (
        layer_name == "navigation_hazard_area"
        and str(properties.get("class_code")) == "427"
        and properties.get("chart_fill_pattern") == "fill-daytime-405"
    ):
        properties["chart_fill_pattern"] = "fill-daytime-427"


def rewrite_tile(
    source: Path,
    target: Path,
    config: dict[str, Any],
    *,
    verify_geometry: bool = False,
) -> dict[str, Any]:
    decoded, extents = decode_tile_with_extents(source.read_bytes())
    changed_features = 0
    icon_counts: Counter[str] = Counter()
    arc_counts: Counter[str] = Counter()
    old_values: Counter[str] = Counter()
    unknown_old_values: Counter[str] = Counter()
    output_layers: list[dict[str, Any]] = []

    for layer_name, layer in decoded.items():
        output_features = []
        for feature in layer["features"]:
            properties = dict(feature.get("properties") or {})
            before = dict(properties)
            normalize_pattern_properties(layer_name, properties)
            legacy = properties.get("chart_icon_image")
            if isinstance(legacy, str) and (
                legacy.startswith("symbol-daytime-") or legacy.startswith("arc-daytime-")
            ):
                old_values[legacy] += 1

            icon_id = class_icon(layer_name, properties.get("class_code"), config)
            if layer_name == "navigation_marks":
                icon_id = icon_id or navigation_icon(properties, config)
            if icon_id is None and isinstance(legacy, str):
                icon_id = config["sprite_key_map"].get(legacy)

            if icon_id:
                properties["icon_id"] = icon_id
                icon_counts[icon_id] += 1
                properties.pop("chart_icon_image", None)

            if layer_name == "navigation_marks":
                arc_id = arc_icon(properties, config)
                if arc_id:
                    properties["arc_id"] = arc_id
                    arc_counts[arc_id] += 1

            for value in properties.values():
                if isinstance(value, str) and (
                    value.startswith("symbol-daytime-") or value.startswith("arc-daytime-")
                ):
                    unknown_old_values[value] += 1

            if properties != before:
                changed_features += 1
            output_feature = {
                "geometry": feature["geometry"],
                "properties": properties,
            }
            if feature.get("id") is not None:
                output_feature["id"] = feature["id"]
            output_features.append(output_feature)
        output_layers.append({"name": layer_name, "features": output_features})

    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes(encode_layers_preserving_extents(output_layers, extents))
    if verify_geometry:
        assert_reencoded_tile_safe(source, target)
    return {
        "changed_features": changed_features,
        "icon_counts": dict(icon_counts),
        "arc_counts": dict(arc_counts),
        "old_values": dict(old_values),
        "unknown_old_values": dict(unknown_old_values),
    }


def rewrite_tile_job(args: tuple[str, str, dict[str, Any], bool]) -> dict[str, Any]:
    source, target, config, verify_geometry = args
    result = rewrite_tile(
        Path(source), Path(target), config, verify_geometry=verify_geometry
    )
    result["source"] = source
    return result


def transform_style(config: dict[str, Any]) -> dict[str, Any]:
    style = json.loads(STYLE_SOURCE.read_text(encoding="utf-8"))
    key_map = dict(config["sprite_key_map"])
    key_map.update(STYLE_EXTRA_MAP)

    def transform(value: Any) -> Any:
        if isinstance(value, dict):
            return {key: transform(item) for key, item in value.items()}
        if isinstance(value, list):
            return [transform(item) for item in value]
        if isinstance(value, str):
            if value == "chart_icon_image":
                return "icon_id"
            return key_map.get(value, value)
        return value

    style = transform(style)
    style["name"] = "NavSea Kyushu Semantic Icon Test v1"
    style["sprite"] = SPRITE_URL
    style["metadata"] = {
        "navsea_test_version": "kyushu-semantic-icon-v1-20260806",
        "source_baseline": str(SOURCE_ROOT),
        "icon_map": str(CONFIG_PATH),
    }
    style["sources"]["navsea_delivery"]["tiles"] = [TARGET_TILE_URL]

    canonical_restore = {
        "lt_hbr": "harbor_lighthouse",
        "lt_bkw": "breakwater_lighthouse",
        "lt_bcn": "light_beacon",
    }

    def restore_canonical_values(value: Any) -> Any:
        if isinstance(value, list):
            if len(value) >= 3 and value[:2] == ["get", "canonical_object_type"]:
                return value
            return [restore_canonical_values(item) for item in value]
        if isinstance(value, dict):
            return {key: restore_canonical_values(item) for key, item in value.items()}
        if isinstance(value, str):
            return canonical_restore.get(value, value)
        return value

    for layer in style["layers"]:
        layer_id = layer.get("id")
        layout = layer.get("layout") or {}
        layer_filter = layer.get("filter")
        if layer_id in SIMPLE_ICON_LAYERS:
            layout["icon-image"] = ["coalesce", ["get", "icon_id"], ""]
            layer["layout"] = layout
        elif layer_id == "nav-light-arc":
            layout["icon-image"] = ["coalesce", ["get", "arc_id"], ""]
            layer["layout"] = layout
        elif layer_id in {
            "nav-marks-harbor-lighthouses",
            "nav-marks-breakwater-lighthouses",
            "nav-marks-light-beacons",
            "nav-marks",
        }:
            if layer_filter is not None:
                layer["filter"] = restore_canonical_values(layer_filter)
            if layer_id == "nav-marks":
                layer["layout"] = restore_canonical_values(layout)
    return style


def write_style(style: dict[str, Any], path: Path, deploy_path: Path) -> None:
    payload = json.dumps(style, ensure_ascii=False, indent=2) + "\n"
    path.write_text(payload, encoding="utf-8")
    deploy_path.parent.mkdir(parents=True, exist_ok=True)
    deploy_path.write_text(payload, encoding="utf-8")


def write_newpec_style() -> None:
    style = json.loads(NEWPEC_STYLE_SOURCE.read_text(encoding="utf-8"))
    style["name"] = "NavSea Newpec Kyushu Icon Test v1"
    style["metadata"] = {
        "navsea_test_version": "kyushu-semantic-icon-v1-20260806",
        "role": "left_original_newpec_reference",
    }
    write_style(style, NEWPEC_STYLE_OUTPUT, NEWPEC_STYLE_DEPLOY)


def collect_literal_sprite_refs(style: dict[str, Any], sprite_keys: set[str] | None = None) -> set[str]:
    refs: set[str] = set()

    def walk(value: Any) -> None:
        if isinstance(value, dict):
            for child in value.values():
                walk(child)
        elif isinstance(value, list):
            for child in value:
                walk(child)
        elif isinstance(value, str):
            if sprite_keys is not None and value in sprite_keys:
                refs.add(value)
            elif (
                value.startswith("pattern_fill_")
                or value.startswith("fill-daytime-")
                or value.startswith("fill_")
                or value.startswith("special_pattern_")
                or value in {"tide_spot"}
            ):
                refs.add(value)

    for layer in style.get("layers", []):
        layout = layer.get("layout") or {}
        paint = layer.get("paint") or {}
        for field in ("icon-image",):
            if field in layout:
                walk(layout[field])
        for field in ("fill-pattern", "line-pattern"):
            if field in paint:
                walk(paint[field])
    return refs


def build_sprite(config: dict[str, Any], style: dict[str, Any]) -> dict[str, Any]:
    if SPRITE_TARGET.exists():
        shutil.rmtree(SPRITE_TARGET)
    SPRITE_TARGET.mkdir(parents=True)
    source_meta: dict[str, dict[str, dict[str, Any]]] = {}
    for filename in ("sprite.json", "sprite@2x.json"):
        source_meta[filename] = json.loads((SPRITE_SOURCE / filename).read_text(encoding="utf-8"))
    source_sprite_keys = set().union(*(meta.keys() for meta in source_meta.values()))

    key_map = dict(config["sprite_key_map"])
    key_map.update(STYLE_EXTRA_MAP)
    fallback_sources = {"lm_cus": "symbol-daytime-660"}
    passthrough_keys = collect_literal_sprite_refs(style, source_sprite_keys)
    added: dict[str, str] = {}
    for filename, meta in source_meta.items():
        # 测试包只改名图标/灯弧；fill/line pattern 等非图标纹理必须按 style 原名保留。
        output: dict[str, dict[str, Any]] = {}
        data_driven_pattern_keys = {
            key
            for key in meta
            if (
                key.startswith("pattern_fill_")
                or key.startswith("fill-daytime-")
                or key.startswith("fill_")
                or key.startswith("special_pattern_")
            )
        }
        for key in sorted(passthrough_keys | data_driven_pattern_keys):
            if key in meta:
                output[key] = dict(meta[key])
                added[key] = key
        for old_key, new_key in key_map.items():
            source_key = old_key if old_key in meta else None
            if source_key is None:
                for candidate, candidate_new in key_map.items():
                    if candidate_new == new_key and candidate in meta:
                        source_key = candidate
                        break
            if source_key is None:
                source_key = fallback_sources.get(new_key)
            if source_key and source_key in meta:
                output[new_key] = dict(meta[source_key])
                added[new_key] = source_key
        (SPRITE_TARGET / filename).write_text(
            json.dumps(output, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
        )

    for filename in ("sprite.png", "sprite@2x.png"):
        shutil.copy2(SPRITE_SOURCE / filename, SPRITE_TARGET / filename)
    return {
        "added": added,
        "passthrough_keys": sorted(passthrough_keys),
        "fallbacks": {"lm_cus": fallback_sources["lm_cus"]},
    }


def main() -> None:
    config = load_config()
    tiles = collect_tiles()
    if TARGET_ROOT.exists():
        raise RuntimeError(f"refusing to overwrite existing PBF target: {TARGET_ROOT}")
    TARGET_ROOT.mkdir(parents=True)

    totals: Counter[str] = Counter()
    old_values: Counter[str] = Counter()
    unknown_old_values: Counter[str] = Counter()
    icon_counts: Counter[str] = Counter()
    arc_counts: Counter[str] = Counter()
    rewritten = 0
    candidate_needles = [
        key.encode("utf-8") for key in config["sprite_key_map"]
    ] + [
        b"navigation_marks",
        b"navigation_hazard_point",
        b"anchor_caution_hazard_point",
        b"anchorage_point",
        b"facility_boundary_point",
        b"onshore_structure_point",
    ]
    rewrite_jobs: list[tuple[str, str, dict[str, Any], bool]] = []
    candidate_count = 0
    for z, x, y, rel in tiles:
        source = SOURCE_ROOT / rel
        target = TARGET_ROOT / rel
        raw = source.read_bytes()
        if not any(needle in raw for needle in candidate_needles):
            copy_link(source, target)
            continue
        rewrite_jobs.append(
            (str(source), str(target), config, candidate_count < 3)
        )
        candidate_count += 1

    worker_count = min(os.cpu_count() or 4, 12)
    with ProcessPoolExecutor(max_workers=worker_count) as executor:
        for result in executor.map(rewrite_tile_job, rewrite_jobs, chunksize=8):
            if result["changed_features"]:
                rewritten += 1
            totals["features_changed"] += result["changed_features"]
            old_values.update(result["old_values"])
            unknown_old_values.update(result["unknown_old_values"])
            icon_counts.update(result["icon_counts"])
            arc_counts.update(result["arc_counts"])

    style = transform_style(config)
    style["metadata"]["navsea_test_version"] = ASSET_VERSION
    style["sprite"] = SPRITE_URL
    sprite = build_sprite(config, style)
    write_style(style, STYLE_OUTPUT, STYLE_DEPLOY)
    write_newpec_style()

    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    summary = {
        "version": "kyushu-semantic-icon-v1-20260806",
        "source_root": str(SOURCE_ROOT),
        "target_root": str(TARGET_ROOT),
        "bbox": KYUSHU_BBOX,
        "zooms": list(ZOOMS),
        "tiles": len(tiles),
        "tiles_rewritten": rewritten,
        "candidate_tiles": candidate_count,
        **totals,
        "old_values": dict(old_values),
        "unknown_old_values_after_rewrite": dict(unknown_old_values),
        "icon_counts": dict(icon_counts),
        "arc_counts": dict(arc_counts),
        "sprite": sprite,
        "style": str(STYLE_OUTPUT),
        "newpec_style": str(NEWPEC_STYLE_OUTPUT),
    }
    (REPORT_DIR / "kyushu_semantic_icon_v1.json").write_text(
        json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )
    lines = [
        "# 九州语义图标替换测试版 v1",
        "",
        f"- PBF 来源：`{SOURCE_ROOT}`",
        f"- PBF 输出：`{TARGET_ROOT}`",
        f"- 范围：`{KYUSHU_BBOX}`",
        f"- zoom：`{ZOOMS[0]}-{ZOOMS[-1]}`",
        f"- 瓦片数：`{len(tiles)}`",
        f"- 重写瓦片：`{rewritten}`",
        f"- 改动 feature：`{totals['features_changed']}`",
        f"- sprite：`{SPRITE_TARGET}`",
        f"- delivery style：`{STYLE_OUTPUT}`",
        f"- newpec style：`{NEWPEC_STYLE_OUTPUT}`",
        "",
        "## 旧图标值",
        "",
        "```json",
        json.dumps(dict(old_values), ensure_ascii=False, indent=2),
        "```",
        "",
        "## 审计关注",
        "",
        "- `unknown_old_values_after_rewrite` 必须为空。",
        "- `lm_cus` 当前使用 `symbol-daytime-660` 的临时视觉回退，需人工确认。",
        "- 几何使用原 tile extent 重编码，并执行 extent 安全检查。",
    ]
    (REPORT_DIR / "kyushu_semantic_icon_v1.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()