Spaces:
Running on Zero
Running on Zero
File size: 5,612 Bytes
a74054f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """
Topological feature extractor for basin characteristics.
Single Responsibility: Extract elevation, slope, drainage density, TWI, area.
"""
import numpy as np
import geopandas as gpd
from typing import Dict, Any, Optional
from pathlib import Path
from .base import BaseFeatureExtractor
class TopologicalExtractor(BaseFeatureExtractor):
"""
Extracts topological/terrain features from basin geometries and DEM data.
Features extracted:
- Basin area (km²)
- Mean/min/max elevation (m)
- Mean basin slope (degrees)
- Drainage density (km/km²)
- Topographic Wetness Index (TWI)
"""
def __init__(self, dem_path: Optional[Path] = None, **kwargs):
"""
Initialize topological extractor.
Args:
dem_path: Path to DEM raster (optional, for elevation/slope/TWI)
**kwargs: Additional parameters
"""
super().__init__(**kwargs)
self.dem_path = Path(dem_path) if dem_path else None
def extract(self, basin_gdf: gpd.GeoDataFrame) -> Dict[str, Any]:
"""
Extract topological features from basin.
Args:
basin_gdf: GeoDataFrame with basin geometry
Returns:
Dictionary with topological features
"""
if not self.validate_inputs(basin_gdf):
raise ValueError("Invalid basin GeoDataFrame")
features = {}
# Basic geometric features
features.update(self._extract_basic_geometry(basin_gdf))
# DEM-based features (if DEM provided)
if self.dem_path and self.dem_path.exists():
features.update(self._extract_dem_features(basin_gdf))
return features
def _extract_basic_geometry(self, basin_gdf: gpd.GeoDataFrame) -> Dict[str, float]:
"""Extract basic geometric features."""
# Ensure CRS is projected (for area calculation)
if basin_gdf.crs and basin_gdf.crs.is_geographic:
# Reproject to appropriate UTM zone
basin_gdf = basin_gdf.to_crs(basin_gdf.estimate_utm_crs())
# Basin area in km²
area_km2 = basin_gdf.geometry.area.sum() / 1e6
# Basin centroid
centroid = basin_gdf.geometry.centroid.iloc[0]
return {
"basin_area_km2": area_km2,
"centroid_lon": centroid.x,
"centroid_lat": centroid.y
}
def _extract_dem_features(self, basin_gdf: gpd.GeoDataFrame) -> Dict[str, float]:
"""
Extract DEM-based features: elevation, slope, TWI.
Note: This is a placeholder for DEM processing.
Full implementation would use rasterio to:
1. Clip DEM to basin boundary
2. Calculate zonal statistics (mean, min, max elevation)
3. Derive slope from DEM
4. Calculate TWI using flow accumulation and slope
"""
try:
import rasterio
from rasterio.mask import mask
import rasterio.features
with rasterio.open(self.dem_path) as src:
# Ensure basin is in same CRS as DEM
basin_gdf_reproj = basin_gdf.to_crs(src.crs)
# Clip DEM to basin
geoms = [mapping for mapping in basin_gdf_reproj.geometry]
out_image, out_transform = mask(src, geoms, crop=True)
elevation = out_image[0]
# Remove nodata values
nodata = src.nodata
if nodata is not None:
elevation = elevation[elevation != nodata]
# Calculate statistics
features = {
"elevation_mean_m": float(np.mean(elevation)),
"elevation_min_m": float(np.min(elevation)),
"elevation_max_m": float(np.max(elevation)),
"elevation_range_m": float(np.max(elevation) - np.min(elevation))
}
# Slope calculation (simplified - would need proper implementation)
# This is a placeholder
features["mean_slope_deg"] = self._estimate_slope(elevation)
return features
except ImportError:
# If rasterio not available, return placeholder values
return {
"elevation_mean_m": None,
"elevation_min_m": None,
"elevation_max_m": None,
"elevation_range_m": None,
"mean_slope_deg": None
}
except Exception as e:
print(f"Warning: Could not extract DEM features: {e}")
return {}
def _estimate_slope(self, elevation: np.ndarray) -> float:
"""
Estimate mean slope from elevation data.
Simplified calculation - full version would use proper gradient computation.
"""
if len(elevation) < 2:
return 0.0
# Simple approximation using elevation variability
slope_proxy = np.std(elevation) / np.mean(elevation) * 100 if np.mean(elevation) > 0 else 0
return float(slope_proxy)
def get_feature_names(self) -> list:
"""Get list of feature names."""
base_features = ["basin_area_km2", "centroid_lon", "centroid_lat"]
dem_features = [
"elevation_mean_m", "elevation_min_m", "elevation_max_m",
"elevation_range_m", "mean_slope_deg"
]
return base_features + (dem_features if self.dem_path else [])
|