81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Shared GDAL helpers for the GeoData exporters."""
|
|
from __future__ import annotations
|
|
|
|
import glob
|
|
import os
|
|
from typing import Iterable, Sequence
|
|
|
|
from osgeo import gdal
|
|
|
|
gdal.UseExceptions()
|
|
|
|
|
|
def ensure_dir(path: str) -> None:
|
|
"""Create a directory path if missing."""
|
|
os.makedirs(path, exist_ok=True)
|
|
|
|
|
|
def ensure_parent(path: str) -> None:
|
|
"""Create the parent directory for a file path."""
|
|
parent = os.path.dirname(path)
|
|
if parent:
|
|
ensure_dir(parent)
|
|
|
|
|
|
def open_dataset(path: str, purpose: str):
|
|
"""Open a dataset and fail fast with context."""
|
|
try:
|
|
ds = gdal.Open(path)
|
|
except RuntimeError as exc:
|
|
raise SystemExit(f"{purpose}: {exc}") from exc
|
|
if ds is None:
|
|
raise SystemExit(f"{purpose}: GDAL returned None for {path}")
|
|
return ds
|
|
|
|
|
|
def build_vrt(vrt_path: str, sources: Sequence[str]) -> bool:
|
|
"""Build a VRT if missing; returns True if built."""
|
|
if os.path.exists(vrt_path):
|
|
return False
|
|
if not sources:
|
|
raise SystemExit(f"No sources available to build VRT {vrt_path}.")
|
|
ensure_parent(vrt_path)
|
|
print(f"Building {vrt_path} from {len(sources)} files...")
|
|
try:
|
|
gdal.BuildVRT(vrt_path, list(sources))
|
|
except RuntimeError as exc:
|
|
raise SystemExit(f"Could not build {vrt_path}: {exc}") from exc
|
|
return True
|
|
|
|
|
|
def safe_remove(path: str) -> bool:
|
|
"""Remove a file if present; return True when deleted."""
|
|
try:
|
|
os.remove(path)
|
|
return True
|
|
except FileNotFoundError:
|
|
return False
|
|
except OSError as exc:
|
|
print(f"Warning: could not remove {path}: {exc}")
|
|
return False
|
|
|
|
|
|
def cleanup_aux_files(patterns: Iterable[str]) -> int:
|
|
"""Clear GDAL sidecars and temp files matching glob patterns."""
|
|
removed = 0
|
|
for pattern in patterns:
|
|
for match in glob.glob(pattern):
|
|
if safe_remove(match):
|
|
removed += 1
|
|
return removed
|
|
|
|
|
|
def resolve_first_existing(paths: Sequence[str], label: str) -> str:
|
|
"""Return the first existing path in order; fail with a clear message."""
|
|
for path in paths:
|
|
if path and os.path.exists(path):
|
|
return path
|
|
tried = ", ".join(paths)
|
|
raise SystemExit(f"No existing path found for {label}. Checked: {tried}")
|