Real ingestion is not one file with one frame. It is fourteen sources, four of which declare a projection correctly, six of which declare one that is technically wrong, three of which declare nothing, and one that changes convention halfway through the year. This guide is about running that reality through a single gate without either rejecting everything or quietly assuming your way to a plausible-looking corpus. It is the working implementation of coordinate reference system normalization.
When to Use This Approach
Use it at the boundary where external data becomes internal data, and nowhere else. Normalizing repeatedly downstream is wasted work and creates several places where the rules can diverge.
| Source behaviour | Handling | Recorded as |
|---|---|---|
| Declares a valid frame | Validate and transform | Declared |
| Declares a wrong frame consistently | Per-source override, with evidence | Overridden |
| Declares nothing, frame known | Per-source default, with evidence | Assumed |
| Declares nothing, frame unknown | Reject | Rejected |
| Mixes frames within one file | Reject the file, not the rows | Rejected |
The distinction between the second and third rows matters operationally. An override says the source is wrong and we know better; an assumption says the source is silent and we have external evidence. Both are declarations someone made, and both should be reviewable — which is why the provenance field records which one applied.
Implementation
The ingester resolves a frame per source, transforms in batches for efficiency, and returns both accepted geometries and a rejection list that a person is expected to read.
import logging
from collections import defaultdict
from dataclasses import dataclass
from typing import Iterable, Optional, Sequence
from pyproj import CRS, Transformer
from pyproj.exceptions import CRSError
from shapely.ops import transform as shapely_transform
from shapely.validation import make_valid
from shapely.errors import GEOSException
log = logging.getLogger("mixed_frame_ingest")
CANONICAL_EPSG = 4326
@dataclass(frozen=True)
class SourceRule:
source_id: str
override_epsg: Optional[int] = None # the source declares wrongly
default_epsg: Optional[int] = None # the source declares nothing
evidence: str = "" # why we believe either of the above
@dataclass(frozen=True)
class Accepted:
geometry: object
source_epsg: int
provenance: str # declared | overridden | assumed
repaired: bool
evidence: str
@dataclass(frozen=True)
class Rejected:
source_id: str
identifier: str
reason: str
def resolve_frame(declared: Optional[int], rule: SourceRule) -> tuple[Optional[int], str, str]:
"""Return (epsg, provenance, evidence). None means reject."""
if rule.override_epsg is not None:
return rule.override_epsg, "overridden", rule.evidence
if declared is not None:
return declared, "declared", ""
if rule.default_epsg is not None:
return rule.default_epsg, "assumed", rule.evidence
return None, "rejected", "source declares no frame and no default is configured"Batching the transforms matters more than it looks on a large ingest. Building a transformer is expensive relative to using one, and a naive loop that constructs a fresh transformer per geometry can spend most of its time on setup.
def ingest(records: Iterable[dict], rules: dict[str, SourceRule]
) -> tuple[list[Accepted], list[Rejected]]:
"""Normalize a mixed batch. Returns accepted geometries and an explicit rejection list."""
accepted: list[Accepted] = []
rejected: list[Rejected] = []
by_frame: dict[tuple[int, str, str], list[dict]] = defaultdict(list)
for rec in records:
rule = rules.get(rec["source_id"], SourceRule(rec["source_id"]))
epsg, provenance, evidence = resolve_frame(rec.get("declared_epsg"), rule)
if epsg is None:
rejected.append(Rejected(rec["source_id"], rec["id"], evidence))
continue
try:
CRS.from_epsg(epsg)
except CRSError as exc:
rejected.append(Rejected(rec["source_id"], rec["id"],
f"EPSG:{epsg} will not construct: {exc}"))
continue
by_frame[(epsg, provenance, evidence)].append(rec)
target = CRS.from_epsg(CANONICAL_EPSG)
for (epsg, provenance, evidence), group in by_frame.items():
source = CRS.from_epsg(epsg)
tf = None if source.equals(target) else Transformer.from_crs(
source, target, always_xy=True)
for rec in group:
geom, repaired = _repair(rec["geometry"])
if geom is None:
rejected.append(Rejected(rec["source_id"], rec["id"],
"geometry could not be repaired"))
continue
try:
moved = geom if tf is None else shapely_transform(tf.transform, geom)
except Exception as exc:
rejected.append(Rejected(rec["source_id"], rec["id"],
f"transform from EPSG:{epsg} failed: {exc}"))
continue
accepted.append(Accepted(moved, epsg, provenance, repaired, evidence))
log.info("ingest: %d accepted, %d rejected across %d frame group(s)",
len(accepted), len(rejected), len(by_frame))
return accepted, rejected
def _repair(geom):
if geom is None or geom.is_empty:
return None, False
if geom.is_valid:
return geom, False
try:
fixed = make_valid(geom)
except GEOSException:
return None, False
return (fixed, True) if not fixed.is_empty else (None, False)The rejection list is a return value rather than a log line because it needs to reach a person. A pipeline that logs rejections and returns only successes will run for months with a fifth of its input silently discarded, and the symptom — “the corpus seems thin in the north” — arrives long after the cause.
Validation & Testing
def test_rejections_are_returned_not_only_logged():
records = [{"source_id": "unknown", "id": "a", "geometry": POINT, "declared_epsg": None}]
accepted, rejected = ingest(records, rules={})
assert not accepted and len(rejected) == 1
assert "no frame" in rejected[0].reason
def test_provenance_distinguishes_assumed_from_declared():
rules = {"silent": SourceRule("silent", default_epsg=27700, evidence="agency confirmed 2024")}
records = [
{"source_id": "silent", "id": "a", "geometry": POINT, "declared_epsg": None},
{"source_id": "silent", "id": "b", "geometry": POINT, "declared_epsg": 4326},
]
accepted, _ = ingest(records, rules)
assert {a.provenance for a in accepted} == {"assumed", "declared"}
def test_one_transformer_per_frame_group(monkeypatch):
built = []
original = Transformer.from_crs
monkeypatch.setattr(Transformer, "from_crs",
lambda *a, **k: (built.append(a) or original(*a, **k)))
ingest(HUNDRED_RECORDS_TWO_FRAMES, rules={})
assert len(built) <= 2The third test guards a performance property with a correctness-shaped assertion, which is the only way this kind of regression gets caught. Nothing about a per-geometry transformer produces wrong output; it simply makes reprocessing expensive enough that people stop doing it.
Gotchas & Edge Cases
An override applied to a source that later fixes itself. The upstream export starts declaring correctly, the override keeps forcing the old value, and the geometry is now wrong in the opposite direction. Review overrides on a schedule, and log when an override contradicts a declaration rather than silently winning.
Mixed frames within one file. Rejecting the file rather than the offending rows is deliberate: a file with two conventions usually means a concatenation, and the rows you can identify as wrong are rarely all of them.
Repair applied before the frame is known. Repairing geometry in a frame that later turns out to be wrong bakes the repair into the wrong coordinates. Resolve the frame first, repair second, transform third — the order in the code is the order that survives contact with bad data.
Rejections that nobody owns. A queue with no reader is a discard with extra steps. Route rejections to whoever owns the source, with the source identifier in the message, and track the queue depth as a monitored value.
Evidence fields left empty. An assumption with no recorded evidence is indistinguishable from a guess six months later. Make the evidence string required on any rule that overrides or defaults, even if the evidence is only “confirmed by the data owner on this date”.
Frequently Asked Questions
Should a source override be per file or per source?
Per source, with the file identifier recorded on the accepted geometry so a per-file exception can be traced later. Per-file overrides multiply quickly and end up as a directory of one-off rules nobody can reason about; a per-source rule with a dated evidence note stays reviewable. Where one file genuinely differs, that is usually a sign the source has changed its convention, which is worth handling as a change rather than as an exception.
How should the rejection queue be sized and monitored?
Track depth and age, and alert on age rather than depth. A queue with two hundred items that are all from this morning is a busy ingest; a queue with five items that have been there for three weeks is an ownership problem. Aging is also the signal that distinguishes a genuine data issue from a rule that needs adding — items that sit are usually items nobody knows what to do with.
Is it worth normalizing during a bulk backfill differently from streaming ingest?
The rules should be identical; only the batching changes. A backfill can group aggressively across a whole file, while a streaming ingest groups within a window. What must not differ is the resolution logic — two code paths that decide frames differently will produce a corpus whose geometry depends on when it arrived, which is the hardest kind of inconsistency to diagnose because nothing about the data records it.
What should happen when a transform succeeds but produces implausible output?
Treat it as a rejection with a distinct reason. A transform that lands geometry in the wrong hemisphere has technically succeeded and is certainly wrong, and the round-trip check described in the parent topic is what catches it. Recording it as its own rejection class rather than as a generic failure is what lets you see, later, that one source accounts for all of them.
Should accepted geometry keep its original coordinates as well?
Keep the source frame code and the transformation note, which together let the original be reconstructed, rather than storing both geometries. Duplicating the geometry doubles the storage for a value that is almost never read, and the two copies drift the moment anything edits one of them. The exception is a corpus where the original is legally the record of truth, in which case store the original and treat the canonical copy as derived.