187 lines
7.9 KiB
Python
187 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Build the checked, compact country fallback from existing source geometry."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from earth_boundary_policy import REQUIRED_RULES, apply_admin0_policy, load_policy
|
|
|
|
|
|
def point_in_ring(lon: float, lat: float, ring: list[list[float]]) -> bool:
|
|
inside = False
|
|
for index, (right_lon, right_lat, *_) in enumerate(ring):
|
|
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(feature: dict[str, Any], lon: float, lat: float) -> bool:
|
|
geometry = feature["geometry"]
|
|
polygons = (
|
|
[geometry["coordinates"]]
|
|
if geometry["type"] == "Polygon"
|
|
else geometry["coordinates"]
|
|
)
|
|
return any(
|
|
point_in_ring(lon, lat, rings[0])
|
|
and all(not point_in_ring(lon, lat, hole) for hole in rings[1:])
|
|
for rings in polygons
|
|
)
|
|
|
|
|
|
def source_islands(admin0: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""Existing small China polygons in the Shandong/Liaodong screenshot area."""
|
|
china = [
|
|
feature for feature in admin0["features"]
|
|
if (feature.get("properties") or {}).get("ADM0_A3") == "CHN"
|
|
]
|
|
if len(china) != 1 or china[0]["geometry"]["type"] != "MultiPolygon":
|
|
raise ValueError("audited China multipolygon not found exactly once")
|
|
islands = []
|
|
for index, rings in enumerate(china[0]["geometry"]["coordinates"]):
|
|
xs = [point[0] for point in rings[0]]
|
|
ys = [point[1] for point in rings[0]]
|
|
if (
|
|
118 <= min(xs) <= 123
|
|
and 36 <= min(ys) <= 40
|
|
and max(xs) - min(xs) < 0.5
|
|
and max(ys) - min(ys) < 0.5
|
|
):
|
|
islands.append({
|
|
"type": "Feature",
|
|
"properties": {
|
|
"NAME": "China", "NAME_EN": "China", "NAME_ZH": "中国",
|
|
"ADMIN": "China", "ISO_A3": "CHN", "ADM0_A3": "CHN",
|
|
"ISO_A2": "CN", "PLANET_GEOMETRY_SUPPLEMENT": "shandong-liaodong-islands",
|
|
"PLANET_SOURCE_NE_ID": china[0]["properties"].get("NE_ID"),
|
|
"PLANET_SOURCE_COMPONENT": index,
|
|
},
|
|
"geometry": {"type": "Polygon", "coordinates": rings},
|
|
})
|
|
if not islands:
|
|
raise ValueError("audited Shandong/Liaodong island polygons unavailable")
|
|
return islands
|
|
|
|
|
|
def source_jeju(admin0: dict[str, Any]) -> dict[str, Any]:
|
|
korea = [
|
|
feature for feature in admin0["features"]
|
|
if (feature.get("properties") or {}).get("ADM0_A3") == "KOR"
|
|
and (feature.get("properties") or {}).get("NE_ID") == 1159320985
|
|
]
|
|
if len(korea) != 1 or korea[0]["geometry"]["type"] != "MultiPolygon":
|
|
raise ValueError("audited South Korea multipolygon not found exactly once")
|
|
matches = []
|
|
for index, rings in enumerate(korea[0]["geometry"]["coordinates"]):
|
|
candidate = {"type": "Feature", "geometry": {"type": "Polygon", "coordinates": rings}}
|
|
if contains(candidate, 126.53, 33.36):
|
|
matches.append((index, rings))
|
|
if len(matches) != 1:
|
|
raise ValueError("audited Jeju polygon not found exactly once")
|
|
index, rings = matches[0]
|
|
xs = [point[0] for point in rings[0]]
|
|
ys = [point[1] for point in rings[0]]
|
|
if not (126.1 < min(xs) < 126.2 and 126.9 < max(xs) < 127.0
|
|
and 33.1 < min(ys) < 33.3 and 33.5 < max(ys) < 33.7):
|
|
raise ValueError("audited Jeju polygon bounds changed")
|
|
return {
|
|
"type": "Feature",
|
|
"properties": {
|
|
"NAME": "South Korea", "NAME_EN": "South Korea", "NAME_ZH": "韩国",
|
|
"ADMIN": "South Korea", "ISO_A3": "KOR", "ADM0_A3": "KOR",
|
|
"ISO_A2": "KR", "PLANET_GEOMETRY_SUPPLEMENT": "jeju-island",
|
|
"PLANET_SOURCE_NE_ID": 1159320985,
|
|
"PLANET_SOURCE_COMPONENT": index,
|
|
},
|
|
"geometry": {"type": "Polygon", "coordinates": rings},
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--source", type=Path, required=True)
|
|
parser.add_argument("--policy", type=Path, required=True)
|
|
parser.add_argument("--audited-admin0", type=Path, required=True)
|
|
parser.add_argument("--pov-geometry", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
|
|
policy = load_policy(args.policy)
|
|
source = json.loads(args.source.read_text(encoding="utf-8"))
|
|
audited = json.loads(args.audited_admin0.read_text(encoding="utf-8"))
|
|
supplements = {
|
|
code: [feature for feature in audited["features"]
|
|
if (feature.get("properties") or {}).get("ADM0_A3") == code
|
|
and (feature.get("properties") or {}).get("NE_ID") == ne_id]
|
|
for code, ne_id in (("HKG", 1159320473), ("MAC", 1159320475))
|
|
}
|
|
if any(len(features) != 1 for features in supplements.values()):
|
|
raise ValueError("audited Hong Kong/Macao polygons not found exactly once")
|
|
source["features"].extend([supplements["HKG"][0], supplements["MAC"][0]])
|
|
geometry_package = json.loads(args.pov_geometry.read_text(encoding="utf-8"))
|
|
result, coverage = apply_admin0_policy(source, policy, geometry_package=geometry_package)
|
|
if any(contains(feature, 126.53, 33.36) for feature in result["features"]):
|
|
raise ValueError("fallback already covers Jeju; refusing duplicate island geometry")
|
|
jeju = source_jeju(audited)
|
|
result["features"].append(jeju)
|
|
island_features = source_islands(audited)
|
|
coverage["geometrySupplement"] = {
|
|
"regions": ["china-hong-kong", "china-macao"],
|
|
"source": "Natural Earth 1:10m admin-0 countries",
|
|
"sourceSha256": hashlib.sha256(args.audited_admin0.read_bytes()).hexdigest(),
|
|
"featureNeId": 1159320475,
|
|
"hongKongFeatureNeId": 1159320473,
|
|
"shandongLiaodongIslandFeaturesInReviewedChina": len(island_features),
|
|
"jejuFeatureNeId": 1159320985,
|
|
"jejuComponent": jeju["properties"]["PLANET_SOURCE_COMPONENT"],
|
|
"jejuVertices": len(jeju["geometry"]["coordinates"][0]),
|
|
}
|
|
# These samples confirm the fallback has the required regions after the
|
|
# reviewed geometry package is applied; they do not prove all tiny reefs.
|
|
inherited = {
|
|
"china-hong-kong": ("CHN", 114.1694, 22.3193),
|
|
"kosovo": ("SRB", 21.1, 42.6),
|
|
}
|
|
for rule, (iso, lon, lat) in inherited.items():
|
|
if any(
|
|
feature["properties"].get("ISO_A3") == iso and contains(feature, lon, lat)
|
|
for feature in result["features"]
|
|
):
|
|
if rule not in coverage["appliedRuleIds"]:
|
|
coverage["appliedRuleIds"].append(rule)
|
|
else:
|
|
raise ValueError(f"fallback polygon does not cover {rule} sample at {lon},{lat}")
|
|
coverage["appliedRuleIds"].sort()
|
|
if not any(
|
|
feature["properties"].get("PLANET_POLICY_RULE") == "china-macao"
|
|
and contains(feature, 113.5439, 22.1987)
|
|
for feature in result["features"]
|
|
):
|
|
raise ValueError("audited Macao polygon does not cover its sample")
|
|
if not REQUIRED_RULES <= set(coverage["appliedRuleIds"]):
|
|
raise ValueError("fallback is missing a required display-policy region")
|
|
coverage["applied"] = True
|
|
coverage["coverage"] = "partial"
|
|
coverage["unresolvedRuleIds"] = sorted(
|
|
{rule["id"] for rule in policy["rules"]} - set(coverage["appliedRuleIds"])
|
|
)
|
|
coverage["inheritedOwnershipSamples"] = list(inherited)
|
|
result["povPolicy"] = coverage
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(
|
|
json.dumps(result, ensure_ascii=False, separators=(",", ":")) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|