Automating coordinate validation with Python and Shapely

import logging
from typing import Optional
from dataclasses import dataclass
import numpy as np
from shapely.geometry import Point, Polygon

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)s | %(module)s | %(message)s'
)
logger = logging.getLogger(__name__)


@dataclass
class ValidationResult:
    is_valid: bool
    error_code: Optional[str] = None
    message: Optional[str] = None
    cleaned_lat: Optional[float] = None
    cleaned_lon: Optional[float] = None


def validate_coordinate(
    lat: float,
    lon: float,
    operational_boundary: Optional[Polygon] = None,
    allow_null_island: bool = False
) -> ValidationResult:
    """
    Validates a single latitude/longitude pair against WGS84 constraints
    and an optional operational boundary. Returns a structured result object.
    """
    # 1. Type coercion & numeric safety
    try:
        lat_val = float(lat)
        lon_val = float(lon)
    except (ValueError, TypeError):
        return ValidationResult(False, "TYPE_MISMATCH", "Non-numeric coordinate values detected.")

    if np.isnan(lat_val) or np.isnan(lon_val) or np.isinf(lat_val) or np.isinf(lon_val):
        return ValidationResult(False, "NUMERIC_INVALID", "Coordinate contains NaN or infinite values.")

    # 2. WGS84 bounds enforcement
    if not (-90.0 <= lat_val <= 90.0) or not (-180.0 <= lon_val <= 180.0):
        # Heuristic: detect common (lon, lat) swap from broker feeds
        if (-180.0 <= lat_val <= 180.0) and (-90.0 <= lon_val <= 90.0):
            return ValidationResult(
                False, "AXIS_SWAP_DETECTED",
                f"Coordinates appear swapped: ({lat_val}, {lon_val}). Expected (lat, lon)."
            )
        return ValidationResult(
            False, "OUT_OF_BOUNDS",
            f"Coordinates exceed WGS84 limits: ({lat_val}, {lon_val})."
        )

    # 3. Null Island / default geocoding filter
    if not allow_null_island and lat_val == 0.0 and lon_val == 0.0:
        return ValidationResult(
            False, "NULL_ISLAND",
            "Coordinate defaults to (0.0, 0.0). Verify upstream geocoder."
        )

    # 4. Operational boundary containment (optional)
    # Point(x, y) = Point(longitude, latitude) per GIS convention
    if operational_boundary is not None:
        point = Point(lon_val, lat_val)
        if not operational_boundary.contains(point):
            return ValidationResult(
                False, "OUTSIDE_BOUNDARY",
                "Coordinate falls outside the designated operational polygon."
            )

    # 5. Precision normalization (~11 cm accuracy at equator)
    return ValidationResult(
        is_valid=True,
        cleaned_lat=round(lat_val, 6),
        cleaned_lon=round(lon_val, 6)
    )

import pandas as pd
from shapely.geometry import Polygon
from typing import Optional


def batch_validate_coordinates(
    df: pd.DataFrame,
    lat_col: str,
    lon_col: str,
    boundary: Optional[Polygon] = None
) -> pd.DataFrame:
    """
    Applies coordinate validation across a DataFrame and appends structured results.
    For datasets exceeding 100k rows, replace apply() with vectorized shapely.contains_xy
    once the schema is stabilized.
    """
    logger.info("Starting batch coordinate validation on %d records.", len(df))

    results = df.apply(
        lambda row: validate_coordinate(row[lat_col], row[lon_col], boundary),
        axis=1
    )

    df = df.copy()
    df["validation_status"] = results.apply(lambda x: x.is_valid)
    df["error_code"] = results.apply(lambda x: x.error_code)
    df["error_message"] = results.apply(lambda x: x.message)
    df["cleaned_lat"] = results.apply(lambda x: x.cleaned_lat)
    df["cleaned_lon"] = results.apply(lambda x: x.cleaned_lon)

    valid_count = df["validation_status"].sum()
    logger.info("Validation complete. %d/%d records passed spatial checks.", valid_count, len(df))
    return df