tei/CM95
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: integrate S-52 lookup fallback into CM95 builder

Browse Source
This commit is contained in:
OpenAI Codex
2026-09-16 13:23:16 +08:00
parent fb228fb0ed
commit 8e51e388c1
7 changed files with 266 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
config/s52/cm95_s52_poc_v1.json
View File

@@ -16,5 +16,6 @@
"static_lookup": "materialize_in_builder",
"conditional_symbology": "explicit_builder_derived_or_runtime_renderer",
"production_release": "do_not_touch_in_poc"
}
},
"point_table": "Paper"
}

91
docs/报告/CM95_S52_MapLibre_PoC阶段报告_v2.md Normal file
View File

@@ -0,0 +1,91 @@
# CM95 S-52 → MapLibre PoC 阶段报告 v2
日期：2026-09-16
## 阶段结论
S-52 已正式接入 CM95 PBF Builder 的可选路径，并在 Linux + 真实 CM93JAPAN Dataset ID=1 上完成 PBF 对照验证。
第一阶段提交：`fb228fb0ed6345a5bf87915f93bc3e321c63336e`。
本阶段保持 fail-safe：未传入 `--s52-lookups` 时，现有 Builder 行为不变；正式 production release 尚未切换到 S-52。
## 接入内容
- `Cm95TileBuilder` 新增可选 S-52 Lookup Registry；
- CLI `build-tile` / `build-sample-aoi` 新增：
- `--s52-lookups`；
- `--s52-point-table Simplified|Paper`；
- delivery PBF 允许保留 `s52_*` 属性；
- 已有映射对象自动附加 S-52 Lookup/Symbol/CSP 元数据；
- 当前无人工 mapping 的 `BOYLAT` / `TOPMAR` 可通过 S-52 fallback 进入 `navigation_marks`。
## 点符号模式选择
数据库对照显示 `Paper` 更适合当前目标：
- `LIGHTS`：31978 / 31978 → `LIGHTS05`；
- `BCNLAT`：3744 / 3744 静态 Symbol；
- `BOYLAT`：5565 / 5565 静态 Symbol；
- `TOPMAR`：11166 / 11166 命中，其中 5320 为静态 Symbol，5846 为 `TOPMAR01`；
- `Simplified` 模式下 TOPMAR 大多落入空指令默认规则。
因此本阶段默认使用 `Paper`，同时保留 `Simplified` 作为可选模式。
## z10 对照
Tile：`10/914/385`
- control：355 Feature / 15 Layer；
- S-52：355 Feature / 15 Layer；
- 原有 Feature ID 无增删；
- 已有 navigation mark 获得 S-52 元数据；
- `LIGHTS05` CSP 已进入 PBF。
## z12 高密度对照
Tile：`12/3537/1637`
- control：1424 Feature；
- S-52：1489 Feature；
- 净增：65；
- 删除：0；
- 新增 `BOYLAT`：29；
- 新增 `TOPMAR`：36；
- `NO_RULE`：97 → 32；
- `S52_FALLBACK`：65。
新增对象全部进入 `navigation_marks`，保持确定性 FID。
该 tile 中 S-52 字段统计：
- `s52_lookup_rcid`：97；
- `s52_symbol_code`：36；
- `s52_csp`：88；
- `LIGHTS05`：58；
- `TOPMAR01`：30。
## Style / Sprite 闭合
该 z12 tile 实际引用的静态 S-52 Symbol：
- `BCNGEN61`；
- `BOYCAN61`；
- `BOYCON60`；
- `BOYPIL60`；
- `BOYPIL61`；
- `TOPMAR01`。
全部存在于当前 PoC `sprite.json`，缺失数为 0。
验证结论：`s52_style_sprite_tile_contract=PASS`。
## 当前明确未完成
1. `LIGHTS05` 尚未实现为 NavSea 动态 S-52 Renderer；
2. `TOPMAR01` 的平台 rigid/floating 组合逻辑尚未在 Builder/Renderer 执行；
3. `BOYSPH79` 仍是上游 Symbol 定义缺口，继续 fail-closed；
4. CM93 `TOWERS/CHIMNY/MONUMT` → S-57 `LNDMRK` 规范化尚未实施；
5. 正式 production plan / release 尚未启用 S-52 参数。
本阶段验证了技术路线和 Builder 接口，不宣称已完成全部 S-52 CSP。

94
src/cm95pbf/builder.py
View File

@@ -56,6 +56,7 @@ from .policy import (
mercator_to_lonlat,
xyz_tile_bounds,
)
from .s52 import S52LookupRegistry
@dataclass(frozen=True)
@@ -269,6 +270,8 @@ class Cm95TileBuilder:
output_root: Path,
report_root: Path,
profile: str = "delivery",
s52_lookups_path: Path | None = None,
s52_point_table: str = "Paper",
):
self.db = db
self.dataset_id = int(dataset_id)
@@ -278,6 +281,12 @@ class Cm95TileBuilder:
self.output_root = Path(output_root)
self.report_root = Path(report_root)
self.profile = str(profile)
self.s52 = (
S52LookupRegistry(Path(s52_lookups_path), point_table=s52_point_table)
if s52_lookups_path is not None
else None
)
self.s52_point_table = str(s52_point_table)
if self.profile not in {"delivery", "engineering"}:
raise ValueError("profile must be delivery or engineering")
@@ -454,6 +463,14 @@ class Cm95TileBuilder:
rule = self.mapping.resolve(row.object_class, row.geometry_type)
if rule is None:
if self._append_s52_fallback_output(
layer_features,
manifest,
row,
trace_rows,
):
status_counts["S52_FALLBACK"] += 1
continue
status_counts["NO_RULE"] += 1
unmapped_counts[row.object_class] += 1
if self.profile == "engineering":
@@ -1635,6 +1652,23 @@ class Cm95TileBuilder:
}
)
if self.s52 is not None:
base_s52 = self.s52.properties(
base.object_class,
base.geometry_type,
battrs,
)
for key, value in base_s52.items():
props.setdefault(key, value)
light_s52 = self.s52.properties(
light.object_class,
light.geometry_type,
lattrs,
)
if light_s52.get("s52_csp"):
props["s52_csp"] = light_s52["s52_csp"]
clean_props = {
key: value
for key, value in props.items()
@@ -1646,6 +1680,63 @@ class Cm95TileBuilder:
"conflict": name_conflict,
}
def _s52_properties(self, row: SourceFeature) -> dict[str, Any]:
if self.s52 is None:
return {}
return self.s52.properties(
row.object_class,
row.geometry_type,
row.attrs,
)
def _append_s52_fallback_output(
self,
layer_features,
manifest,
row: SourceFeature,
trace_rows,
) -> bool:
if self.s52 is None or row.geometry_type != "POINT":
return False
if row.object_class not in {"BOYLAT", "TOPMAR"}:
return False
s52_props = self._s52_properties(row)
if not s52_props.get("s52_symbol_code") and not s52_props.get("s52_csp"):
return False
source_id = self.source_id(row)
fid = manifest.mvt_id(source_id, ROLE_PRIMARY)
target = "navigation_marks"
props = {
"canonical_object_type": f"s52_{row.object_class.lower()}",
**s52_props,
}
layer_features[target].append(
{
"id": fid,
"geometry": shapely_mapping(row.geom_wgs84),
"properties": self._final_properties(
props,
row,
source_id,
mapping_rule_id=f"S52-FALLBACK-{row.object_class}",
mapping_status="S52_FALLBACK",
),
}
)
trace_rows.append(
{
"output_id": fid,
"output_layer": target,
"source_id": source_id,
"native_scale": row.native_scale,
"mapping_rule_id": f"S52-FALLBACK-{row.object_class}",
"mapping_status": "S52_FALLBACK",
}
)
return True
def _append_main_output(
self,
layer_features,
@@ -1908,6 +1999,9 @@ class Cm95TileBuilder:
mapping_status: str,
extra_engineering: dict[str, Any] | None = None,
) -> dict[str, Any]:
props = dict(props)
for key, value in self._s52_properties(row).items():
props.setdefault(key, value)
props = _sanitize_properties(props)
if self.profile == "delivery":

16
src/cm95pbf/cli.py
View File

@@ -57,6 +57,12 @@ def build_parser() -> argparse.ArgumentParser:
tile.add_argument("--manifest", default=str(DEFAULT_MANIFEST))
tile.add_argument("--output-root", default="build/pbf")
tile.add_argument("--report-root", default="build/pbf-reports")
tile.add_argument("--s52-lookups", default=None)
tile.add_argument(
"--s52-point-table",
choices=["Simplified", "Paper"],
default="Paper",
)
sample = sub.add_parser(
"build-sample-aoi",
@@ -75,6 +81,12 @@ def build_parser() -> argparse.ArgumentParser:
sample.add_argument("--manifest", default=str(DEFAULT_MANIFEST))
sample.add_argument("--output-root", default="build/pbf")
sample.add_argument("--report-root", default="build/pbf-reports")
sample.add_argument("--s52-lookups", default=None)
sample.add_argument(
"--s52-point-table",
choices=["Simplified", "Paper"],
default="Paper",
)
prod_plan = sub.add_parser(
@@ -296,6 +308,8 @@ def main() -> None:
output_root=Path(args.output_root) / args.profile,
report_root=Path(args.report_root),
profile=args.profile,
s52_lookups_path=(Path(args.s52_lookups) if args.s52_lookups else None),
s52_point_table=args.s52_point_table,
)
result = builder.build_tile(
TileJob(args.z, args.x, args.y)
@@ -328,6 +342,8 @@ def main() -> None:
output_root=Path(args.output_root) / profile,
report_root=Path(args.report_root),
profile=profile,
s52_lookups_path=(Path(args.s52_lookups) if args.s52_lookups else None),
s52_point_table=args.s52_point_table,
)
for z, x, y in jobs:
result = builder.build_tile(TileJob(z, x, y))

6
src/cm95pbf/mapping.py
View File

@@ -32,6 +32,12 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset(
"arc_source_code",
"name_ja",
"place_name_en",
"s52_csp",
"s52_display_category",
"s52_display_priority",
"s52_lookup_rcid",
"s52_symbol_code",
"s52_table_name",
"shape_class_code",
}
)

56
src/cm95pbf/s52.py
View File

@@ -150,3 +150,59 @@ 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 []))
class S52LookupRegistry:
"""Read extracted S-52 lookups once and expose PBF-ready properties."""
def __init__(self, path, point_table: str = "Paper"):
import json
from pathlib import Path
self.path = Path(path)
self.point_table = str(point_table)
rows = json.loads(self.path.read_text(encoding="utf-8"))
self.by_class: dict[str, list[dict[str, Any]]] = {}
for row in rows:
object_class = str(row.get("object_class") or "")
self.by_class.setdefault(object_class, []).append(row)
def resolve(
self,
object_class: str,
geometry_type: str,
attrs: dict[str, Any],
) -> dict[str, Any] | None:
geometry = str(geometry_type).upper()
if geometry != "POINT":
return None
return find_best_lookup(
self.by_class.get(str(object_class), []),
attrs,
table_name=self.point_table,
geometry_type="Point",
)
def properties(
self,
object_class: str,
geometry_type: str,
attrs: dict[str, Any],
) -> dict[str, Any]:
row = self.resolve(object_class, geometry_type, attrs)
if row is None:
return {}
out: dict[str, Any] = {
"s52_lookup_rcid": str(row.get("rcid") or ""),
"s52_display_category": str(row.get("display_category") or ""),
"s52_display_priority": str(row.get("display_priority") or ""),
"s52_table_name": str(row.get("table_name") or ""),
}
symbol = primary_symbol_code(row)
if symbol:
out["s52_symbol_code"] = symbol
procedures = conditional_procedures(row)
if procedures:
out["s52_csp"] = ",".join(procedures)
return {key: value for key, value in out.items() if value != ""}

2
tools/cm95_s52_database_match_audit.py
View File

@@ -35,7 +35,7 @@ def main() -> int:
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("--table", default="Paper")
parser.add_argument("--classes", default=",".join(DEFAULT_CLASSES))
args = parser.parse_args()