Schema Validation Pipelines

A route optimizer is only as trustworthy as the last payload it read. Municipal waste operations emit heterogeneous telemetry all day — vehicle CAN buses, onboard load-cell scales, ultrasonic and infrared fill sensors — and every source speaks a slightly different JSON dialect at a different cadence, with missing fields, coerced-to-string numbers, and out-of-sequence timestamps. Feed that raw into a constraint solver and the failures are silent, not loud: a null weight quietly relaxes an axle limit, a future-dated weigh-in reorders a manifest, a string "88" fill level compares wrong against a numeric threshold, and the dispatch that comes out is infeasible in ways no exception ever flagged. A schema validation pipeline is the boundary where those payloads are proven — or quarantined — before any downstream system trusts them. It is the first checkpoint in the Telematics & Sensor Data Ingestion framework, and everything the routing side does depends on it holding the line.

This page builds that boundary in Python: a strict Pydantic v2 contract, geospatial and temporal gates, deterministic deduplication, and a compliance-coded quarantine ledger that maps every rejection back to a DOT, FMCSA, or EPA rule. The VRP Route Optimization Algorithms that consume the output never see a value a vehicle could not physically have reported.

Prerequisites

This page targets Python 3.10+ and treats every inbound record as untrusted until a Pydantic v2 model proves it. Pin the toolchain before wiring anything into a live pipeline:

python -m pip install \
  pydantic==2.7.1 \
  orjson==3.10.6 \
  shapely==2.0.4

orjson gives deterministic, fast serialization for the audit hash; shapely handles polygon containment for the geospatial gate. The pipeline reasons about one canonical record shape — vendor wire formats are decoded into it at the edge (the low-level frame decoding for fill sensors is covered in Parsing IoT Fill-Level Sensor Payloads), and every stage downstream reads only the canonical model. The contract carries exactly the fields the routing and compliance sides both depend on:

Two rules deserve emphasis before any code. First, unknown keys are rejected, not ignored — a vendor firmware update that adds a field must be reviewed, never silently absorbed. Second, invalid values are quarantined, never clamped: a latitude of 91.0 is a bug to be diagnosed, and clamping it to 90.0 would fabricate a location and corrupt the audit trail.

Core Implementation

The pipeline is assembled in four steps: a structured log formatter, the canonical contract with field and model validators, the geospatial gate, and the batch runner that deduplicates, validates, and routes failures to a dead-letter queue.

Step 1 — Structured audit logging

Every validation decision must be machine-readable and replayable, so logs are emitted as JSON through a custom formatter rather than free text. This is the same JSONFormatter pattern used across the ingestion framework, so audit records from validation, sync, and polling all parse with one reader.

import logging
import orjson
from datetime import datetime, timezone


class JSONFormatter(logging.Formatter):
    """Emit each log record as a single-line JSON object for audit ingestion."""

    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "event": record.getMessage(),
            "detail": getattr(record, "detail", None),
        }
        return orjson.dumps(entry).decode()


audit_logger = logging.getLogger("waste_ops.validation")
if not audit_logger.handlers:
    _handler = logging.StreamHandler()
    _handler.setFormatter(JSONFormatter())
    audit_logger.addHandler(_handler)
audit_logger.setLevel(logging.INFO)

Step 2 — The canonical contract

The TelemetryRecord model does the type coercion and the invariant checks the solver cannot do for itself. A field_validator enforces timezone-awareness and rejects future-dated events; a model_validator binds the cross-field compliance rules — the axle-weight requirement for roll-offs and the overfill alert threshold — into the type itself, so no caller can forget them.

from typing import Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator


class TelemetryRecord(BaseModel):
    # forbid unknown keys so vendor firmware changes surface as errors, not drift
    model_config = ConfigDict(extra="forbid")

    vehicle_id: str = Field(..., pattern=r"^VH-\d{4,8}$")
    asset_type: str = Field(..., pattern=r"^(curbside|roll_off|compactor)$")
    latitude: float = Field(..., ge=-90.0, le=90.0)
    longitude: float = Field(..., ge=-180.0, le=180.0)
    fill_level_pct: Optional[float] = Field(None, ge=0.0, le=100.0)
    weight_kg: Optional[float] = Field(None, ge=0.0, le=15000.0)
    event_timestamp: datetime
    service_status: str = Field(..., pattern=r"^(en_route|servicing|completed|idle)$")

    @field_validator("event_timestamp")
    @classmethod
    def enforce_utc_and_not_future(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("event_timestamp must be timezone-aware (UTC)")
        if v > datetime.now(timezone.utc):
            raise ValueError("event_timestamp cannot be in the future")
        return v.astimezone(timezone.utc)

    @model_validator(mode="after")
    def enforce_cross_field_rules(self) -> "TelemetryRecord":
        # roll-off containers ride on axles that DOT weight rules govern
        if self.asset_type == "roll_off" and self.weight_kg is None:
            raise ValueError("roll_off requires weight_kg for axle-weight reporting")
        if self.fill_level_pct is not None and self.fill_level_pct > 95.0:
            audit_logger.warning(
                "overfill_threshold",
                extra={"detail": {"vehicle_id": self.vehicle_id,
                                  "fill_level_pct": self.fill_level_pct}},
            )
        return self

Per-vehicle timestamp monotonicity is stateful, so it belongs in the batch runner rather than the model — a single record cannot know what preceded it. The runner keeps the last accepted timestamp per vehicle and rejects any event that regresses.

Step 3 — The geospatial gate

Coordinate range checks (ge=-90, le=90) only prove a point is on Earth, not that it is inside a serviceable municipal zone. The geospatial gate cross-references each point against a pre-compiled GeoJSON boundary polygon and quarantines points that fall outside it — a dead zone, a private easement, or a GPS multipath artifact. The heavy kinematic filtering of coordinate jumps is delegated to the companion reference on validating GPS coordinates in Python; here the gate is the final containment check before a location is trusted.

from shapely.geometry import Point, shape


class ServiceBoundary:
    """Polygon containment gate compiled once from a municipal GeoJSON feature."""

    def __init__(self, geojson_geometry: dict):
        self._polygon = shape(geojson_geometry)

    def contains(self, latitude: float, longitude: float) -> bool:
        # GeoJSON is (lon, lat) order
        return self._polygon.covers(Point(longitude, latitude))

For distance-based dead-zone exclusion the pipeline uses the great-circle (Haversine) distance between a point and a known depot or restricted centroid:

$$d = 2r\,\arcsin\!\sqrt{\sin^2\!\left(\tfrac{\varphi_2-\varphi_1}{2}\right) + \cos\varphi_1\cos\varphi_2\,\sin^2\!\left(\tfrac{\lambda_2-\lambda_1}{2}\right)}$$

where $\varphi$ is latitude, $\lambda$ is longitude in radians, and $r = 6371\text{ km}$ is the mean Earth radius.

Step 4 — Dedup, validate, and route failures

The runner ties the stages together. High-frequency broadcasts and replayed packets are filtered first by a SHA-256 of the record’s identity fields, so duplicates never reach the validator or the solver. Surviving records are validated; failures are converted into a compliance-coded QuarantineRecord and pushed to a dead-letter queue instead of being dropped.

import hashlib
from pydantic import ValidationError


class QuarantineRecord(BaseModel):
    payload_hash: str
    error_code: str
    field_path: str
    constraint_violated: str
    compliance_ref: str
    quarantined_at: datetime


def _compliance_ref(loc: str) -> str:
    if "timestamp" in loc:
        return "EPA_MANIFEST_TS_MISMATCH"   # 40 CFR 262 e-Manifest ordering
    if "weight" in loc:
        return "DOT_FMCSA_AXLE_DISCREPANCY"  # 49 CFR 658 axle weight
    if "lat" in loc or "lon" in loc:
        return "GEO_BOUNDARY_VIOLATION"      # municipal service ordinance
    return "SOLVER_CONSTRAINT_BREACH"


class ValidationPipeline:
    def __init__(self, boundary: ServiceBoundary):
        self.boundary = boundary
        self.dlq: list[QuarantineRecord] = []
        self._seen: set[str] = set()
        self._last_ts: dict[str, datetime] = {}

    @staticmethod
    def _hash(raw: dict) -> str:
        key = {k: raw.get(k) for k in ("vehicle_id", "event_timestamp", "service_status")}
        return hashlib.sha256(orjson.dumps(key, option=orjson.OPT_SORT_KEYS)).hexdigest()

    def _quarantine(self, raw: dict, loc: str, msg: str, code: str) -> None:
        rec = QuarantineRecord(
            payload_hash=self._hash(raw),
            error_code=code,
            field_path=loc,
            constraint_violated=msg,
            compliance_ref=_compliance_ref(loc),
            quarantined_at=datetime.now(timezone.utc),
        )
        self.dlq.append(rec)
        audit_logger.error("payload_quarantined", extra={"detail": rec.model_dump(mode="json")})

    def validate_batch(self, payloads: list[dict]) -> list[dict]:
        accepted: list[dict] = []
        for raw in payloads:
            h = self._hash(raw)
            if h in self._seen:
                continue  # deterministic dedup at the edge
            self._seen.add(h)
            try:
                rec = TelemetryRecord(**raw)
            except ValidationError as e:
                err = e.errors()[0]
                self._quarantine(raw, str(err["loc"]), str(err["msg"]), "SCHEMA_VALIDATION_FAILURE")
                continue
            if not self.boundary.contains(rec.latitude, rec.longitude):
                self._quarantine(raw, "('latitude',)", "outside service boundary", "GEO_BOUNDARY_VIOLATION")
                continue
            last = self._last_ts.get(rec.vehicle_id)
            if last is not None and rec.event_timestamp < last:
                self._quarantine(raw, "('event_timestamp',)", "non-monotonic per vehicle", "TEMPORAL_REGRESSION")
                continue
            self._last_ts[rec.vehicle_id] = rec.event_timestamp
            accepted.append(rec.model_dump(mode="json"))
        return accepted

The accepted list is the only thing the routing engine ever reads. Bursty reconnects and overlapping broadcasts are absorbed by the async batch processing layer, which calls validate_batch on bounded chunks so a reconnect flood cannot exhaust memory. Polling cadence is tuned upstream via GPS Polling Strategies so the validator sees a manageable telemetry density.

Regulatory Mapping

Every rejection carries a compliance reference so an auditor can trace a quarantined payload straight to the rule it broke. Validation is not a convenience layer — it is where the data that feeds regulatory reports is proven defensible.

The DOT and FMCSA mappings above are developed in depth in the DOT/FMCSA rule mapping reference, and the schema shape that carries these fields end-to-end is specified in route schema design. Emitting DOT_FMCSA_AXLE_DISCREPANCY here is only the trigger; the downstream report generation reads the code and populates the corresponding regulatory field.

Validation & Verification

Confirming the pipeline works means proving each gate fires on a bad record and that a clean record passes untouched. The tests below use pytest and exercise the three gates that catch the most production incidents: type coercion, the cross-field roll-off rule, and per-vehicle monotonicity.

import pytest
from datetime import timedelta


NOW = datetime.now(timezone.utc)
BOUNDARY = ServiceBoundary({
    "type": "Polygon",
    "coordinates": [[[-74.05, 40.68], [-73.90, 40.68],
                     [-73.90, 40.80], [-74.05, 40.80], [-74.05, 40.68]]],
})


def _record(**overrides) -> dict:
    base = {
        "vehicle_id": "VH-1042", "asset_type": "curbside",
        "latitude": 40.7128, "longitude": -74.0060,
        "fill_level_pct": 60.0, "event_timestamp": NOW.isoformat(),
        "service_status": "servicing",
    }
    base.update(overrides)
    return base


def test_clean_record_is_accepted():
    p = ValidationPipeline(BOUNDARY)
    assert len(p.validate_batch([_record()])) == 1
    assert p.dlq == []


def test_roll_off_without_weight_is_quarantined():
    p = ValidationPipeline(BOUNDARY)
    accepted = p.validate_batch([_record(asset_type="roll_off", weight_kg=None)])
    assert accepted == []
    assert p.dlq[0].compliance_ref == "DOT_FMCSA_AXLE_DISCREPANCY"


def test_future_timestamp_is_quarantined():
    p = ValidationPipeline(BOUNDARY)
    future = (NOW + timedelta(hours=1)).isoformat()
    p.validate_batch([_record(event_timestamp=future)])
    assert p.dlq[0].compliance_ref == "EPA_MANIFEST_TS_MISMATCH"


def test_out_of_order_event_is_quarantined():
    p = ValidationPipeline(BOUNDARY)
    earlier = (NOW - timedelta(minutes=5)).isoformat()
    p.validate_batch([_record(), _record(event_timestamp=earlier)])
    assert any(q.error_code == "TEMPORAL_REGRESSION" for q in p.dlq)

In a running system, verification is continuous, not one-off. Watch three structured log fields on the audit stream: the count of payload_quarantined events per compliance code (a spike in GEO_BOUNDARY_VIOLATION usually means a stale boundary polygon), the overfill_threshold warning rate (which should track collection-day demand), and the ratio of deduplicated to accepted records (a rising ratio signals an upstream polling misconfiguration).

Failure Modes & Edge Cases

The pipeline’s job is to fail loudly and recoverably, never silently. The cases below are the ones that actually reach production.

A whole batch is malformed. When an entire chunk fails — a vendor ships a breaking schema change — validate_batch returns an empty accepted list and a full DLQ. The routing side must not interpret “zero valid records” as “no work”; it should hold the last-known-good route and alert. That degradation is owned by the fallback routing logic, which the pipeline signals by emitting an empty batch with a non-empty DLQ.

The dead-letter queue needs retry, not just storage. Transient failures (a clock that was briefly skewed, a boundary file mid-reload) deserve a bounded retry with exponential backoff before a payload is declared permanently dead:

import time


def drain_dlq(pipeline: ValidationPipeline, reprocess, max_retries: int = 3) -> None:
    survivors: list[QuarantineRecord] = []
    for rec in pipeline.dlq:
        for attempt in range(max_retries):
            if reprocess(rec):          # returns True if the record now validates
                break
            time.sleep(2 ** attempt)    # 1s, 2s, 4s backoff
        else:
            survivors.append(rec)       # exhausted retries → permanent DLQ
            audit_logger.error("dlq_exhausted", extra={"detail": rec.model_dump(mode="json")})
    pipeline.dlq = survivors

Unbounded dedup memory. The _seen set and _last_ts map grow without limit in a long-lived worker. In production, back them with a TTL store (Redis, or a unique constraint on payload_hash) so a horizontally scaled second worker shares the dedup guarantee and a restart does not re-admit an already-processed burst.

Null vs. missing vs. zero. A weight_kg of 0.0 is a valid empty container; None is “not reported”; a missing key is a contract violation. The model distinguishes all three — extra="forbid" rejects unexpected keys, Optional[...] = Field(None) allows explicit nulls, and the cross-field rule decides when a null is fatal. Collapsing these three states is the most common silent-corruption bug in ad-hoc validators.

Integration Checklist

• Pin pydantic==2.7.1, orjson, and shapely; no unpinned validation dependencies in production.
• Compile the service-boundary GeoJSON once at startup and reload it on a versioned signal, never per-record.
• Set extra="forbid" on the canonical model so vendor field additions surface as errors.
• Back _seen and _last_ts with a durable TTL store before scaling past one worker.
• Route the DLQ to append-only, hash-stamped storage with a retry drain and an exhaustion alert.
• Wire each compliance code to its downstream regulatory report field (HOS, axle weight, e-Manifest).
• Alarm on payload_quarantined rate per code and on the dedup-to-accepted ratio.
• Confirm an empty accepted batch with a non-empty DLQ triggers fallback routing, not a no-op.

FAQ

Why quarantine invalid coordinates instead of clamping them to the boundary?

Clamping fabricates data. A latitude of 91.0 or a point 200 meters outside the service polygon is evidence of a receiver fault, a multipath jump, or a mislabeled asset — clamping it to the nearest legal value hides the fault and writes a location the vehicle was never at into the compliance record. Quarantine preserves the original payload for diagnosis and keeps the audit trail truthful, which is the entire point of a defensible pipeline.

Should timestamp monotonicity live in the Pydantic model?

No — a single record cannot know what preceded it, so monotonicity is stateful and belongs in the batch runner, which holds the last accepted timestamp per vehicle. The model owns everything a record can prove about itself (timezone-awareness, not-in-the-future, range bounds); the runner owns the cross-record ordering. Mixing the two into the model forces global state into what should be a pure, testable type.

Why hash only identity fields for deduplication?

Hashing vehicle_id, event_timestamp, and service_status keeps the dedup key stable against metadata that jitters between retransmissions — signal strength, battery voltage, transport headers. If the whole payload were hashed, a replayed broadcast with a different RSSI would look novel and slip past the filter, re-triggering the exact duplicate route recalculation the pipeline exists to prevent.

What is the difference between a null, a missing, and a zero value?

They are three distinct states and the pipeline treats them as such: 0.0 is a genuine measured zero (an empty container), None is an explicitly unreported optional value, and a missing key is a contract breach that extra="forbid" rejects. Whether a None is fatal is decided by the cross-field rule — a roll-off with weight_kg=None is quarantined, a curbside bin with it is fine. Collapsing these three is the classic silent-corruption bug.

How does this pipeline relate to the GPS coordinate validator?

They compose. The GPS coordinate validator is the kinematic pre-filter — it rejects impossible velocity jumps and low-accuracy fixes before a point is even considered. This pipeline is the contract and containment layer that runs after, proving the surviving point is well-typed, inside a serviceable polygon, and temporally ordered. One filters physics; the other enforces schema and jurisdiction.

Up: Telematics & Sensor Data Ingestion

Related

• Validating GPS Coordinates in Python — the kinematic pre-filter that runs before this contract layer.
• Bin Sensor API Sync — idempotent multi-vendor ingestion that feeds the canonical record this pipeline validates.
• GPS Polling Strategies — tuning telemetry density so the validator sees a manageable load.
• Async Batch Processing — bounded queues that call validate_batch on chunks and absorb reconnect bursts.
• DOT/FMCSA Rule Mapping — the regulatory codes each quarantine reference resolves to.