14 min read 3087 words

Network Attribute Validation for Grid Infrastructure

Network attribute validation is the quality gate that stands between raw spatial ingestion and every downstream interconnection decision. The specific failure mode it addresses is silent attribute and topology drift in multi-source grid datasets: a transmission line arrives with a blank voltage_kv, a substation node sits a few metres off its feeder, a status field reads Decom. instead of decommissioned, or a layer carries no projection at all — and none of these raise an exception. They flow straight through a load-flow model, a setback buffer, or an interconnection-queue screen and surface only as a wrong capacity number or a violated statutory clearance, at the point in the project lifecycle where rework is most expensive. This page sits within the broader Grid Infrastructure & Network Proximity Analysis architecture and defines the validation contract that every later routing, buffering, and compliance stage depends on.

The goal is not a one-time cleanup script. It is a deterministic, auditable gate: every feature is checked against an explicit schema and domain, every geometry is proven valid and snapped to a known coordinate reference system, and every rejection or repair is logged with an asset identifier and a timestamp so the dataset’s lineage is defensible. That log is what lets an analyst answer an independent engineer’s question — how do you know this capacity surface is built on clean network data? — without re-running the pipeline.

Why naive attribute validation fails

The intuition that “the data loaded without error, so it is usable” is the root cause of most downstream grid-modeling defects. GeoPandas and the underlying GDAL readers are permissive by design: they will happily admit a layer with a null geometry, a self-intersecting service-territory polygon, an undefined CRS, or a free-text status column. Three failure paths recur, and each is invisible at read time.

The first is schema and domain violation. A required column such as voltage_kv or capacity_mva is missing, or present but populated with out-of-range thermal ratings and inconsistent status strings (active, Active, ACT, in-service). A load-flow solver that keys on status == "active" silently drops every row that spells it differently, understating available network capacity.

The second is topological invalidity. Digitised utility maps routinely contain self-intersecting line geometries, disconnected feeder segments, and orphaned substation nodes. These corrupt any graph traversal built on the network and break the spatial joins that proximity and distance calculations rely on, because an invalid geometry has undefined intersects and distance behaviour.

The third is CRS misalignment. A layer with no .prj, or a stack mixing EPSG:4326 (degrees) with a projected UTM frame, makes every distance and buffer wrong. Planar distance is only meaningful in a projected CRS; run a 5 km substation setback against geographic coordinates and the radius is interpreted as 5000 degrees. The magnitude of the degree-vs-metre error scales with latitude — at latitude the east–west ground distance of one degree of longitude is metres — so the same unprojected buffer is wrong by a different amount in every part of a portfolio. The validation gate must normalise CRS before any geometry is measured.

Chunked validation pipeline data flow A raw vector dataset is read by an async chunk reader in 5,000-feature slices. Each chunk fans out to three row-local checks run in parallel — schema enforcement, CRS normalization to EPSG:32618, and make_valid topology repair — which all append to an anomaly log CSV. The log feeds a clean, projected GeoDataFrame that is handed to downstream buffer analysis. Raw vectordataset Async chunkreader Schema check CRS normalize make_valid Anomaly logCSV CleanGeoDataFrame 5k features required cols + statuses to EPSG:32618 + drop null geometries → buffer analysis

Prerequisites & data requirements

This workflow assumes a Python 3.11+ environment with geopandas>=0.14, shapely>=2.0, pyproj>=3.6, and pyogrio>=0.7 as the default I/O engine. The pyproj install must carry its bundled PROJ data so datum-shift grids resolve; for air-gapped permitting environments, pin the PROJ grid package and set PROJ_NETWORK=OFF to keep transformations reproducible.

Inputs are vector layers — transmission lines (LineString/MultiLineString) and substation or tap nodes (Point) — in any GDAL-readable format, ideally GeoPackage or GeoParquet so the embedded CRS travels with the geometry. These typically originate from transmission line and substation mapping pipelines, where features extracted from OpenStreetMap or utility GIS exports inherit whatever projection and attribute hygiene the source had. Each feature must carry, at minimum, asset_id, voltage_kv, status, capacity_mva, and a geometry.

Two decisions are made deliberately before any row runs. First, a target CRS is chosen per study area rather than inherited — a local UTM zone (for example EPSG:32618 for longitudes around 75°W) or a national grid, never a geographic frame, because the output feeds distance and buffer work. Second, the attribute contract — required columns, the valid status domain, and the acceptable range for thermal ratings — is declared explicitly so the gate fails loudly on drift. Geometry validity is handled jointly with the broader spatial data quality and validation stage, since an invalid polygon or line will distort under to_crs().

A compact pre-flight that surfaces all three failure modes before the main pipeline runs:

python
import geopandas as gpd

REQUIRED_COLS = {"asset_id", "voltage_kv", "status", "capacity_mva", "geometry"}


def preflight_network_layer(layer_path: str) -> dict:
    """Surface schema, CRS, and topology risks before validation runs."""
    gdf = gpd.read_file(layer_path)
    missing = REQUIRED_COLS - set(gdf.columns)
    return {
        "missing_columns": sorted(missing),
        "crs_defined": gdf.crs is not None,
        "is_geographic": bool(gdf.crs and gdf.crs.is_geographic),  # degrees => unsafe
        "invalid_geometries": int((~gdf.geometry.is_valid).sum()),
        "null_geometries": int(gdf.geometry.isna().sum()),
        "status_values": sorted(set(gdf.get("status", []))),  # spot domain drift
    }

Core implementation: the validation pipeline

The pipeline below performs schema enforcement, domain coercion, CRS normalization, and topology repair, orchestrated as chunked async work so national-scale network layers never have to fit in memory at once. It decouples I/O from CPU-bound spatial validation, caches the target CRS once, and records every anomaly to an in-memory log that is exported for audit. Variable names and thresholds are grid-specific throughout.

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

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

# Suppress non-critical geopandas warnings for cleaner audit 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 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 — metric frame for distance/buffer work
    MAX_RATING_MVA = 5000.0    # domain ceiling; flags fat-fingered thermal ratings

    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]] = []
        # Cache the target CRS once: avoids re-parsing the string per chunk.
        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("Initializing validation pipeline for %s", self.source_path)

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

        clean_gdf = gpd.GeoDataFrame(
            pd.concat(list(validated_chunks), ignore_index=True), crs=self.TARGET_CRS
        )
        await self._export_validation_report()

        logger.info("Pipeline complete. %d features validated.", len(clean_gdf))
        return clean_gdf

    async def _read_chunks_async(self) -> List[gpd.GeoDataFrame]:
        """Read a large GeoPackage/Shapefile in memory-safe chunks.

        skip_features / max_features are pyogrio kwargs that GeoPandas forwards
        to the engine, enabling bounded slices without an iterator.
        """
        loop = asyncio.get_event_loop()
        info = await loop.run_in_executor(None, pyogrio.read_info, self.source_path)
        total_features = info.get("features", 0)
        chunks = []
        for offset in range(0, total_features, self.chunk_size):
            chunk = await loop.run_in_executor(
                None,
                lambda o=offset: gpd.read_file(
                    self.source_path, skip_features=o, max_features=self.chunk_size
                ),
            )
            chunks.append(chunk)
        return chunks

    async def _validate_chunk(self, gdf_chunk: gpd.GeoDataFrame, idx: int) -> gpd.GeoDataFrame:
        """Apply schema, domain, CRS, and topology validation to one chunk."""
        logger.info("Processing chunk %d (%d features)", idx, len(gdf_chunk))

        # 1. Schema enforcement — fail loudly; a missing column is not recoverable.
        missing = self.REQUIRED_COLS - set(gdf_chunk.columns)
        if missing:
            raise ValueError(f"Missing required columns in chunk {idx}: {missing}")

        gdf_chunk = gdf_chunk.copy()

        # 2. Domain coercion — normalize status strings, quarantine the unknown.
        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"].tolist(), "invalid_status"
            )
            gdf_chunk.loc[invalid_status, "status"] = "unknown"

        # 3. CRS normalization — assign a fallback only if undefined, then project.
        if gdf_chunk.crs is None:
            logger.warning("Chunk %d has undefined CRS; assuming EPSG:4326.", idx)
            gdf_chunk = gdf_chunk.set_crs("EPSG:4326")
        if not gdf_chunk.crs.equals(self.target_crs):
            gdf_chunk = gdf_chunk.to_crs(self.target_crs)  # carries datum-shift grid

        # 4. Topology repair — make_valid the broken geometries, then drop nulls.
        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)
            )

        gdf_chunk = gdf_chunk.dropna(subset=["geometry"])
        return gdf_chunk

    def _log_anomalies(self, asset_ids: List[str], anomaly_type: str) -> None:
        """Append validation failures to the in-memory audit log."""
        for aid in asset_ids:
            self.validation_log.append({
                "asset_id": aid,
                "anomaly_type": anomaly_type,
                "timestamp": pd.Timestamp.now(tz="UTC").isoformat(),
            })

    async def _export_validation_report(self) -> None:
        """Async export of the anomaly log 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("Exported %d validation anomalies to CSV.", len(report_df))


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

The happy path is deliberately explicit at every decision point: schema gaps raise immediately because they are unrecoverable, domain drift is quarantined to unknown rather than dropped so nothing disappears silently, CRS is normalised through pyproj’s transformation machinery rather than a re-label, and topology is repaired before the chunk is ever measured.

Error handling & edge cases

Three failure modes from the problem framing need explicit coverage, and the cost of each is that it does not throw on its own.

Domain drift in status and rating fields. Free-text status columns and out-of-range thermal ratings pass type checks while corrupting capacity logic. Coerce against the declared domain and flag — never delete — the offenders:

python
def guard_domain(gdf: gpd.GeoDataFrame, valid_statuses: set, max_mva: float) -> gpd.GeoDataFrame:
    bad_status = ~gdf["status"].str.lower().str.strip().isin(valid_statuses)
    bad_rating = (gdf["capacity_mva"] <= 0) | (gdf["capacity_mva"] > max_mva)
    if bad_status.any() or bad_rating.any():
        logging.warning(
            "Domain violations: %d status, %d rating — quarantined, not dropped.",
            int(bad_status.sum()), int(bad_rating.sum()),
        )
    gdf.loc[bad_status, "status"] = "unknown"
    gdf.loc[bad_rating, "capacity_mva"] = pd.NA  # NA propagates; a wrong number lies
    return gdf

Undefined or geographic CRS (the silent-degree trap). A layer with no .prj is admitted without complaint and then treated as whatever the next operation assumes. Refuse to measure in a geographic frame:

python
def guard_crs(gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
    if gdf.crs is None:
        raise ValueError(
            "Undefined CRS — confirm the source projection from metadata before "
            "assigning. A wrong assignment shifts every node and is invisible downstream."
        )
    if gdf.crs.is_geographic:
        logging.warning(
            "Geographic CRS (EPSG:%s): distance/buffer math is invalid until reprojected.",
            gdf.crs.to_epsg(),
        )
    return gdf

Irreparable topology and orphaned nodes. make_valid fixes self-intersections, but it can return an empty geometry, and it cannot reconnect a feeder segment that was digitised disconnected from its substation. Detect both so they are quarantined rather than fed to a graph build:

python
def guard_topology(gdf: gpd.GeoDataFrame, snap_tol_m: float = 1.0) -> gpd.GeoDataFrame:
    repaired = gdf.geometry.apply(make_valid)
    empty_after = repaired.is_empty | repaired.isna()
    if empty_after.any():
        logging.warning("%d geometries collapsed to empty under repair.", int(empty_after.sum()))
    gdf = gdf.assign(geometry=repaired).loc[~empty_after].copy()
    # Flag endpoints that fall outside the snapping tolerance of any node — likely orphans.
    gdf["suspect_orphan"] = ~gdf.geometry.is_valid  # placeholder for a node-join check
    return gdf
Mapping each failure mode to its guard and outcome Three silent failure modes each route to a guard function and an outcome branch: domain drift to guard_domain (quarantine status to 'unknown', set bad ratings to NA); undefined or geographic CRS to guard_crs (raise if undefined, warn if geographic); irreparable topology and orphaned nodes to guard_topology (drop geometries that collapse to empty, flag suspect orphans). All outcomes converge on a clean GeoDataFrame plus immutable anomaly log. FAILURE MODE GUARD FUNCTION OUTCOME Domain drift Undefined / Irreparable topology free-text status,out-of-range MVA geographic CRS & orphaned nodes guard_domain() guard_crs() guard_topology() Quarantine, never drop Refuse to measure Drop collapsed geometry status='unknown', MVA→NA raise if none; warn if degrees flag suspect_orphan Clean GeoDataFrame + immutable anomaly log

Performance & scalability

The chunk size is the primary memory dial. For multi-gigabyte national transmission layers, 5,000–10,000 features per chunk balances per-read I/O overhead against heap stability; smaller chunks lower the ceiling at the cost of more pyogrio round-trips. Because reads are dispatched to a thread executor while validation runs as awaited coroutines, disk and object-store latency overlap with CPU-bound geometry repair rather than serialising behind it — the practical win when each read_file against a cloud bucket carries hundreds of milliseconds of latency.

The cached target CRS is the second lever: re-parsing the projection string per chunk is a measurable cost at national scale, and building it once in __init__ removes it. When a downstream stage needs nearest-asset queries against this validated output — for example feeding grid proximity and distance calculations — build a spatial index (gdf.sindex) once on the projected frame so topology relationships are not recomputed per query. For datasets that exceed even chunked single-machine limits, the _validate_chunk contract drops into a dask-geopandas partition map almost unchanged, because schema, domain, and topology checks are all row-local and parallelise with no cross-partition shuffle.

Validation & audit trail

A validated dataset is only defensible if its rejections are recorded. Interconnection studies, environmental impact reports, and project-finance due diligence can each be rejected when the underlying network data lacks documented lineage. Every run should emit an immutable record covering, per layer: the count and identity of features quarantined for invalid_status, the count repaired or dropped for invalid_geometry, the source-and-target CRS pair, and the final feature count carried forward.

python
import json


def audit_summary(clean_gdf: gpd.GeoDataFrame, log: list, target_epsg: int) -> str:
    by_type: Dict[str, int] = {}
    for entry in log:
        by_type[entry["anomaly_type"]] = by_type.get(entry["anomaly_type"], 0) + 1
    rec = {
        "features_out": int(len(clean_gdf)),
        "anomalies_total": len(log),
        "anomalies_by_type": by_type,
        "target_epsg": target_epsg,
        "target_unit": clean_gdf.crs.axis_info[0].unit_name,
    }
    # Gate: a metric target reporting degree units means the reprojection never ran.
    assert rec["target_unit"] != "degree", "CRS drift: output still in degrees"
    return json.dumps(rec)

In CI/CD, wrap the validator in pytest with small spatial fixtures and gate on these assertions: an output unit of degree, an anomaly rate that jumps above a baseline, or a non-empty set of irreparable geometries each signals an upstream ETL regression that must block release rather than ship into a permitting submission. The cleaned output then transitions safely into spatial buffering — analysts can feed it to grid capacity buffer analysis without risking projection-induced distance errors or topology breaks — and the same anomaly log threads forward as the provenance record. By embedding schema enforcement, domain coercion, CRS normalization, topology repair, memory-aware chunking, and an immutable audit log into the ingestion layer, grid teams eliminate the silent failures that otherwise surface only at interconnection review.

Explore this section

Back to Grid Infrastructure & Network Proximity Analysis