feat: add S-52 MapLibre compiler poc

This commit is contained in:
OpenAI Codex
2026-09-16 12:27:27 +08:00
parent b52b3447cf
commit fb228fb0ed
5 changed files with 711 additions and 0 deletions

View File

@@ -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"
}
}

View File

@@ -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。

152
src/cm95pbf/s52.py Normal file
View File

@@ -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 []))

View File

@@ -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())

View File

@@ -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())