GeoJSON Processor and Validator

GeoJSON Processor and Validator

Problem Statement

GeoJSON is the de-facto interchange format for geographic vector data, used by everything from PostGIS exports to web mapping libraries like Leaflet and Mapbox. In practice, GeoJSON files arriving from third-party sources are frequently malformed: coordinates out of range, self-intersecting polygons, wrong winding order, mixed geometry types, or simply enormous files with sub-meter precision that bloats payloads and slows down web clients.

This script provides a single, dependency-light command-line tool that acts as a gatekeeper for GeoJSON data. It validates structural and geometric integrity (catching bad geometries before they hit a database or a map renderer), reports useful diagnostics, and optionally simplifies geometries using Douglas-Peucker to shrink file size — all while preserving the standard RFC 7946 output format.

How It Works

The script follows a pipeline architecture: load → validate → transform → write. Input is read with the geojson library, which parses the JSON and exposes typed geometry objects. Each geometry is then converted to a Shapely geometry for geometric operations (validity checks, simplification, bounds computation). Validation runs on the Shapely representation because Shapely's is_valid correctly detects self-intersections and ring issues that raw JSON parsing won't catch.

Key design decisions:

• Fail-soft validation: validation collects all problems into a report rather than raising on the first one, so users can fix everything in one pass.
• Preserve properties: only geometry is mutated during simplification; feature properties and IDs are passed through untouched.
• Bounding-box guard: coordinates are checked against [-180, 180] longitude and [-90, 90] latitude to catch swapped lat/lon or projected coordinates mistakenly labeled as WGS84.
• Exit codes matter: 0 on success, 1 on validation failure, 2 on I/O or parse errors — so the script is CI-friendly.

The argparse layer exposes --input, --output, --validate, --simplify, and --report, keeping the CLI composable and scriptable.

Full Script

#!/usr/bin/env python3
"""
GeoJSON Processor and Validator

Loads a GeoJSON file, validates its geometries (structural + geometric),
optionally simplifies geometries with Douglas-Peucker, and writes the
result to a new GeoJSON file. Prints a validation report to stderr.

Usage:
    python3 geojson_tools.py --input boundaries.geojson --validate \
        --simplify 0.001 --output simplified.geojson
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
from pathlib import Path
from typing import Any, Iterable

import geojson
from shapely.geometry import shape, mapping
from shapely.geometry.base import BaseGeometry

# ---------------------------------------------------------------------------
# Configuration (environment variables with sensible defaults)
# ---------------------------------------------------------------------------
LOG_LEVEL: str = os.environ.get("GEOJSON_LOG_LEVEL", "INFO").upper()
DEFAULT_TOLERANCE: float = float(os.environ.get("GEOJSON_SIMPLIFY_TOLERANCE", "0.001"))
MAX_BBOX_LON: float = float(os.environ.get("GEOJSON_MAX_LON", "180.0"))
MAX_BBOX_LAT: float = float(os.environ.get("GEOJSON_MAX_LAT", "90.0"))
PRESERVE_TOPO: bool = os.environ.get("GEOJSON_PRESERVE_TOPOLOGY", "true").lower() == "true"

logging.basicConfig(
    level=getattr(logging, LOG_LEVEL, logging.INFO),
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
log = logging.getLogger("geojson_tools")


class GeoJSONProcessingError(Exception):
    """Raised for unrecoverable I/O or parse errors."""


# ---------------------------------------------------------------------------
# Loading
# ---------------------------------------------------------------------------
def load_geojson(path: Path) -> dict[str, Any]:
    """Read and JSON-parse a GeoJSON file, returning a plain dict."""
    if not path.exists():
        raise GeoJSONProcessingError(f"Input file not found: {path}")
    if not path.is_file():
        raise GeoJSONProcessingError(f"Input path is not a file: {path}")

    try:
        with path.open("r", encoding="utf-8") as fh:
            data = json.load(fh)
    except json.JSONDecodeError as exc:
        raise GeoJSONProcessingError(f"Invalid JSON in {path}: {exc}") from exc
    except OSError as exc:
        raise GeoJSONProcessingError(f"Could not read {path}: {exc}") from exc

    if not isinstance(data, dict) or data.get("type") not in {
        "FeatureCollection", "Feature", "GeometryCollection",
        "Point", "MultiPoint", "LineString", "MultiLineString",
        "Polygon", "MultiPolygon",
    }:
        raise GeoJSONProcessingError(
            f"{path} does not look like GeoJSON (missing/unknown 'type')."
        )
    return data


def iter_features(doc: dict[str, Any]) -> Iterable[dict[str, Any]]:
    """Yield every Feature dict from a FeatureCollection, Feature, or bare geometry."""
    dtype = doc.get("type")
    if dtype == "FeatureCollection":
        yield from doc.get("features", [])
    elif dtype == "Feature":
        yield doc
    else:
        # Bare geometry: wrap it so downstream code has a uniform shape.
        yield {"type": "Feature", "geometry": doc, "properties": {}}


# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
def check_bbox(geom: BaseGeometry) -> list[str]:
    """Return a list of bbox-related warnings for a geometry."""
    problems: list[str] = []
    if geom.is_empty:
        return ["empty geometry"]

    minx, miny, maxx, maxy = geom.bounds
    if minx < -MAX_BBOX_LON or maxx > MAX_BBOX_LON:
        problems.append(f"longitude out of range: [{minx:.4f}, {maxx:.4f}]")
    if miny < -MAX_BBOX_LAT or maxy > MAX_BBOX_LAT:
        problems.append(f"latitude out of range: [{miny:.4f}, {maxy:.4f}]")
    return problems


def validate_feature(index: int, feature: dict[str, Any]) -> list[str]:
    """Validate a single Feature dict; return a list of human-readable issues."""
    issues: list[str] = []
    geom_obj = feature.get("geometry")
    if geom_obj is None:
        issues.append(f"feature[{index}]: null geometry")
        return issues

    try:
        shp = shape(geom_obj)
    except (ValueError, TypeError, KeyError, AttributeError) as exc:
        issues.append(f"feature[{index}]: cannot parse geometry ({exc})")
        return issues

    if not shp.is_valid:
        issues.append(f"feature[{index}]: invalid geometry ({shp.geom_type})")
    for prob in check_bbox(shp):
        issues.append(f"feature[{index}]: {prob}")

    return issues


def validate_document(doc: dict[str, Any]) -> list[str]:
    """Run validation across all features; return every issue found."""
    all_issues: list[str] = []
    for i, feat in enumerate(iter_features(doc)):
        if not isinstance(feat, dict):
            all_issues.append(f"feature[{i}]: not a JSON object")
            continue
        all_issues.extend(validate_feature(i, feat))
    return all_issues


# ---------------------------------------------------------------------------
# Transformation
# ---------------------------------------------------------------------------
def simplify_feature(feature: dict[str, Any], tolerance: float) -> dict[str, Any]:
    """Return a new Feature with its geometry simplified. Properties are preserved."""
    geom_obj = feature.get("geometry")
    if geom_obj is None:
        return feature

    try:
        shp = shape(geom_obj)
    except (ValueError, TypeError, KeyError, AttributeError):
        # Unparseable geometry: pass through unchanged so we don't silently drop data.
        log.warning("Skipping simplify for unparseable geometry")
        return feature

    simplified = shp.simplify(tolerance, preserve_topology=PRESERVE_TOPO)

    # Build a fresh feature to avoid mutating the caller's dict.
    new_feature: dict[str, Any] = {
        "type": "Feature",
        "geometry": mapping(simplified),
        "properties": feature.get("properties", {}),
    }
    if "id" in feature:
        new_feature["id"] = feature["id"]
    return new_feature


def simplify_document(doc: dict[str, Any], tolerance: float) -> dict[str, Any]:
    """Apply simplification to every feature in the document."""
    dtype = doc.get("type")
    if dtype == "FeatureCollection":
        return {
            "type": "FeatureCollection",
            "features": [simplify_feature(f, tolerance) for f in doc.get("features", [])],
        }
    if dtype == "Feature":
        return simplify_feature(doc, tolerance)
    # Bare geometry: wrap, simplify, unwrap.
    wrapped = simplify_feature({"type": "Feature", "geometry": doc}, tolerance)
    return wrapped["geometry"]


# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------
def write_geojson(doc: dict[str, Any], path: Path) -> None:
    """Serialise a document to disk as compact GeoJSON."""
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("w", encoding="utf-8") as fh:
            # separators trims whitespace — smaller payloads for web use.
            json.dump(doc, fh, ensure_ascii=False, separators=(",", ":"))
    except OSError as exc:
        raise GeoJSONProcessingError(f"Could not write {path}: {exc}") from exc


def print_report(issues: list[str], feature_count: int) -> None:
    """Print a validation summary to stderr."""
    print(f"\n=== Validation report ===", file=sys.stderr)
    print(f"Features checked: {feature_count}", file=sys.stderr)
    print(f"Issues found:     {len(issues)}", file=sys.stderr)
    for issue in issues:
        print(f"  - {issue}", file=sys.stderr)
    print("=========================\n", file=sys.stderr)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Validate and simplify GeoJSON files.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument("--input", "-i", required=True, type=Path,
                        help="Input GeoJSON file path.")
    parser.add_argument("--output", "-o", type=Path,
                        help="Output GeoJSON file path (required with --simplify).")
    parser.add_argument("--validate", action="store_true",
                        help="Run validation and print a report.")
    parser.add_argument("--simplify", type=float, metavar="TOLERANCE",
                        default=None,
                        help="Douglas-Peucker tolerance in degrees (e.g. 0.001).")
    parser.add_argument("--report", action="store_true",
                        help="Print the validation report even without --validate.")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    try:
        doc = load_geojson(args.input)
    except GeoJSONProcessingError as exc:
        log.error(str(exc))
        return 2

    features = list(iter_features(doc))
    feature_count = len(features)
    log.info("Loaded %d feature(s) from %s", feature_count, args.input)

    # --- Validation ---------------------------------------------------------
    issues: list[str] = []
    if args.validate or args.report:
        issues = validate_document(doc)
        print_report(issues, feature_count)
        if issues and args.validate:
            log.error("Validation failed with %d issue(s).", len(issues))

    # --- Simplification -----------------------------------------------------
    if args.simplify is not None:
        if args.simplify < 0:
            log.error("--simplify tolerance must be non-negative.")
            return 2
        if args.output is None:
            log.error("--output is required when --simplify is used.")
            return 2
        log.info("Simplifying with tolerance=%.6f (preserve_topology=%s)",
                 args.simplify, PRESERVE_TOPO)
        doc = simplify_document(doc, args.simplify)

    # --- Output -------------------------------------------------------------
    if args.output is not None:
        try:
            write_geojson(doc, args.output)
        except GeoJSONProcessingError as exc:
            log.error(str(exc))
            return 2
        log.info("Wrote %s", args.output)

    # Exit code: 1 if validation was requested and found problems.
    if args.validate and issues:
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

The script reads a handful of environment variables so behaviour can be tuned without editing code (useful in CI or containerised deployments):

CLI flags: --input/-i (required), --output/-o, --validate, --simplify TOLERANCE, --report.

Usage Examples

1. Validate only — report issues, no output file:

$ python3 geojson_tools.py --input boundaries.geojson --validate
2024-06-01T10:12:03 [INFO] Loaded 42 feature(s) from boundaries.geojson

=== Validation report ===
Features checked: 42
Issues found:     2
  - feature[7]: invalid geometry (Polygon)
  - feature[19]: longitude out of range: [-190.0000, -170.0000]
=========================
2024-06-01T10:12:03 [ERROR] Validation failed with 2 issue(s).
$ echo $?
1

2. Simplify and write a smaller file:

$ python3 geojson_tools.py --input boundaries.geojson \
    --simplify 0.001 --output simplified.geojson
2024-06-01T10:13:11 [INFO] Loaded 42 feature(s) from boundaries.geojson
2024-06-01T10:13:11 [INFO] Simplifying with tolerance=0.001000 (preserve_topology=True)
2024-06-01T10:13:11 [INFO] Wrote simplified.geojson
$ ls -lh boundaries.geojson simplified.geojson
-rw-r--r-- 1 user user 2.4M boundaries.geojson
-rw-r--r-- 1 user user 812K simplified.geojson

3. Combined validate + simplify for a CI gate:

$ GEOJSON_LOG_LEVEL=DEBUG python3 geojson_tools.py \
    --input boundaries.geojson --validate --simplify 0.0005 \
    --output dist/boundaries.min.geojson

Error Handling

• Missing or unreadable input → GeoJSONProcessingError, logged, exit code 2.
• Malformed JSON → caught json.JSONDecodeError, re-raised as GeoJSONProcessingError, exit 2.
• Non-GeoJSON JSON (missing/unknown type) → rejected during load with a clear message.
• Unparseable individual geometry → logged as a warning; the feature is passed through unchanged rather than silently dropped.
• Invalid geometry (self-intersecting rings, etc.) → reported by --validate; exit code 1 so CI pipelines can fail the build.
• Coordinates out of range → flagged in the report; the file is still processed so you can inspect the rest.
• --simplify without --output → explicit error, exit 2.
• Negative tolerance → rejected before any work is done.

Extending Ideas

• CRS reprojection: add a --reproject EPSG:3857 flag using pyproj to convert WGS84 GeoJSON to/from projected coordinate systems before simplification.
• Streaming for huge files: swap json.load for ijson or geojson-stream to process multi-gigabyte FeatureCollections without loading them fully into memory.
• Topology repair: use shapely.make_valid() or shapely.validation.explain_validity() to auto-fix polygons and surface a precise reason for each invalid ring.
• Multi-format output: add GeoPackage/FlatGeobuf writers via fiona or ogr2ogr subprocess calls, so the same validation pipeline serves non-web consumers.
• CI integration: emit the report as JSON (--report-format json) so GitHub Actions or GitLab CI can annotate specific features inline.