Skills › Scientific Agent Skills › geopandas › geometry_validity_report.py
geometry_validity_report.py
227 lines · skills/geopandas/scripts/geometry_validity_report.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit and simulate repair of bounded local geometries with optional new output."""
4from __future__ import annotations
6import argparse
7import hashlib
8import sys
9from typing import Any
11from _common import (
12 DEFAULT_MAX_FEATURES,
13 DEFAULT_MAX_INPUT_BYTES,
14 DEFAULT_MAX_OUTPUT_BYTES,
15 CliError,
16 checked_input_file,
17 checked_output_file,
18 crs_summary,
19 duplicate_column_state,
20 emit_json,
21 fail_json,
22 finite_number,
23 geometry_state,
24 load_geodataframe,
25 native_versions,
26 package_versions,
27 positive_int,
28 sha256_file,
29 write_new_geopackage,
30)
32TOOL = "geometry_validity_report"
35def build_parser() -> argparse.ArgumentParser:
36 parser = argparse.ArgumentParser(
37 description=(
38 "Audit null/empty/invalid geometry and simulate make_valid locally. "
39 "Default is dry-run; --repair-output writes only a new GeoPackage."
40 ),
41 epilog=(
42 "No coordinates, paths, IDs, or validity-reason coordinates are emitted. "
43 "URLs, archives, symlinks, traversal, and overwrite are rejected."
44 ),
45 )
46 parser.add_argument("input", help="Allowlisted local GDAL vector file")
47 parser.add_argument("--root", default=".", help="Existing local I/O root")
48 parser.add_argument("--layer", help="Exact input layer name")
49 parser.add_argument("--id-column", help="Optional stable ID column to audit")
50 parser.add_argument(
51 "--method",
52 choices=("linework", "structure"),
53 default="linework",
54 help="Shapely make_valid algorithm (default: linework)",
55 )
56 parser.add_argument(
57 "--drop-collapsed",
58 action="store_true",
59 help="For structure repair only, drop lower-dimensional collapsed parts",
60 )
61 parser.add_argument(
62 "--precision-grid",
63 help=(
64 "Optional positive post-repair precision grid in projected CRS units; "
65 "geographic or missing CRS is rejected"
66 ),
67 )
68 parser.add_argument(
69 "--repair-output",
70 help="Optional new .gpkg path; existing paths are never overwritten",
71 )
72 parser.add_argument(
73 "--max-input-bytes",
74 type=positive_int,
75 default=DEFAULT_MAX_INPUT_BYTES,
76 )
77 parser.add_argument(
78 "--max-output-bytes",
79 type=positive_int,
80 default=DEFAULT_MAX_OUTPUT_BYTES,
81 )
82 parser.add_argument(
83 "--max-features",
84 type=positive_int,
85 default=DEFAULT_MAX_FEATURES,
86 )
87 return parser
90def _type_transition_count(before: Any, after: Any) -> int:
91 before_types = [
92 "missing" if value is None else str(value) for value in before.geom_type
93 ]
94 after_types = [
95 "missing" if value is None else str(value) for value in after.geom_type
96 ]
97 return sum(left != right for left, right in zip(before_types, after_types))
100def report(args: argparse.Namespace) -> dict[str, Any]:
101 if args.drop_collapsed and args.method != "structure":
102 raise CliError("--drop-collapsed is valid only with --method structure")
103 path = checked_input_file(
104 args.input,
105 root=args.root,
106 max_bytes=args.max_input_bytes,
107 )
108 frame = load_geodataframe(
109 path,
110 layer=args.layer,
111 max_features=args.max_features,
112 )
113 before_geometry = frame.geometry.copy()
114 before = geometry_state(frame)
115 id_state = duplicate_column_state(frame, args.id_column)
116 crs = crs_summary(frame.crs)
118 grid_size: float | None = None
119 if args.precision_grid is not None:
120 grid_size = finite_number(args.precision_grid, name="precision_grid")
121 if grid_size <= 0:
122 raise CliError("precision_grid must be greater than zero")
123 if not crs["present"] or crs["geographic"] or not crs["projected"]:
124 raise CliError("precision_grid requires a known projected CRS")
126 repaired_frame = frame.copy()
127 repaired_geometry = repaired_frame.geometry.make_valid(
128 method=args.method,
129 keep_collapsed=not args.drop_collapsed,
130 )
131 if grid_size is not None:
132 repaired_geometry = repaired_geometry.set_precision(
133 grid_size,
134 mode="valid_output",
135 )
136 repaired_frame = repaired_frame.set_geometry(repaired_geometry)
137 simulated = geometry_state(repaired_frame)
139 output_report: dict[str, Any] = {
140 "requested": args.repair_output is not None,
141 "written": False,
142 "path_emitted": False,
143 "basename_sha256": None,
144 "sha256": None,
145 "roundtrip_verified": False,
146 }
147 if args.repair_output is not None:
148 destination = checked_output_file(
149 args.repair_output,
150 root=args.root,
151 allowed_suffixes={".gpkg"},
152 )
153 write_new_geopackage(
154 repaired_frame,
155 destination,
156 max_output_bytes=args.max_output_bytes,
157 )
158 roundtrip = load_geodataframe(
159 destination,
160 layer="repaired",
161 max_features=args.max_features,
162 )
163 roundtrip_state = geometry_state(roundtrip)
164 if roundtrip_state != simulated:
165 destination.unlink(missing_ok=True)
166 raise CliError("roundtrip geometry-state verification failed; output removed")
167 output_report = {
168 "requested": True,
169 "written": True,
170 "path_emitted": False,
171 "basename_sha256": hashlib.sha256(
172 destination.name.encode("utf-8")
173 ).hexdigest(),
174 "sha256": sha256_file(destination),
175 "roundtrip_verified": True,
176 }
178 return {
179 "ok": True,
180 "tool": TOOL,
181 "dry_run": args.repair_output is None,
182 "source": {
183 "primary_file_sha256": sha256_file(path),
184 "primary_file_bytes": path.stat().st_size,
185 "path_emitted": False,
186 },
187 "crs": crs,
188 "stable_id_audit": id_state,
189 "before": before,
190 "simulated_after": simulated,
191 "repair_contract": {
192 "method": args.method,
193 "keep_collapsed": not args.drop_collapsed,
194 "precision_grid": grid_size,
195 "precision_grid_units": (
196 "source_projected_crs_units" if grid_size is not None else None
197 ),
198 "type_transition_rows": _type_transition_count(
199 before_geometry,
200 repaired_geometry,
201 ),
202 "source_modified": False,
203 },
204 "output": output_report,
205 "network_accessed": False,
206 "coordinates_emitted": False,
207 "identifiers_emitted": False,
208 "stack": {"packages": package_versions(), "native": native_versions()},
209 "warnings": [
210 "Repair can change type, dimension, component count, area, and emptiness.",
211 "Validity reasons were reduced to categories; defect coordinates were not emitted.",
212 "A written GeoPackage retains source attributes and coordinates and remains sensitive.",
213 ],
214 }
217def main(argv: list[str] | None = None) -> int:
218 args = build_parser().parse_args(argv)
219 try:
220 emit_json(report(args))
221 return 0
222 except Exception as exc: # noqa: BLE001 - errors are redacted at CLI boundary
223 return fail_json(TOOL, exc)
226if __name__ == "__main__":
227 sys.exit(main())