shuishen
4 hours ago 01776511b66bfd87b4f8ef57d3fefc1d99e1e8f6
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
"""Convert irregular change polygons into product-friendly rectangles."""
 
from __future__ import annotations
 
import json
from pathlib import Path
 
from shapely.geometry import box, mapping, shape
 
 
def rectangularize_vector(source: Path, destination: Path, wgs84_destination: Path | None = None) -> int:
    # geoai.masks_to_vector may omit the file when a threshold produces no
    # features. Keep the artifact contract stable by materializing an empty
    # GeoJSON collection so downstream promotion and downloads still work.
    if source.is_file():
        payload = json.loads(source.read_text(encoding="utf-8"))
    else:
        payload = {"type": "FeatureCollection", "features": []}
        source.parent.mkdir(parents=True, exist_ok=True)
        source.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    rectangles = []
    for feature in payload.get("features", []):
        geometry = feature.get("geometry")
        if not geometry:
            continue
        source_shape = shape(geometry)
        if source_shape.is_empty:
            continue
        rectangle = box(*source_shape.bounds)
        properties = dict(feature.get("properties") or {})
        properties["geometry_mode"] = "axis_aligned_rectangle"
        properties["source_geometry_type"] = source_shape.geom_type
        rectangles.append({"type": "Feature", "properties": properties, "geometry": mapping(rectangle)})
    output = {"type": "FeatureCollection", "features": rectangles}
    if payload.get("crs"):
        output["crs"] = payload["crs"]
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(json.dumps(output, ensure_ascii=False, indent=2), encoding="utf-8")
    if wgs84_destination is not None and payload.get("crs") and rectangles:
        import geopandas as gpd
 
        frame = gpd.GeoDataFrame.from_features(output["features"], crs=payload["crs"].get("properties", {}).get("name"))
        if frame.crs:
            frame.to_crs("EPSG:4326").to_file(wgs84_destination, driver="GeoJSON")
    return len(rectangles)