"""Materialize full GeoJSON artifacts for selected parameter-scan masks."""
|
|
from __future__ import annotations
|
|
import argparse
|
import json
|
from pathlib import Path
|
|
import importlib.util
|
|
|
def load_demo():
|
path = Path(__file__).with_name("run_change_detection.py")
|
spec = importlib.util.spec_from_file_location("change_detection_demo", path)
|
if spec is None or spec.loader is None:
|
raise RuntimeError("Unable to load change-detection vector helpers.")
|
module = importlib.util.module_from_spec(spec)
|
spec.loader.exec_module(module)
|
return module
|
|
|
def source_probability_path(scan_dir: Path) -> Path:
|
summary = json.loads((scan_dir / "scan_summary.json").read_text(encoding="utf-8"))
|
source_run = str(summary.get("source_run") or "")
|
output_root = scan_dir.parent.parent if scan_dir.parent.name == "parameter-scans" else scan_dir.parent
|
probability = output_root / "runs" / source_run / "change_probability.tif"
|
if not probability.is_file():
|
probability = output_root / source_run / "change_probability.tif"
|
if not probability.is_file():
|
raise FileNotFoundError("The source run probability raster is unavailable for vector export.")
|
return probability
|
|
|
def materialize(scan_dir: Path, candidates: list[str]) -> list[dict]:
|
from rectangularize_vectors import rectangularize_vector
|
|
demo = load_demo()
|
probability_path = source_probability_path(scan_dir)
|
results: list[dict] = []
|
for candidate in candidates:
|
item_dir = scan_dir / candidate
|
mask_path = item_dir / "change_mask.tif"
|
summary_path = item_dir / "summary.json"
|
if not mask_path.is_file() or not summary_path.is_file():
|
raise FileNotFoundError(f"Missing scan result: {candidate}")
|
summary = json.loads(summary_path.read_text(encoding="utf-8"))
|
vector_path = item_dir / "changes.geojson"
|
_, vector_count = demo.vectorize_cleaned_mask(mask_path, probability_path, vector_path)
|
rectangle_path = item_dir / "changes_rectangles.geojson"
|
rectangle_wgs84_path = item_dir / "changes_rectangles_wgs84.geojson"
|
rectangle_count = rectangularize_vector(vector_path, rectangle_path, rectangle_wgs84_path)
|
record = {
|
**summary,
|
"full_vector_feature_count": vector_count,
|
"full_vector": vector_path.name,
|
"rectangle_vector_feature_count": rectangle_count,
|
"rectangle_vector": rectangle_path.name,
|
"rectangle_vector_wgs84": rectangle_wgs84_path.name if rectangle_wgs84_path.is_file() else None,
|
}
|
(item_dir / "full_result.json").write_text(json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8")
|
results.append(record)
|
output_path = scan_dir / "candidate_results.json"
|
output_path.write_text(json.dumps({"candidates": results}, ensure_ascii=False, indent=2), encoding="utf-8")
|
return results
|
|
|
def main() -> None:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--scan-dir", type=Path, required=True)
|
parser.add_argument(
|
"--candidate",
|
action="append",
|
dest="candidates",
|
default=None,
|
)
|
args = parser.parse_args()
|
candidates = args.candidates or [
|
"threshold-0.40_area-256",
|
"threshold-0.40_area-64",
|
"threshold-0.30_area-256",
|
]
|
print(json.dumps({"candidates": materialize(args.scan_dir, candidates)}, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
