254 lines
10 KiB
Python
254 lines
10 KiB
Python
"""Apply the versioned Planet display policy to audited admin-0 geometry.
|
|
|
|
This changes ownership of existing polygons only. Unavailable disputed-area
|
|
geometry must remain explicitly unresolved, never synthesized from a name.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
PROFILE = "china-pov-v1"
|
|
REQUIRED_RULES = {"china-taiwan-penghu", "china-hong-kong", "china-macao", "kosovo"}
|
|
CANONICAL = {
|
|
"china-taiwan-penghu": ("CHN", "CN", "China", "中国"),
|
|
"china-hong-kong": ("CHN", "CN", "China", "中国"),
|
|
"china-macao": ("CHN", "CN", "China", "中国"),
|
|
"kosovo": ("SRB", "RS", "Serbia", "塞尔维亚"),
|
|
}
|
|
|
|
|
|
def load_policy(path: Path) -> dict[str, Any]:
|
|
policy = json.loads(path.read_text(encoding="utf-8"))
|
|
ids = {rule.get("id") for rule in policy.get("rules", [])}
|
|
if policy.get("profile") != PROFILE or not REQUIRED_RULES <= ids:
|
|
raise ValueError(f"unsupported or incomplete Earth display policy: {path}")
|
|
return policy
|
|
|
|
|
|
def policy_digest(policy: dict[str, Any]) -> str:
|
|
encoded = json.dumps(policy, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _rule_for(properties: dict[str, Any], code_to_rule: dict[str, str]) -> str | None:
|
|
codes = {
|
|
str(properties.get(field) or "").upper()
|
|
for field in ("ISO_A3", "ADM0_A3", "BRK_A3", "SOV_A3")
|
|
}
|
|
matches = {code_to_rule[code] for code in codes if code in code_to_rule}
|
|
if len(matches) > 1:
|
|
raise ValueError(f"ambiguous Earth policy source codes: {sorted(codes)}")
|
|
return next(iter(matches), None)
|
|
|
|
|
|
def _polygons(geometry: dict[str, Any]) -> list[list[list[list[float]]]]:
|
|
return (
|
|
[geometry["coordinates"]]
|
|
if geometry["type"] == "Polygon"
|
|
else geometry["coordinates"]
|
|
)
|
|
|
|
|
|
def _ring_contains(ring: list[list[float]], lon: float, lat: float) -> bool:
|
|
inside = False
|
|
for index, point in enumerate(ring):
|
|
right_lon, right_lat = point[:2]
|
|
left_lon, left_lat = ring[index - 1][:2]
|
|
if (right_lat > lat) != (left_lat > lat) and lon < (
|
|
(left_lon - right_lon) * (lat - right_lat) / (left_lat - right_lat) + right_lon
|
|
):
|
|
inside = not inside
|
|
return inside
|
|
|
|
|
|
def _contains(geometry: dict[str, Any], lon: float, lat: float) -> bool:
|
|
return any(
|
|
_ring_contains(rings[0], lon, lat)
|
|
and all(not _ring_contains(hole, lon, lat) for hole in rings[1:])
|
|
for rings in _polygons(geometry)
|
|
)
|
|
|
|
|
|
def _apply_reviewed_geometry(
|
|
features: list[dict[str, Any]], package: dict[str, Any], policy: dict[str, Any]
|
|
) -> tuple[list[dict[str, Any]], set[str]]:
|
|
if package.get("schema") != "planet-earth-boundary-reviewed-geometry/v1":
|
|
raise ValueError("unsupported Earth boundary geometry package")
|
|
owners = package["countryGeometries"]
|
|
expected_country_ids = {
|
|
"CHN": 1159320471, "IND": 1159320847, "JPN": 1159320937,
|
|
"SRB": 1159321267, "PSE": 1159320899,
|
|
}
|
|
if package.get("countryFeatureIds") != expected_country_ids:
|
|
raise ValueError("Earth boundary geometry package country IDs changed")
|
|
absorbed_source_codes = {
|
|
code
|
|
for rule in policy["rules"]
|
|
if rule["id"] in {"china-taiwan-penghu", "kosovo"}
|
|
for code in rule.get("sourceIsoA3", [])
|
|
}
|
|
results = []
|
|
replaced = set()
|
|
for feature in features:
|
|
props = feature.get("properties") or {}
|
|
code = props.get("ISO_A3") if props.get("ISO_A3") == "PSE" else props.get("ADM0_A3")
|
|
if code in absorbed_source_codes | {"PGA", "SCR"}:
|
|
continue
|
|
if code in owners:
|
|
if code in replaced:
|
|
raise ValueError(f"duplicate source country geometry: {code}")
|
|
replaced.add(code)
|
|
feature = {**feature, "geometry": owners[code]}
|
|
if code == "PSE":
|
|
feature = {**feature, "properties": {**props, "ADM0_A3": "PSE"}}
|
|
results.append(feature)
|
|
if replaced != set(owners):
|
|
raise ValueError(f"missing source countries for reviewed geometry: {set(owners) - replaced}")
|
|
|
|
china = next(feature for feature in results if feature["properties"].get("ADM0_A3") == "CHN")
|
|
china_geometry = china["geometry"]
|
|
china_polygons = list(_polygons(china_geometry))
|
|
signatures = {json.dumps(polygon, separators=(",", ":")) for polygon in china_polygons}
|
|
added = {}
|
|
for rule, geometry in package["disputedGeometries"].items():
|
|
added[rule] = 0
|
|
for polygon in _polygons(geometry):
|
|
signature = json.dumps(polygon, separators=(",", ":"))
|
|
if signature in signatures:
|
|
continue
|
|
signatures.add(signature)
|
|
china_polygons.append(polygon)
|
|
added[rule] += 1
|
|
china["geometry"] = {"type": "MultiPolygon", "coordinates": china_polygons}
|
|
|
|
by_code = {feature["properties"].get("ISO_A3"): feature for feature in results}
|
|
checks = {
|
|
"china-zangnan": ("CHN", 93.85, 28.35),
|
|
"china-aksai-chin": ("CHN", 79.5, 35.5),
|
|
"china-taiwan-penghu": ("CHN", 121, 23.5),
|
|
"kosovo": ("SRB", 21.1, 42.6),
|
|
"gaza": ("PSE", 34.35, 31.4),
|
|
}
|
|
for rule, (code, lon, lat) in checks.items():
|
|
owners = {
|
|
feature["properties"].get("ISO_A3")
|
|
for feature in results
|
|
if _contains(feature["geometry"], lon, lat)
|
|
}
|
|
if code not in by_code or owners != {code}:
|
|
raise ValueError(f"reviewed geometry does not contain {rule} sample: {lon},{lat}")
|
|
expected_ids = {
|
|
"china-diaoyu-chiwei": 1159320959,
|
|
"china-paracels": 1159320477,
|
|
"china-spratlys": 1159321165,
|
|
"china-huangyan": 1159321227,
|
|
}
|
|
if package.get("disputedFeatureIds") != expected_ids or not all(added.values()):
|
|
raise ValueError("reviewed island geometry missing or IDs changed")
|
|
return results, set(checks) | {"china-diaoyu-chiwei"}
|
|
|
|
|
|
def apply_admin0_policy(
|
|
collection: dict[str, Any],
|
|
policy: dict[str, Any],
|
|
*,
|
|
require_regions: bool = True,
|
|
geometry_package: dict[str, Any] | None = None,
|
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
"""Return transformed polygons and an honest inventory of rule coverage."""
|
|
if collection.get("type") != "FeatureCollection":
|
|
raise ValueError("admin-0 source must be a FeatureCollection")
|
|
rule_ids = {rule["id"] for rule in policy["rules"]}
|
|
code_to_rule = {
|
|
code: rule["id"]
|
|
for rule in policy["rules"]
|
|
if rule["id"] in REQUIRED_RULES
|
|
for code in rule.get("sourceIsoA3", [])
|
|
}
|
|
source_features = collection.get("features", [])
|
|
geometry_rules: set[str] = set()
|
|
if geometry_package is not None:
|
|
source_features, geometry_rules = _apply_reviewed_geometry(
|
|
source_features, geometry_package, policy
|
|
)
|
|
counts = {rule: 0 for rule in REQUIRED_RULES}
|
|
features: list[dict[str, Any]] = []
|
|
for feature in source_features:
|
|
geometry = feature.get("geometry") or {}
|
|
if geometry.get("type") not in {"Polygon", "MultiPolygon"}:
|
|
raise ValueError("admin-0 features must have Polygon/MultiPolygon geometry")
|
|
properties = dict(feature.get("properties") or {})
|
|
rule = _rule_for(properties, code_to_rule)
|
|
if rule:
|
|
counts[rule] += 1
|
|
iso_a3, iso_a2, english, chinese = CANONICAL[rule]
|
|
properties.update(
|
|
ISO_A3=iso_a3,
|
|
ADM0_A3=iso_a3,
|
|
BRK_A3=iso_a3,
|
|
SOV_A3=iso_a3,
|
|
ISO_A2=iso_a2,
|
|
NAME=english,
|
|
NAME_EN=english,
|
|
NAME_ZH=chinese,
|
|
ADMIN=english,
|
|
SOVEREIGNT=english,
|
|
PLANET_POLICY_RULE=rule,
|
|
)
|
|
# Source-specific alternate naming/identifiers must not leak back
|
|
# into cards or a future tile client as a competing ownership.
|
|
for field in ("NAME_LONG", "BRK_NAME", "ADM0_ISO", "ADM0_TLC", "TLC"):
|
|
if field in properties:
|
|
properties[field] = english if field.endswith("NAME") or field == "NAME_LONG" else iso_a3
|
|
features.append({**feature, "properties": properties})
|
|
applied = sorted({rule for rule, count in counts.items() if count} | geometry_rules)
|
|
if require_regions and not REQUIRED_RULES <= set(applied):
|
|
raise ValueError(f"required policy polygon missing (source codes TWN/KOS): {counts}")
|
|
|
|
coverage = {
|
|
"profile": PROFILE,
|
|
"sha256": policy_digest(policy),
|
|
"applied": REQUIRED_RULES <= set(applied),
|
|
"appliedRuleIds": applied,
|
|
"unresolvedRuleIds": sorted(rule_ids - set(applied)),
|
|
"matchedFeatureCounts": counts,
|
|
"coverage": "partial" if rule_ids - set(applied) else "complete",
|
|
"publicationReady": False,
|
|
"unresolvedReasons": {
|
|
"china-south-china-sea-islands": (
|
|
"Natural Earth supplies reviewed Paracels, Spratlys and Scarborough polygons; "
|
|
"Dongsha land and Zhongsha reef/island geometry remain unverified. "
|
|
"Submerged Zengmu Ansha needs a verified geographic point annotation, not a land polygon."
|
|
),
|
|
"china-maritime-claim-line": (
|
|
"The configured source is disputed administrative boundary lines, not an audited "
|
|
"maritime claim line. No land polygon is inferred from a maritime claim."
|
|
),
|
|
},
|
|
}
|
|
if geometry_package is not None:
|
|
coverage["reviewedGeometrySourceCommit"] = geometry_package["sourceCommit"]
|
|
coverage["geometrySha256"] = policy_digest(geometry_package)
|
|
coverage["reviewedIslandComponents"] = {
|
|
"diaoyuChiwei": len(_polygons(geometry_package["disputedGeometries"]["china-diaoyu-chiwei"])),
|
|
"paracels": len(_polygons(geometry_package["disputedGeometries"]["china-paracels"])),
|
|
"spratlys": len(_polygons(geometry_package["disputedGeometries"]["china-spratlys"])),
|
|
"huangyan": len(_polygons(geometry_package["disputedGeometries"]["china-huangyan"])),
|
|
}
|
|
return {**collection, "features": features}, coverage
|
|
|
|
|
|
def policy_verified(manifest: dict[str, Any], policy: dict[str, Any]) -> bool:
|
|
coverage = manifest.get("povPolicy") or {}
|
|
return (
|
|
coverage.get("profile") == PROFILE
|
|
and coverage.get("sha256") == policy_digest(policy)
|
|
and coverage.get("applied") is True
|
|
and REQUIRED_RULES <= set(coverage.get("appliedRuleIds") or [])
|
|
)
|