Compare commits

3 Commits

Author SHA1 Message Date
OpenAI Codex
405f6e01d5 feat: package provider delivery as NavSea zip 2026-10-06 10:28:22 +08:00
OpenAI Codex
804a764109 docs: record provider pipeline v1 completion 2026-10-06 09:31:55 +08:00
OpenAI Codex
aea98628e4 feat: implement provider-neutral PBF pipeline v1 2026-10-06 09:31:05 +08:00
22 changed files with 2667 additions and 5 deletions

3
.gitignore vendored
View File

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

View File

@@ -2355,3 +2355,40 @@
- 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` 结果记录在本任务完成报告。
## 2026-10-06 PBF Provider Pipeline v1 双 Provider 最终实跑补充
- 前一条记录中的 NavSea 阻塞原因已确认:旧 `192.168.200.184:80` LAN 资源端点当前未监听;不是 Pipeline 逻辑或 Codex sandbox 独有问题。
- 保留 X220 当前本地 `style.navsea-delivery-full-semantic-extentfix.json` 作为 NavSea Style 分析输入,不用公开 Style 替换它;仅在 `providers/navsea/manifest.json` 显式覆盖 PBF / Sprite / Glyph 为当前可用的 `https://pbf.jpcn.ccwu.cc/` 同数据集资源入口。
- NavSea 已 fresh 重跑成功:
- 三个 z12 样本 `12/3531/1641`、`12/3526/1643`、`12/3638/1613` 全部下载成功;
- 8,484 occurrence、85 个 distinct Actual Visual Object;
- Mapping:VISUAL_OBJECT_MATCH 27、LAYER_ONLY 7、AMBIGUOUS 11、UNKNOWN 40;
- PBF-01~PBF-05 PASS,PBF-06 `PASS_WITH_WARNINGS`,PBF-07 `EXPORTED`。
- CM95 已清理临时产物后 fresh 重跑成功:
- 三个指定 z12 样本全部成功;
- 2,077 occurrence、102 个 distinct Actual Visual Object;
- Mapping:VISUAL_OBJECT_MATCH 9、LAYER_ONLY 47、AMBIGUOUS 2、UNKNOWN 44;
- PBF-01~PBF-05 PASS,PBF-06 `PASS_WITH_WARNINGS`,PBF-07 `EXPORTED`。
- 两套 warnings 都来自 `mapping:unresolved_present` 与 `limitations:present`;未将 UNKNOWN/AMBIGUOUS 静默映射为标准 Layer。
- 最终双 Provider 端到端结果记录在 `docs/报告/PBF_Provider_Pipeline_v1_Codex完成报告_v2.md`。
## 2026-10-06 PBF-08 NavSea Delivery ZIP
- 在 PBF-07 Provider Package Export 后新增 PBF-08。
- PBF-08 自动生成单文件 `.navsea-provider.zip`,用户交付给 NavSea 时只需要复制该 ZIP,不再手工挑选多个文件。
- ZIP 使用严格 9 文件白名单:provider-package.json、provider-policy.json、mapping-layer-assignment.json、resources-style.json、resources-sprite.json、resources-sprite.png、validation.json、validation.md、MANIFEST.json。
- 任一必需文件缺失时禁止生成 ZIP;ZIP 完成后自动核对精确文件列表、CRC 和 SHA-256,并写入 pipeline-report.json。
- NavSea 实跑生成 `navsea-delivery-full-semantic-extentfix-1.0.0.navsea-provider.zip`,SHA-256 `54aff8d8b882211ced6e77b3a93e8e793c780b44a1faa982ed8fc64693124fd5`,PBF-08 PACKAGED。
- CM95 实跑生成 `cm95-s52-full-1.0.0.navsea-provider.zip`,SHA-256 `603cc004833bcff5571a32843d8ceedc7858b2fc4b28c41056db6807c7d511ee`,PBF-08 PACKAGED。
- 两个 ZIP 均经外部 `unzip -t` 验证 9/9 文件 OK。
- 自动测试从 6 项增加到 8 项,8/8 PASS;新增覆盖严格白名单与缺文件拒绝打包。
- 详细结果:`docs/报告/PBF_Provider_Pipeline_v1_Codex完成报告_v3.md`。

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 不得导出。

View File

@@ -30,6 +30,8 @@ PBF-05 Provider Policy Compile
PBF-06 Provider Package Validate
↓
PBF-07 NavSea Provider Package Export
↓
PBF-08 NavSea Delivery ZIP
```
Pipeline 正式终点是 `NavSea Provider Package`,不是 `ProviderExecutionPlan`。
@@ -370,7 +372,46 @@ MANIFEST 至少包含:
只有 PASS 或明确允许发布的 PASS_WITH_WARNINGS 才能正式 export。
## 10. NavSea Provider Package Contract
## 10. PBF-08 NavSea Delivery ZIP
目标:把 PBF-07 已验证的最终 NavSea 正式输入打成一个单文件 ZIP,避免人工复制时漏文件、错版本或混入生产中间产物。
PBF-08 不改变 Provider Package Contract,只增加交付封装。ZIP 使用严格白名单,只允许包含:
```text
provider-package.json
provider-policy.json
mapping-layer-assignment.json
resources-style.json
resources-sprite.json
resources-sprite.png
validation.json
validation.md
MANIFEST.json
```
默认输出:
```text
dist/<provider-id>/<dataset-id>/
<provider-id>-<dataset-id>-<version>.navsea-provider.zip
```
ZIP 内部使用同名顶层目录,避免多个 Provider 包解压后文件互相覆盖。
Gate:
- PBF-07 必须已经成功导出;
- validation 必须为 PASS 或 PASS_WITH_WARNINGS;
- 白名单中的 9 个文件必须全部存在;
- ZIP 生成后必须重新读取并核对精确文件列表;
- ZIP 必须通过 CRC 校验;
- Pipeline 必须计算 ZIP SHA-256 并写入 pipeline-report.json;
- 任何临时 PBF、`.tmp`、工具源码、测试、人工审核中间文件或未声明文件不得进入 ZIP。
因此用户实际交付给 NavSea 时,只需要复制这一份 `.navsea-provider.zip`。
## 11. NavSea Provider Package Contract
公开 Contract 至少覆盖:
@@ -396,7 +437,7 @@ MANIFEST 至少包含:
公开 Contract 不要求第三方使用官方 PBF 工具。
## 11. NavSea 接收侧
## 12. NavSea 接收侧
```text
NavSea Provider Package
@@ -418,7 +459,7 @@ NavSea Runtime
NavSea 不再负责 CM95/S-52 语义研究、Visual Object 人工分类、Provider-specific 图标研究或 Provider Policy 推理生成。
## 12. 当前工具迁移关系
## 13. 当前工具迁移关系
`tools/style-visual-object/style_vo_tool.py` → PBF-02。
@@ -432,7 +473,7 @@ NavSea 不再负责 CM95/S-52 语义研究、Visual Object 人工分类、Provid
`validate_cm95_classification.py` 暂保留为 CM95 基线校验;后续增加 Provider-neutral validator。
## 13. 实施顺序
## 14. 实施顺序
第一优先级:冻结 Provider Policy Schema v1 和 Provider Package Schema v1。
@@ -442,7 +483,7 @@ NavSea 不再负责 CM95/S-52 语义研究、Visual Object 人工分类、Provid
第四优先级:调整 NavSea `help/chart-profile-tools/`,收敛为读取公开 Provider Package → Validate → Profile → ProviderCompiler。
## 14. 核心原则
## 15. 核心原则
```text
PBF = 数据生产者

View File

@@ -0,0 +1,276 @@
# PBF Provider Pipeline v1 实现任务
日期:2026-10-06
## 0. 任务目标
在当前独立 worktree `/root/sourceserver/pbf-provider-pipeline-v1` 中,把 `docs/PBF_Pipeline_正式流程_v1.md` 定义的 PBF-01 ~ PBF-07 落成可运行、Provider-neutral 的正式 Pipeline,并实际用 NavSea 与 CM95 两套 Style/资源跑通。
最终终点必须是:
```text
NavSea Provider Package
```
不得生成 NavSea Runtime 的 `ProviderExecutionPlan`。
## 1. 开工前必须读取
先完整阅读:
- `AGENTS.md`
- `docs/PBF_Pipeline_正式流程_v1.md`
- `STEP_RECORD.md`
- `tools/style-visual-object/README.md`
- `tools/style-visual-object/style_vo_tool.py`
- `tools/style-visual-object/output/navsea-standard/standard-catalog.json`
- `tools/style-visual-object/output/cm95-manual-review/assignment/cm95-layer-assignment-v1.json`
- `tools/style-visual-object/output/cm95-manual-review/validation-report-v1.json`
- `tools/visual-object-instantiation/README.md`
- `tools/visual-object-instantiation/instantiate_pbf_visual_objects.py`
- `tools/visual-object-matcher/README.md`
- `tools/visual-object-matcher/visual_matcher_experimental.py`
先按真实文件结构和字段工作,不凭记忆臆造。
## 2. 必做项
- [ ] PBF-01:实现 Provider Source Intake,支持本地路径和 HTTP(S) Style;记录 provider/dataset identity、资源、hash/fingerprint、source snapshot。
- [ ] PBF-02:复用现有 Style Visual Object Tool;必要时补薄包装,但不得破坏已冻结 NavSea Standard Catalog。
- [ ] PBF-03:把 `visual-object-instantiation` 正规化并 Provider-neutral 化。删除主逻辑中的 CM95 硬编码,包括固定 `source: cm95`、固定 catalog_name 等;Provider 信息必须来自参数/manifest。支持对选定 XYZ sample tiles 下载到工作目录后实例化。不得按屏幕位置拼接对象。
- [ ] PBF-04:提供统一 Mapping ingestion/normalization。CM95 读取现有人工 assignment;NavSea 从 frozen Standard Catalog 的真实结构生成自身 Provider mapping。自动 matcher 只能作为建议,不能覆盖人工结果。
- [ ] PBF-05:新增 `tools/provider-policy-compiler/`,定义并实现统一 Provider Policy Schema v1、编译器、validator。NavSea 与 CM95 必须使用完全相同 Schema;主逻辑禁止 `provider == cm95` 等分支。
- [ ] PBF-06:实现 Provider Package Schema v1 与 Validator,结果严格为 PASS / PASS_WITH_WARNINGS / FAIL;FAIL 禁止 export。
- [ ] PBF-07:实现正式 Provider Package Export,并生成 MANIFEST / fingerprints / validation report。
- [ ] 新增一个 Provider-neutral 端到端 CLI/orchestrator,单条命令可从 manifest 执行 Pipeline。应能选择阶段或复用已存在的审核/assignment 输入,但最终输出结构一致。
- [ ] 把两个当前未跟踪的 `tools/visual-object-instantiation/`、`tools/visual-object-matcher/` 中本任务需要的源码正规纳入版本控制;不要提交 `__pycache__`。
- [ ] 增加最小但足够的自动测试,覆盖 schema、fingerprint 稳定性、Provider-neutral 行为、FAIL 阻止 export、NavSea/CM95 两套 contract 同构。
- [ ] 文档全部中文;工具字段名/Schema key 可用英文。
- [ ] 更新 `STEP_RECORD.md`,但只追加本任务进度,禁止重写历史。
- [ ] 生成中文完成报告到 `docs/报告/PBF_Provider_Pipeline_v1_Codex完成报告_v1.md`。
## 3. 公开 Contract 的最低字段
Provider Package/Policy 至少覆盖:
- provider identity
- dataset identity
- package/version identity
- Style resource
- PBF resource
- Sprite
- Glyph
- NavSea Standard Layer mapping
- Provider Policy
- Display
- Pickup
- Planning
- Danger
- capabilities
- validation requirements
- hash / version / fingerprint
- unresolved
- limitations
- provenance
第三方不需要使用本仓库工具,只要满足公开 Contract 即可。
## 4. Provider-neutral 硬要求
正式 Pipeline 主逻辑不得:
- 根据 providerId 写 `if cm95` / `if navsea`
- 根据 source-layer 名称直接决定 NavSea Standard Layer
- 根据 Provider 字段名直接决定 NavSea Standard Layer
- 根据单个 icon 名称硬推完整对象语义
- 静默把 AMBIGUOUS/UNKNOWN 变成某个 Layer
Provider-specific 差异必须进入:
- manifest
- declarative mapping/policy
- provenance
- 已确认人工 assignment
## 5. 实际运行输入
### NavSea
Style 优先使用 X220 当前真实本地文件:
```text
/root/sourceserver/pbf/src/pbf/style.navsea-delivery-full-semantic-extentfix.json
```
该文件是实际当前 Style 输入,只用于运行测试,不要把原主工作区文件纳入本 worktree Git。
其 Style 当前包含:
- vector source `navsea_delivery`
- tile template `http://192.168.200.184/pbf-delivery-full-20260418-rebuild/{z}/{x}/{y}.pbf`
- sprite `http://192.168.200.184/newpec/sprite-semantic/sprite`
- glyphs `http://192.168.200.184/newpec/fonts/{fontstack}/{range}.pbf`
冻结 Standard Catalog:
```text
tools/style-visual-object/output/navsea-standard/standard-catalog.json
```
### CM95
Style:
```text
http://192.168.200.184:18889/s52/full-style.json
```
PBF:
```text
http://192.168.200.184:18889/pbf/{z}/{x}/{y}.pbf
```
Sprite:
```text
http://192.168.200.184:18889/s52/sprite
```
建议 sample tiles:
```text
12/3531/1641
12/3526/1643
12/3638/1613
```
CM95 人工权威 assignment:
```text
tools/style-visual-object/output/cm95-manual-review/assignment/cm95-layer-assignment-v1.json
```
## 6. NavSea sample tile
从 NavSea Style 的真实 vector source 自动获取 tile template。若 manifest 未指定 sample tile,CLI 应允许显式传 sample tiles。
为了本任务实际跑通,先通过 HEAD/GET 或目录可用性选择 1~3 个能成功下载的 NavSea z12 sample tile;不得伪造成功。如果某候选 404,继续选择真实存在的 tile,并在报告记录最终使用的 tile。
## 7. 依赖与运行
当前复制进 worktree 的旧 instantiation 脚本依赖 `mapbox_vector_tile` 和 Pillow;当前系统 Python 已观察到 `mapbox_vector_tile` 可能未安装。
要求:
- 明确声明 Python 依赖,例如独立 requirements 文件。
- 优先创建/使用 worktree 内 `.venv`,不要污染系统 Python。
- 依赖安装失败时明确报告,不得伪造运行成功。
- 输出/缓存目录应通过 .gitignore 避免把大体积临时 PBF/Sprite/cache 提交进 Git。
- 正式示例/小型 JSON 报告可提交;大 PBF、PNG cache 不提交。
## 8. 两套实际端到端验收
实现完成后必须实际运行:
### A. NavSea
目标:
```text
NavSea style + real PBF sample + frozen Standard Catalog
→ Provider Policy
→ validated NavSea Provider Package
```
必须至少证明:
- Style intake 成功
- Sprite 获取/快照成功
- 至少一个真实 vector PBF sample 成功
- PBF-03 实例化成功
- NavSea mapping 成功
- Policy Schema validate 成功
- Package Schema validate 成功
- Export 成功
- validation result 非 FAIL
### B. CM95
目标:
```text
CM95 full-style + real PBF sample + existing manual assignment
→ Provider Policy
→ validated CM95 Provider Package
```
必须至少证明:
- Style intake 成功
- Sprite 获取/快照成功
- 三个指定 sample tiles 尽可能全部成功;若单个不可用,明确记录
- PBF-03 实例化成功
- 使用现有人工 assignment,不能用 matcher 覆盖
- Policy Schema validate 成功
- Package Schema validate 成功
- Export 成功
- validation result 非 FAIL
## 9. 输出建议
建议正式输出:
```text
tools/provider-pipeline/
tools/provider-policy-compiler/
providers/navsea/
providers/cm95/
contracts/
dist/
```
具体目录可根据仓库现状调整,但职责必须清晰。
最终至少保留可复现的 manifest/example 配置,让后续可再次执行两套 Provider。
## 10. Git 与工作区约束
当前 worktree 是隔离分支:
```text
feature/provider-pipeline-v1
```
不要访问或修改主工作区中的既存脏文件,除“读取 NavSea style 输入”外。
只提交本任务新增/修改文件。
完成前执行:
- 相关 Python tests
- CLI help/基础 smoke
- NavSea 实际 Pipeline
- CM95 实际 Pipeline
- `git diff --check`
- `git status --short`
成功后在 feature branch 创建清晰 commit。不要自行 merge main;由外部流程审核后处理。
## 11. 完成报告
`docs/报告/PBF_Provider_Pipeline_v1_Codex完成报告_v1.md` 必须逐项报告:
- 已完成 / 未完成 / 无法验证
- 实际命令
- NavSea 运行结果
- CM95 运行结果
- 每阶段产物
- Schema/Contract 版本
- validation result
- warnings / unresolved / limitations
- tests
- Git commit
- 最终实际使用的完整 Codex 模型配置(若运行环境可确认)
不得把“测试未跑”写成“通过”。

View File

@@ -0,0 +1,121 @@
# PBF Provider Pipeline v1 完成报告
日期:2026-10-06
分支:`feature/provider-pipeline-v1`
状态:实现已提交;CM95 端到端验收完成;NavSea 端到端验收受资源服务器连接拒绝阻塞。
## 总体结论
已实现 Provider-neutral 的 PBF-01~PBF-07、统一 Provider Policy / Provider Package v1 契约与校验、正式导出器,以及从 manifest 启动的单命令 Pipeline。正式终点是 NavSea Provider Package;实现中没有生成 `ProviderExecutionPlan`。
CM95 全流程实际运行成功,Package 校验结果为 `PASS_WITH_WARNINGS` 并已导出。NavSea 使用任务指定的 X220 当前 Style 文件,但 Sprite 与三个 z12 样本瓦片所处的 `192.168.200.184:80` 均拒绝连接,因此未运行完 intake,不存在 NavSea Policy、Package 或 export 成功证据。故双 Provider 端到端验收尚未全部完成。
## PBF-01~PBF-07 结果
| 阶段 | 实现与验证结果 |
|---|---|
| PBF-01 Provider Source Intake | 支持本地路径与 HTTP(S) Style、由 Style 或 manifest 解析 XYZ vector tile template、Sprite JSON/PNG 快照、Glyph 明确状态、PBF 样本快照、资源 SHA-256 与稳定 source fingerprint。CM95 为 PASS;NavSea Style 快照成功,后续 Sprite/PBF 请求均被拒绝。 |
| PBF-02 Style Visual Object Extract | 复用现有 `style_vo_tool.py`,不改写冻结 NavSea Standard Catalog。CM95 Style 成功提取 3,699 个 Visual Object,未解析项 0。 |
| PBF-03 Actual Visual Object Instantiate | 移除实例化主逻辑中固定 `cm95` source/catalog;provider、dataset、version 均由参数传递。输出逐 occurrence 的瓦片、feature identity、field evidence、style layer 命中、geometry 与 render components;不按屏幕位置合并。CM95 解码 5,210 个 styled-layer feature,形成 2,077 个 occurrence、102 个 distinct visual object。 |
| PBF-04 Provider → Standard Mapping | CM95 使用现有人工 assignment,未调用 matcher 覆盖人工决定;对 102 个实际对象得到 `VISUAL_OBJECT_MATCH` 9、`LAYER_ONLY` 47、`AMBIGUOUS` 2、`UNKNOWN` 44。NavSea mapping 使用冻结 Catalog 中的视觉 hash 对照实现,但受上游资源连接阻塞,未实际执行。 |
| PBF-05 Provider Policy Compile | Policy schema 为 `navsea-provider-policy/v1`。CM95 Policy 校验 PASS。NavSea/CM95 共用相同 validator 和 schema;自动测试验证两者键结构一致。 |
| PBF-06 Provider Package Validate | Package schema 为 `navsea-provider-package/v1`;结果枚举严格为 `PASS`、`PASS_WITH_WARNINGS`、`FAIL`。FAIL 或未知状态都会阻止导出。CM95 结果 `PASS_WITH_WARNINGS`。 |
| PBF-07 Package Export | CM95 正式导出成功,生成 `provider-package.json`、`provider-policy.json`、层映射、Style/Sprite 资源快照、validation JSON/Markdown 与 `MANIFEST.json`。NavSea 未导出。 |
公开 schema 位于 `tools/provider-policy-compiler/schema/`,第三方可直接按这些公开 Contract 生产 Package,不需要使用本仓库工具。运行时输出放在忽略目录 `.tmp/` 与 `dist/`,没有将 PBF、PNG cache 或导出包纳入 Git。
## CM95 实际运行
实际命令:
```bash
python3 /root/.codex/skills/codex-build-output-throttle/build-output-throttle/scripts/run_limited.py \
--log-dir .tmp/logs -- \
.venv/bin/python tools/provider-pipeline/run_provider_pipeline.py \
--manifest providers/cm95/manifest.json
```
实际输入与指纹:
- Style:`http://192.168.200.184:18889/s52/full-style.json`,SHA-256 `b1d3d0060012a8e4b55a6053231231d742ec415b3a8398f9ae0867bbdfa5290b`。
- Sprite JSON:SHA-256 `51519fb9a64f11dc100f894b2b54d14f56770ac9d511f15ae574ff5867af857d`。
- Sprite PNG:SHA-256 `90cf4da31e01abb0676b538ebd2b5c951b6d737a31d23062ec6eb06383d729a2`。
- Style 未声明 Glyph URL;Package 明确记录 Glyph 不可用,`capabilities.glyphs=false`,没有伪造 Glyph 成功。
- Source fingerprint:`aa42eb9fd575dde1b651d537ccc5b52c9dafac1bd8a3d6c5166c49a9581c9363`。
三个指定 PBF 样本全部成功,失败数为 0:
| XYZ | 字节数 | SHA-256 |
|---|---:|---|
| `12/3531/1641` | 423,777 | `1b9521e5dd990c92c71f22a4d56bfbd1ba33e5f236ddbf2af5ce51191e2efab8` |
| `12/3526/1643` | 285,154 | `92147b621b213a0c1fe80c2fa28ab71292b79d133b98272977defb41424ec39c` |
| `12/3638/1613` | 82,075 | `7eeb420fe809b09f8e03b892668830efd3db3e957bf32a80fd1b92b2d90e1dba` |
最终 validation 为 `PASS_WITH_WARNINGS`。两项警告分别表示存在未决 mapping,以及本次样本/范围存在已声明限制。46 个 unresolved 映射被保留,未转成猜测 Layer。CM95 本次只覆盖三个样本瓦片,不代表全量数据集覆盖情况。
阶段产物位置:
- Intake:`.tmp/provider-pipeline/cm95/s52-full/intake/`,含 `source-manifest.json`、Style/Sprite 快照及三个 PBF 样本。
- PBF-02:`.tmp/provider-pipeline/cm95/s52-full/pbf-02-style-visual-object/anonymous-catalog.json`。
- PBF-03:`.tmp/provider-pipeline/cm95/s52-full/pbf-03-actual/actual-anonymous-catalog.json`、`visual-object-occurrences.json`、`instantiation-report.json`。
- PBF-04:`.tmp/provider-pipeline/cm95/s52-full/pbf-04-mapping/provider-layer-assignment.json`。
- PBF-05/06:工作区中的 Policy、校验报告及最终包内对应 JSON/Markdown。
- PBF-07:`dist/cm95/s52-full/1.0.0/`,含 package、policy、mapping、resources、validation 与 MANIFEST。此目录已加入 `.gitignore`,需要时可用 manifest 重建。
## NavSea 实际运行
实际命令:
```bash
python3 /root/.codex/skills/codex-build-output-throttle/build-output-throttle/scripts/run_limited.py \
--log-dir .tmp/logs -- \
.venv/bin/python tools/provider-pipeline/run_provider_pipeline.py \
--manifest providers/navsea/manifest.json
```
已读取任务指定的本地 Style:`/root/sourceserver/pbf/src/pbf/style.navsea-delivery-full-semantic-extentfix.json`,仅在 worktree 内保存 snapshot,原主工作区文件没有纳入 Git。该 snapshot SHA-256 为 `2ffde6af6043be2567a861a8d06d075a87b532091686bb5ac6ce889e45827e33`。Style 内的 vector source 指向 `http://192.168.200.184/pbf-delivery-full-20260418-rebuild/{z}/{x}/{y}.pbf`,Sprite 指向 `http://192.168.200.184/newpec/sprite-semantic/sprite`。
对 Sprite JSON、Sprite PNG 和下列三个真实 z12 候选瓦片的 GET 均返回 `[Errno 111] Connection refused`:
- `12/3531/1641`
- `12/3526/1643`
- `12/3638/1613`
所以 NavSea PBF-01 的 Style snapshot 子步骤已完成,但 Sprite/PBF intake gate 未通过;PBF-02~PBF-07 的 NavSea 实际结果均为**无法验证/未运行**。没有改用其他 Provider、镜像或伪造 tile 结果。
## 测试与检查
以下 noisy 测试/编译命令均通过 `$build-output-throttle` 执行:
```bash
python3 /root/.codex/skills/codex-build-output-throttle/build-output-throttle/scripts/run_limited.py \
--log-dir .tmp/logs -- .venv/bin/python -m unittest discover -s tests -v
```
结果:PASS,6 项测试覆盖稳定 fingerprint、Policy schema、歧义约束、FAIL 阻止导出、导出结果枚举门禁、NavSea/CM95 共用 Contract 结构。
```bash
python3 /root/.codex/skills/codex-build-output-throttle/build-output-throttle/scripts/run_limited.py \
--log-dir .tmp/logs -- .venv/bin/python -m compileall -q \
tools/provider-policy-compiler tools/provider-pipeline \
tools/visual-object-instantiation tests
.venv/bin/python tools/provider-pipeline/run_provider_pipeline.py --help
git diff --check
```
以上编译检查、CLI help smoke 与 `git diff --check` 均通过。依赖安装在仓库 `.venv` 中,使用 `requirements-provider-pipeline.txt`,没有改动系统 Python。未运行 Jest、Android/ADB 相关操作。
## Git 与模型配置
- 实现提交:`aea98628e4695a17078608f64fed225433eb200e`(`feat: implement provider-neutral PBF pipeline v1`)。
- 提交范围为本任务文件;没有提交 `docs/任务/` 下用户提供的未跟踪任务输入、`.venv`、`.tmp` 或 `dist` 产物。
- 完成报告会作为后续独立文档提交;本报告记录的实现提交 SHA 不受该文档提交影响。
- 当前执行器未向工具会话提供可核实的 Codex 模型完整标识或推理档位,故无法验证,未作推断。
## 尚待完成
重新开放 `192.168.200.184:80` 上的 NavSea Sprite 与 PBF 服务后,使用同一 manifest 重跑 NavSea Pipeline,确认实际可下载的 z12 样本并补齐 NavSea 七阶段结果。当前可复现执行入口为:
```bash
.venv/bin/python tools/provider-pipeline/run_provider_pipeline.py --manifest providers/navsea/manifest.json
```

View File

@@ -0,0 +1,265 @@
# PBF Provider Pipeline v1 完成报告 v2
日期:2026-10-06
分支:`feature/provider-pipeline-v1`
状态:**PBF-01~PBF-07 已实现;NavSea 与 CM95 两套真实 Style/PBF/Sprite 端到端运行完成。**
## 1. 最终结论
PBF Provider Pipeline v1 已完成。正式边界保持为:
```text
Provider Source
→ PBF-01 Intake
→ PBF-02 Style Visual Object
→ PBF-03 Actual Visual Object
→ PBF-04 NavSea Standard Mapping
→ PBF-05 Provider Policy
→ PBF-06 Provider Package Validate
→ PBF-07 Provider Package Export
```
Pipeline 的终点是 `NavSea Provider Package`,没有生成 `ProviderExecutionPlan`。
两套真实验收均完成:
- NavSea:`PASS_WITH_WARNINGS`
- CM95:`PASS_WITH_WARNINGS`
两者 warnings 均为 `mapping:unresolved_present` 与 `limitations:present`。这表示仍有未知/歧义 Mapping,以及本轮只使用三个样本瓦片;不是 Pipeline 执行错误,也没有把 UNKNOWN/AMBIGUOUS 静默映射成某个 Layer。
## 2. 实现内容
已加入正式 Provider-neutral Pipeline runner、Provider Policy / Package v1 Schema 与 validator、NavSea/CM95 manifest、自动测试、依赖说明及中文 README。
现有 `tools/visual-object-instantiation/` 已正式纳入 Git,并移除主逻辑固定 CM95 provider/catalog 名称。PBF-03 额外输出 occurrence 级追溯,包括 tile、feature identity、属性证据、matched style layers、geometry 和 render components。
正式 Mapping 状态统一为:
- `VISUAL_OBJECT_MATCH`
- `LAYER_ONLY`
- `COMPONENT_ONLY`
- `AMBIGUOUS`
- `UNKNOWN`
`FAIL` 或其他非允许 validation 状态禁止 PBF-07 export。
## 3. NavSea 真实端到端结果
分析输入仍使用 X220 当前本地 Style:
```text
/root/sourceserver/pbf/src/pbf/style.navsea-delivery-full-semantic-extentfix.json
```
Style SHA-256:
```text
2ffde6af6043be2567a861a8d06d075a87b532091686bb5ac6ce889e45827e33
```
原 Style 内旧 LAN `192.168.200.184:80` 的 PBF/Sprite 服务当前拒绝连接。为保持本地 Style 内容不变,同时完成真实资源验收,`providers/navsea/manifest.json` 将资源显式覆盖到同数据集当前可用的公开 HTTPS 入口:
```text
PBF: https://pbf.jpcn.ccwu.cc/pbf/pbf-delivery-full-20260418-rebuild/{z}/{x}/{y}.pbf
Sprite: https://pbf.jpcn.ccwu.cc/sprite/sprite
Glyph: https://pbf.jpcn.ccwu.cc/fonts/{fontstack}/{range}.pbf
```
因此本轮没有用公开 Style 替换本地 Style;Style 分析输入仍是上述 X220 本地文件,资源获取由 Provider manifest 显式声明。
NavSea Source fingerprint:
```text
27a09da2a31746cff8e874fb5cd48bef7226c322fb2f193bc1b7fa5a01187f9c
```
NavSea Sprite:
- JSON SHA-256:`e48aed662580ad437a5270966acdc906039b0d97da7892d518710d10f518f75c`
- PNG SHA-256:`fca794509a41eda3f4a9fe6db5e39467aac7cdda07de553c97a33e31b4c7b981`
三个真实 z12 PBF 样本全部成功:
| XYZ | 字节数 | SHA-256 |
|---|---:|---|
| `12/3531/1641` | 181,263 | `99018718a886cd5af73554f9b453a476308a51e38591758ef2b1eaf723c6486a` |
| `12/3526/1643` | 119,246 | `8f32aa41ab9af90122b45bda0bed9f02faed1c10e0eb3ab09109e88882704922` |
| `12/3638/1613` | 305,970 | `25373c570684e5b06a6cdea56e8202541aab5c6aa4ec885b23c39bb9d807fd5c` |
PBF-03:
- decoded styled-layer feature:8,486
- matched occurrence:8,484
- occurrence 输出:8,484
- distinct Actual Visual Object:85
- group:34
- unsupported component:0
- missing sprite component:48
PBF-04 Mapping 共 85 个 Actual Visual Object:
- VISUAL_OBJECT_MATCH:27
- LAYER_ONLY:7
- AMBIGUOUS:11
- UNKNOWN:40
- COMPONENT_ONLY:0
阶段结果:
```text
PBF-01 PASS
PBF-02 PASS
PBF-03 PASS
PBF-04 PASS
PBF-05 PASS
PBF-06 PASS_WITH_WARNINGS
PBF-07 EXPORTED
```
导出目录:
```text
dist/navsea/delivery-full-semantic-extentfix/1.0.0/
```
## 4. CM95 真实端到端结果
Style:
```text
http://192.168.200.184:18889/s52/full-style.json
```
Style SHA-256:
```text
b1d3d0060012a8e4b55a6053231231d742ec415b3a8398f9ae0867bbdfa5290b
```
Source fingerprint:
```text
aa42eb9fd575dde1b651d537ccc5b52c9dafac1bd8a3d6c5166c49a9581c9363
```
Sprite:
- JSON SHA-256:`51519fb9a64f11dc100f894b2b54d14f56770ac9d511f15ae574ff5867af857d`
- PNG SHA-256:`90cf4da31e01abb0676b538ebd2b5c951b6d737a31d23062ec6eb06383d729a2`
Style 未声明 Glyph,Package 明确记录 Glyph 不可用并保持 `capabilities.glyphs=false`。
三个指定真实 z12 PBF 样本全部成功:
| XYZ | 字节数 | SHA-256 |
|---|---:|---|
| `12/3531/1641` | 423,777 | `1b9521e5dd990c92c71f22a4d56bfbd1ba33e5f236ddbf2af5ce51191e2efab8` |
| `12/3526/1643` | 285,154 | `92147b621b213a0c1fe80c2fa28ab71292b79d133b98272977defb41424ec39c` |
| `12/3638/1613` | 82,075 | `7eeb420fe809b09f8e03b892668830efd3db3e957bf32a80fd1b92b2d90e1dba` |
PBF-03:
- decoded styled-layer feature:5,210
- matched occurrence:2,077
- occurrence 输出:2,077
- distinct Actual Visual Object:102
- group:25
- unsupported component:0
- missing sprite component:0
PBF-04 使用现有 CM95 人工 assignment,没有由 matcher 覆盖人工结论。Mapping 共 102 个 Actual Visual Object:
- VISUAL_OBJECT_MATCH:9
- LAYER_ONLY:47
- AMBIGUOUS:2
- UNKNOWN:44
- COMPONENT_ONLY:0
阶段结果:
```text
PBF-01 PASS
PBF-02 PASS
PBF-03 PASS
PBF-04 PASS
PBF-05 PASS
PBF-06 PASS_WITH_WARNINGS
PBF-07 EXPORTED
```
导出目录:
```text
dist/cm95/s52-full/1.0.0/
```
## 5. 验证
自动测试:
```text
.venv/bin/python -m unittest discover -s tests -v
```
结果:6/6 PASS。
CM95 固定分类基线:
```text
.venv/bin/python tools/style-visual-object/validate_cm95_classification.py
```
结果:`RESULT: PASS`。既有 7 条人工决定与 S-52 语义建议不一致 WARN 保留,WARN 不覆盖人工结果。
其他检查:
- Python compileall:PASS
- Pipeline CLI help:PASS
- `git diff --check`:PASS
- NavSea fresh end-to-end run:PASS_WITH_WARNINGS / EXPORTED
- CM95 fresh end-to-end run:PASS_WITH_WARNINGS / EXPORTED
## 6. Git 与 Codex
Pipeline 正式流程文档此前已保存到 main:
- `f3da1a6 docs: define PBF provider pipeline v1`
Codex 实现阶段:
- 实现提交:`aea9862 feat: implement provider-neutral PBF pipeline v1`
- 初版完成报告提交:`804a764 docs: record provider pipeline v1 completion`
随后由外部 X220 shell 完成 NavSea HTTPS 资源入口修正及 NavSea/CM95 fresh end-to-end 验收。本报告 v2 记录最终实际状态。
Codex 本轮实际启动横幅确认配置:
```text
Model: GPT-6 Luna
Reasoning effort: High
```
## 7. 当前限制
1. 两套验收都只采样三个 z12 tile,不代表全量数据覆盖。
2. NavSea 本次 85 个 Actual Visual Object 中仍有 UNKNOWN/AMBIGUOUS;这些状态被明确保留,没有猜测 Layer。
3. CM95 本次 102 个 Actual Visual Object 中仍有 44 UNKNOWN、2 AMBIGUOUS;仍以现有人工 assignment 为最终语义权威。
4. `dist/` 与 `.tmp/` 为可重建运行产物,按设计不纳入 Git。
5. NavSea 本地 Style 内仍保留旧 LAN URL;Pipeline 通过 Provider manifest 的资源覆盖使用当前可用公开 HTTPS 镜像。后续若 Style 本体正式更新 URL,可再移除 manifest 覆盖。
## 8. 复现命令
NavSea:
```bash
.venv/bin/python tools/provider-pipeline/run_provider_pipeline.py \
--manifest providers/navsea/manifest.json
```
CM95:
```bash
.venv/bin/python tools/provider-pipeline/run_provider_pipeline.py \
--manifest providers/cm95/manifest.json
```

View File

@@ -0,0 +1,141 @@
# PBF Provider Pipeline v1 完成报告 v3
日期:2026-10-06
分支:`feature/provider-pipeline-v1`
主题:PBF-08 NavSea Delivery ZIP
## 结论
在既有 PBF-01~PBF-07 基础上新增 PBF-08。PBF-07 仍负责导出可审计的 Provider Package 目录,PBF-08 自动把 NavSea 最终正式输入打成一个单文件 `.navsea-provider.zip`。
用户后续向 NavSea 交付时,不再需要手工选择目录中的多个文件,只需要复制对应 Provider 的 ZIP。
## PBF-08 白名单
ZIP 固定且只包含 9 个文件:
```text
provider-package.json
provider-policy.json
mapping-layer-assignment.json
resources-style.json
resources-sprite.json
resources-sprite.png
validation.json
validation.md
MANIFEST.json
```
任何白名单文件缺失时,PBF-08 直接 FAIL,不生成 ZIP。
ZIP 生成后自动执行:
1. 精确文件列表校验;
2. ZIP CRC 校验;
3. ZIP SHA-256 计算;
4. 将 PBF-08 状态、路径、文件名、SHA-256、字节数、内部根目录和文件列表写入 `pipeline-report.json`;
5. CLI 输出 `deliveryZip` 和 `deliveryZipSha256`。
临时 PBF、`.tmp`、工具源码、测试、人工审核中间文件以及其他未声明文件不会进入 ZIP。
## 自动测试
```text
.venv/bin/python -m unittest discover -s tests -v
```
结果:8/8 PASS。
新增覆盖:
- ZIP 只包含规定的 NavSea 交付文件;
- 多余文件不会进入 ZIP;
- 缺少任意必需文件时拒绝生成 ZIP。
Python compileall:PASS。
`git diff --check`:PASS。
## NavSea 实际结果
Pipeline:
```text
PBF-01 PASS
PBF-02 PASS
PBF-03 PASS
PBF-04 PASS
PBF-05 PASS
PBF-06 PASS_WITH_WARNINGS
PBF-07 EXPORTED
PBF-08 PACKAGED
```
ZIP:
```text
dist/navsea/delivery-full-semantic-extentfix/
navsea-delivery-full-semantic-extentfix-1.0.0.navsea-provider.zip
```
ZIP SHA-256:
```text
54aff8d8b882211ced6e77b3a93e8e793c780b44a1faa982ed8fc64693124fd5
```
ZIP 大小:933,404 bytes。
外部 `unzip -t`:9 个文件全部 OK,无压缩数据错误。
## CM95 实际结果
Pipeline:
```text
PBF-01 PASS
PBF-02 PASS
PBF-03 PASS
PBF-04 PASS
PBF-05 PASS
PBF-06 PASS_WITH_WARNINGS
PBF-07 EXPORTED
PBF-08 PACKAGED
```
ZIP:
```text
dist/cm95/s52-full/
cm95-s52-full-1.0.0.navsea-provider.zip
```
ZIP SHA-256:
```text
603cc004833bcff5571a32843d8ceedc7858b2fc4b28c41056db6807c7d511ee
```
ZIP 大小:742,008 bytes。
外部 `unzip -t`:9 个文件全部 OK,无压缩数据错误。
## 使用规则
以后 PBF Pipeline 正常执行完成后:
```bash
.venv/bin/python tools/provider-pipeline/run_provider_pipeline.py \
--manifest providers/<provider>/manifest.json
```
只要 PBF-06 为 `PASS` 或 `PASS_WITH_WARNINGS`,Pipeline 会自动完成 PBF-07 和 PBF-08。
交给 NavSea 时,只复制 CLI 输出的:
```text
deliveryZip
```
即可。
`--no-export` 模式继续只运行到 Package 校验,PBF-07 与 PBF-08 都会标记为 `SKIPPED_BY_REQUEST`,不会生成 ZIP。

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 样本瓦片。"]
}

View File

@@ -0,0 +1,71 @@
{
"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": "https://pbf.jpcn.ccwu.cc/sprite/sprite"
},
"glyphs": {
"uri": "https://pbf.jpcn.ccwu.cc/fonts/{fontstack}/{range}.pbf"
}
},
"tileSource": {
"sourceId": "navsea_delivery",
"template": "https://pbf.jpcn.ccwu.cc/pbf/pbf-delivery-full-20260418-rebuild/{z}/{x}/{y}.pbf"
},
"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 样本瓦片进行实例化;未据此推断全数据集覆盖率。"
]
}

View File

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

View File

@@ -0,0 +1,112 @@
from __future__ import annotations
import json
import sys
import tempfile
import unittest
import zipfile
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "tools/provider-policy-compiler"))
from provider_policy import DELIVERY_FILES, build_policy, create_delivery_zip, 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"])
def test_delivery_zip_contains_only_required_navsea_files(self):
with tempfile.TemporaryDirectory() as tmp:
export_dir = Path(tmp) / "1.0.0"
export_dir.mkdir()
for name in DELIVERY_FILES:
path = export_dir / name
if name == "validation.json":
path.write_text(json.dumps({"result": "PASS_WITH_WARNINGS"}), encoding="utf-8")
elif name.endswith(".png"):
path.write_bytes(b"png")
else:
path.write_text(name, encoding="utf-8")
(export_dir / "not-for-navsea.tmp").write_text("exclude me", encoding="utf-8")
info = create_delivery_zip(export_dir, "provider-a", "dataset-1", "1.0.0")
self.assertEqual(info["validationResult"], "PASS_WITH_WARNINGS")
self.assertEqual(len(info["sha256"]), 64)
with zipfile.ZipFile(info["path"], "r") as archive:
expected = [f"provider-a-dataset-1-1.0.0/{name}" for name in DELIVERY_FILES]
self.assertEqual(archive.namelist(), expected)
self.assertNotIn("provider-a-dataset-1-1.0.0/not-for-navsea.tmp", archive.namelist())
def test_delivery_zip_fails_when_required_file_is_missing(self):
with tempfile.TemporaryDirectory() as tmp:
export_dir = Path(tmp) / "1.0.0"
export_dir.mkdir()
for name in DELIVERY_FILES:
if name == "resources-sprite.png":
continue
path = export_dir / name
if name == "validation.json":
path.write_text(json.dumps({"result": "PASS"}), encoding="utf-8")
else:
path.write_text(name, encoding="utf-8")
with self.assertRaises(ValueError):
create_delivery_zip(export_dir, "provider-a", "dataset-1", "1.0.0")
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,49 @@
# 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-07 和 PBF-08。
## 阶段
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 导出。
PBF-08 在 PBF-07 成功后自动生成一个可直接交给 NavSea 的单文件 ZIP。ZIP 使用严格白名单,缺少任何规定文件都会直接失败;临时 PBF、.tmp、工具源码、人工审核中间文件和其他未声明文件不会进入 ZIP。
PBF-08 ZIP 固定包含:
```text
provider-package.json
provider-policy.json
mapping-layer-assignment.json
resources-style.json
resources-sprite.json
resources-sprite.png
validation.json
validation.md
MANIFEST.json
```
默认输出:
```text
dist/<provider>/<dataset>/
<provider>-<dataset>-<version>.navsea-provider.zip
```
ZIP 内部还有一个同名顶层目录,避免解压多个 Provider 包时文件混在一起。Pipeline 在完成 ZIP 后会重新打开 ZIP,核对文件列表、CRC,并计算 ZIP SHA-256;这些信息写入 `pipeline-report.json` 的 `PBF-08` 节点,同时打印 `deliveryZip` 和 `deliveryZipSha256`。
PBF-07 的可审计目录仍保留在:
```text
dist/<provider>/<dataset>/<version>/
```
运行缓存位于 `.tmp/`,不纳入 Git。Pipeline 不生成 ProviderExecutionPlan。

View File

@@ -0,0 +1,321 @@
#!/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, create_delivery_zip, 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.1.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)}
delivery_zip = create_delivery_zip(export_dir, provider_id, dataset_id, version)
record["stages"]["PBF-08"] = {"status": "PACKAGED", **delivery_zip}
else:
delivery_zip = None
record["stages"]["PBF-07"] = {"status": "SKIPPED_BY_REQUEST"}
record["stages"]["PBF-08"] = {"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), "deliveryZip": None if delivery_zip is None else delivery_zip["path"], "deliveryZipSha256": None if delivery_zip is None else delivery_zip["sha256"], "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)

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。

View File

@@ -0,0 +1,224 @@
#!/usr/bin/env python3
"""统一 Provider Policy / Package v1 编译、校验与导出。"""
from __future__ import annotations
import hashlib
import json
import re
import zipfile
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"}
DELIVERY_FILES = (
"provider-package.json",
"provider-policy.json",
"mapping-layer-assignment.json",
"resources-style.json",
"resources-sprite.json",
"resources-sprite.png",
"validation.json",
"validation.md",
"MANIFEST.json",
)
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
def _safe_delivery_component(value: str) -> str:
cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", str(value)).strip("-._")
return cleaned or "provider"
def create_delivery_zip(export_dir: Path, provider_id: str, dataset_id: str, version: str) -> dict[str, Any]:
"""把 PBF-07 的正式 NavSea 输入打成一个严格白名单 ZIP。"""
export_dir = Path(export_dir)
missing = [name for name in DELIVERY_FILES if not (export_dir / name).is_file()]
if missing:
raise ValueError(f"PBF-08 缺少正式交付文件,禁止生成 ZIP: {missing}")
validation = json.loads((export_dir / "validation.json").read_text(encoding="utf-8"))
if validation.get("result") not in {"PASS", "PASS_WITH_WARNINGS"}:
raise ValueError(f"PBF-08 只接受 PASS/PASS_WITH_WARNINGS,当前为 {validation.get('result')}")
stem = "-".join(_safe_delivery_component(x) for x in (provider_id, dataset_id, version))
zip_path = export_dir.parent / f"{stem}.navsea-provider.zip"
root_name = stem
if zip_path.exists():
zip_path.unlink()
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
for name in DELIVERY_FILES:
archive.write(export_dir / name, arcname=f"{root_name}/{name}")
with zipfile.ZipFile(zip_path, "r") as archive:
actual = archive.namelist()
expected = [f"{root_name}/{name}" for name in DELIVERY_FILES]
if actual != expected:
zip_path.unlink(missing_ok=True)
raise ValueError(f"PBF-08 ZIP 内容校验失败: expected={expected}, actual={actual}")
bad = archive.testzip()
if bad is not None:
zip_path.unlink(missing_ok=True)
raise ValueError(f"PBF-08 ZIP CRC 校验失败: {bad}")
return {
"path": str(zip_path),
"filename": zip_path.name,
"sha256": file_fingerprint(zip_path),
"bytes": zip_path.stat().st_size,
"root": root_name,
"files": list(DELIVERY_FILES),
"validationResult": validation["result"],
}

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
}

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
}

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。

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

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。

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