Spaces:
Running on Zero
Running on Zero
| """ | |
| 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 |