Add schema validation for shapes

Author: sjpfenninger

workflow/Snakefile

@@ -17,8 +17,9 @@ with open(workflow.source_path("internal/settings.yaml"), "r") as f:
     internal = yaml.safe_load(f)
 
 # Python files that are imported from other scripts and need to be included when accessing the module
-workflow.source_path("scripts/_script_utils.py")
 workflow.source_path("scripts/_geo.py")
+workflow.source_path("scripts/_schemas.py")
+workflow.source_path("scripts/_script_utils.py")
 
 
 wildcard_constraints:

workflow/envs/default.yaml

@@ -7,6 +7,7 @@ dependencies:
   - click=8.2.1
   - geopandas=1.1.0
   - pandas=2.3.0
+  - pandera-geopandas=0.24.0
   - pyarrow=19.0.1
   - xarray=2025.6.1
   - netcdf4=1.7.2
@@ -21,4 +22,4 @@ dependencies:
   - pyproj=3.7.1
   - utm=0.7.0
   - glom=24.11.0
-  - dask=2025.7.0
+  - dask=2025.7.0

workflow/scripts/_schemas.py

@@ -0,0 +1,25 @@
+"""Schemas for tabular data used in the workflow."""
+
+from pandera.pandas import DataFrameModel, Field, check
+from pandera.typing.geopandas import GeoSeries
+from pandera.typing.pandas import Series
+from shapely.geometry import Point
+
+
+class Shapes(DataFrameModel):
+    class Config:
+        coerce = True
+        strict = False
+
+    shape_id: Series[str] = Field(unique=True)
+    "Unique ID for this shape."
+    country_id: Series[str]
+    "ISO alpha-3 code."
+    shape_class: Series[str] = Field(isin=["land", "maritime"])
+    "Shape classifier"
+    geometry: GeoSeries[Point] = Field()
+    "Shape polygon."
+
+    @check("geometry", element_wise=True)
+    def geom_not_empty(cls, geom):
+        return (geom is not None) and (not geom.is_empty) and geom.is_valid

workflow/scripts/breakup_shape.py

@@ -4,6 +4,7 @@
 
 import click
 import geopandas as gpd
+from _schemas import Shapes
 
 
 @click.command()
@@ -25,6 +26,7 @@ def breakup_shape(shapes_path, split_by, output_path):
     output_path = Path(output_path)
     output_path.mkdir(parents=True, exist_ok=True)
     shapes = gpd.read_parquet(shapes_path)
+    shapes = Shapes.validate(shapes)
 
     # Print rows where geometry is empty
     if shapes.geometry.is_empty.any():