Reconciling Geometry Type Changes Across Branches

A precise procedure for handling a vector layer whose geometry type diverged between branches — single versus multi, promoted lines, mixed collections — and normalizing both sides safely before merge, as part of Schema Drift Reconciliation for Spatial Layers.

Concept & Context

Most schema drift is attribute drift, but the sharpest failures come from the geometry token itself changing. One branch edits parcels as Polygon, another writes them back through a tool that promotes everything to MultiPolygon; a road layer gains a few multi-part features and its declared type flips from LineString to MultiLineString; a well-meaning bulk edit leaves a layer holding a GeometryCollection. A merge across that boundary either throws a driver error at write time or, worse, silently coerces geometries and loses parts. Because the attribute-level migration with Fiona deliberately passes geometry through untouched, geometry-type drift needs its own detection and normalization pass, and that is what this page builds with Shapely and GeoPandas.

The governing principle is that normalization may only ever widen: promoting a single-part geometry to its multi-part container is lossless, so it is allowed automatically, while demoting a multi to a single would discard parts and is forbidden. Anything that would change spatial dimension — lines to polygons, or a genuine mix of dimensions in one layer — is not drift at all but a modeling conflict, and gets escalated rather than coerced. Getting these rules right up front is what lets downstream steps like attribute reconciliation for tabular spatial data run against a layer with a single, well-defined geometry type.

Core Normalization Pipeline

  1. Profile geometry types. For each branch, tabulate the distinct Shapely geom_type values actually present, independent of the declared layer token — they often disagree.
  2. Classify the drift. Map the pair of profiles onto one of: identical, safe single↔multi promotion, or an unsafe dimensional mix that must escalate.
  3. Choose a target type. Pick the least-widening common type; within one dimension that is the Multi form (e.g. MultiPolygon for any mix of Polygon and MultiPolygon).
  4. Normalize geometries. Promote singles to multis, explode GeometryCollections into their homogeneous parts, and route null/empty geometries to a flag queue.
  5. Validate before merge. Assert every output geometry matches the target type, is valid, and preserved area or length within tolerance.

Working Implementation

"""Detect and reconcile geometry-type drift between two branches of a vector layer.

Rule: promotion (single -> multi) is lossless and automatic; demotion and cross-
dimension changes are forbidden and escalate. Null/empty geometries are flagged.
"""
from __future__ import annotations
import geopandas as gpd
from shapely.geometry import MultiPoint, MultiLineString, MultiPolygon
from shapely.geometry.base import BaseGeometry

# Base dimension of each geometry type (0=point, 1=line, 2=area).
_DIMENSION = {
    "Point": 0, "MultiPoint": 0,
    "LineString": 1, "MultiLineString": 1, "LinearRing": 1,
    "Polygon": 2, "MultiPolygon": 2,
}
_MULTI_FOR_DIM = {0: "MultiPoint", 1: "MultiLineString", 2: "MultiPolygon"}
_MULTI_CTOR = {0: MultiPoint, 1: MultiLineString, 2: MultiPolygon}


def profile_types(gdf: gpd.GeoDataFrame) -> dict[str, int]:
    """Distinct actual geometry types present and their counts."""
    return gdf.geometry.geom_type.value_counts(dropna=False).to_dict()


def classify_drift(prof_a: dict[str, int], prof_b: dict[str, int]) -> tuple[str, str]:
    """Return (status, detail). status in {'identical','promote','escalate'}."""
    types = {t for t in (prof_a | prof_b) if t is not None}
    dims = {_DIMENSION.get(t) for t in types if t in _DIMENSION}
    if None in {_DIMENSION.get(t) for t in types}:
        return "escalate", f"unrecognized geometry type in {types}"
    if len(dims) > 1:
        return "escalate", f"mixed spatial dimensions {dims}: modeling conflict, not drift"
    if set(prof_a) == set(prof_b):
        return "identical", f"both branches are {types}"
    return "promote", f"unify {types} to {_MULTI_FOR_DIM[dims.pop()]}"


def choose_target_type(prof_a: dict[str, int], prof_b: dict[str, int]) -> str:
    dims = {_DIMENSION[t] for t in (prof_a | prof_b) if t in _DIMENSION}
    if len(dims) != 1:
        raise ValueError("cannot choose a single target type across mixed dimensions")
    return _MULTI_FOR_DIM[dims.pop()]


def _promote(geom: BaseGeometry, target: str) -> BaseGeometry | None:
    if geom is None or geom.is_empty:
        return None                                  # flagged upstream
    gt = geom.geom_type
    if gt == target:
        return geom
    if gt == "GeometryCollection":
        # Keep only parts matching the target dimension; discard foreign-dim debris.
        dim = _DIMENSION[target[len("Multi"):]] if target.startswith("Multi") else None
        parts = [g for g in geom.geoms if _DIMENSION.get(g.geom_type) == dim]
        if not parts:
            return None
        return _MULTI_CTOR[dim]([p for g in parts
                                 for p in (g.geoms if g.geom_type.startswith("Multi") else [g])])
    if _DIMENSION.get(gt) == _DIMENSION.get(target[len("Multi"):]):
        # Single -> Multi promotion: lossless wrap of one part.
        return _MULTI_CTOR[_DIMENSION[gt]]([geom])
    raise ValueError(f"refusing unsafe cast {gt} -> {target}")


def reconcile_geometry(gdf: gpd.GeoDataFrame, target: str
                       ) -> tuple[gpd.GeoDataFrame, gpd.GeoDataFrame]:
    """Return (normalized_gdf, flagged_gdf) where flagged holds null/empty rows."""
    empty_mask = gdf.geometry.is_empty | gdf.geometry.isna()
    flagged = gdf[empty_mask].copy()
    work = gdf[~empty_mask].copy()
    work["geometry"] = work.geometry.apply(lambda g: _promote(g, target))
    work = gpd.GeoDataFrame(work, geometry="geometry", crs=gdf.crs)
    return work, flagged


if __name__ == "__main__":
    branch_a = gpd.read_file("branches/main/parcels.gpkg")     # Polygon
    branch_b = gpd.read_file("branches/edit/parcels.gpkg")     # MultiPolygon

    pa, pb = profile_types(branch_a), profile_types(branch_b)
    status, detail = classify_drift(pa, pb)
    print(f"[{status}] {detail}")
    if status == "escalate":
        raise SystemExit("geometry-type drift requires manual review")
    if status == "promote":
        target = choose_target_type(pa, pb)
        a_norm, a_flag = reconcile_geometry(branch_a, target)
        b_norm, b_flag = reconcile_geometry(branch_b, target)
        a_norm.to_file("normalized/main_parcels.gpkg", driver="GPKG")
        b_norm.to_file("normalized/edit_parcels.gpkg", driver="GPKG")
        print(f"normalized to {target}; "
              f"flagged {len(a_flag) + len(b_flag)} null/empty geometries")

Validation & Output Verification

Prove the normalization was both complete and lossless before the layers reach the merge.

def verify_normalization(orig: gpd.GeoDataFrame, norm: gpd.GeoDataFrame,
                         target: str, dim: int) -> None:
    # 1. Every retained geometry is exactly the target type.
    types = set(norm.geometry.geom_type.unique())
    assert types <= {target}, f"non-target geometries remain: {types}"
    # 2. Nothing became invalid during promotion.
    assert norm.geometry.is_valid.all(), "invalid geometry produced by normalization"
    # 3. Area/length preserved within tolerance (promotion must not change measure).
    measure = (lambda g: g.geometry.area) if dim == 2 else (lambda g: g.geometry.length)
    kept = orig[~(orig.geometry.is_empty | orig.geometry.isna())]
    before, after = measure(kept).sum(), measure(norm).sum()
    assert abs(before - after) <= 1e-6 * max(before, 1.0), "measure drift after promotion"
    print("normalization verified")

Because promotion only rewraps parts, total area (for polygons) or total length (for lines) must be identical to floating-point tolerance — a measure mismatch means a part was dropped and the run should fail. Once both branches share the target type and pass validation, hand off to the parent schema drift reconciliation gate, which folds the unified geometry token into the reconciled schema fingerprint.

Failure Modes

  • refusing unsafe cast MultiPolygon -> Polygon — a demotion was requested. Root cause: the target type was chosen from the wrong (narrower) side. Fix: always target the Multi form; never demote, since multi-part features would lose all but one part.

  • escalate: mixed spatial dimensions — the layer holds both lines and polygons. Root cause: a modeling error or an accidental append of the wrong dataset, not true drift. Fix: split by dimension into separate layers, or designate one dimension authoritative before reconciling.

  • measure drift after promotion — output area or length differs from input. Root cause: a GeometryCollection explode discarded parts of the target dimension, or an invalid ring was silently repaired. Fix: inspect the exploded parts, run make_valid explicitly and re-check, and confirm the collection contained no target-dimension debris.

  • Empty MultiPolygon written to output — a null or empty geometry slipped past the flag queue. Root cause: is_empty was not checked before promotion. Fix: route empties and nulls into the flagged frame and decide their treatment explicitly rather than coercing them into empty multis.