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

feat: implement S-52 hazard presentation

Browse Source
This commit is contained in:
OpenAI Codex
2026-09-17 10:22:56 +08:00
parent ff613884a6
commit 4cbc05315c
5 changed files with 375 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
docs/报告/CM95_S52_危险物阶段报告_v1.md Normal file
View File

@@ -0,0 +1,65 @@
# CM95 S-52 危险物阶段报告 v1
## 范围
本阶段沿 OpenCPN S-52 流程实现危险物条件符号化：
- `OBSTRN04`
- `UDWHAZ03`
- `WRECKS02`
- Point / Line / Area 三种几何分支
- DEPARE / DRGARE 空间关联上下文
基线仍为 Linux 项目 `/root/sourceserver/cm95`。
正式 production release 未修改。
## 实现原则
危险物不能只按对象自身属性选图标。
Builder 先收集当前 scale 的 `DEPARE / DRGARE` 几何与 `DRVAL1/DRVAL2`，
只对 `OBSTRN / UWTROC / WRECKS` 做空间 `intersects`，再把关联水深上下文交给 CSP。
`UDWHAZ03` 根据 safety contour、对象推导深度、`EXPSOU`、`WATLEV` 和关联水深区判断 isolated danger。
## 福冈湾 z12 实测
Tile：`12/3531/1641`
- 危险物相关 Feature：40
- `WRECKS`：21
- `UWTROC`：16
- `OBSTRN`：3
- `OBSTRN04` 已解析：19 / 19
- 未解析 `OBSTRN04`：0
- 本 tile isolated danger：0
- `HAZARD_TILE_CONTRACT=PASS`
本 tile 实际产生的主要符号包括：
- `FOULGND1`
- `DANGER51`
- `DANGER52`
- `UWTROC03`
- `UWTROC04`
- `LNDARE01`
危险物自带的 sounding 继续复用 `SOUNDG03 / SNDFRM02` 符号序列。
## 浏览器验收
测试页：`presentation_v7.html`
Chrome DevTools Protocol 实测状态：
- 页面标题进入 `READY CM95 福冈湾 S52`
- 危险物 S-52：44
- `OBSTRN04 已解析`：19
- `isolated danger`：0
- LIGHTS05、SOUNDG、深度色、等深线仍保持正常
折叠 HUD 后截图确认危险物 S-52 symbol 已实际绘制，不是仅写入 PBF 属性。
## 已知资源闭合缺口
OpenCPN 当前 `s52cnsy.cpp` 的 `OBSTRN04` 会引用 `DANGER53`，
但当前 `chartsymbols.xml` 中没有同名 Symbol 定义。
福冈湾本 tile 未触发该分支，因此不阻塞本阶段。
CM95 对该缺口保持 fail-closed，不自行创造替代图标。

29
src/cm95pbf/builder.py
View File

@@ -1708,10 +1708,18 @@ class Cm95TileBuilder:
floating_relations: set[Any] = set()
floating_points: set[Any] = set()
contour_values: list[float] = []
depth_areas: list[dict[str, Any]] = []
safety_contour = (
self.s52.settings.safety_contour_m if self.s52 is not None else None
)
for row in rows:
if row.object_class in {"DEPARE", "DRGARE"} and row.geometry_type == "AREA":
depth_areas.append({
"geom": row.geom_wgs84,
"geometry_type": row.geometry_type,
"drval1": row.attrs.get("DRVAL1"),
"drval2": row.attrs.get("DRVAL2"),
})
if row.object_class == "DEPCNT" and safety_contour is not None:
try:
value = float(row.attrs.get("VALDCO"))
@@ -1733,6 +1741,7 @@ class Cm95TileBuilder:
"floating_relations": floating_relations,
"floating_points": floating_points,
"next_safe_contour_m": min(contour_values) if contour_values else None,
"depth_areas": depth_areas,
}
@staticmethod
@@ -1745,6 +1754,25 @@ class Cm95TileBuilder:
return (float(geom.x), float(geom.y)) in context["floating_points"]
return False
def _associated_depth_ranges(self, row: SourceFeature) -> list[dict[str, Any]]:
if row.object_class not in {"OBSTRN", "UWTROC", "WRECKS"}:
return []
result: list[dict[str, Any]] = []
for item in self._s52_context.get("depth_areas", []):
geom = item.get("geom")
try:
intersects = geom is not None and geom.intersects(row.geom_wgs84)
except Exception:
intersects = False
if not intersects:
continue
result.append({
"geometry_type": item.get("geometry_type"),
"drval1": item.get("drval1"),
"drval2": item.get("drval2"),
})
return result
def _s52_properties(self, row: SourceFeature) -> dict[str, Any]:
if self.s52 is None:
return {}
@@ -1753,6 +1781,7 @@ class Cm95TileBuilder:
row.geometry_type,
row.attrs,
next_safe_contour_m=self._s52_context.get("next_safe_contour_m"),
associated_depth_ranges=self._associated_depth_ranges(row),
)
def _append_s52_fallback_output(

274
src/cm95pbf/csp.py
View File

@@ -372,14 +372,276 @@ def resolve_soundg03_instruction(
}
return instruction, extra
def _int_attr(attrs: dict[str, Any], key: str, default: int = -9) -> int:
try:
return int(attrs.get(key))
except (TypeError, ValueError):
return default
def _hazard_depth_context(
associated_depth_ranges: list[dict[str, Any]] | None,
) -> list[dict[str, Any]]:
return list(associated_depth_ranges or [])
def resolve_udwhaz03_instruction(
attrs: dict[str, Any],
depth_value: float | None,
settings: S52MarinerSettings,
*,
associated_depth_ranges: list[dict[str, Any]] | None = None,
) -> tuple[str, dict[str, Any]]:
danger = False
expsou = _int_attr(attrs, "EXPSOU", 0)
if depth_value is None and expsou != 1:
danger = True
if not danger and (
expsou == 1
or (depth_value is not None and depth_value <= settings.safety_contour_m)
):
for item in _hazard_depth_context(associated_depth_ranges):
primitive = str(item.get("geometry_type") or "AREA").upper()
try:
drval1 = float(item.get("drval1") or 0.0)
except (TypeError, ValueError):
drval1 = 0.0
try:
drval2 = float(item.get("drval2") or 0.0)
except (TypeError, ValueError):
drval2 = 0.0
if primitive == "LINE":
if drval2 < settings.safety_contour_m:
danger = True
break
elif drval1 >= settings.safety_contour_m and expsou != 1:
danger = True
break
extra = {
"s52_isolated_danger": int(danger),
"s52_display_base_promoted": int(danger),
}
if depth_value is not None:
extra["s52_hazard_depth_m"] = float(depth_value)
if not danger:
return "", extra
watlev = _int_attr(attrs, "WATLEV", 0)
if watlev in {1, 2}:
return "", extra
return "SY(ISODGR51)", extra
def _area_least_depth(
associated_depth_ranges: list[dict[str, Any]] | None,
) -> float | None:
values: list[float] = []
for item in _hazard_depth_context(associated_depth_ranges):
try:
values.append(float(item.get("drval1")))
except (TypeError, ValueError):
pass
return min(values) if values else None
def resolve_obstrn04_instruction(
object_class: str,
geometry_type: str,
attrs: dict[str, Any],
settings: S52MarinerSettings,
*,
associated_depth_ranges: list[dict[str, Any]] | None = None,
) -> tuple[str, dict[str, Any]]:
valsou = _optional_float(attrs.get("VALSOU"))
depth_value = valsou
if depth_value is None and str(geometry_type).upper() == "AREA":
depth_value = _area_least_depth(associated_depth_ranges)
if depth_value is None:
catobs = _int_attr(attrs, "CATOBS", 0)
watlev = _int_attr(attrs, "WATLEV", 0)
expsou = _int_attr(attrs, "EXPSOU", 0)
if expsou != 1:
if catobs == 6:
depth_value = 0.01
elif watlev == 5:
depth_value = 0.0
elif watlev == 3:
depth_value = 0.01
else:
depth_value = -15.0
udw, extra = resolve_udwhaz03_instruction(
attrs,
depth_value,
settings,
associated_depth_ranges=associated_depth_ranges,
)
extra["s52_hazard_procedure"] = "OBSTRN04"
geom = str(geometry_type).upper()
if geom == "POINT":
if udw:
return udw, extra
sounding = False
watlev = _int_attr(attrs, "WATLEV", -9)
if valsou is not None:
if valsou <= 20.0:
if object_class == "UWTROC":
if watlev in {4, 5}:
instruction = "SY(UWTROC04)"
else:
instruction = "SY(DANGER51)"
sounding = True
else:
if watlev in {1, 2}:
instruction = "SY(LNDARE01)"
elif watlev == 3:
instruction = "SY(DANGER52)"
sounding = True
elif watlev in {4, 5}:
instruction = "SY(DANGER53)"
sounding = True
else:
instruction = "SY(DANGER51)"
sounding = True
else:
instruction = "SY(DANGER52)"
sounding = True
else:
if object_class == "UWTROC":
instruction = (
"SY(LNDARE01)" if watlev == 2 else
"SY(UWTROC03)" if watlev == 3 else
"SY(UWTROC04)"
)
else:
instruction = (
"SY(OBSTRN11)" if watlev in {1, 2} else
"SY(OBSTRN03)" if watlev in {4, 5} else
"SY(OBSTRN01)"
)
if sounding and valsou is not None:
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
instruction += ";" + snd if snd else ""
extra.update(snd_extra)
return instruction, extra
if geom == "LINE":
quapos = _int_attr(attrs, "QUAPOS", 0)
if 2 <= quapos < 10:
return ("LC(LOWACC41)" if udw else "LC(LOWACC31)"), extra
if udw:
return "LS(DOTT,2,CHBLK)", extra
if valsou is None or valsou <= 20.0:
return "LS(DOTT,2,CHBLK)", extra
return "LS(DASH,2,CHBLK)", extra
# AREA
if udw:
return "AC(DEPVS);AP(FOULAR01);LS(DOTT,2,CHBLK);" + udw, extra
watlev = _int_attr(attrs, "WATLEV", -9)
if valsou is not None:
line = "LS(DOTT,2,CHBLK)" if valsou <= 20.0 else "LS(DASH,2,CHBLK)"
return line, extra
if watlev in {1, 2}:
return "AC(CHBRN);LS(SOLD,2,CSTLN)", extra
if watlev == 4:
return "AC(DEPIT);LS(DASH,2,CSTLN)", extra
if _int_attr(attrs, "CATOBS", -9) == 6 and watlev in {3, 5}:
return "AC(DEPVS);AP(FOULAR01);LS(DOTT,2,CHBLK)", extra
return "AC(DEPVS);LS(DOTT,2,CHBLK)", extra
def resolve_wrecks02_instruction(
geometry_type: str,
attrs: dict[str, Any],
settings: S52MarinerSettings,
*,
associated_depth_ranges: list[dict[str, Any]] | None = None,
) -> tuple[str, dict[str, Any]]:
valsou = _optional_float(attrs.get("VALSOU"))
watlev = _int_attr(attrs, "WATLEV", -9)
catwrk = _int_attr(attrs, "CATWRK", -9)
depth_value = valsou
if depth_value is None and str(geometry_type).upper() == "AREA":
depth_value = _area_least_depth(associated_depth_ranges)
if depth_value is None:
if catwrk == 1:
depth_value = 20.0
elif catwrk == 2:
depth_value = 0.0
elif catwrk in {4, 5}:
depth_value = -15.0
elif watlev == 3:
depth_value = 0.01
elif watlev == 5:
depth_value = 0.0
else:
depth_value = -15.0
quasou7 = 7 in _int_list(attrs.get("QUASOU"))
if quasou7:
udw, udw_extra = "", {"s52_isolated_danger": 0, "s52_display_base_promoted": 0}
else:
udw, udw_extra = resolve_udwhaz03_instruction(
attrs, depth_value, settings,
associated_depth_ranges=associated_depth_ranges,
)
extra = dict(udw_extra)
extra.update({"s52_hazard_procedure": "WRECKS02", "s52_hazard_depth_m": float(depth_value)})
geom = str(geometry_type).upper()
if geom == "POINT":
if udw:
return udw, extra
if valsou is not None:
symbol = "DANGER51" if valsou < settings.safety_contour_m else "DANGER52"
instruction = f"SY({symbol});TX('Wk',3,1,2,'15110',2,0,CHBLK,21)"
if quasou7:
instruction += ";SY(WRECKS07)"
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
if snd:
instruction += ";" + snd
extra.update(snd_extra)
return instruction, extra
if catwrk == 1 and watlev == 3:
return "SY(WRECKS04)", extra
if catwrk == 2 and watlev == 3:
return "SY(WRECKS05)", extra
if catwrk in {4, 5} or watlev in {1, 2, 4, 5}:
return "SY(WRECKS01)", extra
return "SY(WRECKS05)", extra
quapos = _int_attr(attrs, "QUAPOS", 0)
if 2 <= quapos < 10:
line = "LC(LOWACC41)"
elif udw or valsou is None or valsou <= 20.0:
line = "LS(DOTT,2,CHBLK)" if udw or valsou is not None else "LS(DOTT,2,CSTLN)"
else:
line = "LS(DASH,2,CHBLK)"
instruction = line
if valsou is not None and valsou <= 20.0:
instruction += ";" + udw if udw else ""
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
if snd:
instruction += ";" + snd
extra.update(snd_extra)
elif valsou is None:
color = "CHBRN" if watlev in {1, 2} else "DEPIT" if watlev == 4 else "DEPVS"
instruction += f";AC({color})"
if udw:
instruction += ";" + udw
return instruction, extra
def resolve_csp_hints(
csp_name: str,
object_class: str,
geometry_type: str,
attrs: dict[str, Any],
settings: S52MarinerSettings,
*,
next_safe_contour_m: float | None = None,
sounding_depth_m: float | None = None,
associated_depth_ranges: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
name = str(csp_name or "")[:8]
instruction: str | None = None
@@ -405,6 +667,18 @@ def resolve_csp_hints(
attrs, sounding_depth_m, settings
)
extra.update(sounding_extra)
elif name == "OBSTRN04":
instruction, hazard_extra = resolve_obstrn04_instruction(
object_class, geometry_type=geometry_type,
attrs=attrs, settings=settings,
associated_depth_ranges=associated_depth_ranges,
)
extra.update(hazard_extra)
elif name == "WRECKS02":
instruction, hazard_extra = resolve_wrecks02_instruction(
geometry_type, attrs, settings, associated_depth_ranges=associated_depth_ranges
)
extra.update(hazard_extra)
else:
return {}

4
src/cm95pbf/mapping.py
View File

@@ -48,6 +48,10 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset(
"s52_arc_start_deg",
"s52_arc_width",
"s52_has_text",
"s52_display_base_promoted",
"s52_hazard_depth_m",
"s52_hazard_procedure",
"s52_isolated_danger",
"s52_line_color",
"s52_line_complex_code",
"s52_line_style",

3
src/cm95pbf/s52.py
View File

@@ -341,6 +341,7 @@ class S52LookupRegistry:
*,
next_safe_contour_m: float | None = None,
sounding_depth_m: float | None = None,
associated_depth_ranges: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
row = self.resolve(object_class, geometry_type, attrs)
if row is None:
@@ -364,10 +365,12 @@ class S52LookupRegistry:
resolved = resolve_csp_hints(
procedure,
str(object_class),
str(geometry_type),
normalized_attrs,
self.settings,
next_safe_contour_m=next_safe_contour_m,
sounding_depth_m=sounding_depth_m,
associated_depth_ranges=associated_depth_ranges,
)
if resolved:
out.update(resolved)