191 lines
6.4 KiB
Python
191 lines
6.4 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""CM95 delivery PBF → NavSea style 消费契约审计 v2。
|
||
|
||
检查每个 delivery feature：
|
||
1. source-layer 是否存在于 style；
|
||
2. 属性值是否至少能命中一个 style filter（忽略 min/max zoom）；
|
||
3. 当前 zoom 是否实际可见；
|
||
4. 将“契约不匹配”和“因 minzoom/maxzoom 有意隐藏”分开统计。
|
||
|
||
适用于当前 NavSea unified style 使用的 filter 表达式子集。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import mapbox_vector_tile
|
||
|
||
|
||
def parse_args():
|
||
p = argparse.ArgumentParser()
|
||
p.add_argument("--pbf-root", required=True)
|
||
p.add_argument("--style", required=True)
|
||
p.add_argument("--output", default="cm95_pbf_style_contract_audit_v2.json")
|
||
return p.parse_args()
|
||
|
||
|
||
def ev(expr: Any, props: dict[str, Any]):
|
||
if not isinstance(expr, list):
|
||
return expr
|
||
if not expr:
|
||
return None
|
||
op = expr[0]
|
||
if op == "get":
|
||
return props.get(expr[1])
|
||
if op == "has":
|
||
return expr[1] in props
|
||
if op == "!":
|
||
return not bool(ev(expr[1], props))
|
||
if op == "all":
|
||
return all(bool(ev(x, props)) for x in expr[1:])
|
||
if op == "any":
|
||
return any(bool(ev(x, props)) for x in expr[1:])
|
||
if op == "==":
|
||
return ev(expr[1], props) == ev(expr[2], props)
|
||
if op == "!=":
|
||
return ev(expr[1], props) != ev(expr[2], props)
|
||
if op == "slice":
|
||
value = ev(expr[1], props)
|
||
if value is None:
|
||
return None
|
||
start = int(ev(expr[2], props))
|
||
end = int(ev(expr[3], props)) if len(expr) > 3 else None
|
||
return str(value)[start:end]
|
||
if op == "to-number":
|
||
value = ev(expr[1], props)
|
||
try:
|
||
num = float(value)
|
||
return int(num) if num.is_integer() else num
|
||
except Exception:
|
||
return 0
|
||
if op == "literal":
|
||
return expr[1]
|
||
if op == "match":
|
||
value = ev(expr[1], props)
|
||
pairs = expr[2:-1]
|
||
default = expr[-1]
|
||
for i in range(0, len(pairs), 2):
|
||
label = pairs[i]
|
||
output = pairs[i + 1]
|
||
if isinstance(label, list) and label and label[0] == "literal":
|
||
labels = label[1]
|
||
if value in labels:
|
||
return ev(output, props)
|
||
elif isinstance(label, list):
|
||
if value in label:
|
||
return ev(output, props)
|
||
elif value == label:
|
||
return ev(output, props)
|
||
return ev(default, props)
|
||
raise RuntimeError(f"unsupported style filter op: {op}")
|
||
|
||
|
||
def decode_layers(path: Path):
|
||
decoded = mapbox_vector_tile.decode(path.read_bytes())
|
||
for name, layer in decoded.items():
|
||
yield name, layer
|
||
|
||
|
||
def main():
|
||
a = parse_args()
|
||
pbf_root = Path(a.pbf_root)
|
||
style = json.loads(Path(a.style).read_text(encoding="utf-8"))
|
||
|
||
style_by_source_layer = defaultdict(list)
|
||
for layer in style.get("layers", []):
|
||
if layer.get("source") == "navsea_delivery" and layer.get("source-layer"):
|
||
style_by_source_layer[layer["source-layer"]].append(layer)
|
||
|
||
files = sorted(pbf_root.rglob("*.pbf"))
|
||
contract_unmatched = Counter()
|
||
zoom_hidden = Counter()
|
||
visible = Counter()
|
||
examples = defaultdict(list)
|
||
total_features = 0
|
||
|
||
for path in files:
|
||
try:
|
||
rel = path.relative_to(pbf_root)
|
||
z = int(rel.parts[-3])
|
||
except Exception:
|
||
raise RuntimeError(f"PBF path must end with z/x/y.pbf: {path}")
|
||
|
||
for source_layer, layer in decode_layers(path):
|
||
style_layers = style_by_source_layer.get(source_layer, [])
|
||
for feature in layer.get("features", []):
|
||
total_features += 1
|
||
props = feature.get("properties") or {}
|
||
|
||
filter_matches = []
|
||
for sl in style_layers:
|
||
flt = sl.get("filter")
|
||
if flt is None or bool(ev(flt, props)):
|
||
filter_matches.append(sl)
|
||
|
||
key = (z, source_layer)
|
||
if not filter_matches:
|
||
contract_unmatched[key] += 1
|
||
if len(examples[("contract", key)]) < 10:
|
||
examples[("contract", key)].append(props)
|
||
continue
|
||
|
||
zoom_matches = []
|
||
for sl in filter_matches:
|
||
minz = sl.get("minzoom")
|
||
maxz = sl.get("maxzoom")
|
||
if minz is not None and z < minz:
|
||
continue
|
||
if maxz is not None and z >= maxz:
|
||
continue
|
||
zoom_matches.append(sl)
|
||
|
||
if zoom_matches:
|
||
visible[key] += 1
|
||
else:
|
||
zoom_hidden[key] += 1
|
||
if len(examples[("zoom", key)]) < 10:
|
||
examples[("zoom", key)].append(props)
|
||
|
||
result = {
|
||
"schema": "navsea.cm95.style-contract-audit/v2",
|
||
"pbf_file_count": len(files),
|
||
"feature_count": total_features,
|
||
"style_source_layer_count": len(style_by_source_layer),
|
||
"contract_unmatched_count": sum(contract_unmatched.values()),
|
||
"zoom_hidden_count": sum(zoom_hidden.values()),
|
||
"visible_count": sum(visible.values()),
|
||
"contract_unmatched": [
|
||
{"z": z, "source_layer": sl, "count": count,
|
||
"examples": examples.get(("contract", (z, sl)), [])}
|
||
for (z, sl), count in sorted(contract_unmatched.items())
|
||
],
|
||
"zoom_hidden": [
|
||
{"z": z, "source_layer": sl, "count": count,
|
||
"examples": examples.get(("zoom", (z, sl)), [])}
|
||
for (z, sl), count in sorted(zoom_hidden.items())
|
||
],
|
||
}
|
||
result["overall"] = "PASS" if result["contract_unmatched_count"] == 0 else "FAIL"
|
||
|
||
Path(a.output).write_text(
|
||
json.dumps(result, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
print(f"PBF files : {result['pbf_file_count']}")
|
||
print(f"Features : {result['feature_count']}")
|
||
print(f"Contract unmatched : {result['contract_unmatched_count']}")
|
||
print(f"Zoom-hidden (expected) : {result['zoom_hidden_count']}")
|
||
print(f"Visible : {result['visible_count']}")
|
||
print(f"OVERALL : {result['overall']}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|