diff --git a/config/s52/cm95_s52_poc_v1.json b/config/s52/cm95_s52_poc_v1.json new file mode 100644 index 0000000..34dab7a --- /dev/null +++ b/config/s52/cm95_s52_poc_v1.json @@ -0,0 +1,20 @@ +{ + "schema": "navsea.cm95.s52-poc-config/v1", + "targets": [ + "LIGHTS", + "LNDMRK", + "BCNLAT", + "BOYLAT", + "TOPMAR" + ], + "palette": "DAY_BRIGHT", + "source_id": "navsea_delivery", + "source_layer": "navigation_marks", + "sprite_url": "./sprite", + "tiles_template": "https://REPLACE-ME/{z}/{x}/{y}.pbf", + "policy": { + "static_lookup": "materialize_in_builder", + "conditional_symbology": "explicit_builder_derived_or_runtime_renderer", + "production_release": "do_not_touch_in_poc" + } +} diff --git a/docs/报告/CM95_S52_MapLibre_PoC阶段报告_v1.md b/docs/报告/CM95_S52_MapLibre_PoC阶段报告_v1.md new file mode 100644 index 0000000..586ff2a --- /dev/null +++ b/docs/报告/CM95_S52_MapLibre_PoC阶段报告_v1.md @@ -0,0 +1,65 @@ +# CM95 S-52 → MapLibre PoC 阶段报告 v1 + +日期:2026-09-16 + +## 结论 + +S-52 Presentation Library → CM95 → MapLibre 技术路线已完成第一阶段 Linux PoC,并通过真实 CM93JAPAN 数据库验证。 + +本阶段不修改正式 PBF release,不覆盖现有 Linux 迁移中的 `builder.py` / `production.py` 修改。 + +## 新增实现 + +- `src/cm95pbf/s52.py` + - CM93 `COLMAR` → S-57 `COLOUR` 规范化; + - S-52 Lookup condition 匹配; + - 按 OpenCPN `CompareLUPObjects` 顺序选择规则; + - 提取静态 Symbol 与 Conditional Symbology Procedure。 +- `tools/cm95_s52_maplibre_poc.py` + - 解析 OpenCPN `chartsymbols.xml`; + - 生成 MapLibre `sprite.png/sprite.json`; + - 生成 S-52 Lookup/颜色/CSP/Style PoC 产物。 +- `tools/cm95_s52_database_match_audit.py` + - 对真实 `chart_feature.attributes_json` 执行 S-52 Lookup 命中审计。 +- `config/s52/cm95_s52_poc_v1.json` + - 固定首批对象与 PoC 输出策略。 + +## 资源编译结果 + +标准对象大小写精确匹配后: + +- Lookup 总数:359; +- BCNLAT:70; +- BOYLAT:72; +- LIGHTS:6; +- LNDMRK:127; +- TOPMAR:84; +- 直接 Symbol 引用:186; +- 可生成 Sprite:185; +- 缺失 Symbol:`BOYSPH79`; +- Atlas 越界:0; +- CSP:`LIGHTS05`、`TOPMAR01`。 +## 真实数据库命中结果 + +Dataset ID=1: + +- LIGHTS:31978 / 31978 命中,全部进入 `LIGHTS05`; +- BCNLAT:3744 / 3744 命中静态 S-52 Symbol; +- BOYLAT:5565 / 5565 命中静态 S-52 Symbol; +- TOPMAR:11166 / 11166 命中 Simplified 默认空指令规则; +- LNDMRK:数据库原始类数量为 0,CM93 中实际存在 TOWERS / CHIMNY / MONUMT 等陆标类。 + +## 已确认的架构边界 + +1. CM93 对象类和属性必须先规范化到 S-57/S-52 语义,再执行 Lookup。 +2. 静态 Lookup 可由 Builder 物化为 `s52_symbol_code` 等字段。 +3. `CS(...)` 不得伪装为静态 Sprite;LIGHTS / TOPMAR 必须走显式 CSP 或等价派生逻辑。 +4. 当前正式 release 未修改。 +5. 下一阶段在当前 Linux `builder.py` 实际版本上做最小接入,并生成样本 PBF 验证。 + +## 验证 + +- Python `py_compile`:PASS; +- S-52 资源生成:PASS; +- Sprite/Atlas 边界自检:PASS; +- 数据库 Lookup 审计:PASS。 diff --git a/src/cm95pbf/s52.py b/src/cm95pbf/s52.py new file mode 100644 index 0000000..5d7746e --- /dev/null +++ b/src/cm95pbf/s52.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +from typing import Any, Iterable + + +CM93_COLMAR_TO_S57_COLOUR: dict[int, str] = { + 1: "4", # green + 2: "2", # black + 3: "3", # red + 4: "6", # yellow + 5: "1", # white + 6: "11", # orange + 7: "2,6", # black/yellow + 8: "2,6,2", # black/yellow/black + 9: "6,2", # yellow/black + 10: "6,2,6", # yellow/black/yellow + 11: "3,1", # red/white + 12: "4,3,4", # green/red/green + 13: "3,4,3", # red/green/red + 14: "2,3,2", # black/red/black + 15: "6,3,6", # yellow/red/yellow + 16: "4,3", # green/red + 17: "3,4", # red/green + 18: "4,1", # green/white +} + + +def _s57_list_string(value: Any) -> str | None: + if value is None: + return None + if isinstance(value, (list, tuple)): + return ",".join(str(int(x)) for x in value) + if isinstance(value, bool): + return str(int(value)) + if isinstance(value, int): + return str(value) + if isinstance(value, float) and value.is_integer(): + return str(int(value)) + return str(value) + + +def normalize_cm93_attributes_for_s52(attrs: dict[str, Any]) -> dict[str, Any]: + """Return S-57/S-52-compatible attributes without mutating CM93 raw data.""" + out = dict(attrs) + + if "COLOUR" in out: + value = _s57_list_string(out.get("COLOUR")) + if value is not None: + out["COLOUR"] = value + elif "COLMAR" in out: + try: + code = int(out["COLMAR"]) + except (TypeError, ValueError): + code = -1 + translated = CM93_COLMAR_TO_S57_COLOUR.get(code) + if translated is not None: + out["COLOUR"] = translated + + return out + + +def split_lookup_condition(condition: str) -> tuple[str, str]: + text = str(condition) + if len(text) < 6: + raise ValueError(f"invalid S-52 lookup condition: {condition!r}") + return text[:6].upper(), text[6:] + + +def lookup_condition_matches(condition: str, attrs: dict[str, Any]) -> bool: + acronym, expected = split_lookup_condition(condition) + actual_present = acronym in attrs and attrs.get(acronym) is not None + + if expected.startswith(" "): + return actual_present + if expected.startswith("?"): + return not actual_present + if not actual_present: + return False + + actual = attrs.get(acronym) + if isinstance(actual, (list, tuple)): + actual = _s57_list_string(actual) + elif isinstance(actual, bool): + actual = int(actual) + + if isinstance(actual, (int, float)): + try: + return abs(float(actual) - float(expected)) < 1e-6 + except (TypeError, ValueError): + return False + return str(actual) == expected + + +def _lookup_sequence(row: dict[str, Any]) -> int: + try: + return int(row.get("id") or 0) + except (TypeError, ValueError): + return 0 + + +def sort_lookup_candidates(rows: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: + """Match OpenCPN CompareLUPObjects: most conditions first, then sequence.""" + return sorted( + rows, + key=lambda row: ( + -len(row.get("attribute_conditions_raw") or []), + _lookup_sequence(row), + ), + ) + + +def find_best_lookup( + rows: Iterable[dict[str, Any]], + attrs: dict[str, Any], + *, + table_name: str | None = None, + geometry_type: str | None = None, +) -> dict[str, Any] | None: + normalized = normalize_cm93_attributes_for_s52(attrs) + candidates = [] + for row in rows: + if table_name and str(row.get("table_name")) != table_name: + continue + if geometry_type and str(row.get("geometry_type")) != geometry_type: + continue + candidates.append(row) + + fallback: dict[str, Any] | None = None + for row in sort_lookup_candidates(candidates): + conditions = row.get("attribute_conditions_raw") or [] + if not conditions: + if fallback is None: + fallback = row + continue + if all(lookup_condition_matches(c, normalized) for c in conditions): + return row + return fallback + + +def primary_symbol_code(row: dict[str, Any] | None) -> str | None: + if not row: + return None + symbols = row.get("sy") or [] + if not symbols: + return None + return str(symbols[0]) + + +def conditional_procedures(row: dict[str, Any] | None) -> tuple[str, ...]: + if not row: + return () + return tuple(str(x) for x in (row.get("cs") or [])) diff --git a/tools/cm95_s52_database_match_audit.py b/tools/cm95_s52_database_match_audit.py new file mode 100644 index 0000000..a58617f --- /dev/null +++ b/tools/cm95_s52_database_match_audit.py @@ -0,0 +1,138 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +from collections import Counter, defaultdict +import json +import os +from pathlib import Path +from typing import Any + +from dotenv import load_dotenv +import pymysql + +from cm95pbf.s52 import ( + conditional_procedures, + find_best_lookup, + primary_symbol_code, +) + + +DEFAULT_CLASSES = ("LIGHTS", "BCNLAT", "BOYLAT", "TOPMAR", "LNDMRK") + + +def _json_object(value: Any) -> dict[str, Any]: + if value is None: + return {} + if isinstance(value, str): + value = json.loads(value) + return value if isinstance(value, dict) else {} + + +def main() -> int: + parser = argparse.ArgumentParser(description="Audit CM95 DB features against S-52 lookups.") + parser.add_argument("--lookups", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--env-file", default=Path(".env"), type=Path) + parser.add_argument("--dataset-id", default=1, type=int) + parser.add_argument("--table", default="Simplified") + parser.add_argument("--classes", default=",".join(DEFAULT_CLASSES)) + args = parser.parse_args() + + rows = json.loads(args.lookups.read_text(encoding="utf-8")) + by_class: dict[str, list[dict[str, Any]]] = defaultdict(list) + for row in rows: + by_class[str(row.get("object_class") or "")].append(row) + + load_dotenv(args.env_file) + conn = pymysql.connect( + host=os.environ["CM93_DB_HOST"], + port=int(os.environ.get("CM93_DB_PORT", "3306")), + user=os.environ["CM93_DB_USER"], + password=os.environ["CM93_DB_PASSWORD"], + database=os.environ["CM93_DB_NAME"], + charset=os.environ.get("CM93_DB_CHARSET", "utf8mb4"), + autocommit=True, + ) + + classes = [x.strip() for x in args.classes.split(",") if x.strip()] + summary: dict[str, Any] = {} + placeholders = ",".join(["%s"] * len(classes)) + query = ( + "SELECT object_class, attributes_json " + "FROM chart_feature " + "WHERE dataset_id=%s AND object_class IN (" + placeholders + ") " + "ORDER BY object_class, id" + ) + + with conn.cursor() as cur: + cur.execute(query, [args.dataset_id, *classes]) + db_rows = cur.fetchall() + conn.close() + + grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) + for object_class, attrs_raw in db_rows: + grouped[str(object_class)].append(_json_object(attrs_raw)) + + for object_class in classes: + stats = Counter() + symbols = Counter() + csps = Counter() + rcids = Counter() + for attrs in grouped.get(object_class, []): + hit = find_best_lookup( + by_class.get(object_class, []), + attrs, + table_name=args.table, + geometry_type="Point", + ) + if hit is None: + stats["no_lookup"] += 1 + continue + + stats["matched"] += 1 + rcids[str(hit.get("rcid"))] += 1 + + symbol = primary_symbol_code(hit) + if symbol: + stats["static_symbol"] += 1 + symbols[symbol] += 1 + + procedures = conditional_procedures(hit) + if procedures: + stats["conditional"] += 1 + for procedure in procedures: + csps[procedure] += 1 + + if not symbol and not procedures: + stats["empty_instruction_or_non_symbol"] += 1 + + feature_count = len(grouped.get(object_class, [])) + summary[object_class] = { + "feature_count": feature_count, + "stats": dict(stats), + "match_rate": ( + stats["matched"] / feature_count if feature_count else None + ), + "top_symbols": symbols.most_common(20), + "conditional_procedures": dict(csps), + "top_lookup_rcids": rcids.most_common(20), + } + + report = { + "schema": "navsea.cm95.s52-database-match-audit/v1", + "dataset_id": args.dataset_id, + "table_name": args.table, + "classes": classes, + "summary": summary, + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text( + json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" + ) + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/cm95_s52_maplibre_poc.py b/tools/cm95_s52_maplibre_poc.py new file mode 100644 index 0000000..652a95c --- /dev/null +++ b/tools/cm95_s52_maplibre_poc.py @@ -0,0 +1,336 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +from collections import Counter +import json +from pathlib import Path +import re +import shutil +import struct +import xml.etree.ElementTree as ET +from typing import Any + +DEFAULT_TARGETS = ("LIGHTS", "LNDMRK", "BCNLAT", "BOYLAT", "TOPMAR") + + +def _text(node: ET.Element, tag: str) -> str: + child = node.find(tag) + return (child.text or "").strip() if child is not None else "" + + +def _calls(instruction: str, opcode: str) -> list[str]: + pattern = re.compile(rf"(?:^|;)\s*{re.escape(opcode)}\(([^)]*)\)") + out: list[str] = [] + for match in pattern.finditer(instruction or ""): + first = match.group(1).split(",", 1)[0].strip().strip("'\"") + if first: + out.append(first) + return out + + +def _png_size(path: Path) -> tuple[int, int]: + data = path.read_bytes()[:24] + if len(data) < 24 or data[:8] != b"\x89PNG\r\n\x1a\n": + raise RuntimeError(f"not a PNG: {path}") + return struct.unpack(">II", data[16:24]) + + +def _parse_bitmap(node: ET.Element) -> dict[str, int] | None: + bitmap = node.find("bitmap") + if bitmap is None: + return None + loc = bitmap.find("graphics-location") + pivot = bitmap.find("pivot") + if loc is None or pivot is None: + return None + return { + "x": int(loc.attrib["x"]), + "y": int(loc.attrib["y"]), + "width": int(bitmap.attrib["width"]), + "height": int(bitmap.attrib["height"]), + "pivot_x": int(pivot.attrib["x"]), + "pivot_y": int(pivot.attrib["y"]), + } + + +def _parse_symbols(root: ET.Element) -> dict[str, dict[str, Any]]: + out: dict[str, dict[str, Any]] = {} + parent = root.find("symbols") + if parent is None: + return out + for node in parent.findall("symbol"): + name = _text(node, "name") + if not name: + continue + out[name] = { + "name": name, + "description": _text(node, "description"), + "definition": _text(node, "definition") or None, + "color_ref": _text(node, "color-ref") or None, + "bitmap": _parse_bitmap(node), + } + return out + + +def _parse_lookups(root: ET.Element, targets: set[str]) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + parent = root.find("lookups") + if parent is None: + return out + for node in parent.findall("lookup"): + # OpenCPN matches OBCL case-sensitively. Keep standard TOPMAR + # separate from lowercase Inland/extension object names such as topmar. + object_class = node.attrib.get("name", "") + if object_class not in targets: + continue + instruction = _text(node, "instruction") + out.append({ + "id": node.attrib.get("id"), + "rcid": node.attrib.get("RCID"), + "object_class": object_class, + "geometry_type": _text(node, "type"), + "display_priority": _text(node, "disp-prio"), + "radar_priority": _text(node, "radar-prio"), + "table_name": _text(node, "table-name"), + "attribute_conditions_raw": [ + (x.text or "").strip() + for x in node.findall("attrib-code") + if (x.text or "").strip() + ], + "instruction": instruction, + "display_category": _text(node, "display-cat"), + "comment": _text(node, "comment"), + "sy": _calls(instruction, "SY"), + "cs": _calls(instruction, "CS"), + "ap": _calls(instruction, "AP"), + "lc": _calls(instruction, "LC"), + }) + return out + + +def _parse_color_tables(root: ET.Element) -> dict[str, dict[str, Any]]: + out: dict[str, dict[str, Any]] = {} + parent = root.find("color-tables") + if parent is None: + return out + for node in parent.findall("color-table"): + graphics = node.find("graphics-file") + out[node.attrib.get("name", "")] = { + "graphics_file": graphics.attrib.get("name") if graphics is not None else None, + "colors": { + c.attrib["name"]: { + "r": int(c.attrib["r"]), + "g": int(c.attrib["g"]), + "b": int(c.attrib["b"]), + } + for c in node.findall("color") + }, + } + return out + + +def _sprite_entry(bitmap: dict[str, int]) -> dict[str, int]: + return { + "x": bitmap["x"], + "y": bitmap["y"], + "width": bitmap["width"], + "height": bitmap["height"], + "pixelRatio": 1, + } + + +def _icon_offset(bitmap: dict[str, int]) -> list[float]: + return [ + bitmap["width"] / 2.0 - bitmap["pivot_x"], + bitmap["height"] / 2.0 - bitmap["pivot_y"], + ] + + +def _offset_expression(metadata: dict[str, dict[str, Any]]) -> list[Any]: + expr: list[Any] = ["match", ["get", "s52_symbol_code"]] + for name in sorted(metadata): + offset = metadata[name]["maplibre_icon_offset"] + expr.extend([name, ["literal", offset]]) + expr.append(["literal", [0, 0]]) + return expr + + +def _build_style( + source_id: str, + source_layer: str, + sprite_url: str, + tiles_template: str, + metadata: dict[str, dict[str, Any]], +) -> dict[str, Any]: + symbol_names = sorted(metadata) + return { + "version": 8, + "name": "CM95 S-52 MapLibre PoC v1", + "sprite": sprite_url, + "sources": { + source_id: { + "type": "vector", + "tiles": [tiles_template], + "minzoom": 0, + "maxzoom": 12, + } + }, + "layers": [{ + "id": "s52-static-point-symbols", + "type": "symbol", + "source": source_id, + "source-layer": source_layer, + "filter": [ + "all", + ["has", "s52_symbol_code"], + ["match", ["get", "s52_symbol_code"], symbol_names, True, False], + ], + "layout": { + "icon-image": ["get", "s52_symbol_code"], + "icon-anchor": "center", + "icon-offset": _offset_expression(metadata), + "icon-allow-overlap": True, + "icon-ignore-placement": True, + }, + }], + } + + +def _load_config(path: Path | None) -> dict[str, Any]: + if path is None: + return {} + return json.loads(path.read_text(encoding="utf-8")) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Compile OpenCPN S-52 resources into CM95/MapLibre PoC artifacts." + ) + parser.add_argument("--s57data", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--config", type=Path) + parser.add_argument("--targets") + parser.add_argument("--palette") + args = parser.parse_args() + + config = _load_config(args.config) + targets_text = args.targets or ",".join(config.get("targets", DEFAULT_TARGETS)) + targets = {x.strip().upper() for x in targets_text.split(",") if x.strip()} + palette = args.palette or config.get("palette", "DAY_BRIGHT") + source_id = config.get("source_id", "navsea_delivery") + source_layer = config.get("source_layer", "navigation_marks") + sprite_url = config.get("sprite_url", "./sprite") + tiles_template = config.get("tiles_template", "https://REPLACE-ME/{z}/{x}/{y}.pbf") + + xml_path = args.s57data / "chartsymbols.xml" + if not xml_path.is_file(): + raise SystemExit(f"missing chartsymbols.xml: {xml_path}") + root = ET.parse(xml_path).getroot() + symbols = _parse_symbols(root) + lookups = _parse_lookups(root, targets) + color_tables = _parse_color_tables(root) + + if palette not in color_tables: + raise SystemExit(f"unknown palette {palette}; choices={sorted(color_tables)}") + atlas_name = color_tables[palette].get("graphics_file") + if not atlas_name: + raise SystemExit(f"palette {palette} has no graphics-file") + atlas_path = args.s57data / atlas_name + if not atlas_path.is_file(): + raise SystemExit(f"missing atlas: {atlas_path}") + + atlas_width, atlas_height = _png_size(atlas_path) + direct_symbols = sorted({name for row in lookups for name in row["sy"]}) + csp_names = sorted({name for row in lookups for name in row["cs"]}) + metadata: dict[str, dict[str, Any]] = {} + missing_symbols: list[str] = [] + vector_only_symbols: list[str] = [] + out_of_bounds_symbols: list[str] = [] + + for name in direct_symbols: + symbol = symbols.get(name) + if symbol is None: + missing_symbols.append(name) + continue + bitmap = symbol.get("bitmap") + if bitmap is None: + vector_only_symbols.append(name) + continue + if ( + bitmap["x"] < 0 + or bitmap["y"] < 0 + or bitmap["x"] + bitmap["width"] > atlas_width + or bitmap["y"] + bitmap["height"] > atlas_height + ): + out_of_bounds_symbols.append(name) + continue + metadata[name] = { + **symbol, + "maplibre_icon_offset": _icon_offset(bitmap), + } + + output = args.output + output.mkdir(parents=True, exist_ok=True) + shutil.copy2(atlas_path, output / "sprite.png") + sprite_json = { + name: _sprite_entry(metadata[name]["bitmap"]) + for name in sorted(metadata) + } + (output / "sprite.json").write_text( + json.dumps(sprite_json, ensure_ascii=False, indent=2), encoding="utf-8" + ) + (output / "s52_symbol_metadata.json").write_text( + json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8" + ) + (output / "s52_target_lookups.json").write_text( + json.dumps(lookups, ensure_ascii=False, indent=2), encoding="utf-8" + ) + (output / "s52_colors.json").write_text( + json.dumps(color_tables, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + conditional_rows = [row for row in lookups if row["cs"]] + static_rows = [row for row in lookups if row["sy"] and not row["cs"]] + (output / "s52_csp_inventory.json").write_text( + json.dumps(conditional_rows, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + style = _build_style( + source_id, source_layer, sprite_url, tiles_template, metadata + ) + (output / "style.poc.json").write_text( + json.dumps(style, ensure_ascii=False, indent=2), encoding="utf-8" + ) + by_class = Counter(row["object_class"] for row in lookups) + report = { + "schema": "navsea.cm95.s52-maplibre-poc/v1", + "targets": sorted(targets), + "palette": palette, + "atlas": {"file": atlas_name, "width": atlas_width, "height": atlas_height}, + "lookup_count": len(lookups), + "lookup_count_by_object_class": dict(sorted(by_class.items())), + "static_direct_symbol_lookup_count": len(static_rows), + "conditional_lookup_count": len(conditional_rows), + "conditional_procedures": csp_names, + "direct_symbol_reference_count": len(direct_symbols), + "sprite_entry_count": len(sprite_json), + "missing_symbol_definitions": missing_symbols, + "vector_only_symbol_definitions": vector_only_symbols, + "atlas_out_of_bounds": out_of_bounds_symbols, + "production_boundary": { + "static_lookup_matching": "materialize_in_cm95_builder", + "pbf_symbol_field": "s52_symbol_code", + "conditional_symbology": "explicit_builder_derived_or_runtime_renderer", + "formal_release": "not_modified_by_this_poc", + }, + } + (output / "audit.json").write_text( + json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" + ) + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())