106 lines
4.4 KiB
Python
106 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
DELIVERY_STATUSES = {"MAPPED", "MAPPED_CONDITIONAL", "COMPOSITE", "MAPPED_DERIVED_POINTS"}
|
|
|
|
|
|
def parse_args():
|
|
p = argparse.ArgumentParser(description="CM95 全 Dataset Mapping Gap 分类收口审计 v2")
|
|
p.add_argument("--gap-audit", default="build/production/cm95_global_mapping_gap_audit_v1.json")
|
|
p.add_argument("--mapping", default="config/mappings/cm95_to_navsea_v1.yaml")
|
|
p.add_argument("--disposition", default="config/mappings/cm95_gap_disposition_v1.yaml")
|
|
p.add_argument("--output", default="build/production/cm95_global_mapping_gap_closure_audit_v2.json")
|
|
return p.parse_args()
|
|
|
|
|
|
def main():
|
|
a = parse_args()
|
|
gap = json.loads(Path(a.gap_audit).read_text(encoding="utf-8"))
|
|
mapping = yaml.safe_load(Path(a.mapping).read_text(encoding="utf-8")) or {}
|
|
disposition = yaml.safe_load(Path(a.disposition).read_text(encoding="utf-8")) or {}
|
|
|
|
rules = {
|
|
(str(r.get("cm95_object_class") or ""), str(r.get("geometry_type") or "").upper()): r
|
|
for r in mapping.get("rules") or []
|
|
if isinstance(r, dict)
|
|
}
|
|
disp = {
|
|
(str(r.get("object_class") or ""), str(r.get("geometry_type") or "").upper()): r
|
|
for r in disposition.get("variants") or []
|
|
if isinstance(r, dict)
|
|
}
|
|
|
|
unclassified = []
|
|
pending = []
|
|
existing_non_delivery = []
|
|
newly_delivery = []
|
|
for item in gap.get("gap_variants") or []:
|
|
key = (str(item.get("object_class") or ""), str(item.get("geometry_type") or "").upper())
|
|
d = disp.get(key)
|
|
if d is None:
|
|
unclassified.append({"object_class": key[0], "geometry_type": key[1], "feature_count": item.get("feature_count")})
|
|
continue
|
|
state = str(d.get("disposition") or "")
|
|
row = {
|
|
"object_class": key[0],
|
|
"geometry_type": key[1],
|
|
"feature_count": int(item.get("feature_count") or 0),
|
|
"disposition": state,
|
|
}
|
|
if state == "ENGINEERING_ONLY_PENDING_SEMANTIC_REVIEW":
|
|
pending.append(row)
|
|
elif state == "DELIVERY_MAPPED_V1_4_3":
|
|
newly_delivery.append(row)
|
|
elif state.startswith("EXISTING_"):
|
|
existing_non_delivery.append(row)
|
|
|
|
fish = rules.get(("FSHHAV", "AREA"))
|
|
fish_ok = bool(
|
|
fish
|
|
and str(fish.get("target_source_layer")) == "anchor_caution_hazard_area"
|
|
and str(fish.get("safety_class")) == "hard_gate"
|
|
and str(fish.get("mapping_status")) in DELIVERY_STATUSES
|
|
)
|
|
|
|
contract = mapping.get("production_change_contract") or {}
|
|
changed_classes = sorted({str(v) for v in contract.get("delivery_output_changed_object_classes") or []})
|
|
changed_ok = changed_classes == ["FSHHAV"]
|
|
|
|
result = {
|
|
"schema": "navsea.cm95.global-mapping-gap-closure-audit/v2",
|
|
"dataset_id": gap.get("dataset_id"),
|
|
"observed_gap_variants": len(gap.get("gap_variants") or []),
|
|
"unclassified_variants": unclassified,
|
|
"newly_delivery_mapped": newly_delivery,
|
|
"existing_explicit_non_delivery": existing_non_delivery,
|
|
"pending_semantic_review": pending,
|
|
"pending_semantic_review_feature_rows": sum(x["feature_count"] for x in pending),
|
|
"fshhav_area_hard_gate_mapping_ok": fish_ok,
|
|
"delivery_output_changed_object_classes": changed_classes,
|
|
"delivery_change_contract_ok": changed_ok,
|
|
"semantic_completeness": "COMPLETE" if not pending else "PENDING_REVIEW",
|
|
"overall": "PASS" if not unclassified and fish_ok and changed_ok else "FAIL",
|
|
}
|
|
Path(a.output).parent.mkdir(parents=True, exist_ok=True)
|
|
Path(a.output).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
print(f"Observed gap variants : {result['observed_gap_variants']}")
|
|
print(f"Unclassified variants : {len(unclassified)}")
|
|
print(f"New delivery mappings : {len(newly_delivery)}")
|
|
print(f"Pending semantic review : {len(pending)} variants / {result['pending_semantic_review_feature_rows']} rows")
|
|
print(f"FSHHAV AREA hard-gate : {'PASS' if fish_ok else 'FAIL'}")
|
|
print(f"Delivery change contract: {'PASS' if changed_ok else 'FAIL'} {changed_classes}")
|
|
print(f"Semantic completeness : {result['semantic_completeness']}")
|
|
print(f"OVERALL classification : {result['overall']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|