Skills › Scientific Agent Skills › geopandas › _common.py
_common.py
616 lines · skills/geopandas/scripts/_common.py at 5f2bc03
1"""Shared local-only safety and reporting helpers for GeoPandas CLIs."""
3from __future__ import annotations
5import hashlib
6import json
7import math
8import os
9import re
10import tempfile
11from importlib.metadata import PackageNotFoundError, version
12from pathlib import Path
13from typing import Any
15PINNED_STACK = {
16 "geopandas": "1.2.0",
17 "numpy": "2.5.3",
18 "packaging": "26.3",
19 "pandas": "3.0.6",
20 "pyarrow": "25.0.1",
21 "pyogrio": "0.13.0",
22 "pyproj": "3.8.0",
23 "shapely": "2.1.2",
24}
25DEFAULT_MAX_INPUT_BYTES = 64 * 1024 * 1024
26DEFAULT_MAX_OUTPUT_BYTES = 256 * 1024 * 1024
27DEFAULT_MAX_FEATURES = 100_000
28ABSOLUTE_MAX_INPUT_BYTES = 512 * 1024 * 1024
29ABSOLUTE_MAX_OUTPUT_BYTES = 1024 * 1024 * 1024
30ABSOLUTE_MAX_FEATURES = 1_000_000
31GDAL_SUFFIXES = {".fgb", ".geojson", ".gpkg", ".json", ".shp"}
32PARQUET_SUFFIXES = {".geoparquet", ".parquet"}
33FEATHER_SUFFIXES = {".arrow", ".feather"}
34ALLOWED_INPUT_SUFFIXES = GDAL_SUFFIXES | PARQUET_SUFFIXES | FEATHER_SUFFIXES
35ARCHIVE_SUFFIXES = {
36 ".7z",
37 ".bz2",
38 ".gz",
39 ".kmz",
40 ".rar",
41 ".tar",
42 ".tgz",
43 ".xz",
44 ".zip",
45}
46SENSITIVE_FIELD_TOKENS = {
47 "address",
48 "coordinate",
49 "email",
50 "latitude",
51 "location",
52 "longitude",
53 "parcel",
54 "phone",
55 "postcode",
56 "trajectory",
57 "zipcode",
58}
61class CliError(ValueError):
62 """A bounded, user-facing CLI error."""
65def positive_int(value: str) -> int:
66 """Argparse converter for positive integers."""
67 try:
68 number = int(value)
69 except ValueError as exc:
70 raise ValueError("must be an integer") from exc
71 if number < 1:
72 raise ValueError("must be at least 1")
73 return number
76def bounded_limit(value: int, *, name: str, maximum: int) -> int:
77 """Validate a positive resource limit against a hard ceiling."""
78 if value < 1:
79 raise CliError(f"{name} must be at least 1")
80 if value > maximum:
81 raise CliError(f"{name} may not exceed {maximum}")
82 return value
85def finite_number(value: Any, *, name: str) -> float:
86 """Return one finite floating-point value."""
87 try:
88 number = float(value)
89 except (TypeError, ValueError) as exc:
90 raise CliError(f"{name} must be a finite number") from exc
91 if not math.isfinite(number):
92 raise CliError(f"{name} must be finite")
93 return number
96def reject_nonlocal(value: str, *, label: str) -> None:
97 """Reject URLs, GDAL virtual filesystems, archives, and path tricks."""
98 lowered = value.casefold().strip()
99 if not lowered or "\x00" in value:
100 raise CliError(f"{label} is empty or invalid")
101 if "://" in lowered or lowered.startswith(("/vsi", "vsi")):
102 raise CliError(f"{label} must be a local path, not a URL or GDAL VSI path")
103 if "!" in value or lowered.startswith(("zip+", "tar+")):
104 raise CliError(f"{label} may not address an archive member")
105 suffixes = {suffix.casefold() for suffix in Path(lowered).suffixes}
106 if suffixes & ARCHIVE_SUFFIXES:
107 raise CliError(f"{label} may not be an archive or compressed file")
110def _root_path(value: str | Path) -> Path:
111 raw = str(value)
112 reject_nonlocal(raw, label="root")
113 candidate = Path(raw).expanduser()
114 if candidate.is_symlink():
115 raise CliError("root may not be a symbolic link")
116 try:
117 root = candidate.resolve(strict=True)
118 except OSError as exc:
119 raise CliError("root does not exist or cannot be resolved") from exc
120 if not root.is_dir():
121 raise CliError("root must be a directory")
122 return root
125def _reject_symlink_components(candidate: Path, root: Path) -> None:
126 current = candidate
127 while True:
128 if current.is_symlink():
129 raise CliError("symbolic-link paths are not accepted")
130 if current == root:
131 return
132 parent = current.parent
133 if parent == current:
134 raise CliError("path escapes the configured root")
135 current = parent
138def _candidate_path(value: str | Path, root: Path, *, label: str) -> Path:
139 raw = str(value)
140 reject_nonlocal(raw, label=label)
141 supplied = Path(raw).expanduser()
142 if ".." in supplied.parts:
143 raise CliError(f"{label} may not contain '..'")
144 candidate = supplied if supplied.is_absolute() else root / supplied
145 resolved = candidate.resolve(strict=False)
146 if not resolved.is_relative_to(root):
147 raise CliError(f"{label} must remain inside the configured root")
148 _reject_symlink_components(candidate, root)
149 return candidate
152def checked_input_file(
153 value: str | Path,
154 *,
155 root: str | Path = ".",
156 max_bytes: int = DEFAULT_MAX_INPUT_BYTES,
157 allowed_suffixes: set[str] | None = None,
158) -> Path:
159 """Resolve one bounded, allowlisted local regular input file."""
160 max_bytes = bounded_limit(
161 max_bytes,
162 name="max_input_bytes",
163 maximum=ABSOLUTE_MAX_INPUT_BYTES,
164 )
165 resolved_root = _root_path(root)
166 candidate = _candidate_path(value, resolved_root, label="input")
167 try:
168 path = candidate.resolve(strict=True)
169 except OSError as exc:
170 raise CliError("input does not exist or cannot be resolved") from exc
171 if not path.is_file():
172 raise CliError("input must be a regular file")
173 suffixes = allowed_suffixes or ALLOWED_INPUT_SUFFIXES
174 if path.suffix.casefold() not in suffixes:
175 raise CliError("input suffix is not on the vector-data allowlist")
176 size = path.stat().st_size
177 if size > max_bytes:
178 raise CliError(f"input exceeds the {max_bytes}-byte limit")
179 return path
182def checked_output_file(
183 value: str | Path,
184 *,
185 root: str | Path = ".",
186 allowed_suffixes: set[str],
187) -> Path:
188 """Resolve a new local output path without creating or overwriting it."""
189 resolved_root = _root_path(root)
190 candidate = _candidate_path(value, resolved_root, label="output")
191 if candidate.suffix.casefold() not in allowed_suffixes:
192 raise CliError("output suffix is not allowed for this operation")
193 if candidate.exists() or candidate.is_symlink():
194 raise CliError("output already exists; choose a new path")
195 parent = candidate.parent
196 if parent.is_symlink() or not parent.exists() or not parent.is_dir():
197 raise CliError("output parent must be an existing non-symlink directory")
198 return candidate.resolve(strict=False)
201def sha256_file(path: Path) -> str:
202 """Hash a bounded local file without loading it into memory."""
203 digest = hashlib.sha256()
204 with path.open("rb") as handle:
205 while chunk := handle.read(1024 * 1024):
206 digest.update(chunk)
207 return digest.hexdigest()
210def package_versions() -> dict[str, str | None]:
211 """Return distribution versions without importing optional packages."""
212 found: dict[str, str | None] = {}
213 for name in PINNED_STACK:
214 try:
215 found[name] = version(name)
216 except PackageNotFoundError:
217 found[name] = None
218 return found
221def native_versions() -> dict[str, str | None]:
222 """Return native-library versions after callers have accepted the boundary."""
223 try:
224 import pyogrio
225 import pyproj
226 import shapely
227 except ImportError as exc:
228 raise CliError(
229 "runtime analysis requires the pinned GeoPandas stack from SKILL.md"
230 ) from exc
231 gdal_geos = getattr(pyogrio, "__gdal_geos_version__", None)
232 return {
233 "gdal": ".".join(str(part) for part in pyogrio.__gdal_version__),
234 "gdal_geos": (
235 ".".join(str(part) for part in gdal_geos) if gdal_geos else None
236 ),
237 "geos": str(shapely.geos_version_string),
238 "proj": str(pyproj.proj_version_str),
239 }
242def disable_proj_network() -> None:
243 """Disable PROJ network access for this process."""
244 try:
245 from pyproj import network
246 except ImportError as exc:
247 raise CliError("pyproj is required for CRS analysis") from exc
248 network.set_network_enabled(False)
251def crs_summary(value: Any) -> dict[str, Any]:
252 """Summarize CRS semantics without serializing sensitive data extents."""
253 try:
254 from pyproj import CRS
255 except ImportError as exc:
256 raise CliError("pyproj is required for CRS analysis") from exc
257 if value is None:
258 return {
259 "present": False,
260 "authority": None,
261 "geographic": None,
262 "projected": None,
263 "axes": [],
264 }
265 try:
266 crs = CRS.from_user_input(value)
267 except (TypeError, ValueError) as exc:
268 raise CliError("CRS metadata cannot be parsed") from exc
269 authority = crs.to_authority()
270 axes = []
271 for axis in crs.axis_info:
272 factor = axis.unit_conversion_factor
273 axes.append(
274 {
275 "abbreviation": axis.abbrev,
276 "direction": axis.direction,
277 "unit": axis.unit_name,
278 "unit_to_si": float(factor) if factor is not None else None,
279 }
280 )
281 return {
282 "present": True,
283 "authority": f"{authority[0]}:{authority[1]}" if authority else None,
284 "geographic": bool(crs.is_geographic),
285 "projected": bool(crs.is_projected),
286 "axes": axes,
287 }
290def _json_scalar(value: Any) -> Any:
291 """Convert common array scalars to strict JSON values."""
292 if value is None or isinstance(value, (bool, int, str)):
293 return value
294 if isinstance(value, float):
295 return value if math.isfinite(value) else None
296 if hasattr(value, "item"):
297 return _json_scalar(value.item())
298 return str(value)
301def _strict_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
302 result: dict[str, Any] = {}
303 for key, value in pairs:
304 if key in result:
305 raise CliError(f"duplicate JSON metadata key: {key!r}")
306 result[key] = value
307 return result
310def _reject_constant(value: str) -> None:
311 raise CliError(f"non-finite JSON metadata value is not allowed: {value}")
314def inspect_geoparquet(path: Path) -> dict[str, Any]:
315 """Inspect Parquet metadata without reading feature coordinates."""
316 try:
317 from pyarrow import parquet
318 except ImportError as exc:
319 raise CliError("PyArrow is required to inventory GeoParquet") from exc
320 try:
321 parquet_file = parquet.ParquetFile(path)
322 except (OSError, ValueError) as exc:
323 raise CliError("PyArrow could not read Parquet metadata") from exc
324 file_metadata = parquet_file.metadata
325 key_values = file_metadata.metadata or {}
326 raw_geo = key_values.get(b"geo")
327 geo: dict[str, Any] | None = None
328 if raw_geo is not None:
329 try:
330 decoded = raw_geo.decode("utf-8")
331 parsed = json.loads(
332 decoded,
333 object_pairs_hook=_strict_object,
334 parse_constant=_reject_constant,
335 )
336 except (UnicodeDecodeError, json.JSONDecodeError) as exc:
337 raise CliError("GeoParquet geo metadata is invalid JSON") from exc
338 if not isinstance(parsed, dict):
339 raise CliError("GeoParquet geo metadata must be an object")
340 geo = parsed
341 columns = geo.get("columns", {}) if geo else {}
342 if columns is not None and not isinstance(columns, dict):
343 raise CliError("GeoParquet columns metadata must be an object")
344 if geo is not None:
345 if not isinstance(geo.get("version"), str):
346 raise CliError("GeoParquet metadata requires a version string")
347 primary = geo.get("primary_column")
348 if not isinstance(primary, str) or not columns or primary not in columns:
349 raise CliError("GeoParquet metadata requires a declared primary geometry column")
350 if any(name not in parquet_file.schema_arrow.names for name in columns):
351 raise CliError("GeoParquet geometry metadata refers to an absent column")
352 crs_states = {"missing": 0, "null": 0, "present": 0}
353 covering_columns = 0
354 geometry_types: dict[str, int] = {}
355 for metadata in (columns or {}).values():
356 if not isinstance(metadata, dict):
357 raise CliError("GeoParquet geometry column metadata must be an object")
358 if "crs" not in metadata:
359 crs_states["missing"] += 1
360 elif metadata["crs"] is None:
361 crs_states["null"] += 1
362 else:
363 crs_states["present"] += 1
364 if metadata.get("covering") is not None:
365 covering_columns += 1
366 for geometry_type in metadata.get("geometry_types", []):
367 key = str(geometry_type)
368 geometry_types[key] = geometry_types.get(key, 0) + 1
369 arrow_types: dict[str, int] = {}
370 for field in parquet_file.schema_arrow:
371 key = str(field.type)
372 arrow_types[key] = arrow_types.get(key, 0) + 1
373 return {
374 "kind": "geoparquet" if geo else "parquet_without_geo_metadata",
375 "declared_features": int(file_metadata.num_rows),
376 "row_groups": int(file_metadata.num_row_groups),
377 "field_count": len(parquet_file.schema_arrow),
378 "field_type_counts": dict(sorted(arrow_types.items())),
379 "geometry_column_count": len(columns or {}),
380 "primary_geometry_declared": bool(geo and geo.get("primary_column")),
381 "schema_version": _json_scalar(geo.get("version")) if geo else None,
382 "crs_metadata_states": crs_states,
383 "geometry_type_metadata_counts": dict(sorted(geometry_types.items())),
384 "covering_geometry_columns": covering_columns,
385 "bounds_redacted": True,
386 "feature_data_loaded": False,
387 "dataframe_attrs_present": b"PANDAS_ATTRS" in key_values,
388 }
391def inspect_gdal_vector(path: Path, *, layer: str | None = None) -> dict[str, Any]:
392 """Inspect one GDAL vector layer without reading feature coordinates."""
393 try:
394 import pyogrio
395 except ImportError as exc:
396 raise CliError("pyogrio is required to inventory this vector format") from exc
397 try:
398 layers = pyogrio.list_layers(path)
399 except (OSError, RuntimeError, ValueError) as exc:
400 raise CliError("GDAL could not enumerate local vector layers") from exc
401 layer_count = len(layers)
402 if layer_count > 1 and layer is None:
403 raise CliError("multi-layer input requires an explicit --layer")
404 try:
405 info = pyogrio.read_info(
406 path,
407 layer=layer,
408 force_feature_count=False,
409 force_total_bounds=False,
410 )
411 except (OSError, RuntimeError, ValueError) as exc:
412 raise CliError("GDAL could not inspect the selected local layer") from exc
413 fields = [str(item) for item in info.get("fields", [])]
414 dtypes: dict[str, int] = {}
415 for dtype in info.get("dtypes", []):
416 key = str(dtype)
417 dtypes[key] = dtypes.get(key, 0) + 1
418 sensitive_names = 0
419 for field in fields:
420 normalized = re.sub(r"[^a-z0-9]+", "_", field.casefold())
421 if any(token in normalized for token in SENSITIVE_FIELD_TOKENS):
422 sensitive_names += 1
423 declared = info.get("features", -1)
424 try:
425 declared_features = int(declared)
426 except (TypeError, ValueError):
427 declared_features = -1
428 return {
429 "kind": "gdal_vector",
430 "driver": _json_scalar(info.get("driver")),
431 "layer_count": layer_count,
432 "layer_explicitly_selected": layer is not None,
433 "declared_features": declared_features if declared_features >= 0 else None,
434 "field_count": len(fields),
435 "field_type_counts": dict(sorted(dtypes.items())),
436 "sensitive_field_name_count": sensitive_names,
437 "geometry_type": _json_scalar(info.get("geometry_type")),
438 "encoding": _json_scalar(info.get("encoding")),
439 "crs": crs_summary(info.get("crs")),
440 "bounds_available": info.get("total_bounds") is not None,
441 "bounds_redacted": True,
442 "feature_data_loaded": False,
443 }
446def inspect_local_vector(path: Path, *, layer: str | None = None) -> dict[str, Any]:
447 """Dispatch to a metadata-only local vector inventory."""
448 suffix = path.suffix.casefold()
449 if suffix in PARQUET_SUFFIXES:
450 if layer is not None:
451 raise CliError("--layer is not valid for GeoParquet")
452 return inspect_geoparquet(path)
453 if suffix in FEATHER_SUFFIXES:
454 raise CliError("Feather/Arrow inventory is not implemented; convert a vetted copy")
455 return inspect_gdal_vector(path, layer=layer)
458def load_geodataframe(
459 path: Path,
460 *,
461 layer: str | None,
462 max_features: int,
463) -> Any:
464 """Load at most max_features local geometries from an allowlisted format."""
465 max_features = bounded_limit(
466 max_features,
467 name="max_features",
468 maximum=ABSOLUTE_MAX_FEATURES,
469 )
470 if path.suffix.casefold() not in GDAL_SUFFIXES:
471 raise CliError("this analysis accepts only allowlisted GDAL vector files")
472 try:
473 import geopandas as gpd
474 except ImportError as exc:
475 raise CliError("GeoPandas is required for geometry analysis") from exc
476 info = inspect_gdal_vector(path, layer=layer)
477 declared = info["declared_features"]
478 if declared is not None and declared > max_features:
479 raise CliError(f"input declares more than {max_features} features")
480 try:
481 frame = gpd.read_file(
482 path,
483 layer=layer,
484 engine="pyogrio",
485 rows=slice(0, max_features + 1),
486 use_arrow=True,
487 )
488 except (OSError, RuntimeError, TypeError, ValueError) as exc:
489 raise CliError("GeoPandas could not read the bounded local layer") from exc
490 if not isinstance(frame, gpd.GeoDataFrame):
491 raise CliError("selected layer has no geometry column")
492 if not frame.columns.is_unique:
493 raise CliError("duplicate input column names are not accepted")
494 if len(frame) > max_features:
495 raise CliError(f"input exceeds the {max_features}-feature limit")
496 return frame
499def geometry_state(frame: Any) -> dict[str, Any]:
500 """Return aggregate geometry state without values, IDs, or coordinates."""
501 geometry = frame.geometry
502 missing = geometry.isna()
503 empty = geometry.is_empty
504 invalid = (~missing) & (~empty) & (~geometry.is_valid)
505 type_counts: dict[str, int] = {}
506 for value in geometry.geom_type:
507 key = "missing" if value is None else str(value)
508 type_counts[key] = type_counts.get(key, 0) + 1
509 reason_counts: dict[str, int] = {}
510 if bool(invalid.any()):
511 for reason in geometry[invalid].is_valid_reason():
512 category = str(reason).split("[", 1)[0].strip()[:120]
513 reason_counts[category] = reason_counts.get(category, 0) + 1
514 has_m = getattr(geometry, "has_m", None)
515 return {
516 "features": len(frame),
517 "missing": int(missing.sum()),
518 "empty": int(empty.sum()),
519 "invalid": int(invalid.sum()),
520 "valid_nonempty": int((~missing & ~empty & ~invalid).sum()),
521 "geometry_type_counts": dict(sorted(type_counts.items())),
522 "validity_reason_categories": dict(sorted(reason_counts.items())),
523 "has_z": int(geometry.has_z.fillna(False).sum()),
524 "has_m": int(has_m.fillna(False).sum()) if has_m is not None else None,
525 "duplicate_index_rows": int(frame.index.duplicated(keep=False).sum()),
526 "coordinates_emitted": False,
527 "identifiers_emitted": False,
528 }
531def duplicate_column_state(frame: Any, name: str | None) -> dict[str, Any]:
532 """Report null/duplicate counts for an optional ID without emitting values."""
533 if name is None:
534 return {"provided": False, "null_rows": None, "duplicate_rows": None}
535 if name not in frame.columns:
536 raise CliError("requested ID column is not present")
537 values = frame[name]
538 return {
539 "provided": True,
540 "null_rows": int(values.isna().sum()),
541 "duplicate_rows": int(values.duplicated(keep=False).sum()),
542 }
545def write_new_geopackage(
546 frame: Any,
547 destination: Path,
548 *,
549 max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES,
550) -> None:
551 """Write a single-layer GeoPackage and link it atomically to a new path."""
552 max_output_bytes = bounded_limit(
553 max_output_bytes,
554 name="max_output_bytes",
555 maximum=ABSOLUTE_MAX_OUTPUT_BYTES,
556 )
557 descriptor, temporary_name = tempfile.mkstemp(
558 prefix=".geopandas-repair-",
559 suffix=".gpkg",
560 dir=destination.parent,
561 )
562 os.close(descriptor)
563 temporary = Path(temporary_name)
564 try:
565 temporary.unlink()
566 frame.to_file(
567 temporary,
568 layer="repaired",
569 driver="GPKG",
570 engine="pyogrio",
571 index=False,
572 use_arrow=True,
573 )
574 if temporary.stat().st_size > max_output_bytes:
575 raise CliError(f"derived output exceeds the {max_output_bytes}-byte limit")
576 try:
577 os.link(temporary, destination)
578 except FileExistsError as exc:
579 raise CliError("output appeared concurrently; nothing was overwritten") from exc
580 finally:
581 temporary.unlink(missing_ok=True)
584def json_text(payload: Any) -> str:
585 """Serialize strict, deterministic JSON."""
586 return json.dumps(
587 payload,
588 allow_nan=False,
589 ensure_ascii=False,
590 indent=2,
591 sort_keys=True,
592 )
595def emit_json(payload: Any) -> None:
596 """Print a strict JSON report."""
597 print(json_text(payload))
600def fail_json(tool: str, exc: BaseException) -> int:
601 """Emit a redacted error and return the CLI usage/data error status."""
602 if isinstance(exc, CliError):
603 message = str(exc)
604 else:
605 message = f"{type(exc).__name__}: operation failed; details redacted"
606 emit_json(
607 {
608 "ok": False,
609 "tool": tool,
610 "error": message[:500],
611 "network_accessed": False,
612 "coordinates_emitted": False,
613 "identifiers_emitted": False,
614 }
615 )
616 return 2