184 lines
5.2 KiB
Python
184 lines
5.2 KiB
Python
"""
|
||
geo_tools.utils.validators
|
||
~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||
数据验证工具：CRS 合法性、几何有效性、文件格式等。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING
|
||
|
||
if TYPE_CHECKING:
|
||
import geopandas as gpd
|
||
from shapely.geometry.base import BaseGeometry
|
||
|
||
|
||
# ── CRS 校验 ──────────────────────────────────────────────────────────────────
|
||
|
||
def is_valid_crs(crs_input: str | int) -> bool:
|
||
"""检查 CRS 是否可以被 pyproj 正常解析。
|
||
|
||
Parameters
|
||
----------
|
||
crs_input:
|
||
EPSG 代码（整数或 ``"EPSG:4326"`` 字符串）或 proj 字符串。
|
||
|
||
Returns
|
||
-------
|
||
bool
|
||
"""
|
||
try:
|
||
from pyproj import CRS
|
||
CRS.from_user_input(crs_input)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def validate_crs(crs_input: str | int) -> str:
|
||
"""校验并标准化 CRS，返回 EPSG 代码字符串。
|
||
|
||
Raises
|
||
------
|
||
ValueError
|
||
如果 CRS 无法被 pyproj 解析。
|
||
"""
|
||
from pyproj import CRS
|
||
try:
|
||
crs_obj = CRS.from_user_input(crs_input)
|
||
# 尝试返回简洁的 EPSG 字符串
|
||
epsg = crs_obj.to_epsg()
|
||
if epsg:
|
||
return f"EPSG:{epsg}"
|
||
return crs_obj.to_string()
|
||
except Exception as exc:
|
||
raise ValueError(f"无效的 CRS：{crs_input!r}。原因：{exc}") from exc
|
||
|
||
|
||
# ── 几何校验 ──────────────────────────────────────────────────────────────────
|
||
|
||
def validate_geometry(gdf: "gpd.GeoDataFrame", *, raise_on_invalid: bool = False) -> dict[str, int]:
|
||
"""检查 GeoDataFrame 中几何对象的有效性。
|
||
|
||
Parameters
|
||
----------
|
||
gdf:
|
||
待检查的 GeoDataFrame。
|
||
raise_on_invalid:
|
||
若为 True，当存在无效几何时抛出 ``ValueError``。
|
||
|
||
Returns
|
||
-------
|
||
dict
|
||
包含 ``total``、``valid``、``invalid``、``null`` 计数。
|
||
"""
|
||
null_count = gdf.geometry.isna().sum()
|
||
non_null = gdf.geometry.dropna()
|
||
invalid_mask = ~non_null.is_valid # type: ignore
|
||
invalid_count = int(invalid_mask.sum())
|
||
valid_count = len(non_null) - invalid_count
|
||
|
||
result = {
|
||
"total": len(gdf),
|
||
"valid": valid_count,
|
||
"invalid": invalid_count,
|
||
"null": int(null_count),
|
||
}
|
||
|
||
if raise_on_invalid and (invalid_count > 0 or null_count > 0):
|
||
raise ValueError(
|
||
f"GeoDataFrame 存在 {invalid_count} 个无效几何、{null_count} 个空几何。"
|
||
)
|
||
return result
|
||
|
||
|
||
# ── 文件格式校验 ───────────────────────────────────────────────────────────────
|
||
|
||
#: 支持读取的矢量文件扩展名（fiona 驱动映射）
|
||
SUPPORTED_VECTOR_EXTENSIONS: dict[str, str] = {
|
||
".shp": "ESRI Shapefile",
|
||
".geojson": "GeoJSON",
|
||
".json": "GeoJSON",
|
||
".gpkg": "GPKG",
|
||
".gdb": "OpenFileGDB",
|
||
".kml": "KML",
|
||
".kmz": "KML",
|
||
".csv": "CSV",
|
||
".gml": "GML",
|
||
".dxf": "DXF",
|
||
".fgb": "FlatGeobuf",
|
||
}
|
||
|
||
|
||
def is_supported_vector_format(path: str | Path) -> bool:
|
||
"""判断路径是否为已知的矢量格式。
|
||
|
||
Parameters
|
||
----------
|
||
path:
|
||
输入路径，可以是字符串或 Path 对象。
|
||
|
||
Returns
|
||
-------
|
||
bool
|
||
如果路径是支持的矢量格式，返回 True；否则返回 False。
|
||
"""
|
||
path = Path(path)
|
||
suffix = path.suffix.lower()
|
||
# .gdb 可能是目录（FileGDB）
|
||
if path.is_dir() and suffix == ".gdb":
|
||
return True
|
||
return suffix in SUPPORTED_VECTOR_EXTENSIONS
|
||
|
||
|
||
def validate_vector_path(path: str | Path) -> Path:
|
||
"""校验矢量数据路径，返回 Path 对象。
|
||
|
||
Raises
|
||
------
|
||
FileNotFoundError
|
||
文件或目录不存在。
|
||
ValueError
|
||
文件格式不受支持。
|
||
"""
|
||
path = Path(path)
|
||
# GDB 是目录
|
||
if not path.exists():
|
||
raise FileNotFoundError(f"路径不存在：{path}")
|
||
if not is_supported_vector_format(path):
|
||
raise ValueError(
|
||
f"不支持的矢量格式：{path.suffix!r}。"
|
||
f"支持的格式：{list(SUPPORTED_VECTOR_EXTENSIONS.keys())}"
|
||
)
|
||
return path
|
||
|
||
|
||
def ensure_valid_geometry(geom: 'BaseGeometry', verbose: bool = False) -> 'BaseGeometry':
|
||
"""确保几何对象有效，无效时尝试修复。
|
||
|
||
Parameters
|
||
----------
|
||
geom:
|
||
输入几何。
|
||
verbose:
|
||
是否输出修复几何的警告信息。
|
||
|
||
Returns
|
||
-------
|
||
BaseGeometry
|
||
有效的几何对象（可能是修复后的）。
|
||
"""
|
||
from app.core.geometry import fix_geometry
|
||
from app.utils.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
if not geom.is_valid:
|
||
fixed_geom = fix_geometry(geom)
|
||
if fixed_geom is not None:
|
||
if verbose:
|
||
logger.warning("几何对象无效，已自动修复")
|
||
return fixed_geom
|
||
return geom
|