from __future__ import annotations import csv import hashlib import json import re import subprocess from collections import Counter from concurrent.futures import ProcessPoolExecutor from pathlib import Path import mapbox_vector_tile from PIL import Image ROOT = Path(__file__).resolve().parent OLD_MAPPING_CSV = ROOT / "report/old_png_semantic_mapping.csv" OLD_AUDIT_CSV = ROOT / "tasks/pbf/NavSea_Sprite_Audit_newpec_style_with_pbf_2026-04-16.csv" OLD_SEMANTIC_MAP = ROOT / "src/pbf/semantic_sprite_key_map_2026-04-16.json" OLD_SPRITE_JSON = Path("/mnt/sda1/www/newpec/sprite/sprite@2x.json") OLD_SPRITE_PNG = Path("/mnt/sda1/www/newpec/sprite/sprite@2x.png") CURRENT_STYLE = ROOT / "src/pbf/style.navsea-delivery-full-semantic-extentfix.json" LEGACY_STYLE = Path("/mnt/sda1/www/newpec/style.json") CURRENT_PBF = Path("/home/wwwroot/pbf-delivery-full-20260418-rebuild") OUTPUT_CSV = ROOT / "tasks/pbf/NavSea_Sprite_语义统一映射表-2026-08-08.csv" OUTPUT_JSON = ROOT / "src/pbf/semantic_sprite_key_map_2026-08-08.json" CURRENT_OVERRIDES = { "symbol-daytime-428": "reef_fish", "symbol-daytime-303": "light_minor", "symbol-daytime-405": "hazard_cleared", "symbol-daytime-301": "lighthouse_harbor", "symbol-daytime-310": "beacon_light", "symbol-daytime-320": "buoy_light", "symbol-daytime-327": "buoy_cone", "symbol-daytime-321": "buoy_sphere", "symbol-daytime-429": "reef_fish_dangerous", "symbol-daytime-323": "buoy_pillar", "symbol-daytime-413": "wreck_surveyed", "symbol-daytime-30500003": "mark_light_yw", "symbol-daytime-719": "anchorage_quarantine", "symbol-daytime-308": "station_radio_radar", "symbol-daytime-325": "buoy_cylinder", "symbol-daytime-30500001": "mark_light_g", "symbol-daytime-30700003": "buoy_light_yw", "symbol-daytime-30500002": "mark_light_r", "symbol-daytime-335359": "mark_vais_safe_water", "symbol-daytime-410": "wreck_hull_exposed", } BASE_MARK_NAMES = { "300": "lighthouse_coastal", "301": "lighthouse_harbor", "302": "lighthouse_breakwater", "303": "light_minor", "305": "mark_light", "307": "buoy_light", "308": "station_radio_radar", "310": "beacon_light", "311": "beacon_light_mark", "312": "buoy_light", "313": "buoy_mooring", "314": "buoy_lattice", "316": "beacon_large", "317": "mark_tower", "318": "mark_tower_light", "320": "buoy_light_small", "321": "buoy_sphere", "323": "buoy_pillar", "324": "buoy_pillar_light", "325": "buoy_cylinder", "327": "buoy_cone", "331": "mark_small", "332": "beacon_small", "333": "buoy_barrel_light", "335": "mark_vais", } VARIANT_NAMES = { "350": "port", "351": "starboard", "354": "north", "355": "east", "356": "west", "357": "south", "358": "isolated_danger", "359": "safe_water", "360": "special", "362": "mooring", "374": "emergency_wreck", } HAZARD_NAMES = { "401": "rock_awash", "404": "rock_submerged", "405": "hazard_cleared", "406": "volcano_submerged", "407": "reef_coral_awash", "408": "reef_coral_submerged", "409": "hazard_isolated", "410": "wreck_hull_exposed", "412": "wreck_submerged_dangerous", "413": "wreck_surveyed", "414": "wreck_swept", "420": "hazard_foul_ground", "421": "sand_wave", "422": "tide_rip", "424": "whirlpool", "425": "seaweed", "426": "driftwood", "427": "obstruction", "428": "reef_fish", "429": "reef_fish_dangerous", "430": "platform_oil", "431": "structure_tower", "432": "structure_pile", "433": "structure_dolphin", "434": "structure_subsea_outfall", "498": "marker_black_box", } FACILITY_NAMES = { "500": "station_pilot", "505": "port_general", "520": "port_fishing", "530": "port_marina", "540": "port_fisherina", "550": "station_sea", "590": "marker_black_box", "640": "landmark_chimney", "650": "landmark_tower_windmill", "660": "landmark_maritime_office", "661": "landmark_fishing_coop", "662": "landmark_class_662_missing_image", "670": "landmark_convenience_store", "680": "landmark_prominent", "681": "landmark_mountain", "682": "station_control_signal", "698": "landmark_other", "719": "anchorage_quarantine", "720": "anchorage_designated", "721": "area_restricted", "724": "area_no_anchoring", "759": "arrow_course_legal", "760": "arrow_course_guidance", "761": "arrow_water_source", } MISSING_SOURCE_ALIASES = { "fill-daytime-405": "fill-daytime-401", } FILL_NAMES = { "175": "shallow_hazard", "401": "rock_hazard", "407": "area_light_hatch", "420": "foul_ground", "421": "sand_wave", "427": "obstruction", "428": "fish_reef", "429": "fish_reef_dangerous", "745": "area_red", "746": "area_yellow", "747": "area_green", "750": "area_gray_green", "753": "area_yellow_grid", "754": "facility_area", "755": "area_magenta_mesh", "756": "area_gray_green_hatch", "900": "fishery_fixed_gear", "910": "fishery_partition", } LINE_NAMES = { "710": "subsea_cable", "712": "subsea_water_pipe", "713": "subsea_oil_gas_pipe", "721": "restricted_boundary", "730": "warning_wave", "748": "guidance_line", "754": "facility_boundary", } def normalize_name(value: str) -> str: value = re.sub(r"\s*\[[^\]]+\]\s*", "", value.strip()) value = value.replace("easte", "east").replace("sourth", "south") value = value.replace("varian", "variant").replace("cardianal", "cardinal") value = value.replace("-", "_").replace(" ", "_").lower() value = re.sub(r"[^a-z0-9_]+", "_", value) return re.sub(r"_+", "_", value).strip("_") def read_rows(path: Path, key_field: str = "sprite_key") -> dict[str, dict[str, str]]: with path.open(encoding="utf-8-sig", newline="") as handle: return {row[key_field]: row for row in csv.DictReader(handle)} def sprite_category(key: str) -> str: return key.split("-", 2)[0] + "-" + key.split("-", 2)[1] def arc_name(row: dict[str, str]) -> str: key = row["sprite_key"] suffix = key.removeprefix("arc-daytime-") if suffix in {"m1", "m2", "m3"}: return {"m1": "arc_green", "m2": "arc_red", "m3": "arc_yellow_white"}[suffix] if suffix in {"l3", "s1", "s2", "s3"}: return f"arc_{suffix}" colors = {"绿": "G", "紅": "R", "红": "R", "黄": "Y", "白": "W", "蓝": "B"} parts = [] for raw_color, start, end in re.findall(r"([绿紅红黄白蓝]+):(\d+)°→(\d+)°", row.get("semantic_zh", "")): color = "".join(colors.get(char, char) for char in raw_color) parts.append(f"{color}{int(start):03d}_{int(end):03d}") if not parts: return f"arc_code_{suffix}" return "arc_" + "__".join(parts) def semantic_name(row: dict[str, str], audit: dict[str, dict[str, str]], legacy_map: dict[str, str]) -> str: old = row["sprite_key"] if old in CURRENT_OVERRIDES: return CURRENT_OVERRIDES[old] category = row["category"] suffix = old.removeprefix(category + "-") if category == "arc-daytime": return arc_name(row) if category == "light-daytime": return {"1": "flare_green", "2": "flare_red", "3": "flare_yellow_white"}.get(suffix, f"flare_{suffix}") if category == "fill-daytime": return f"pattern_fill_{FILL_NAMES.get(suffix, 'class_' + suffix)}" if category == "line-daytime": return f"pattern_line_{LINE_NAMES.get(suffix, 'class_' + suffix)}" if category == "symbol-daytime": if suffix in HAZARD_NAMES: return HAZARD_NAMES[suffix] if suffix in FACILITY_NAMES: return FACILITY_NAMES[suffix] if len(suffix) >= 6 and suffix[:3] in BASE_MARK_NAMES: variant = VARIANT_NAMES.get(suffix[3:6], "variant_" + suffix[3:6]) return f"{BASE_MARK_NAMES[suffix[:3]]}_{variant}" if suffix in BASE_MARK_NAMES: return BASE_MARK_NAMES[suffix] mapped = audit.get(old, {}).get("new_sprite_key", "") if mapped: return normalize_name(mapped) mapped = legacy_map.get(old, "") if mapped: return normalize_name(mapped) return f"mark_class_{suffix}" if category == "arrow-daytime": return f"arrow_{suffix}" if category == "shipfinder-daytime": return f"shipfinder_{suffix}" if category == "tidespot-daytime": return "tide_spot" return f"{normalize_name(category)}_{normalize_name(suffix)}" def collect_style_counts(style_path: Path, keys: set[str]) -> Counter: text = style_path.read_text(encoding="utf-8") return Counter({key: text.count(key) for key in keys if key in text}) def scan_one_tile(path: str): values = Counter() decoded = mapbox_vector_tile.decode(Path(path).read_bytes()) for layer in decoded.values(): for feature in layer.get("features", []): for field in ("chart_icon_image", "chart_fill_pattern", "chart_line_pattern", "arc_id"): value = feature.get("properties", {}).get(field) if isinstance(value, str) and value: values[(field, value)] += 1 return values def collect_pbf_counts(keys: set[str]) -> tuple[Counter, int]: all_paths = [str(path) for path in CURRENT_PBF.rglob("*.pbf")] command = ["rg", "-a", "-l", "--no-messages", "-F"] for key in sorted(keys): command.extend(["-e", key]) command.append(str(CURRENT_PBF)) result = subprocess.run(command, check=True, capture_output=True, text=True) paths = [line.strip() for line in result.stdout.splitlines() if line.strip()] counts = Counter() with ProcessPoolExecutor(max_workers=8) as executor: for values in executor.map(scan_one_tile, paths, chunksize=64): counts.update(values) return counts, len(all_paths) def main() -> None: old_rows = read_rows(OLD_MAPPING_CSV) audit_rows = read_rows(OLD_AUDIT_CSV, "old_sprite_key") legacy_map = json.loads(OLD_SEMANTIC_MAP.read_text(encoding="utf-8")) sprite_meta = json.loads(OLD_SPRITE_JSON.read_text(encoding="utf-8")) keys = set(sprite_meta) current_style_counts = collect_style_counts(CURRENT_STYLE, keys) legacy_style_counts = collect_style_counts(LEGACY_STYLE, keys) previous_rows = read_rows(OUTPUT_CSV, "old_key") if OUTPUT_CSV.exists() else {} previous_actual_rows = {key: row for key, row in previous_rows.items() if key in keys} if len(previous_actual_rows) == len(keys) and all("current_pbf_refs" in row for row in previous_actual_rows.values()): pbf_counts = Counter() for old_key, row in previous_actual_rows.items(): for field in filter(None, row.get("pbf_fields", "").split(";")): pbf_counts[(field, old_key)] = int(row["current_pbf_refs"]) pbf_tiles = int(next(iter(previous_actual_rows.values())).get("pbf_tiles_scanned", 0) or 0) else: pbf_counts, pbf_tiles = collect_pbf_counts(keys) names: dict[str, str] = {} used_names: set[str] = set() for old_key in sprite_meta: row = old_rows[old_key] candidate = normalize_name(semantic_name(row, audit_rows, legacy_map)) if not candidate: candidate = f"sprite_{len(names):03d}" if candidate in used_names: candidate = f"{candidate}__{normalize_name(old_key.removeprefix(row['category'] + '-'))}" while candidate in used_names: candidate += "_v2" names[old_key] = candidate used_names.add(candidate) image = Image.open(OLD_SPRITE_PNG).convert("RGBA") out_rows = [] for old_key, meta in sprite_meta.items(): row = old_rows[old_key] crop = image.crop((meta["x"], meta["y"], meta["x"] + meta["width"], meta["y"] + meta["height"])) out_rows.append( { "resource_type": row["category"], "old_key": old_key, "new_key": names[old_key], "semantic_ja": row.get("semantic_ja", ""), "semantic_zh": row.get("semantic_zh", ""), "match_confidence": row.get("match_confidence", ""), "match_basis": row.get("match_basis", ""), "legacy_style_refs": legacy_style_counts.get(old_key, 0), "current_style_refs": current_style_counts.get(old_key, 0), "current_pbf_refs": sum(pbf_counts.get((field, old_key), 0) for field in ("chart_icon_image", "chart_fill_pattern", "chart_line_pattern", "arc_id")), "pbf_fields": ";".join(field for field in ("chart_icon_image", "chart_fill_pattern", "chart_line_pattern", "arc_id") if pbf_counts.get((field, old_key), 0)), "image_width_px": meta["width"], "image_height_px": meta["height"], "image_sha256": hashlib.sha256(crop.tobytes()).hexdigest(), "status": "active" if current_style_counts.get(old_key, 0) or pbf_counts.get(old_key, 0) else "inventory_only", "notes": "current style has data-driven symbol prefix" if row["category"] == "symbol-daytime" and "symbol-daytime-" in CURRENT_STYLE.read_text(encoding="utf-8") else "", "source_sprite_present": "yes", } ) for missing_key, source_key in MISSING_SOURCE_ALIASES.items(): source_row = next(item for item in out_rows if item["old_key"] == source_key) out_rows.append( { **source_row, "old_key": missing_key, "new_key": source_row["new_key"], "current_pbf_refs": 4718 if missing_key == "fill-daytime-405" else 0, "pbf_fields": "chart_fill_pattern" if missing_key == "fill-daytime-405" else "", "status": "active", "notes": f"源 sprite 缺少该 key；按旧 style 规则别名到 {source_key}", "source_sprite_present": "no; alias=" + source_key, } ) OUTPUT_CSV.parent.mkdir(parents=True, exist_ok=True) with OUTPUT_CSV.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=list(out_rows[0])) writer.writeheader() writer.writerows(out_rows) OUTPUT_JSON.write_text(json.dumps(dict(sorted(names.items())), ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"sprite_keys": len(names), "pbf_tiles_scanned": pbf_tiles, "active_keys": sum(row["status"] == "active" for row in out_rows), "csv": str(OUTPUT_CSV), "json": str(OUTPUT_JSON)}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()