309 lines
15 KiB
Python
309 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""在实际原始 PBF/Style 上重放候选逐层 selector，报告误命中；仅用于人工审核。"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
from collections import Counter, defaultdict
|
||
import hashlib
|
||
import importlib.util
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from PIL import Image
|
||
from mapbox_vector_tile.Mapbox import vector_tile_pb2
|
||
|
||
from feature_lineage_replay_audit import load_instantiator
|
||
|
||
KEY_EXPRESSIONS: dict[str, tuple[str, str]] = {
|
||
"icon": ("layout", "icon-image"),
|
||
"fill": ("paint", "fill-pattern"),
|
||
"line": ("paint", "line-pattern"),
|
||
}
|
||
ELIGIBLE = {"LAYER_ONLY", "VISUAL_OBJECT_MATCH"}
|
||
|
||
|
||
def predicate(layer: dict[str, Any], comp: dict[str, Any]) -> tuple[str, str] | None:
|
||
"""Only original direct-get expressions are candidates, never inferred names."""
|
||
role = comp.get("role")
|
||
property_location = KEY_EXPRESSIONS.get(role)
|
||
key = comp.get("resourceKey")
|
||
if property_location is None or not isinstance(key, str) or not key:
|
||
return None
|
||
section, property_name = property_location
|
||
expression = layer.get(section, {}).get(property_name)
|
||
if isinstance(expression, list) and len(expression) == 2 and expression[0] == "get" and isinstance(expression[1], str) and expression[1]:
|
||
return expression[1], key
|
||
return None
|
||
|
||
|
||
def build_candidates(mapping: dict[str, Any], style: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Output candidate per *original* Style Layer, not one selector shared across layers."""
|
||
layers = style["layers"]
|
||
seen: set[str] = set()
|
||
result: list[dict[str, Any]] = []
|
||
for assignment in sorted(mapping["assignments"], key=lambda row: row["visualObjectId"]):
|
||
oid = assignment["visualObjectId"]
|
||
if oid in seen:
|
||
raise ValueError(f"visualObjectId 重复: {oid}")
|
||
seen.add(oid)
|
||
evidence = assignment.get("renderEvidence") or {}
|
||
comps = evidence.get("components") or []
|
||
eligible = (assignment["status"] in ELIGIBLE and
|
||
isinstance(assignment.get("standardLayer"), str) and
|
||
bool(assignment.get("standardLayer")))
|
||
source = None
|
||
source_layer = evidence.get("sourceLayer")
|
||
reasons: set[str] = set()
|
||
fields: dict[str, str] = {}
|
||
component_layers: set[int] = set()
|
||
own_direct: set[int] = set()
|
||
if not eligible:
|
||
reasons.add("UNRESOLVED_STANDARD_CLASSIFICATION")
|
||
if not comps or not isinstance(source_layer, str):
|
||
reasons.add("NO_COMPONENTS_OR_SOURCE_LAYER")
|
||
for c in comps:
|
||
index = c.get("styleIndex")
|
||
if type(index) is not int or index < 0 or index >= len(layers):
|
||
raise ValueError(f"{oid}: 无效 Style Index")
|
||
layer = layers[index]
|
||
if layer.get("id") != c.get("styleLayerId") or layer.get("source-layer") != source_layer:
|
||
raise ValueError(f"{oid}: Style component/source-layer 不匹配")
|
||
if source is None:
|
||
source = layer.get("source")
|
||
if layer.get("source") != source or not isinstance(source, str):
|
||
reasons.add("MULTIPLE_OR_MISSING_SOURCES")
|
||
component_layers.add(index)
|
||
pair = predicate(layer, c)
|
||
if pair:
|
||
field, value = pair
|
||
if field in fields and fields[field] != value:
|
||
reasons.add("CONTRADICTORY_DIRECT_FIELD")
|
||
fields[field] = value
|
||
own_direct.add(index)
|
||
if not fields:
|
||
reasons.add("NO_DIRECT_RESOURCE_FIELD")
|
||
terms = [["==", ["get", f], v] for f, v in sorted(fields.items())]
|
||
all_predicate = terms[0] if len(terms) == 1 else ["all", *terms] if terms else None
|
||
for index in sorted(component_layers):
|
||
status = "CANDIDATE_ONLY" if not reasons else "BLOCKED_STATIC"
|
||
result.append({
|
||
"visualObjectTypeId": oid,
|
||
"standardLayer": assignment.get("standardLayer"),
|
||
"assignmentStatus": assignment["status"],
|
||
"styleLayerId": layers[index]["id"],
|
||
"styleIndex": index,
|
||
"sourceId": source,
|
||
"sourceLayer": source_layer,
|
||
"selector": all_predicate if status == "CANDIDATE_ONLY" else None,
|
||
"selectorOrigin": "DIRECT_OR_CONJUNCTION" if index in own_direct else "INHERIT_SAME_RAW_FEATURE_CANDIDATE",
|
||
"staticBlockers": sorted(reasons),
|
||
"sampleTargetMatches": 0,
|
||
"sameStandardOtherTypeHits": 0,
|
||
"crossStandardHits": 0,
|
||
"unresolvedOrUnknownHits": 0,
|
||
"outOfCatalogHits": 0,
|
||
"counterexamples": [],
|
||
"runtimeAuthorized": False,
|
||
})
|
||
return result
|
||
|
||
|
||
def _append_counterexample(row: dict[str, Any], category: str, tile: str, fid: int | None, actual: str):
|
||
if len(row["counterexamples"]) < 3:
|
||
row["counterexamples"].append({
|
||
"category": category, "tile": tile, "featureId": fid,
|
||
"actualVisualObjectTypeId": actual,
|
||
})
|
||
|
||
|
||
def evaluate_examples(
|
||
rows: list[dict[str, Any]],
|
||
observations: list[dict[str, Any]],
|
||
assignments: dict[str, dict[str, Any]],
|
||
*,
|
||
eval_selector,
|
||
) -> dict[str, Any]:
|
||
"""Feature observations are original-Style filtered matches; no runtime grant."""
|
||
by_layer: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
||
for row in rows:
|
||
if row["selector"] is not None:
|
||
by_layer[row["styleLayerId"]].append(row)
|
||
hits = Counter()
|
||
for obs in observations:
|
||
actual_id = obs["visualObjectTypeId"]
|
||
actual = assignments.get(actual_id)
|
||
for layer_id in obs["matchedStyleLayerIds"]:
|
||
for row in by_layer.get(layer_id, ()):
|
||
if not eval_selector(row["selector"], obs["properties"], obs["zoom"]):
|
||
continue
|
||
if actual_id == row["visualObjectTypeId"]:
|
||
field = "sampleTargetMatches"
|
||
elif actual is None:
|
||
field = "outOfCatalogHits"
|
||
elif actual.get("status") not in ELIGIBLE or not actual.get("standardLayer"):
|
||
field = "unresolvedOrUnknownHits"
|
||
elif actual["standardLayer"] == row["standardLayer"]:
|
||
field = "sameStandardOtherTypeHits"
|
||
else:
|
||
field = "crossStandardHits"
|
||
row[field] += 1
|
||
hits[field] += 1
|
||
if field not in {"sampleTargetMatches", "sameStandardOtherTypeHits"}:
|
||
_append_counterexample(
|
||
row, field, obs["tile"], obs.get("featureId"), actual_id
|
||
)
|
||
sample_status = Counter()
|
||
for row in rows:
|
||
if row["selector"] is None:
|
||
row["sampleStatus"] = "NO_CANDIDATE"
|
||
elif row["crossStandardHits"] or row["unresolvedOrUnknownHits"] or row["outOfCatalogHits"]:
|
||
row["sampleStatus"] = "CROSS_CLASS_OR_RESIDUAL_COLLISION"
|
||
elif row["sampleTargetMatches"] == 0:
|
||
row["sampleStatus"] = "TARGET_NOT_OBSERVED"
|
||
else:
|
||
row["sampleStatus"] = "SAMPLE_NO_FORBIDDEN_MATCH"
|
||
sample_status[row["sampleStatus"]] += 1
|
||
return {"candidateStatusCounts": dict(sorted(sample_status.items())),
|
||
"observationMatchCounts": dict(sorted(hits.items()))}
|
||
|
||
|
||
def run_audit(style: dict[str, Any], mapping: dict[str, Any], catalog: dict[str, Any],
|
||
pbf_root: Path, sprite_json: dict[str, Any], sprite_hashes: dict[str, str]) -> dict[str, Any]:
|
||
engine = load_instantiator()
|
||
rows = build_candidates(mapping, style)
|
||
catalog_ids = [obj.get("visual_object_id") for obj in catalog["visual_objects"]]
|
||
if len(catalog_ids) != len(set(catalog_ids)):
|
||
raise ValueError("Actual Catalog Visual Object ID 重复")
|
||
mapping_ids = {assignment["visualObjectId"] for assignment in mapping["assignments"]}
|
||
if set(catalog_ids) != mapping_ids:
|
||
raise ValueError("Mapping 与 Catalog 的 Visual Object ID 覆盖不一致")
|
||
relevant_layers: dict[str, list[tuple[int, dict]]] = defaultdict(list)
|
||
candidate_source_layers: set[str] = set()
|
||
for row in rows:
|
||
if row["selector"] is not None:
|
||
candidate_source_layers.add(row["sourceLayer"])
|
||
source_owners: dict[str, set[str]] = defaultdict(set)
|
||
for index, layer in enumerate(style["layers"]):
|
||
sl = layer.get("source-layer")
|
||
if sl in candidate_source_layers:
|
||
relevant_layers[sl].append((index, layer))
|
||
source_owners[sl].add(layer.get("source"))
|
||
if any(len(owners) != 1 or None in owners for owners in source_owners.values()):
|
||
raise ValueError("相同 source-layer 对应多个 Style Source，单 PBF Root 无法唯一验证")
|
||
# Only tiles already referenced by validated Catalog examples, all raw features
|
||
# within each tile, and all original Style components for those features.
|
||
examples = {obj["example"]["tile"] for obj in catalog["visual_objects"]}
|
||
assignment_by_id = {row["visualObjectId"]: row for row in mapping["assignments"]}
|
||
if len(assignment_by_id) != len(mapping["assignments"]):
|
||
raise ValueError("Mapping Visual Object ID 冲突")
|
||
# Limit in-memory observations by processing each tile separately.
|
||
sample_stats = Counter()
|
||
aggregate = Counter()
|
||
for tile in sorted(examples):
|
||
parts = tile.split("/")
|
||
if len(parts) != 3 or not all(p.isdigit() for p in parts):
|
||
raise ValueError(f"无效 tile: {tile}")
|
||
z, x, y = map(int, parts)
|
||
if x >= 1 << z or y >= 1 << z:
|
||
raise ValueError(f"超出 Tile 范围: {tile}")
|
||
source_path = pbf_root / str(z) / str(x) / (str(y) + ".pbf")
|
||
if not source_path.is_file():
|
||
raise ValueError(f"原始 PBF 不存在: {source_path}")
|
||
decoded = vector_tile_pb2.tile()
|
||
decoded.ParseFromString(source_path.read_bytes())
|
||
source_layer_counts = Counter(layer.name for layer in decoded.layers)
|
||
if any(source_layer_counts[sl] > 1 for sl in candidate_source_layers):
|
||
raise ValueError(f"原始 PBF 中 SourceLayer 名称重复: {tile}")
|
||
observations = []
|
||
for tile_layer in decoded.layers:
|
||
style_members = relevant_layers.get(tile_layer.name)
|
||
if not style_members:
|
||
continue
|
||
for feature in tile_layer.features:
|
||
sample_stats["sourceLayerRawFeaturesExamined"] += 1
|
||
props = engine.decode_feature_properties(tile_layer, feature)
|
||
matched_layers = []
|
||
components = []
|
||
for index, layer in style_members:
|
||
if engine.matches(layer, props, z):
|
||
matched_layers.append(layer["id"])
|
||
components.extend(engine.component_from_layer(layer, index, props, z, sprite_json, sprite_hashes))
|
||
if not matched_layers:
|
||
continue
|
||
vo_id = engine.stable_id("VO", {
|
||
"geometry_role": engine.geometry_role(feature.type),
|
||
"components": engine.signature_components(components),
|
||
})
|
||
observations.append({
|
||
"visualObjectTypeId": vo_id,
|
||
"properties": props, "zoom": z,
|
||
"matchedStyleLayerIds": matched_layers,
|
||
"tile": tile, "featureId": int(feature.id) if feature.HasField("id") else None,
|
||
})
|
||
sample_stats["styledRawFeaturesObserved"] += 1
|
||
partial = evaluate_examples(
|
||
rows, observations, assignment_by_id,
|
||
eval_selector=lambda expr, props, zoom: engine.evaluate(expr, props, zoom) is True,
|
||
)
|
||
for key, val in partial["observationMatchCounts"].items():
|
||
aggregate[key] += val
|
||
sample_stats["tilesScanned"] += 1
|
||
# Recompute final row status because some target samples may be found later.
|
||
statuses = Counter()
|
||
for row in rows:
|
||
if row["selector"] is None:
|
||
row["sampleStatus"] = "NO_CANDIDATE"
|
||
elif any(row[k] for k in ("crossStandardHits", "unresolvedOrUnknownHits", "outOfCatalogHits")):
|
||
row["sampleStatus"] = "CROSS_CLASS_OR_RESIDUAL_COLLISION"
|
||
elif not row["sampleTargetMatches"]:
|
||
row["sampleStatus"] = "TARGET_NOT_OBSERVED"
|
||
else:
|
||
row["sampleStatus"] = "SAMPLE_NO_FORBIDDEN_MATCH"
|
||
statuses[row["sampleStatus"]] += 1
|
||
return {
|
||
"schemaVersion": "navsea-selector-collision-evidence-audit/v1",
|
||
"scope": "261_RECORDED_EXAMPLE_TILES_ALL_RAW_FEATURES_NOT_ALL_PROVIDER_TILES",
|
||
"observedTileCount": sample_stats["tilesScanned"],
|
||
"sourceLayerRawFeaturesExamined": sample_stats["sourceLayerRawFeaturesExamined"],
|
||
"styledRawFeaturesObserved": sample_stats["styledRawFeaturesObserved"],
|
||
"candidateStatusCounts": dict(sorted(statuses.items())),
|
||
"observationMatchCounts": dict(sorted(aggregate.items())),
|
||
"displayRuntimeAuthorizedCount": 0,
|
||
"primaryAuthorizedCount": 0,
|
||
"entries": rows,
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
p = argparse.ArgumentParser(description=__doc__)
|
||
for name in ("style", "mapping", "catalog", "pbf-root", "sprite-json", "sprite-png", "output"):
|
||
p.add_argument(f"--{name}", type=Path, required=True)
|
||
args = p.parse_args()
|
||
style_bytes = args.style.read_bytes()
|
||
mapping = json.loads(args.mapping.read_text(encoding="utf-8"))
|
||
expected = (mapping.get("renderEvidenceSource") or {}).get("styleSha256")
|
||
if expected != hashlib.sha256(style_bytes).hexdigest():
|
||
raise ValueError("Mapping 绑定的原 Style SHA-256 不一致")
|
||
catalog = json.loads(args.catalog.read_text(encoding="utf-8"))
|
||
declared = (catalog.get("inputs") or {}).get("style")
|
||
if not isinstance(declared, str) or not Path(declared).is_file():
|
||
raise ValueError("Catalog 声明的原 Style 路径不可校验")
|
||
if hashlib.sha256(Path(declared).read_bytes()).digest() != hashlib.sha256(style_bytes).digest():
|
||
raise ValueError("Catalog 原 Style SHA-256 不一致")
|
||
style = json.loads(style_bytes)
|
||
sprite_json = json.loads(args.sprite_json.read_text(encoding="utf-8"))
|
||
with Image.open(args.sprite_png) as atlas:
|
||
hashes = load_instantiator().sprite_hashes(sprite_json, atlas.convert("RGBA"))
|
||
out = run_audit(style, mapping, catalog, args.pbf_root, sprite_json, hashes)
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
if args.output.exists():
|
||
raise ValueError("审计结果已存在：避免覆盖前一份不可变证据；请增版本")
|
||
args.output.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
print(json.dumps({k: v for k, v in out.items() if k != "entries"}, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|