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

feat: implement provider-neutral PBF pipeline v1

Browse Source
This commit is contained in:
OpenAI Codex
2026-10-06 09:27:59 +08:00
parent f3da1a6a32
commit aea98628e4
17 changed files with 1604 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -35,3 +35,6 @@ dist/
# Local runtime outputs
report/*.tmp
report/*.cache
.tmp/
/dist/
/tools/visual-object-instantiation/output/

7
STEP_RECORD.md
View File

@@ -2355,3 +2355,10 @@
- native geometry audit：extent 与当前 raw source 不一致 0，未修复 0；报告中的 outside-extent 是当前既有坐标溢出表现，且对象审计确认本次没有新增几何变化。
- 博多/唐津 z10、z12 浏览器视觉截图已完成；左右差异较大主要来自原始 newpec 底图与 delivery 海图样式本身不同，不能单独作为图标替换失败依据。
- 机器可复现入口：`navsea_build_kyushu_semantic_icon_test.py`；对象审计入口：`navsea_audit_kyushu_semantic_icon_test.py`。
## 2026-10-06 PBF Provider Pipeline v1
- 在独立分支实现 Provider-neutral PBF-01～PBF-07，公开 Policy/Package Schema v1、校验器、manifest orchestrator 与 NavSea/CM95 示例配置；没有生成 ProviderExecutionPlan。
- 实例化器接受 provider/dataset 参数，逐 occurrence 输出瓦片、feature identity、属性证据、样式命中与 render components；移除固定 `cm95` source/catalog 名称。
- CM95 通过用户指定 Style、Sprite 与三块真实 PBF 样本执行完成，复用权威人工 assignment；2,077 occurrence、102 个实际 Visual Object，Package 校验 `PASS_WITH_WARNINGS` 并导出。警告保留未映射项和样本限制。
- NavSea 的当前本地 Style 可读，但 Sprite 与三块候选 PBF 对 `192.168.200.184:80` 均返回 connection refused；NavSea 端到端未完成，未伪造样本或 PASS。
- 测试、Python 编译、CLI help 与 `git diff --check` 结果记录在本任务完成报告。

8
contracts/README.md Normal file
View File

@@ -0,0 +1,8 @@
# NavSea Provider 公开 Contract
第三方 Provider 可以自行生产符合契约的包，不需要使用本仓库 Pipeline。公开 schema 定义位于：
- `tools/provider-policy-compiler/schema/provider-policy-v1.schema.json`
- `tools/provider-policy-compiler/schema/provider-package-v1.schema.json`
Policy 与 Package 的固定版本分别为 `navsea-provider-policy/v1`、`navsea-provider-package/v1`。字段包括 Provider/Dataset/Package identity、Style/PBF/Sprite/Glyph 资源、Standard Layer mapping、Display/Pickup/Planning/Danger、capabilities、validation requirements、fingerprints、unresolved、limitations 与 provenance。Package 校验结果仅允许 `PASS`、`PASS_WITH_WARNINGS`、`FAIL`；FAIL 不得导出。

22
providers/cm95/manifest.json Normal file
View File

@@ -0,0 +1,22 @@
{
"schemaVersion": "pbf-provider-source-manifest/v1",
"provider": {"providerId": "cm95", "name": "CM95", "version": "full-style"},
"dataset": {"datasetId": "s52-full", "name": "S-52 Full Style"},
"package": {"packageId": "cm95-s52-full", "version": "1.0.0"},
"resources": {
"style": {"uri": "http://192.168.200.184:18889/s52/full-style.json"},
"sprite": {"uri": "http://192.168.200.184:18889/s52/sprite"}
},
"tileSource": {"template": "http://192.168.200.184:18889/pbf/{z}/{x}/{y}.pbf"},
"sampleTiles": ["12/3531/1641", "12/3526/1643", "12/3638/1613"],
"probeGlyph": false,
"mapping": {"mode": "manual-assignment", "assignmentFile": "../../tools/style-visual-object/output/cm95-manual-review/assignment/cm95-layer-assignment-v1.json"},
"policy": {
"display": {"enabled": true, "source": "provider-style"},
"pickup": {"enabled": false, "rules": []},
"planning": {"enabled": false, "rules": []},
"danger": {"enabled": false, "rules": []},
"capabilities": {"vectorTiles": true, "sprite": true, "glyphs": false, "display": true, "pickup": false, "planning": false, "danger": false}
},
"limitations": ["人工 mapping 依 CM95 resource key 与 occurrence 视觉组件对照；本次实例化仅覆盖 manifest 样本瓦片。"]
}

23
providers/navsea/manifest.json Normal file
View File

@@ -0,0 +1,23 @@
{
"schemaVersion": "pbf-provider-source-manifest/v1",
"provider": {"providerId": "navsea", "name": "NavSea", "version": "1"},
"dataset": {"datasetId": "delivery-full-semantic-extentfix", "name": "Delivery Full Semantic ExtentFix"},
"package": {"packageId": "navsea-delivery-full", "version": "1.0.0"},
"resources": {
"style": {"uri": "/root/sourceserver/pbf/src/pbf/style.navsea-delivery-full-semantic-extentfix.json"},
"sprite": {"uri": "http://192.168.200.184/newpec/sprite-semantic/sprite"},
"glyphs": {"uri": "http://192.168.200.184/newpec/fonts/{fontstack}/{range}.pbf"}
},
"tileSource": {"sourceId": "navsea_delivery"},
"sampleTileCandidates": ["12/3531/1641", "12/3526/1643", "12/3638/1613"],
"probeGlyph": false,
"mapping": {"mode": "frozen-standard-visual", "standardCatalog": "../../tools/style-visual-object/output/navsea-standard/standard-catalog.json"},
"policy": {
"display": {"enabled": true, "source": "provider-style"},
"pickup": {"enabled": false, "rules": []},
"planning": {"enabled": false, "rules": []},
"danger": {"enabled": false, "rules": []},
"capabilities": {"vectorTiles": true, "sprite": true, "glyphs": true, "display": true, "pickup": false, "planning": false, "danger": false}
},
"limitations": ["只使用 manifest 列出的 z12 样本瓦片进行实例化；未据此推断全数据集覆盖率。"]
}

2
requirements-provider-pipeline.txt Normal file
View File

@@ -0,0 +1,2 @@
mapbox-vector-tile>=2.1
Pillow>=10.0

75
tests/test_provider_pipeline.py Normal file
View File

@@ -0,0 +1,75 @@
from __future__ import annotations
import json
import sys
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "tools/provider-policy-compiler"))
from provider_policy import build_policy, export_package, fingerprint, validate_package, validate_policy
class ProviderPipelineContractTests(unittest.TestCase):
def data(self, provider_id: str) -> dict:
return {
"provider": {"providerId": provider_id, "name": provider_id, "version": "1"},
"dataset": {"datasetId": "dataset-1"},
"package": {"packageId": f"{provider_id}-package", "version": "1.0.0"},
"standardCatalog": {"id": "navsea-frozen-standard", "fingerprint": "a" * 64},
"resources": {"style": {"uri": "https://example.invalid/style.json"}, "pbfSamples": [{"tile": "1/0/0", "sha256": "b" * 64}]},
"mapping": [{"visualObjectId": "VO-a", "status": "LAYER_ONLY", "standardLayer": "navigation_marks", "candidateStandardLayers": [], "basis": "review"}],
"display": {}, "pickup": {}, "planning": {}, "danger": {}, "capabilities": {},
"validationRequirements": {"requireStyle": True, "requirePbfSample": True},
"unresolved": [], "limitations": [], "provenance": {},
"fingerprints": {"source": "c" * 64, "style": "d" * 64, "mapping": "e" * 64},
}
def test_fingerprint_is_key_order_independent(self):
self.assertEqual(fingerprint({"a": 1, "b": 2}), fingerprint({"b": 2, "a": 1}))
self.assertEqual(fingerprint({"snapshot": "/one", "generatedAt": "now", "sha256": "abc"}), fingerprint({"snapshot": "/two", "generatedAt": "later", "sha256": "abc"}))
def test_policy_schema_same_for_navsea_and_cm95(self):
navsea = build_policy(self.data("navsea"))
cm95 = build_policy(self.data("cm95"))
self.assertEqual(navsea["schemaVersion"], cm95["schemaVersion"])
self.assertEqual(set(navsea), set(cm95))
self.assertEqual(validate_policy(navsea)["result"], "PASS")
self.assertEqual(validate_policy(cm95)["result"], "PASS")
def test_ambiguous_requires_multiple_candidates(self):
policy = build_policy(self.data("provider-a"))
policy["mapping"]["assignments"] = [{"status": "AMBIGUOUS", "standardLayer": None, "candidateStandardLayers": ["one"]}]
self.assertEqual(validate_policy(policy)["result"], "FAIL")
def test_fail_prevents_export(self):
data = self.data("provider-a")
policy = build_policy(data)
package = {"providerPolicy": policy}
invalid = validate_package(package)
self.assertEqual(invalid["result"], "FAIL")
with tempfile.TemporaryDirectory() as tmp:
with self.assertRaises(ValueError):
export_package(Path(tmp) / "forbidden", package, policy, invalid)
self.assertFalse((Path(tmp) / "forbidden").exists())
def test_export_accepts_only_two_non_fail_results(self):
data = self.data("provider-a")
policy = build_policy(data)
package = {"provider": data["provider"], "dataset": data["dataset"], "package": data["package"], "fingerprints": {}, "limitations": []}
with tempfile.TemporaryDirectory() as tmp:
with self.assertRaises(ValueError):
export_package(Path(tmp) / "unknown", package, policy, {"result": "NOT_FAIL"})
def test_both_provider_manifests_share_policy_shape(self):
manifests = [json.loads((ROOT / f"providers/{key}/manifest.json").read_text()) for key in ("navsea", "cm95")]
self.assertEqual({x["mapping"]["mode"] for x in manifests}, {"frozen-standard-visual", "manual-assignment"})
for manifest in manifests:
self.assertTrue(manifest["provider"]["providerId"])
self.assertIn("style", manifest["resources"])
self.assertIn("sprite", manifest["resources"])
if __name__ == "__main__":
unittest.main()

16
tools/provider-pipeline/README.md Normal file
View File

@@ -0,0 +1,16 @@
# Provider-neutral PBF Pipeline
## 执行
在仓库根目录安装 `requirements-provider-pipeline.txt` 后，执行：
```bash
python tools/provider-pipeline/run_provider_pipeline.py --manifest providers/navsea/manifest.json
python tools/provider-pipeline/run_provider_pipeline.py --manifest providers/cm95/manifest.json
```
也可重复传入 `--sample-tile z/x/y` 覆盖 manifest 中的样本候选；Pipeline 复用 manifest 声明的现有人工 assignment 或冻结 Catalog，不会用自动 matcher 覆盖审核结果。已有视觉目录和实例化目录会从 `.tmp/provider-pipeline/` 复用；需要从头重算时删除对应 work 目录。`--no-export` 只运行到 Package 校验。
PBF-01 保存原 Style、Sprite 和真实 PBF 样本并计算 SHA-256 / source fingerprint。PBF-02 直接复用现有 Style Visual Object Tool。PBF-03 接收 provider/dataset 参数并为每个 feature 写 occurrence 及可追溯证据。PBF-04 从冻结 Standard Catalog 的视觉 hash 对照，或读取声明的人工 assignment。PBF-05/06 使用共享契约和校验逻辑。只有 PASS 或 PASS_WITH_WARNINGS 才能 PBF-07 导出。
输出目录默认 `dist/<provider>/<dataset>/<version>/`，运行缓存位于 `.tmp/`，不纳入 Git。Pipeline 不生成 ProviderExecutionPlan。

317
tools/provider-pipeline/run_provider_pipeline.py Normal file
View File

@@ -0,0 +1,317 @@
#!/usr/bin/env python3
"""从 Provider manifest 运行正式 PBF Pipeline，终点为 NavSea Provider Package。"""
from __future__ import annotations
import argparse
import hashlib
import importlib.util
import json
import shutil
import subprocess
import sys
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "tools" / "provider-policy-compiler"))
from provider_policy import build_package, build_policy, fingerprint, validate_package, validate_policy, export_package
def read_resource(location: str, timeout: int = 60) -> bytes:
if location.startswith(("http://", "https://")):
request = urllib.request.Request(location, headers={"User-Agent": "PBF-Provider-Pipeline/1.0"})
with urllib.request.urlopen(request, timeout=timeout) as response:
return response.read()
return Path(location).expanduser().read_bytes()
def resolved(base: str, relative: str) -> str:
if relative.startswith(("http://", "https://")):
return relative
if base.startswith(("http://", "https://")):
return urllib.parse.urljoin(base, relative)
return str((Path(base).expanduser().parent / relative).resolve())
def sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def snapshot(location: str, target: Path) -> dict[str, Any]:
data = read_resource(location)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(data)
return {"uri": location, "snapshot": str(target), "sha256": sha(data), "bytes": len(data), "available": True}
def source_tile_template(style: dict[str, Any], configured: str | None, source_id: str | None) -> str:
if configured:
return configured
sources = style.get("sources", {})
ordered = [sources[source_id]] if source_id in sources else list(sources.values())
for source in ordered:
if source.get("type") == "vector" and source.get("tiles"):
return source["tiles"][0]
raise ValueError("Style 中没有可用的 vector tile template，且 manifest 未提供 tileSource.template")
def sample_one(template: str, samples: list[str], destination: Path) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
successes: list[dict[str, Any]] = []
failures: list[dict[str, Any]] = []
for sample in samples:
tile = sample.strip().strip("/")
url = template.replace("{z}", tile.split("/")[0]).replace("{x}", tile.split("/")[1]).replace("{y}", tile.split("/")[2])
target = destination / (tile + ".pbf")
try:
item = snapshot(url, target)
item.update({"tile": tile})
if not item["bytes"]:
raise ValueError("空 PBF 响应")
successes.append(item)
except Exception as exc: # each configured candidate is independently auditable
failures.append({"tile": tile, "uri": url, "error": f"{type(exc).__name__}: {exc}"})
return successes, failures
def standard_assignment(actual: dict[str, Any], standard: dict[str, Any]) -> list[dict[str, Any]]:
by_visual: dict[tuple[str, str], set[str]] = {}
ref_by_visual: dict[tuple[str, str], set[str]] = {}
for obj in standard.get("visual_objects", []):
layer = obj.get("standard_layer_name")
vhash = obj.get("visual_hash")
role = obj.get("component_role")
if layer and vhash and role:
by_visual.setdefault((role, vhash), set()).add(layer)
ref_by_visual.setdefault((role, vhash), set()).add(obj.get("visual_object_id"))
rows = []
for obj in actual.get("visual_objects", []):
candidates: set[str] = set()
refs: set[str] = set()
for component in obj.get("components", []):
role = {"icon": "icon", "fill": "fill-pattern", "line": "line-pattern"}.get(component.get("role"), component.get("role"))
vhash = component.get("resource_visual_hash")
if role and vhash:
candidates.update(by_visual.get((role, vhash), set()))
refs.update(ref_by_visual.get((role, vhash), set()))
exact_ref = next(iter(refs)) if len(candidates) == 1 and len(refs) == 1 else None
rows.append(_assignment(obj, sorted(candidates), "FROZEN_STANDARD_VISUAL_HASH", exact_ref))
return rows
def _assignment(obj: dict[str, Any], candidates: list[str], basis: str, standard_visual_object_ref: str | None = None) -> dict[str, Any]:
if len(candidates) == 1:
status, layer = ("VISUAL_OBJECT_MATCH", candidates[0]) if standard_visual_object_ref else ("LAYER_ONLY", candidates[0])
elif len(candidates) > 1:
status, layer = "AMBIGUOUS", None
else:
status, layer = "UNKNOWN", None
return {"visualObjectId": obj["visual_object_id"], "occurrenceIds": [], "status": status, "standardLayer": layer, "standardVisualObjectRef": standard_visual_object_ref, "candidateStandardLayers": candidates if status == "AMBIGUOUS" else [], "basis": basis, "provenance": obj.get("provenance", {})}
def manual_assignment(actual: dict[str, Any], assignment_doc: dict[str, Any]) -> list[dict[str, Any]]:
resource_lookup = {}
for row in assignment_doc.get("resource_summary", []):
resource_lookup[row.get("cm95_resource_key")] = row
object_by_id = {x["visual_object_id"]: x for x in actual.get("visual_objects", [])}
components_by_vo: dict[str, set[str]] = {}
for obj in actual.get("visual_objects", []):
components_by_vo[obj["visual_object_id"]] = {c.get("resource_key") for c in obj.get("components", []) if c.get("resource_key")}
rows = []
for vo_id, obj in object_by_id.items():
evidence = [resource_lookup[key] for key in components_by_vo.get(vo_id, set()) if key in resource_lookup]
layer_candidates = sorted({layer for e in evidence for layer in ([e.get("assigned_standard_layer")] + (e.get("candidate_standard_layers") or [])) if layer})
statuses = {e.get("assignment_status") for e in evidence}
if not evidence or not layer_candidates:
status, layer, candidates = "UNKNOWN", None, []
elif "AMBIGUOUS" in statuses or len(layer_candidates) > 1:
status, layer, candidates = "AMBIGUOUS", None, layer_candidates
elif statuses == {"COMPONENT"}:
status, layer, candidates = "COMPONENT_ONLY", None, []
else:
status, layer, candidates = "LAYER_ONLY", layer_candidates[0], []
exact_resources = {x.get("reviewed_standard_resource_key") for x in evidence if x.get("assignment_status") == "MATCH" and x.get("reviewed_standard_resource_key")}
exact_ref = next(iter(exact_resources)) if len(evidence) == 1 and len(exact_resources) == 1 else None
if status == "LAYER_ONLY" and exact_ref:
status = "VISUAL_OBJECT_MATCH"
rows.append({"visualObjectId": vo_id, "occurrenceIds": [], "status": status, "standardLayer": layer, "standardVisualObjectRef": exact_ref, "candidateStandardLayers": candidates, "basis": "HUMAN_REVIEWED_ASSIGNMENT_RESOURCE_KEY", "manualStatuses": sorted(x for x in statuses if x), "provenance": obj.get("provenance", {})})
return rows
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--manifest", required=True, help="Provider manifest JSON")
ap.add_argument("--work-dir", help="Defaults to .tmp/provider-pipeline/<provider>/<dataset>")
ap.add_argument("--output-dir", help="Defaults to dist/<provider>/<dataset>/<version>")
ap.add_argument("--sample-tile", action="append", help="XYZ tile z/x/y; repeat to supply candidates")
ap.add_argument("--no-export", action="store_true", help="Run validation without exporting a package")
args = ap.parse_args()
manifest_path = Path(args.manifest).resolve()
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
provider = manifest["provider"]
dataset = manifest["dataset"]
package = manifest["package"]
provider_id, dataset_id, version = provider["providerId"], dataset["datasetId"], package["version"]
work = Path(args.work_dir).resolve() if args.work_dir else ROOT / ".tmp" / "provider-pipeline" / provider_id / dataset_id
export_dir = Path(args.output_dir).resolve() if args.output_dir else ROOT / "dist" / provider_id / dataset_id / version
record: dict[str, Any] = {"manifest": str(manifest_path), "stages": {}, "provider": provider, "dataset": dataset, "package": package}
intake = work / "intake"
intake.mkdir(parents=True, exist_ok=True)
style_spec = manifest["resources"]["style"]
style_uri = style_spec["uri"]
style_info = snapshot(style_uri, intake / "style.snapshot.json")
style = json.loads((intake / "style.snapshot.json").read_text(encoding="utf-8"))
template = source_tile_template(style, manifest.get("tileSource", {}).get("template"), manifest.get("tileSource", {}).get("sourceId"))
tile_source = {"uriTemplate": template, "sourceId": manifest.get("tileSource", {}).get("sourceId"), "protocol": "XYZ"}
sprite_base = manifest.get("resources", {}).get("sprite", {}).get("uri") or style.get("sprite")
if not sprite_base:
sprite_info = {"available": False, "uri": None, "reason": "Style 未声明 Sprite"}
else:
sprite_base = resolved(style_uri, sprite_base) if not sprite_base.startswith(("http://", "https://")) else sprite_base
sprite_dir = intake / "sprite"
sprite_dir.mkdir(parents=True, exist_ok=True)
sprite_info = snapshot(sprite_base + ".json", sprite_dir / "sprite.json")
png_info = snapshot(sprite_base + ".png", sprite_dir / "sprite.png")
sprite_info["png"] = png_info
glyph_template = manifest.get("resources", {}).get("glyphs", {}).get("uri") or style.get("glyphs")
glyph_info = {"uriTemplate": glyph_template, "available": bool(glyph_template), "snapshot": None}
if not glyph_template:
glyph_info["reason"] = "Style 未声明 Glyph 资源"
if glyph_template and manifest.get("probeGlyph", False):
probe = glyph_template.replace("{fontstack}", urllib.parse.quote(manifest.get("glyphProbe", {}).get("fontstack", "Noto Sans Regular"), safe="")).replace("{range}", manifest.get("glyphProbe", {}).get("range", "0-255"))
try:
glyph_info.update(snapshot(probe, intake / "glyphs" / "probe.pbf"))
except Exception as exc:
glyph_info.update({"available": False, "error": str(exc)})
samples = args.sample_tile or manifest.get("sampleTiles", []) or manifest.get("sampleTileCandidates", [])
if not samples:
raise ValueError("manifest 未提供 sampleTiles/sampleTileCandidates；请用 --sample-tile 明确指定")
successful_tiles, failed_tiles = sample_one(template, samples, intake / "samples")
source_manifest = {
"schemaVersion": "pbf-source-manifest/v1", "provider": provider, "dataset": dataset, "package": package,
"resources": {"style": style_info, "tileSource": tile_source, "sprite": sprite_info, "glyphs": glyph_info},
"bounds": manifest.get("bounds"), "zoom": manifest.get("zoom", {}),
"sampleTiles": successful_tiles,
"failedSampleTiles": failed_tiles, "semanticSources": manifest.get("semanticSources", []),
}
source_manifest["sourceFingerprint"] = fingerprint({k: v for k, v in source_manifest.items() if k != "sourceFingerprint"})
(intake / "source-manifest.json").write_text(json.dumps(source_manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
record["stages"]["PBF-01"] = {"status": "PASS" if successful_tiles and style_info["available"] and sprite_info.get("available") else "PASS_WITH_WARNINGS", "sourceFingerprint": source_manifest["sourceFingerprint"], "samples": successful_tiles, "failedSamples": failed_tiles}
if not successful_tiles:
raise RuntimeError("PBF-01 未取得任何真实 PBF tile，停止后续阶段")
vo_out = work / "pbf-02-style-visual-object"
if not (vo_out / "anonymous-catalog.json").exists():
spec = importlib.util.spec_from_file_location("style_vo_tool", ROOT / "tools/style-visual-object/style_vo_tool.py")
module = importlib.util.module_from_spec(spec); assert spec and spec.loader; spec.loader.exec_module(module)
catalog = module.extract(str(intake / "style.snapshot.json"), str(intake / "sprite" / "sprite") if sprite_info.get("available") else None, vo_out, f"{provider_id}-anonymous")
else:
catalog = json.loads((vo_out / "anonymous-catalog.json").read_text(encoding="utf-8"))
record["stages"]["PBF-02"] = {"status": "PASS", "visualObjects": catalog["summary"]["visual_object_count"], "unresolved": catalog["summary"]["unresolved_count"]}
if not (work / "pbf-03-actual" / "actual-anonymous-catalog.json").exists():
instance_output = work / "pbf-03-actual"
for item in successful_tiles:
z, x, y = item["tile"].split("/")
dest = instance_output / "tiles" / z / x / f"{y}.pbf"
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(item["snapshot"], dest)
command = [sys.executable, str(ROOT / "tools/visual-object-instantiation/instantiate_pbf_visual_objects.py"), "--style", str(intake / "style.snapshot.json"), "--sprite-json", str(intake / "sprite/sprite.json"), "--sprite-png", str(intake / "sprite/sprite.png"), "--pbf-root", str(instance_output / "tiles"), "--output", str(instance_output), "--provider-id", provider_id, "--dataset-id", dataset_id, "--provider-version", str(provider.get("version", "unknown"))]
if not sprite_info.get("available"):
raise RuntimeError("PBF-03 当前要求 Sprite JSON/PNG 可用")
completed = subprocess.run(command, cwd=ROOT, text=True, capture_output=True)
(instance_output / "instantiation.log").write_text(completed.stdout + completed.stderr, encoding="utf-8")
if completed.returncode:
raise RuntimeError(f"PBF-03 实例化失败，详见 {instance_output / 'instantiation.log'}")
actual = json.loads((work / "pbf-03-actual" / "actual-anonymous-catalog.json").read_text(encoding="utf-8"))
for row in actual.get("visual_objects", []):
matching_occurrences = [x for x in actual.get("occurrences", []) if x["visual_object_id"] == row["visual_object_id"]]
row["occurrence_ids"] = [x["occurrence_id"] for x in matching_occurrences]
record["stages"]["PBF-03"] = {"status": "PASS", **actual["summary"], "occurrenceCount": len(actual.get("occurrences", []))}
mapping_mode = manifest.get("mapping", {}).get("mode", "frozen-standard-visual")
if mapping_mode == "manual-assignment":
assignment_path = (manifest_path.parent / manifest["mapping"]["assignmentFile"]).resolve()
assignment_doc = json.loads(assignment_path.read_text(encoding="utf-8"))
assignments = manual_assignment(actual, assignment_doc)
mapping_source = str(assignment_path)
else:
standard_path = (manifest_path.parent / manifest.get("mapping", {}).get("standardCatalog", "../../tools/style-visual-object/output/navsea-standard/standard-catalog.json")).resolve()
if not standard_path.exists():
standard_path = ROOT / "tools/style-visual-object/output/navsea-standard/standard-catalog.json"
standard = json.loads(standard_path.read_text(encoding="utf-8"))
assignments = standard_assignment(actual, standard)
mapping_source = str(standard_path)
occurrence_lookup: dict[str, list[str]] = {}
for occ in actual.get("occurrences", []):
occurrence_lookup.setdefault(occ["visual_object_id"], []).append(occ["occurrence_id"])
for row in assignments:
row["occurrenceIds"] = occurrence_lookup.get(row["visualObjectId"], [])
mapping_doc = {"schemaVersion": "navsea-standard-layer-mapping/v1", "source": mapping_source, "assignments": assignments}
mapping_path = work / "pbf-04-mapping" / "provider-layer-assignment.json"
mapping_path.parent.mkdir(parents=True, exist_ok=True)
mapping_path.write_text(json.dumps(mapping_doc, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
unresolved = [{"visualObjectId": x["visualObjectId"], "status": x["status"], "candidateStandardLayers": x["candidateStandardLayers"]} for x in assignments if x["status"] in {"AMBIGUOUS", "UNKNOWN", "COMPONENT_ONLY"}]
record["stages"]["PBF-04"] = {"status": "PASS", "mappingMode": mapping_mode, "assignments": len(assignments), "unresolved": len(unresolved), "source": mapping_source}
fingerprints = {
"source": source_manifest["sourceFingerprint"], "style": style_info["sha256"],
"sprite": sprite_info.get("sha256"), "mapping": fingerprint(mapping_doc),
}
if mapping_mode != "manual-assignment":
fingerprints["standardCatalog"] = fingerprint(standard)
data = {
"provider": provider, "dataset": dataset, "package": package,
"standardCatalog": {"id": "navsea-frozen-standard", "schemaVersion": "style-visual-object-standard/v1", "fingerprint": fingerprints.get("standardCatalog")},
"resources": {
"style": {"uri": style_uri, "mediaType": "application/vnd.mapbox.style+json", "sha256": style_info["sha256"], "version": style.get("version")},
"pbfTemplate": {"uriTemplate": template, "protocol": "XYZ"},
"pbfSamples": [{"tile": x["tile"], "uri": x["uri"], "sha256": x["sha256"], "bytes": x["bytes"]} for x in successful_tiles],
"sprite": {"uri": sprite_base, "jsonSha256": sprite_info.get("sha256"), "pngSha256": sprite_info.get("png", {}).get("sha256")},
"glyphs": glyph_info,
},
"mapping": assignments,
"display": manifest.get("policy", {}).get("display", {}), "pickup": manifest.get("policy", {}).get("pickup", {}),
"planning": manifest.get("policy", {}).get("planning", {}), "danger": manifest.get("policy", {}).get("danger", {}),
"capabilities": manifest.get("policy", {}).get("capabilities", {}),
"validationRequirements": {"requireStyle": True, "requirePbfSample": True, "requireSprite": True, "minimumMappingCount": 1},
"unresolved": unresolved, "limitations": manifest.get("limitations", []),
"provenance": {"manifest": str(manifest_path), "mappingSource": mapping_source, "pipelineVersion": "1.0.0", "generatedAt": datetime.now(timezone.utc).isoformat()},
"fingerprints": fingerprints,
}
record["stages"]["PBF-05"] = {"status": "PASS", "schemaVersion": "navsea-provider-policy/v1"}
policy = build_policy(data)
policy_validation = validate_policy(policy)
if policy_validation["result"] == "FAIL":
raise RuntimeError(f"Provider Policy 校验 FAIL: {policy_validation['errors']}")
(work / "pbf-05-policy.json").write_text(json.dumps(policy, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
record["stages"]["PBF-05"]["validation"] = policy_validation
package_doc = build_package(data, policy, policy_validation)
validation = validate_package(package_doc)
package_doc["validation"] = validation
record["stages"]["PBF-06"] = {"status": validation["result"], "schemaVersion": "navsea-provider-package/v1", "validation": validation}
record["validationResult"] = validation["result"]
if not args.no_export:
export_package(export_dir, package_doc, policy, validation)
shutil.copy2(intake / "style.snapshot.json", export_dir / "resources-style.json")
shutil.copy2(intake / "sprite/sprite.json", export_dir / "resources-sprite.json")
shutil.copy2(intake / "sprite/sprite.png", export_dir / "resources-sprite.png")
shutil.copy2(mapping_path, export_dir / "mapping-layer-assignment.json")
record["stages"]["PBF-07"] = {"status": "EXPORTED", "path": str(export_dir)}
else:
record["stages"]["PBF-07"] = {"status": "SKIPPED_BY_REQUEST"}
report_path = work / "pipeline-report.json"
report_path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(json.dumps({"providerId": provider_id, "datasetId": dataset_id, "validationResult": validation["result"], "sampleTiles": [x["tile"] for x in successful_tiles], "output": None if args.no_export else str(export_dir), "report": str(report_path)}, ensure_ascii=False, indent=2))
return 0 if validation["result"] != "FAIL" else 2
if __name__ == "__main__":
try:
raise SystemExit(main())
except Exception as error:
print(f"ERROR: {type(error).__name__}: {error}", file=sys.stderr)
raise SystemExit(2)

9
tools/provider-policy-compiler/README.md Normal file
View File

@@ -0,0 +1,9 @@
# Provider Policy Compiler
该目录定义公开的 Provider Policy v1 与 NavSea Provider Package v1 契约，并提供编译、校验和 FAIL 阻止导出逻辑。NavSea、CM95 与第三方输入共用相同 schema；来源差异由 manifest、mapping、policy 声明和 provenance 表达。
Policy 不包含 ProviderExecutionPlan、Runtime cache、query bucket、运行时 layer handle 或界面对象。最终数据交付物是 NavSea Provider Package。
统一 mapping 状态为 `VISUAL_OBJECT_MATCH`、`LAYER_ONLY`、`COMPONENT_ONLY`、`AMBIGUOUS`、`UNKNOWN`。AMBIGUOUS 至少带两个标准 Layer 候选；UNKNOWN/COMPONENT_ONLY 不得暗填标准 Layer。
执行 `python tools/provider-pipeline/run_provider_pipeline.py --manifest providers/navsea/manifest.json` 可使用本编译器构建相同契约的 Policy 与 Package。

162
tools/provider-policy-compiler/provider_policy.py Normal file
View File

@@ -0,0 +1,162 @@
#!/usr/bin/env python3
"""统一 Provider Policy / Package v1 编译、校验与导出。"""
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
POLICY_SCHEMA = "navsea-provider-policy/v1"
PACKAGE_SCHEMA = "navsea-provider-package/v1"
ASSIGNMENT_STATUSES = {"VISUAL_OBJECT_MATCH", "LAYER_ONLY", "COMPONENT_ONLY", "AMBIGUOUS", "UNKNOWN"}
def canonical(value: Any) -> bytes:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
def stable_projection(value: Any) -> Any:
if isinstance(value, dict):
return {k: stable_projection(v) for k, v in value.items() if k not in {"snapshot", "createdAt", "generatedAt"}}
if isinstance(value, list):
return [stable_projection(v) for v in value]
return value
def fingerprint(value: Any) -> str:
return hashlib.sha256(canonical(stable_projection(value))).hexdigest()
def file_fingerprint(path: Path) -> str:
h = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
def validate_policy(policy: dict[str, Any]) -> dict[str, Any]:
errors: list[str] = []
required = ("schemaVersion", "provider", "dataset", "package", "standardCatalog", "resources", "mapping", "display", "pickup", "planning", "danger", "capabilities", "validationRequirements", "unresolved", "limitations", "provenance", "fingerprints")
errors += [f"missing:{key}" for key in required if key not in policy]
if policy.get("schemaVersion") != POLICY_SCHEMA:
errors.append("schemaVersion:unsupported")
for key in ("provider", "dataset", "package", "standardCatalog", "resources", "mapping", "display", "pickup", "planning", "danger", "capabilities", "validationRequirements", "provenance", "fingerprints"):
if key in policy and not isinstance(policy[key], dict):
errors.append(f"type:{key}:object_required")
for section, field in (("provider", "providerId"), ("dataset", "datasetId"), ("package", "packageId"), ("package", "version")):
value = policy.get(section)
if not isinstance(value, dict) or not isinstance(value.get(field), str) or not value[field].strip():
errors.append(f"identity:{section}.{field}:required")
mapping = policy.get("mapping", {}).get("assignments", []) if isinstance(policy.get("mapping"), dict) else []
if not isinstance(mapping, list):
errors.append("type:mapping.assignments:array_required")
else:
for i, row in enumerate(mapping):
if not isinstance(row, dict) or row.get("status") not in ASSIGNMENT_STATUSES:
errors.append(f"mapping[{i}]:invalid_status")
continue
if row["status"] in {"VISUAL_OBJECT_MATCH", "LAYER_ONLY"} and not row.get("standardLayer"):
errors.append(f"mapping[{i}]:standardLayer_required")
if row["status"] == "VISUAL_OBJECT_MATCH" and not row.get("standardVisualObjectRef"):
errors.append(f"mapping[{i}]:standardVisualObjectRef_required")
if row["status"] in {"AMBIGUOUS", "UNKNOWN", "COMPONENT_ONLY"} and row.get("standardLayer") is not None:
errors.append(f"mapping[{i}]:unresolved_status_cannot_set_standardLayer")
if row["status"] == "AMBIGUOUS" and len(row.get("candidateStandardLayers", [])) < 2:
errors.append(f"mapping[{i}]:ambiguous_requires_candidates")
return {"schemaVersion": "navsea-provider-validation-report/v1", "result": "FAIL" if errors else "PASS", "errors": errors, "warnings": []}
def validate_package(package: dict[str, Any]) -> dict[str, Any]:
errors: list[str] = []
warnings: list[str] = []
required = ("schemaVersion", "provider", "dataset", "package", "resources", "standardLayerMapping", "providerPolicy", "display", "pickup", "planning", "danger", "capabilities", "validationRequirements", "fingerprints", "unresolved", "limitations", "provenance")
errors += [f"missing:{key}" for key in required if key not in package]
if package.get("schemaVersion") != PACKAGE_SCHEMA:
errors.append("schemaVersion:unsupported")
for section, field in (("provider", "providerId"), ("dataset", "datasetId"), ("package", "packageId")):
value = package.get(section)
if not isinstance(value, dict) or not isinstance(value.get(field), str) or not value[field].strip():
errors.append(f"identity:{section}.{field}:required")
package_identity = package.get("package", {})
if not package_identity.get("version"):
errors.append("identity:package.version:required")
policy_result = validate_policy(package.get("providerPolicy", {}))
if policy_result["result"] == "FAIL":
errors.append("providerPolicy:invalid")
requirements = package.get("validationRequirements", {})
if requirements.get("requireStyle") and not package.get("resources", {}).get("style"):
errors.append("resource:style:required")
if requirements.get("requirePbfSample") and not package.get("resources", {}).get("pbfSamples"):
errors.append("resource:pbfSamples:required")
if requirements.get("requireSprite") and not package.get("resources", {}).get("sprite", {}).get("uri"):
errors.append("resource:sprite:required")
if requirements.get("requireGlyphs") and not package.get("resources", {}).get("glyphs", {}).get("uriTemplate"):
errors.append("resource:glyphs:required")
for resource in package.get("resources", {}).get("pbfSamples", []):
if not resource.get("sha256"):
warnings.append("resource:pbfSample:hash_missing")
mapping = package.get("standardLayerMapping", {}).get("assignments", [])
if not mapping:
warnings.append("mapping:empty")
if package.get("unresolved"):
warnings.append("mapping:unresolved_present")
if package.get("limitations"):
warnings.append("limitations:present")
return {"schemaVersion": "navsea-provider-validation-report/v1", "result": "FAIL" if errors else ("PASS_WITH_WARNINGS" if warnings else "PASS"), "errors": errors, "warnings": sorted(set(warnings))}
def build_policy(data: dict[str, Any]) -> dict[str, Any]:
policy = {
"schemaVersion": POLICY_SCHEMA,
"provider": data["provider"], "dataset": data["dataset"], "package": data["package"],
"standardCatalog": data["standardCatalog"], "resources": data["resources"],
"mapping": {"schemaVersion": "navsea-standard-layer-mapping/v1", "assignments": data["mapping"]},
"display": data.get("display", {}), "pickup": data.get("pickup", {}),
"planning": data.get("planning", {}), "danger": data.get("danger", {}),
"capabilities": data.get("capabilities", {}),
"validationRequirements": data.get("validationRequirements", {}),
"unresolved": data.get("unresolved", []), "limitations": data.get("limitations", []),
"provenance": data.get("provenance", {}), "fingerprints": data["fingerprints"],
}
policy["fingerprints"]["policy"] = fingerprint({k: v for k, v in policy.items() if k != "fingerprints"})
return policy
def build_package(data: dict[str, Any], policy: dict[str, Any], validation: dict[str, Any]) -> dict[str, Any]:
return {
"schemaVersion": PACKAGE_SCHEMA,
"provider": data["provider"], "dataset": data["dataset"], "package": data["package"],
"resources": data["resources"],
"standardLayerMapping": policy["mapping"], "providerPolicy": policy,
"display": policy["display"], "pickup": policy["pickup"], "planning": policy["planning"], "danger": policy["danger"],
"capabilities": policy["capabilities"], "validationRequirements": policy["validationRequirements"],
"fingerprints": {**data["fingerprints"], "policy": policy["fingerprints"]["policy"]},
"unresolved": policy["unresolved"], "limitations": policy["limitations"], "provenance": policy["provenance"],
"validation": validation,
}
def export_package(output: Path, package: dict[str, Any], policy: dict[str, Any], validation: dict[str, Any]) -> Path:
if validation.get("result") not in {"PASS", "PASS_WITH_WARNINGS"}:
raise ValueError("只有 PASS 或 PASS_WITH_WARNINGS 可以导出 Provider Package")
output.mkdir(parents=True, exist_ok=True)
(output / "provider-package.json").write_text(json.dumps(package, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
(output / "provider-policy.json").write_text(json.dumps(policy, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
(output / "validation.json").write_text(json.dumps(validation, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
md = ["# Provider Package 验证报告", "", f"- 结果：**{validation['result']}**", f"- Provider：`{package['provider']['providerId']}`", f"- 数据集：`{package['dataset']['datasetId']}`", "", "## 错误", ""]
md += [f"- {x}" for x in validation.get("errors", [])] or ["- 无"]
md += ["", "## 警告", ""]
md += [f"- {x}" for x in validation.get("warnings", [])] or ["- 无"]
(output / "validation.md").write_text("\n".join(md) + "\n", encoding="utf-8")
manifest = {
"manifestSchemaVersion": "navsea-provider-package-manifest/v1",
"provider": package["provider"], "dataset": package["dataset"], "package": package["package"],
"createdAt": datetime.now(timezone.utc).isoformat(), "fingerprints": package["fingerprints"],
"validationResult": validation["result"], "knownLimitations": package["limitations"],
"files": {name: file_fingerprint(output / name) for name in ("provider-package.json", "provider-policy.json", "validation.json", "validation.md")},
}
(output / "MANIFEST.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
return output

15
tools/provider-policy-compiler/schema/provider-package-v1.schema.json Normal file
View File

@@ -0,0 +1,15 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://navsea.example/contracts/provider-package-v1.schema.json",
"title": "NavSea Provider Package v1",
"type": "object",
"required": ["schemaVersion", "provider", "dataset", "package", "resources", "standardLayerMapping", "providerPolicy", "display", "pickup", "planning", "danger", "capabilities", "validationRequirements", "fingerprints", "unresolved", "limitations", "provenance"],
"properties": {
"schemaVersion": {"const": "navsea-provider-package/v1"},
"provider": {"type": "object", "required": ["providerId"]}, "dataset": {"type": "object", "required": ["datasetId"]}, "package": {"type": "object", "required": ["packageId", "version"]},
"resources": {"type": "object"}, "standardLayerMapping": {"type": "object"}, "providerPolicy": {"type": "object", "required": ["schemaVersion"]},
"display": {"type": "object"}, "pickup": {"type": "object"}, "planning": {"type": "object"}, "danger": {"type": "object"}, "capabilities": {"type": "object"},
"validationRequirements": {"type": "object"}, "fingerprints": {"type": "object"}, "unresolved": {"type": "array"}, "limitations": {"type": "array"}, "provenance": {"type": "object"}, "validation": {"type": "object"}
},
"additionalProperties": false
}

18
tools/provider-policy-compiler/schema/provider-policy-v1.schema.json Normal file
View File

@@ -0,0 +1,18 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://navsea.example/contracts/provider-policy-v1.schema.json",
"title": "NavSea Provider Policy v1",
"type": "object",
"required": ["schemaVersion", "provider", "dataset", "package", "standardCatalog", "resources", "mapping", "display", "pickup", "planning", "danger", "capabilities", "validationRequirements", "unresolved", "limitations", "provenance", "fingerprints"],
"properties": {
"schemaVersion": {"const": "navsea-provider-policy/v1"},
"provider": {"type": "object", "required": ["providerId"], "properties": {"providerId": {"type": "string", "minLength": 1}, "name": {"type": "string"}, "version": {"type": "string"}}},
"dataset": {"type": "object", "required": ["datasetId"], "properties": {"datasetId": {"type": "string", "minLength": 1}}},
"package": {"type": "object", "required": ["packageId", "version"], "properties": {"packageId": {"type": "string"}, "version": {"type": "string"}}},
"standardCatalog": {"type": "object"}, "resources": {"type": "object"},
"mapping": {"type": "object", "required": ["schemaVersion", "assignments"], "properties": {"schemaVersion": {"const": "navsea-standard-layer-mapping/v1"}, "assignments": {"type": "array"}}},
"display": {"type": "object"}, "pickup": {"type": "object"}, "planning": {"type": "object"}, "danger": {"type": "object"},
"capabilities": {"type": "object"}, "validationRequirements": {"type": "object"}, "unresolved": {"type": "array"}, "limitations": {"type": "array"}, "provenance": {"type": "object"}, "fingerprints": {"type": "object"}
},
"additionalProperties": false
}

14
tools/visual-object-instantiation/README.md Normal file
View File

@@ -0,0 +1,14 @@
# Visual Object Instantiation
该工具是 Style Visual Object Tool 的下游阶段。
职责：读取已经完成 Style 解析的视觉规则，并用真实 PBF Feature 的实际属性把动态表达式实例化为“实际出现过的 Anonymous Visual Object”。
边界：
- 不给 Layer 命名；
- 不解释 provider 字段语义；
- source-layer 只用于选择对应 Style Layer，并在输出中作为 provenance；
- 不把 source-layer 名称作为 Semantic；
- 不修改 Style Visual Object Tool 的 Provider-neutral 规则。
流程：Style Definition Catalog + PBF -> Actual Anonymous Visual Object Catalog。

544
tools/visual-object-instantiation/instantiate_pbf_visual_objects.py Normal file
View File

@@ -0,0 +1,544 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import hashlib
import json
import math
import time
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any
from mapbox_vector_tile.Mapbox import vector_tile_pb2
from PIL import Image
def canonical(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
def stable_id(prefix: str, value: Any) -> str:
return prefix + "-" + hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()[:16]
def protobuf_value(value: Any) -> Any:
fields = value.ListFields()
if not fields:
return None
return fields[0][1]
def decode_feature_properties(layer: Any, feature: Any) -> dict[str, Any]:
props: dict[str, Any] = {}
tags = feature.tags
for i in range(0, len(tags), 2):
if i + 1 >= len(tags):
break
key_index = tags[i]
value_index = tags[i + 1]
if key_index >= len(layer.keys) or value_index >= len(layer.values):
continue
props[layer.keys[key_index]] = protobuf_value(layer.values[value_index])
return props
def geometry_role(feature_type: int) -> str:
return {1: "Point", 2: "LineString", 3: "Polygon"}.get(int(feature_type), "Unknown")
def numeric(value: Any) -> float | None:
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return float(value)
if isinstance(value, str):
try:
return float(value)
except ValueError:
return None
return None
def evaluate(expr: Any, props: dict[str, Any], zoom: int) -> Any:
if not isinstance(expr, list):
return expr
if not expr:
return expr
head = expr[0]
if not isinstance(head, str):
return [evaluate(x, props, zoom) for x in expr]
if head == "get":
return props.get(expr[1]) if len(expr) > 1 else None
if head == "has":
if len(expr) < 2:
return False
key = expr[1]
return key in props and props.get(key) not in (None, "")
if head == "!":
return not bool(evaluate(expr[1], props, zoom)) if len(expr) > 1 else True
if head == "all":
return all(bool(evaluate(x, props, zoom)) for x in expr[1:])
if head == "any":
return any(bool(evaluate(x, props, zoom)) for x in expr[1:])
if head in {"==", "!=", ">", "<", ">=", "<="}:
if len(expr) < 3:
return False
a = evaluate(expr[1], props, zoom)
b = evaluate(expr[2], props, zoom)
if head == "==":
return a == b
if head == "!=":
return a != b
an = numeric(a)
bn = numeric(b)
if an is not None and bn is not None:
a, b = an, bn
try:
return {
">": a > b,
"<": a < b,
">=": a >= b,
"<=": a <= b,
}[head]
except TypeError:
return False
if head == "literal":
return expr[1] if len(expr) > 1 else None
if head == "to-number":
value = evaluate(expr[1], props, zoom) if len(expr) > 1 else None
n = numeric(value)
if n is not None:
return n
return evaluate(expr[2], props, zoom) if len(expr) > 2 else None
if head == "to-string":
value = evaluate(expr[1], props, zoom) if len(expr) > 1 else None
return "" if value is None else str(value)
if head == "concat":
parts = []
for item in expr[1:]:
value = evaluate(item, props, zoom)
parts.append("" if value is None else str(value))
return "".join(parts)
if head == "coalesce":
for item in expr[1:]:
value = evaluate(item, props, zoom)
if value not in (None, ""):
return value
return None
if head == "case":
clauses = expr[1:]
for i in range(0, len(clauses) - 1, 2):
if bool(evaluate(clauses[i], props, zoom)):
return evaluate(clauses[i + 1], props, zoom)
return evaluate(clauses[-1], props, zoom) if clauses else None
if head == "match":
if len(expr) < 4:
return None
value = evaluate(expr[1], props, zoom)
arms = expr[2:]
fallback = arms[-1]
pairs = arms[:-1]
for i in range(0, len(pairs), 2):
if i + 1 >= len(pairs):
break
labels, result = pairs[i], pairs[i + 1]
if isinstance(labels, list):
matched = value in labels
else:
matched = value == labels
if matched:
return evaluate(result, props, zoom)
return evaluate(fallback, props, zoom)
if head == "zoom":
return zoom
if head == "step":
if len(expr) < 3:
return None
input_value = numeric(evaluate(expr[1], props, zoom))
result = evaluate(expr[2], props, zoom)
if input_value is None:
return result
stops = expr[3:]
for i in range(0, len(stops), 2):
if i + 1 >= len(stops):
break
stop = numeric(evaluate(stops[i], props, zoom))
if stop is None:
continue
if input_value >= stop:
result = evaluate(stops[i + 1], props, zoom)
else:
break
return result
# Unknown operators stay explicit instead of silently becoming truthy.
return {"unsupported_expression": expr}
def visible(layer: dict[str, Any], zoom: int) -> bool:
if zoom < float(layer.get("minzoom", 0)):
return False
maxzoom = layer.get("maxzoom")
if maxzoom is not None and zoom >= float(maxzoom):
return False
if (layer.get("layout") or {}).get("visibility") == "none":
return False
return True
def matches(layer: dict[str, Any], props: dict[str, Any], zoom: int) -> bool:
if not visible(layer, zoom):
return False
filt = layer.get("filter")
if filt is None:
return True
value = evaluate(filt, props, zoom)
return value is True
def clean_value(value: Any) -> Any:
if isinstance(value, float) and math.isfinite(value):
return round(value, 6)
return value
def sprite_hashes(meta: dict[str, Any], atlas: Image.Image) -> dict[str, str]:
out = {}
for key, m in meta.items():
crop = atlas.crop((
int(m["x"]), int(m["y"]),
int(m["x"]) + int(m["width"]),
int(m["y"]) + int(m["height"]),
))
out[key] = hashlib.sha256(crop.tobytes()).hexdigest()
return out
def component_from_layer(
layer: dict[str, Any],
style_index: int,
props: dict[str, Any],
zoom: int,
sprite_meta: dict[str, Any],
visual_hashes: dict[str, str],
) -> list[dict[str, Any]]:
ltype = layer.get("type")
layout = layer.get("layout") or {}
paint = layer.get("paint") or {}
out: list[dict[str, Any]] = []
if ltype == "symbol":
if "icon-image" in layout:
key = evaluate(layout["icon-image"], props, zoom)
if isinstance(key, str) and key:
out.append({
"style_index": style_index,
"style_layer_id": layer.get("id"),
"role": "icon",
"resource_key": key,
"resource_visual_hash": visual_hashes.get(key),
"resource_present": key in sprite_meta,
"icon_size": clean_value(evaluate(layout.get("icon-size", 1.0), props, zoom)),
"icon_rotate": clean_value(evaluate(layout.get("icon-rotate", 0.0), props, zoom)),
"icon_anchor": clean_value(evaluate(layout.get("icon-anchor", "center"), props, zoom)),
})
if "text-field" in layout:
text_value = evaluate(layout["text-field"], props, zoom)
if text_value not in (None, ""):
out.append({
"style_index": style_index,
"style_layer_id": layer.get("id"),
"role": "text",
"text_present": True,
"text_size": clean_value(evaluate(layout.get("text-size"), props, zoom)),
"text_anchor": clean_value(evaluate(layout.get("text-anchor"), props, zoom)),
"text_offset": clean_value(evaluate(layout.get("text-offset"), props, zoom)),
"text_color": clean_value(evaluate(paint.get("text-color"), props, zoom)),
"text_halo_color": clean_value(evaluate(paint.get("text-halo-color"), props, zoom)),
"text_halo_width": clean_value(evaluate(paint.get("text-halo-width"), props, zoom)),
})
elif ltype == "line":
key = evaluate(paint.get("line-pattern"), props, zoom) if "line-pattern" in paint else None
out.append({
"style_index": style_index,
"style_layer_id": layer.get("id"),
"role": "line",
"resource_key": key if isinstance(key, str) and key else None,
"resource_visual_hash": visual_hashes.get(key) if isinstance(key, str) else None,
"resource_present": key in sprite_meta if isinstance(key, str) else None,
"line_color": clean_value(evaluate(paint.get("line-color"), props, zoom)),
"line_width": clean_value(evaluate(paint.get("line-width"), props, zoom)),
"line_dasharray": clean_value(evaluate(paint.get("line-dasharray"), props, zoom)),
"line_opacity": clean_value(evaluate(paint.get("line-opacity"), props, zoom)),
})
elif ltype == "fill":
key = evaluate(paint.get("fill-pattern"), props, zoom) if "fill-pattern" in paint else None
out.append({
"style_index": style_index,
"style_layer_id": layer.get("id"),
"role": "fill",
"resource_key": key if isinstance(key, str) and key else None,
"resource_visual_hash": visual_hashes.get(key) if isinstance(key, str) else None,
"resource_present": key in sprite_meta if isinstance(key, str) else None,
"fill_color": clean_value(evaluate(paint.get("fill-color"), props, zoom)),
"fill_outline_color": clean_value(evaluate(paint.get("fill-outline-color"), props, zoom)),
"fill_opacity": clean_value(evaluate(paint.get("fill-opacity"), props, zoom)),
})
elif ltype == "circle":
out.append({
"style_index": style_index,
"style_layer_id": layer.get("id"),
"role": "circle",
"circle_color": clean_value(evaluate(paint.get("circle-color"), props, zoom)),
"circle_radius": clean_value(evaluate(paint.get("circle-radius"), props, zoom)),
"circle_stroke_color": clean_value(evaluate(paint.get("circle-stroke-color"), props, zoom)),
"circle_stroke_width": clean_value(evaluate(paint.get("circle-stroke-width"), props, zoom)),
})
return out
def signature_components(components: list[dict[str, Any]]) -> list[dict[str, Any]]:
# Exclude provider/resource names and Style layer IDs from the visual signature.
result = []
for c in components:
x = {k: v for k, v in c.items() if k not in {"style_layer_id", "resource_key"}}
result.append(x)
return result
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--style", required=True)
ap.add_argument("--sprite-json", required=True)
ap.add_argument("--sprite-png", required=True)
ap.add_argument("--pbf-root", required=True)
ap.add_argument("--output", required=True)
ap.add_argument("--provider-id", required=True)
ap.add_argument("--dataset-id", required=True)
ap.add_argument("--provider-version", default="unknown")
ap.add_argument("--catalog-name")
ap.add_argument("--progress-every", type=int, default=5000)
args = ap.parse_args()
style = json.loads(Path(args.style).read_text(encoding="utf-8"))
sprite_meta = json.loads(Path(args.sprite_json).read_text(encoding="utf-8"))
atlas = Image.open(args.sprite_png).convert("RGBA")
visual_hashes = sprite_hashes(sprite_meta, atlas)
indexed: dict[str, list[tuple[int, dict[str, Any]]]] = defaultdict(list)
for i, layer in enumerate(style.get("layers", [])):
sl = layer.get("source-layer")
if sl:
indexed[sl].append((i, layer))
root = Path(args.pbf_root)
paths = sorted(root.glob("*/*/*.pbf"))
output = Path(args.output)
output.mkdir(parents=True, exist_ok=True)
catalog_name = args.catalog_name or f"{args.provider_id}-actual-anonymous"
distinct: dict[str, dict[str, Any]] = {}
group_stats: dict[str, Counter] = defaultdict(Counter)
source_group_map: dict[str, str] = {}
layer_seen = Counter()
feature_count = 0
matched_feature_count = 0
unsupported_count = 0
missing_sprite_count = 0
occurrences: list[dict[str, Any]] = []
started = time.time()
for tile_idx, path in enumerate(paths, 1):
parts = path.relative_to(root).parts
z, x, yfile = int(parts[0]), int(parts[1]), parts[2]
y = int(Path(yfile).stem)
try:
decoded = vector_tile_pb2.tile()
decoded.ParseFromString(path.read_bytes())
except Exception as exc:
print(f"DECODE_FAIL {path}: {exc}", flush=True)
continue
for tile_layer in decoded.layers:
layer_name = tile_layer.name
layer_seen[layer_name] += len(tile_layer.features)
style_layers = indexed.get(layer_name, [])
if not style_layers:
continue
group_id = source_group_map.setdefault(
layer_name,
stable_id("GRP", {"provider_id": args.provider_id, "source_layer": layer_name}),
)
for feature in tile_layer.features:
feature_count += 1
props = decode_feature_properties(tile_layer, feature)
components: list[dict[str, Any]] = []
for style_index, style_layer in style_layers:
if not matches(style_layer, props, z):
continue
comps = component_from_layer(
style_layer, style_index, props, z, sprite_meta, visual_hashes
)
for comp in comps:
if isinstance(comp, dict):
if any(
isinstance(v, dict) and "unsupported_expression" in v
for v in comp.values()
):
unsupported_count += 1
if comp.get("resource_key") and comp.get("resource_present") is False:
missing_sprite_count += 1
components.extend(comps)
if not components:
continue
matched_feature_count += 1
visual_signature = {
"geometry_role": geometry_role(feature.type),
"components": signature_components(components),
}
vo_id = stable_id("VO", visual_signature)
feature_id = int(feature.id) if feature.HasField("id") else None
tile_id = f"{z}/{x}/{y}"
occurrence_id = stable_id("OCC", {
"tile": tile_id,
"source_layer": layer_name,
"feature_id": feature_id,
"feature_ordinal": feature_count,
"visual_object_id": vo_id,
})
item = distinct.get(vo_id)
if item is None:
item = {
"visual_object_id": vo_id,
"group_id": group_id,
"geometry_role": visual_signature["geometry_role"],
"components": components,
"occurrence_count": 0,
"provenance": {
"source_layer": layer_name,
},
"example": {
"tile": tile_id,
"feature_id": feature_id,
},
}
distinct[vo_id] = item
item["occurrence_count"] += 1
occurrences.append({
"occurrence_id": occurrence_id,
"visual_object_id": vo_id,
"tile": tile_id,
"feature_id": feature_id,
"feature_ordinal": feature_count,
"source": style_layers[0][1].get("source") if style_layers else None,
"source_layer": layer_name,
"matched_style_layers": [c.get("style_layer_id") for c in components],
"components": signature_components(components),
"geometry_role": geometry_role(feature.type),
"zoom": z,
"field_evidence": decode_feature_properties(tile_layer, feature),
})
group_stats[group_id]["tile_feature_occurrences"] += 1
group_stats[group_id]["distinct_vo_ids"] = 0
if tile_idx % args.progress_every == 0 or tile_idx == len(paths):
elapsed = time.time() - started
print(
f"PROGRESS tiles={tile_idx}/{len(paths)} features={feature_count} "
f"distinct_vo={len(distinct)} elapsed={elapsed:.1f}s",
flush=True,
)
# Count distinct VOs per group after the scan.
per_group_distinct = Counter(item["group_id"] for item in distinct.values())
groups = []
for source_layer, group_id in sorted(source_group_map.items()):
groups.append({
"group_id": group_id,
"distinct_visual_object_count": per_group_distinct[group_id],
"tile_feature_occurrences": group_stats[group_id]["tile_feature_occurrences"],
"provenance": {"source_layer": source_layer},
})
catalog = {
"schema": "actual-visual-object-catalog/v1",
"catalog_name": catalog_name,
"principle": (
"PBF is used only to instantiate Style visual expressions and occurrence combinations; "
"provider semantic names are not used for classification."
),
"inputs": {
"provider_id": args.provider_id,
"dataset_id": args.dataset_id,
"provider_version": args.provider_version,
"style": str(Path(args.style)),
"sprite_json": str(Path(args.sprite_json)),
"sprite_png": str(Path(args.sprite_png)),
"pbf_root": str(root),
"tile_count": len(paths),
},
"summary": {
"decoded_feature_count_in_styled_layers": feature_count,
"matched_feature_occurrence_count": matched_feature_count,
"distinct_visual_object_count": len(distinct),
"group_count": len(groups),
"unsupported_component_value_count": unsupported_count,
"missing_sprite_component_count": missing_sprite_count,
"source_layer_occurrence_counts": dict(layer_seen),
},
"groups": groups,
"visual_objects": sorted(distinct.values(), key=lambda x: x["visual_object_id"]),
"occurrences": occurrences,
}
(output / "actual-anonymous-catalog.json").write_text(
json.dumps(catalog, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
(output / "visual-object-occurrences.json").write_text(
json.dumps({
"schema": "visual-object-occurrences/v1",
"provider_id": args.provider_id,
"dataset_id": args.dataset_id,
"occurrences": occurrences,
}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
(output / "instantiation-report.json").write_text(
json.dumps({
"schema": "visual-object-instantiation-report/v1",
"provider_id": args.provider_id,
"dataset_id": args.dataset_id,
"provider_version": args.provider_version,
"inputs": catalog["inputs"],
"summary": catalog["summary"],
"occurrence_count": len(occurrences),
}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
sanitized = {
"schema": catalog["schema"],
"catalog_name": catalog["catalog_name"],
"occurrences": catalog["occurrences"],
"summary": catalog["summary"],
"groups": [
{k: v for k, v in g.items() if k != "provenance"}
for g in catalog["groups"]
],
"visual_objects": [
{k: v for k, v in obj.items() if k not in {"provenance", "example"}}
for obj in catalog["visual_objects"]
],
}
(output / "matcher-input-sanitized.json").write_text(
json.dumps(sanitized, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
print(json.dumps(catalog["summary"], ensure_ascii=False, indent=2), flush=True)
if __name__ == "__main__":
main()

11
tools/visual-object-matcher/README.md Normal file
View File

@@ -0,0 +1,11 @@
# Visual Object Matcher
输入：
1. 已冻结的 NavSea Standard Visual Object Catalog；
2. Provider 的 Actual Anonymous Visual Object Catalog。
匹配依据只使用视觉结构：Sprite 图像、组件角色、组件顺序、几何角色、line/fill/pattern/text role 等。
禁止使用 Provider source-layer 名称、Provider 名称、对象类名称或字段名称直接决定标准 Layer。
Provider source-layer 仅允许在匹配完成后用于 blind validation / audit，不能进入 matcher input。

358
tools/visual-object-matcher/visual_matcher_experimental.py Normal file
View File

@@ -0,0 +1,358 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import math
from collections import defaultdict
from pathlib import Path
from typing import Any
import numpy as np
from PIL import Image, ImageFilter
def load_catalog(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def load_atlas(catalog_dir: Path) -> Image.Image:
return Image.open(catalog_dir / "_sprite" / "sprite.png").convert("RGBA")
def visual_feature(atlas: Image.Image, meta: dict[str, Any]) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
crop = atlas.crop((
int(meta["x"]), int(meta["y"]),
int(meta["x"]) + int(meta["width"]),
int(meta["y"]) + int(meta["height"]),
))
alpha = np.asarray(crop.getchannel("A"))
ys, xs = np.where(alpha > 8)
if len(xs):
crop = crop.crop((int(xs.min()), int(ys.min()), int(xs.max()) + 1, int(ys.max()) + 1))
w, h = crop.size
scale = min(54.0 / max(w, 1), 54.0 / max(h, 1))
nw = max(1, round(w * scale))
nh = max(1, round(h * scale))
crop = crop.resize((nw, nh), Image.Resampling.LANCZOS)
canvas = Image.new("RGBA", (64, 64), (255, 255, 255, 0))
canvas.alpha_composite(crop, ((64 - nw) // 2, (64 - nh) // 2))
arr = np.asarray(canvas, dtype=np.float32) / 255.0
a = arr[:, :, 3]
rgb = arr[:, :, :3] * a[:, :, None] + (1.0 - a[:, :, None])
edge = np.asarray(canvas.convert("L").filter(ImageFilter.FIND_EDGES), dtype=np.float32) / 255.0
return rgb, a, edge
def cosine(a: np.ndarray, b: np.ndarray) -> float:
aa = a.ravel()
bb = b.ravel()
denom = float(np.linalg.norm(aa) * np.linalg.norm(bb))
return float(np.dot(aa, bb) / denom) if denom else 0.0
def image_similarity(a: tuple[np.ndarray, np.ndarray, np.ndarray], b: tuple[np.ndarray, np.ndarray, np.ndarray]) -> float:
rgb1, alpha1, edge1 = a
rgb2, alpha2, edge2 = b
rgb_score = 1.0 - float(np.mean((rgb1 - rgb2) ** 2))
alpha_score = cosine(alpha1, alpha2)
edge_score = cosine(edge1, edge2)
return 0.45 * rgb_score + 0.35 * alpha_score + 0.20 * edge_score
def unique_sprite_objects(catalog: dict[str, Any]) -> list[dict[str, Any]]:
seen: set[tuple[str, str, str | None]] = set()
out: list[dict[str, Any]] = []
for obj in catalog.get("visual_objects", []):
if not obj.get("resource_key") or not obj.get("resource_meta"):
continue
key = (
obj.get("component_role") or "",
obj.get("visual_hash") or "",
obj.get("standard_layer_name"),
obj.get("group_id"),
)
if key in seen:
continue
seen.add(key)
out.append(obj)
return out
def build_features(catalog: dict[str, Any], catalog_dir: Path) -> tuple[list[dict[str, Any]], list[tuple[np.ndarray, np.ndarray, np.ndarray]]]:
atlas = load_atlas(catalog_dir)
objects = unique_sprite_objects(catalog)
features = [visual_feature(atlas, obj["resource_meta"]) for obj in objects]
return objects, features
def compatible(cm: dict[str, Any], std: dict[str, Any]) -> bool:
if cm.get("component_role") != std.get("component_role"):
return False
cg = cm.get("geometry_role")
sg = std.get("geometry_role")
if cg and sg and cg != sg:
return False
return True
def classify_object(
cm_obj: dict[str, Any],
cm_feat: tuple[np.ndarray, np.ndarray, np.ndarray],
std_objects: list[dict[str, Any]],
std_features: list[tuple[np.ndarray, np.ndarray, np.ndarray]],
) -> dict[str, Any]:
layer_best: dict[str, tuple[float, str]] = {}
for std, feat in zip(std_objects, std_features):
layer = std.get("standard_layer_name")
if not layer or not compatible(cm_obj, std):
continue
score = image_similarity(cm_feat, feat)
prev = layer_best.get(layer)
if prev is None or score > prev[0]:
layer_best[layer] = (score, std["visual_object_id"])
ranked = sorted(
(
{"layer": layer, "score": score, "standard_visual_object_id": vo_id}
for layer, (score, vo_id) in layer_best.items()
),
key=lambda x: x["score"],
reverse=True,
)
top = ranked[0] if ranked else None
second = ranked[1] if len(ranked) > 1 else None
if not top:
status = "UNKNOWN"
else:
margin = top["score"] - (second["score"] if second else 0.0)
if top["score"] >= 0.90 and margin >= 0.025:
status = "MATCH"
elif top["score"] >= 0.82:
status = "PARTIAL"
else:
status = "UNKNOWN"
return {
"visual_object_id": cm_obj["visual_object_id"],
"group_id": cm_obj["group_id"],
"component_role": cm_obj.get("component_role"),
"geometry_role": cm_obj.get("geometry_role"),
"resource_key": cm_obj.get("resource_key"),
"status": status,
"best": top,
"alternatives": ranked[1:4],
}
def group_scores(object_matches: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
for item in object_matches:
grouped[item["group_id"]].append(item)
results: dict[str, dict[str, Any]] = {}
for group_id, items in grouped.items():
evidence: dict[str, list[float]] = defaultdict(list)
for item in items:
for cand in ([item["best"]] if item.get("best") else []) + item.get("alternatives", []):
if cand and cand["score"] >= 0.78:
evidence[cand["layer"]].append(cand["score"])
ranking = []
for layer, scores in evidence.items():
scores = sorted(scores, reverse=True)
strongest = scores[: min(12, len(scores))]
coverage = min(1.0, len(scores) / max(3.0, len(items) * 0.25))
score = (sum(strongest) / len(strongest)) * (0.65 + 0.35 * coverage)
ranking.append({
"layer": layer,
"score": score,
"evidence_count": len(scores),
"best_evidence": scores[:5],
})
ranking.sort(key=lambda x: x["score"], reverse=True)
top = ranking[0] if ranking else None
second = ranking[1] if len(ranking) > 1 else None
if top:
margin = top["score"] - (second["score"] if second else 0.0)
dominance = top["evidence_count"] / max(1, sum(x["evidence_count"] for x in ranking))
if top["score"] >= 0.86 and margin >= 0.035 and dominance >= 0.40:
status = "MATCH"
elif top["score"] >= 0.80:
status = "PARTIAL"
else:
status = "UNKNOWN"
else:
status = "UNKNOWN"
results[group_id] = {
"group_id": group_id,
"status": status,
"best": top,
"alternatives": ranking[1:4],
"object_count": len(items),
}
return results
def apply_group_result(object_matches: list[dict[str, Any]], groups: dict[str, dict[str, Any]]) -> None:
for item in object_matches:
g = groups.get(item["group_id"])
if g and g["status"] == "MATCH":
item["final_layer"] = g["best"]["layer"]
item["final_basis"] = "GROUP_VISUAL_CONSENSUS"
item["final_status"] = "MATCH"
elif item["status"] == "MATCH":
item["final_layer"] = item["best"]["layer"]
item["final_basis"] = "OBJECT_VISUAL_MATCH"
item["final_status"] = "MATCH"
elif item["status"] == "PARTIAL":
item["final_layer"] = item["best"]["layer"] if item.get("best") else None
item["final_basis"] = "OBJECT_VISUAL_PARTIAL"
item["final_status"] = "PARTIAL"
else:
item["final_layer"] = None
item["final_basis"] = "INSUFFICIENT_VISUAL_EVIDENCE"
item["final_status"] = "UNKNOWN"
def blind_validation(
cm_catalog: dict[str, Any],
group_results: dict[str, dict[str, Any]],
) -> dict[str, Any]:
# This is deliberately post-match audit only. Raw source-layer names are read
# after predictions have already been produced.
provenance = {g["group_id"]: g.get("provenance", {}) for g in cm_catalog.get("groups", [])}
rows = []
comparable = correct = 0
for group_id, pred in sorted(group_results.items()):
raw_name = provenance.get(group_id, {}).get("source_layer")
predicted = pred.get("best", {}).get("layer") if pred.get("best") else None
row = {
"group_id": group_id,
"raw_source_layer_for_audit_only": raw_name,
"predicted_standard_layer": predicted,
"status": pred.get("status"),
"same_name_after_blind_match": bool(raw_name and predicted and raw_name == predicted),
}
if raw_name and predicted and not str(raw_name).startswith("s52_"):
comparable += 1
if raw_name == predicted:
correct += 1
rows.append(row)
return {
"note": "Raw CM95 source-layer is used only after matching for blind validation; it is not matcher input.",
"comparable_group_count": comparable,
"same_name_count": correct,
"same_name_rate": (correct / comparable) if comparable else None,
"rows": rows,
}
def html_report(result: dict[str, Any]) -> str:
stats = result["summary"]
val = result["blind_validation"]
rows = []
for g in result["group_matches"]:
best = g.get("best") or {}
alts = ", ".join(f'{x["layer"]} {x["score"]:.3f}' for x in g.get("alternatives", []))
rows.append(
"<tr>"
f"<td><code>{g['group_id']}</code></td>"
f"<td>{g['status']}</td>"
f"<td>{best.get('layer','')}</td>"
f"<td>{best.get('score',0):.3f}</td>"
f"<td>{g['object_count']}</td>"
f"<td>{alts}</td>"
"</tr>"
)
return f"""<!doctype html><meta charset='utf-8'><title>CM95 Visual Object Layer Match</title>
<style>
body{{font-family:Arial,sans-serif;margin:24px;color:#17202a}} table{{border-collapse:collapse;width:100%}}
th,td{{border:1px solid #d8dee6;padding:7px;text-align:left;font-size:13px}} th{{background:#eef2f6}}
code{{font-size:11px}} .box{{display:inline-block;margin:0 12px 12px 0;padding:10px 14px;border:1px solid #d8dee6;border-radius:8px}}
</style>
<h1>CM95 Visual Object → NavSea Standard Layer</h1>
<p>匹配程序不读取 CM95 source-layer 名称；Layer 名仅来自 NavSea 已冻结 Standard Visual Object Catalog。</p>
<div class='box'>CM95 Visual Objects <b>{stats['cm95_sprite_visual_objects']}</b></div>
<div class='box'>MATCH <b>{stats['object_match']}</b></div>
<div class='box'>PARTIAL <b>{stats['object_partial']}</b></div>
<div class='box'>UNKNOWN <b>{stats['object_unknown']}</b></div>
<div class='box'>Blind audit {val['same_name_count']}/{val['comparable_group_count']}</div>
<h2>匿名 Group 的视觉匹配</h2>
<table><thead><tr><th>匿名 Group</th><th>状态</th><th>预测标准 Layer</th><th>分数</th><th>VO 数</th><th>候选</th></tr></thead>
<tbody>{''.join(rows)}</tbody></table>
<p>详细到每个 Visual Object 的匹配、候选和 blind validation 见 <code>match-results.json</code>。</p>
"""
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--standard", required=True)
ap.add_argument("--cm95", required=True)
ap.add_argument("--output", required=True)
args = ap.parse_args()
std_path = Path(args.standard)
cm_path = Path(args.cm95)
output = Path(args.output)
output.mkdir(parents=True, exist_ok=True)
std_catalog = load_catalog(std_path)
cm_catalog = load_catalog(cm_path)
std_objects, std_features = build_features(std_catalog, std_path.parent.parent / "navsea-anonymous")
cm_objects, cm_features = build_features(cm_catalog, cm_path.parent)
# Sanitized match inputs deliberately omit CM95 provenance/source-layer.
sanitized = [
{
"visual_object_id": o["visual_object_id"],
"group_id": o["group_id"],
"component_role": o.get("component_role"),
"geometry_role": o.get("geometry_role"),
"resource_key": o.get("resource_key"),
}
for o in cm_objects
]
(output / "cm95-match-input-sanitized.json").write_text(
json.dumps(sanitized, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
object_matches = [
classify_object(obj, feat, std_objects, std_features)
for obj, feat in zip(cm_objects, cm_features)
]
groups = group_scores(object_matches)
apply_group_result(object_matches, groups)
validation = blind_validation(cm_catalog, groups)
result = {
"schema": "style-visual-object-match/v1",
"method": {
"cm95_source_layer_used_for_matching": False,
"provider_name_used_for_matching": False,
"resource_key_text_semantics_used_for_matching": False,
"visual_inputs": ["sprite pixels", "component role", "geometry role", "anonymous group membership"],
"standard_labels_from": "frozen NavSea Standard Visual Object Catalog",
},
"summary": {
"standard_sprite_visual_objects": len(std_objects),
"cm95_sprite_visual_objects": len(cm_objects),
"object_match": sum(x["final_status"] == "MATCH" for x in object_matches),
"object_partial": sum(x["final_status"] == "PARTIAL" for x in object_matches),
"object_unknown": sum(x["final_status"] == "UNKNOWN" for x in object_matches),
"group_match": sum(x["status"] == "MATCH" for x in groups.values()),
"group_partial": sum(x["status"] == "PARTIAL" for x in groups.values()),
"group_unknown": sum(x["status"] == "UNKNOWN" for x in groups.values()),
},
"group_matches": sorted(groups.values(), key=lambda x: x["group_id"]),
"visual_object_matches": object_matches,
"blind_validation": validation,
}
(output / "match-results.json").write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
(output / "report.html").write_text(html_report(result), encoding="utf-8")
print(json.dumps(result["summary"], ensure_ascii=False, indent=2))
print(json.dumps({k: validation[k] for k in ("comparable_group_count", "same_name_count", "same_name_rate")}, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()