Files
CM95/tools/cm95_s52_maplibre_poc.py
2026-09-19 08:32:51 +08:00

633 lines
22 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
import argparse
from collections import Counter
import json
import math
from pathlib import Path
import re
import shutil
import struct
import subprocess
import tempfile
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_vector(node: ET.Element) -> dict[str, Any] | None:
vector = node.find("vector")
if vector is None:
return None
origin = vector.find("origin")
pivot = vector.find("pivot")
if origin is None or pivot is None:
return None
return {
"width": int(vector.attrib.get("width", "0")),
"height": int(vector.attrib.get("height", "0")),
"origin_x": float(origin.attrib["x"]),
"origin_y": float(origin.attrib["y"]),
"pivot_x": float(pivot.attrib["x"]),
"pivot_y": float(pivot.attrib["y"]),
"hpgl": _text(vector, "HPGL"),
}
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
candidate = {
"name": name,
"description": _text(node, "description"),
"definition": _text(node, "definition") or None,
"color_ref": _text(node, "color-ref") or None,
"bitmap": _parse_bitmap(node),
"vector": _parse_vector(node),
}
current = out.get(name)
if current is None:
out[name] = candidate
continue
# OpenCPN chartsymbols.xml contains duplicate vector/raster definitions
# for several sounding symbols. Prefer a real bitmap when available so
# the MapLibre sprite matches the raster symbol library exactly.
if current.get("bitmap") is None and candidate.get("bitmap") is not None:
out[name] = candidate
elif current.get("bitmap") is None and current.get("vector") is None:
out[name] = candidate
return out
def _parse_patterns(root: ET.Element) -> dict[str, dict[str, Any]]:
out: dict[str, dict[str, Any]] = {}
parent = root.find("patterns")
if parent is None:
return out
for node in parent.findall("pattern"):
name = _text(node, "name")
if not name:
continue
out[name] = {
"name": name,
"description": _text(node, "description"),
"definition": _text(node, "definition") or None,
"filltype": _text(node, "filltype") or None,
"spacing": _text(node, "spacing") 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 _pen_colors(color_ref: str | None, colors: dict[str, Any]) -> dict[str, str]:
compact = "".join(str(color_ref or "").split())
out: dict[str, str] = {}
for i in range(0, len(compact), 6):
token = compact[i : i + 6]
if len(token) != 6:
continue
rgb = colors.get(token[1:])
if isinstance(rgb, dict):
out[token[0]] = (
f"#{int(rgb['r']):02x}{int(rgb['g']):02x}{int(rgb['b']):02x}"
)
return out or {"A": "#000000"}
def _hpgl_points(text: str) -> list[tuple[float, float]]:
nums = [float(x) for x in text.split(",") if x.strip()]
return [(nums[i], nums[i + 1]) for i in range(0, len(nums) - 1, 2)]
def _vector_symbol_svg(
symbol: dict[str, Any],
colors: dict[str, Any],
*,
dpi: float = 96.0,
) -> tuple[str, dict[str, float]]:
vector = symbol.get("vector")
if not isinstance(vector, dict):
raise RuntimeError(f"symbol {symbol.get('name')} has no vector definition")
unit = (dpi / 25.4) / 100.0
pen_colors = _pen_colors(symbol.get("color_ref"), colors)
pen = next(iter(pen_colors))
pen_width = 1.0
current: tuple[float, float] | None = None
primitives: list[tuple[str, Any]] = []
minx = miny = float("inf")
maxx = maxy = float("-inf")
def include(x: float, y: float) -> None:
nonlocal minx, miny, maxx, maxy
minx = min(minx, x)
miny = min(miny, y)
maxx = max(maxx, x)
maxy = max(maxy, y)
for raw in str(vector.get("hpgl") or "").split(";"):
raw = raw.strip()
if not raw:
continue
cmd, arg = raw[:2], raw[2:]
if cmd == "SP" and arg:
pen = arg[0]
elif cmd == "SW":
try:
pen_width = max(1.0, float(arg))
except ValueError:
pen_width = 1.0
elif cmd == "PU":
pts = _hpgl_points(arg)
if pts:
current = pts[-1]
include(*current)
elif cmd == "PD":
pts = _hpgl_points(arg)
for target in pts:
if current is None:
current = target
include(*current)
continue
primitives.append(
("line", (current, target, pen_colors.get(pen, "#000000"), pen_width))
)
include(*current)
include(*target)
current = target
elif cmd == "CI" and current is not None:
radius = float(arg)
primitives.append(
("circle", (current, radius, pen_colors.get(pen, "#000000"), pen_width))
)
include(current[0] - radius, current[1] - radius)
include(current[0] + radius, current[1] + radius)
else:
raise RuntimeError(
f"unsupported HPGL {cmd} in vector symbol {symbol.get('name')}"
)
if not primitives or not math.isfinite(minx) or not math.isfinite(miny):
raise RuntimeError(f"vector symbol {symbol.get('name')} rendered no primitives")
pad = 1.5
width = max(1, int(math.ceil((maxx - minx) * unit + 2 * pad)))
height = max(1, int(math.ceil((maxy - miny) * unit + 2 * pad)))
def cv(point: tuple[float, float]) -> tuple[float, float]:
return (
(point[0] - minx) * unit + pad,
(point[1] - miny) * unit + pad,
)
parts: list[str] = []
for kind, data in primitives:
if kind == "line":
p1, p2, color, width_hpgl = data
x1, y1 = cv(p1)
x2, y2 = cv(p2)
stroke = max(1.0, float(width_hpgl))
parts.append(
f'<line x1="{x1:.3f}" y1="{y1:.3f}" x2="{x2:.3f}" y2="{y2:.3f}" '
f'stroke="{color}" stroke-width="{stroke:.3f}" stroke-linecap="round" />'
)
else:
center, radius, color, width_hpgl = data
cx, cy = cv(center)
r = max(0.5, float(radius) * unit)
stroke = max(1.0, float(width_hpgl))
parts.append(
f'<circle cx="{cx:.3f}" cy="{cy:.3f}" r="{r:.3f}" fill="none" '
f'stroke="{color}" stroke-width="{stroke:.3f}" />'
)
pivot = cv((float(vector["pivot_x"]), float(vector["pivot_y"])))
svg = (
f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" '
f'viewBox="0 0 {width} {height}">' + "".join(parts) + "</svg>"
)
return svg, {
"width": width,
"height": height,
"pivot_x": pivot[0],
"pivot_y": pivot[1],
}
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")
sprite_scope = str(config.get("sprite_scope", "all"))
if sprite_scope not in {"all", "target"}:
raise SystemExit(f"invalid sprite_scope: {sprite_scope}")
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)
patterns = _parse_patterns(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"]})
direct_patterns = sorted({name for row in lookups for name in row["ap"]})
csp_names = sorted({name for row in lookups for name in row["cs"]})
sprite_symbols = (
sorted(symbols)
if sprite_scope == "all"
else direct_symbols
)
sprite_patterns = (
sorted(name for name, pattern in patterns.items() if pattern.get("bitmap") is not None)
if sprite_scope == "all"
else direct_patterns
)
metadata: dict[str, dict[str, Any]] = {}
pattern_metadata: dict[str, dict[str, Any]] = {}
missing_symbols: list[str] = []
vector_only_symbols: list[str] = []
vector_render_names: list[str] = []
out_of_bounds_symbols: list[str] = []
for name in sprite_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)
if symbol.get("vector") is not None:
vector_render_names.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),
}
for name in sprite_patterns:
pattern = patterns.get(name)
if pattern is None:
continue
bitmap = pattern.get("bitmap")
if bitmap is None:
continue
if (
bitmap["x"] < 0
or bitmap["y"] < 0
or bitmap["x"] + bitmap["width"] > atlas_width
or bitmap["y"] + bitmap["height"] > atlas_height
):
continue
pattern_metadata[name] = dict(pattern)
output = args.output
output.mkdir(parents=True, exist_ok=True)
sprite_json = {
name: _sprite_entry(metadata[name]["bitmap"])
for name in sorted(metadata)
}
sprite_json.update({
name: _sprite_entry(pattern_metadata[name]["bitmap"])
for name in sorted(pattern_metadata)
})
if vector_render_names:
if shutil.which("convert") is None:
raise SystemExit("ImageMagick convert is required for vector-only S-52 symbols")
colors = color_tables[palette].get("colors") or {}
with tempfile.TemporaryDirectory(prefix="cm95-s52-symbol-") as td:
tmp = Path(td)
pngs: list[Path] = []
y = atlas_height
for index, name in enumerate(vector_render_names):
symbol = symbols[name]
svg, vector_meta = _vector_symbol_svg(symbol, colors)
svg_path = tmp / f"{index:03d}-{name}.svg"
png_path = tmp / f"{index:03d}-{name}.png"
svg_path.write_text(svg, encoding="utf-8")
subprocess.run(
["convert", "-background", "none", str(svg_path), str(png_path)],
check=True,
)
width, height = _png_size(png_path)
bitmap = {
"x": 0,
"y": y,
"width": width,
"height": height,
"pivot_x": int(round(vector_meta["pivot_x"])),
"pivot_y": int(round(vector_meta["pivot_y"])),
}
materialized = {
**symbol,
"bitmap": bitmap,
"maplibre_icon_offset": _icon_offset(bitmap),
"materialized_from_vector": True,
}
metadata[name] = materialized
sprite_json[name] = _sprite_entry(bitmap)
y += height
pngs.append(png_path)
subprocess.run(
[
"convert",
str(atlas_path),
*[str(p) for p in pngs],
"-background",
"none",
"-gravity",
"west",
"-append",
str(output / "sprite.png"),
],
check=True,
)
else:
shutil.copy2(atlas_path, output / "sprite.png")
collisions = sorted(set(metadata) & set(pattern_metadata))
if collisions:
raise SystemExit(f"sprite symbol/pattern name collision: {collisions}")
(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_pattern_metadata.json").write_text(
json.dumps(pattern_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)
direct_missing = sorted(name for name in direct_symbols if name not in symbols)
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),
"direct_pattern_reference_count": len(direct_patterns),
"sprite_scope": sprite_scope,
"sprite_symbol_entry_count": len(metadata),
"sprite_pattern_entry_count": len(pattern_metadata),
"sprite_entry_count": len(sprite_json),
"missing_symbol_definitions": direct_missing,
"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())