Network Attribute Validation: Quality Gates for Grid Infrastructure Pipelines #

import asyncio
import logging
import warnings
from pathlib import Path
from typing import Any, Dict, List, Tuple

import geopandas as gpd
import pandas as pd
import pyproj
from shapely.geometry import mapping
from shapely.validation import make_valid

# Suppress non-critical geopandas warnings for cleaner logs
warnings.filterwarnings("ignore", category=UserWarning, module="geopandas")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("grid_validator")

class GridAttributeValidator:
    """Production-ready validator for grid infrastructure vector datasets."""

    REQUIRED_COLS = {"asset_id", "voltage_kv", "status", "capacity_mva", "geometry"}
    VALID_STATUSES = {"active", "decommissioned", "planned", "maintenance"}
    TARGET_CRS = "EPSG:32618"  # UTM Zone 18N (example working projection)

    def __init__(self, source_path: Path, chunk_size: int = 5000):
        self.source_path = source_path
        self.chunk_size = chunk_size
        self.validation_log: List[Dict[str, Any]] = []
        self.target_crs = pyproj.CRS.from_string(self.TARGET_CRS)

    async def run_pipeline(self) -> gpd.GeoDataFrame:
        """Orchestrate chunked validation with async I/O."""
        logger.info(f"Initializing validation pipeline for {self.source_path}")

        # Async file read with chunking
        chunks = await self._read_chunks_async()

        # Process chunks concurrently
        tasks = [self._validate_chunk(chunk, i) for i, chunk in enumerate(chunks)]
        validated_chunks = await asyncio.gather(*tasks)

        # Merge and finalize
        clean_gdf = gpd.pd.concat(validated_chunks, ignore_index=True)
        await self._export_validation_report()

        logger.info(f"Pipeline complete. {len(clean_gdf)} features validated and exported.")
        return clean_gdf

    async def _read_chunks_async(self) -> List[gpd.GeoDataFrame]:
        """Read large GeoPackage/Shapefile in memory-safe chunks."""
        chunks = []
        # Using synchronous read_file inside executor to avoid blocking event loop
        loop = asyncio.get_event_loop()
        reader = await loop.run_in_executor(
            None, lambda: list(gpd.read_file(self.source_path, chunksize=self.chunk_size))
        )
        return reader

    async def _validate_chunk(self, gdf_chunk: gpd.GeoDataFrame, idx: int) -> gpd.GeoDataFrame:
        """Apply schema, CRS, and topology validation to a single chunk."""
        logger.info(f"Processing chunk {idx} ({len(gdf_chunk)} features)")

        # 1. Schema Enforcement
        missing = self.REQUIRED_COLS - set(gdf_chunk.columns)
        if missing:
            raise ValueError(f"Missing required columns in chunk {idx}: {missing}")

        gdf_chunk["status"] = gdf_chunk["status"].str.lower().str.strip()
        invalid_status = ~gdf_chunk["status"].isin(self.VALID_STATUSES)
        if invalid_status.any():
            self._log_anomalies(gdf_chunk.loc[invalid_status, "asset_id"], "invalid_status")
            gdf_chunk.loc[invalid_status, "status"] = "unknown"

        # 2. CRS Normalization & Spatial Validation
        if gdf_chunk.crs is None:
            gdf_chunk.set_crs("EPSG:4326", inplace=True)
        if not gdf_chunk.crs.equals(self.target_crs):
            gdf_chunk = gdf_chunk.to_crs(self.target_crs)

        # 3. Topology Repair & Validation
        is_valid_mask = gdf_chunk.geometry.is_valid
        if not is_valid_mask.all():
            invalid_ids = gdf_chunk.loc[~is_valid_mask, "asset_id"].tolist()
            self._log_anomalies(invalid_ids, "invalid_geometry")
            gdf_chunk.loc[~is_valid_mask, "geometry"] = gdf_chunk.loc[~is_valid_mask, "geometry"].apply(make_valid)

        # Drop null geometries post-repair
        gdf_chunk = gdf_chunk.dropna(subset=["geometry"])
        return gdf_chunk

    def _log_anomalies(self, asset_ids: List[str], anomaly_type: str):
        """Thread-safe logging of validation failures."""
        for aid in asset_ids:
            self.validation_log.append({
                "asset_id": aid,
                "anomaly_type": anomaly_type,
                "timestamp": pd.Timestamp.utcnow().isoformat()
            })

    async def _export_validation_report(self):
        """Async export of validation logs for compliance auditing."""
        if not self.validation_log:
            logger.info("No anomalies detected. Skipping report export.")
            return

        report_df = pd.DataFrame(self.validation_log)
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(
            None, lambda: report_df.to_csv("grid_validation_report.csv", index=False)
        )
        logger.info(f"Exported {len(report_df)} validation anomalies to CSV.")

# Example execution
if __name__ == "__main__":
    validator = GridAttributeValidator(Path("grid_network_raw.gpkg"), chunk_size=10000)
    asyncio.run(validator.run_pipeline())