Spaces:
Running on Zero
Running on Zero
File size: 5,251 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 156 157 | """
Shapefile loader for watershed boundaries.
Single Responsibility: Load geospatial watershed data.
"""
import geopandas as gpd
import matplotlib.pyplot as plt
from pathlib import Path
from typing import Optional
from .base import BaseDataLoader
class ShapefileLoader(BaseDataLoader):
"""
Loads watershed shapefiles for La Risle and La Eure basins.
"""
def __init__(self, data_path: Path, crs: Optional[str] = None):
"""
Initialize shapefile loader.
Args:
data_path: Path to shapefile (.shp) or directory containing shapefiles
crs: Coordinate reference system to reproject to (optional)
"""
super().__init__(data_path)
self.crs = crs
def load(self) -> gpd.GeoDataFrame:
"""
Load shapefile(s) into GeoDataFrame.
Returns:
GeoDataFrame with watershed geometries and attributes
"""
if self.data_path.suffix == ".shp":
gdf = gpd.read_file(self.data_path)
elif self.data_path.is_dir():
# Find .shp files in directory
shp_files = list(self.data_path.glob("*.shp"))
if not shp_files:
raise FileNotFoundError(f"No shapefiles found in {self.data_path}")
# Load first shapefile found (or merge if multiple)
gdf = gpd.read_file(shp_files[0])
else:
raise ValueError(f"Invalid shapefile path: {self.data_path}")
# Reproject if CRS specified
if self.crs and gdf.crs != self.crs:
gdf = gdf.to_crs(self.crs)
return gdf
def get_metadata(self) -> dict:
"""Get shapefile metadata."""
meta = super().get_metadata()
meta.update({
"data_type": "watershed_shapefile",
"target_crs": self.crs
})
return meta
def print_summary(self, gdf: Optional[gpd.GeoDataFrame] = None) -> None:
"""
Print the shapefile's data and metadata in a readable format.
Args:
gdf: Optional pre-loaded GeoDataFrame. If not provided, the
shapefile will be loaded from disk.
"""
if gdf is None:
gdf = self.load()
meta = self.get_metadata()
print("=" * 60)
print("SHAPEFILE METADATA")
print("=" * 60)
for key, value in meta.items():
print(f"{key:>15}: {value}")
print()
print("=" * 60)
print("SHAPEFILE DATA SUMMARY")
print("=" * 60)
print(f"{'Feature count':>15}: {len(gdf)}")
print(f"{'CRS':>15}: {gdf.crs}")
print(f"{'Columns':>15}: {list(gdf.columns)}")
print(f"{'Bounds':>15}: {tuple(gdf.total_bounds)}")
print(f"{'Geometry types':>15}: {gdf.geom_type.unique().tolist()}")
print()
print("-" * 60)
print("Attribute preview:")
print("-" * 60)
print(gdf.drop(columns="geometry").head())
print("=" * 60)
def plot(
self,
gdf: Optional[gpd.GeoDataFrame] = None,
column: Optional[str] = None,
title: Optional[str] = None,
figsize: tuple = (10, 10),
save_path: Optional[Path] = None,
cmap: str = "Blues",
edgecolor: str = "black",
) -> plt.Axes:
"""
Plot the watershed polygon(s).
Args:
gdf: Optional pre-loaded GeoDataFrame. If not provided, the
shapefile will be loaded from disk.
column: Optional column name to color/shade features by
(e.g. a category or numeric attribute). If None,
all features are drawn with a single fill color.
title: Optional plot title. Defaults to the data path stem.
figsize: Figure size in inches (width, height).
save_path: If provided, saves the figure to this path
instead of (or in addition to) displaying it.
cmap: Matplotlib colormap used when `column` is set.
edgecolor: Outline color for the polygons.
Returns:
The matplotlib Axes object, for further customization.
"""
if gdf is None:
gdf = self.load()
fig, ax = plt.subplots(figsize=figsize)
if column and column in gdf.columns:
gdf.plot(column=column, ax=ax, cmap=cmap, edgecolor=edgecolor, legend=True)
else:
gdf.plot(ax=ax, color="steelblue", edgecolor=edgecolor, alpha=0.6)
ax.set_title(title or f"Watershed: {self.data_path.stem}")
ax.set_xlabel("Easting")
ax.set_ylabel("Northing")
ax.set_aspect("equal")
crs_label = str(gdf.crs) if gdf.crs else "Unknown CRS"
ax.annotate(
crs_label,
xy=(0.01, 0.01),
xycoords="axes fraction",
fontsize=8,
color="gray",
)
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches="tight")
print(f"Plot saved to {save_path}")
return ax |