overlay_gpd
Perform spatial overlay analysis on two geospatial files using intersection, union, identity, symmetric difference, or difference methods. Returns overlay results as a new GeoDataFrame.
Instructions
Overlay two GeoDataFrames using geopandas.overlay. Args: gdf1_path: Path to the first geospatial file. gdf2_path: Path to the second geospatial file. how: Overlay method ('intersection', 'union', 'identity', 'symmetric_difference', 'difference'). output_path: Optional path to save the result. Returns: Dictionary with status, message, and output info.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| gdf1_path | Yes | ||
| gdf2_path | Yes | ||
| how | No | intersection | |
| output_path | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The `overlay_gpd` function is the core implementation of the overlay_gpd tool. It reads two GeoDataFrames from file paths, optionally reprojects to match CRS, performs a spatial overlay using geopandas.overlay() with configurable 'how' (intersection, union, identity, symmetric_difference, difference), optionally saves the result, and returns status, feature count, CRS, columns, and a WKT-based preview.
def overlay_gpd(gdf1_path: str, gdf2_path: str, how: str = "intersection", output_path: str = None) -> Dict[str, Any]: """ Overlay two GeoDataFrames using geopandas.overlay. Args: gdf1_path: Path to the first geospatial file. gdf2_path: Path to the second geospatial file. how: Overlay method ('intersection', 'union', 'identity', 'symmetric_difference', 'difference'). output_path: Optional path to save the result. Returns: Dictionary with status, message, and output info. """ try: gdf1 = gpd.read_file(gdf1_path) gdf2 = gpd.read_file(gdf2_path) if gdf1.crs != gdf2.crs: gdf2 = gdf2.to_crs(gdf1.crs) result = gpd.overlay(gdf1, gdf2, how=how) if output_path: output_path_resolved = resolve_path(output_path, relative_to_storage=True) output_path_resolved.parent.mkdir(parents=True, exist_ok=True) result.to_file(str(output_path_resolved)) output_path = str(output_path_resolved) # Convert geometry to WKT for serialization preview_df = result.head(5).copy() if 'geometry' in preview_df.columns: preview_df['geometry'] = preview_df['geometry'].apply(lambda g: g.wkt if g is not None else None) preview = preview_df.to_dict(orient="records") return { "status": "success", "message": f"Overlay ({how}) completed successfully.", "num_features": len(result), "crs": str(result.crs), "columns": list(result.columns), "preview": preview, "output_path": output_path, } except Exception as e: logger.error(f"Error in overlay_gpd: {str(e)}") return {"status": "error", "message": str(e)} - src/gis_mcp/geopandas_functions.py:183-184 (registration)The `@gis_mcp.tool()` decorator on line 183 registers `overlay_gpd` as an MCP tool. The `gis_mcp` instance is a FastMCP object from fastmcp, created in mcp.py.
@gis_mcp.tool() def overlay_gpd(gdf1_path: str, gdf2_path: str, how: str = "intersection", output_path: str = None) -> Dict[str, Any]: - src/gis_mcp/geopandas_functions.py:13-26 (registration)The resource listing at 'gis://geopandas/io' includes 'overlay_gpd' as an available operation, providing discovery for clients.
@gis_mcp.resource("gis://geopandas/io") def get_geopandas_io() -> Dict[str, List[str]]: """List available GeoPandas I/O operations.""" return { "operations": [ "read_file_gpd", "to_file_gpd", "overlay_gpd", "dissolve_gpd", "explode_gpd", "clip_vector", "write_file_gpd" ] } - src/gis_mcp/storage_config.py:72-90 (helper)The `resolve_path` helper function is used by overlay_gpd to resolve output paths relative to the storage directory when saving result files.
def resolve_path(file_path: str, relative_to_storage: bool = True) -> Path: """ Resolve a file path, optionally making it relative to the storage directory. If the path is absolute, it's used as-is. If relative and relative_to_storage is True, it's resolved relative to the storage directory. Args: file_path: The file path to resolve relative_to_storage: If True and path is relative, resolve relative to storage Returns: Resolved Path object """ path = Path(file_path) # If absolute path, use as-is if path.is_absolute(): return path.expanduser().resolve() - The function signature defines the input schema for the overlay_gpd tool: gdf1_path (str), gdf2_path (str), how (str, default 'intersection'), output_path (str, optional). The return type is Dict[str, Any] with status, message, num_features, crs, columns, preview, output_path.
def overlay_gpd(gdf1_path: str, gdf2_path: str, how: str = "intersection", output_path: str = None) -> Dict[str, Any]: """ Overlay two GeoDataFrames using geopandas.overlay. Args: gdf1_path: Path to the first geospatial file. gdf2_path: Path to the second geospatial file. how: Overlay method ('intersection', 'union', 'identity', 'symmetric_difference', 'difference'). output_path: Optional path to save the result. Returns: Dictionary with status, message, and output info. """ try: gdf1 = gpd.read_file(gdf1_path) gdf2 = gpd.read_file(gdf2_path) if gdf1.crs != gdf2.crs: gdf2 = gdf2.to_crs(gdf1.crs) result = gpd.overlay(gdf1, gdf2, how=how) if output_path: output_path_resolved = resolve_path(output_path, relative_to_storage=True) output_path_resolved.parent.mkdir(parents=True, exist_ok=True) result.to_file(str(output_path_resolved)) output_path = str(output_path_resolved) # Convert geometry to WKT for serialization preview_df = result.head(5).copy() if 'geometry' in preview_df.columns: preview_df['geometry'] = preview_df['geometry'].apply(lambda g: g.wkt if g is not None else None) preview = preview_df.to_dict(orient="records") return { "status": "success", "message": f"Overlay ({how}) completed successfully.", "num_features": len(result), "crs": str(result.crs), "columns": list(result.columns), "preview": preview, "output_path": output_path, } except Exception as e: logger.error(f"Error in overlay_gpd: {str(e)}") return {"status": "error", "message": str(e)}