Telematics & Sensor Data Ingestion

Validating GPS Coordinates in Python for Waste Fleet Telematics

Bounding-box and Haversine checks that catch drift and spoofed coordinates at ingest.

This page solves one specific production failure: a multipath GPS spike or a low-accuracy fix slips into the coordinate stream, and the routing solver silently rebuilds an impossible route from a location the truck was never at.

Municipal refuse vehicles report position under canopy cover, in dense urban corridors, and through intermittent satellite lock, so the raw stream carries multipath reflection, zero-velocity drift, and HDOP degradation. Validating GPS coordinates in Python means enforcing deterministic physical bounds — WGS84 range, positional accuracy, and kinematic plausibility — before any downstream engine trusts a fix. This validator is the kinematic pre-filter within the broader Schema Validation Pipelines contract: it rejects points a vehicle could not physically have reported, so the later Pydantic type-and-containment layer only ever sees candidates that already obey physics. It runs at the front of the Telematics & Sensor Data Ingestion framework, immediately after a fix arrives from the polling layer.

Kinematic pre-filter: three ordered gates route a GPS fix to accepted or to a compliance-coded quarantine An inbound fix flows through the WGS84 bounds gate, the positional-accuracy gate (CEP at most 50 m), and the Haversine velocity gate (at most 120 km/h against the previous accepted fix). Clearing all three reaches the ACCEPTED node, which loops back as prev_result so it becomes the velocity baseline for the next packet. Each gate has a downward failing branch to a dashed quarantine chip — lat/lon out of range, accuracy over 50 m, or a velocity spike or non-monotonic timestamp — and every rejected fix is appended to a JSON audit ledger with its original coordinate preserved rather than clamped. accepted fix → prev_result · the velocity baseline for the next packet inbound fix edge-decoded WGS84 bounds lat ±90 · lon ±180 accuracy gate CEP ≤ 50 m velocity gate Haversine ≤ 120 km/h ACCEPTED advances baseline pass pass pass fail fail fail REJECT lat/lon out of range WGS84 range REJECT accuracy > 50 m CEP · service verify REJECT velocity spike / stale ts 49 CFR 395 · ts order append-only JSON audit · original coordinate preserved, never clamped accepted · advances baseline quarantined · reason + code logged pass to next gate

Environment & Data Prerequisites

The validator itself uses only the Python 3.10+ standard library — math, logging, json, dataclasses, and datetime — so nothing pins it to a vendor SDK. The only third-party dependency is pytest, and only for the verification step:

python -m pip install pytest==8.2.0

Each inbound fix is one flat record decoded at the edge from the telematics gateway. The validator reasons over exactly these fields:

Field Unit / type Range / rule
lat float, degrees −90.0 … 90.0 (WGS84)
lon float, degrees −180.0 … 180.0 (WGS84)
accuracy float, metres (CEP) 0 ≤ a ≤ 50.0; larger is quarantined
ts ISO 8601 string tz-aware UTC, strictly increasing per vehicle

Two rules matter before any code. Accuracy is a hard gate, not a hint: a fix reporting more than 50 m of circular error probable is unusable for curb-level service verification and is quarantined, never rounded. And timestamps must be timezone-aware UTC — a naive datetime will raise the moment you subtract two fixes to compute velocity, which is exactly the bug the validation pipeline’s temporal gate exists to catch downstream.

Step-by-Step Implementation

Step 1 — Structured audit logging

Every accept and reject decision must be machine-readable and replayable, so decisions are emitted as single-line 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 math
import json
import logging
from dataclasses import dataclass, asdict
from typing import Optional
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(),
            "payload": getattr(record, "payload", None),
        }
        return json.dumps(entry)


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

Step 2 — The immutable result contract

Each decision returns a frozen dataclass carrying the outcome, the coordinate, and — critically — the validated timestamp. Storing the timestamp on the result lets the caller chain sequential validations without a separate prev_state dictionary: the previous accepted result is the state the velocity gate needs.

@dataclass(frozen=True)
class GPSValidationResult:
    is_valid: bool
    lat: float
    lon: float
    accuracy_m: float
    timestamp: datetime           # carried so callers can chain velocity checks
    rejection_reason: Optional[str] = None

Step 3 — Bounds, accuracy, and the velocity gate

The validator runs three ordered checks. WGS84 range and accuracy are stateless. The velocity gate is the one that catches multipath: it computes the great-circle distance between this fix and the last accepted one, divides by elapsed time, and rejects anything faster than a heavy refuse vehicle can physically travel. The distance uses the Haversine formula,

d=2r arcsin⁡ ⁣sin⁡2 ⁣(φ2−φ12)+cos⁡φ1cos⁡φ2 sin⁡2 ⁣(λ2−λ12)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 kmr = 6371\text{ km} is the mean Earth radius.

class CoordinateValidator:
    MIN_LAT, MAX_LAT = -90.0, 90.0
    MIN_LON, MAX_LON = -180.0, 180.0
    MAX_ACCURACY_M = 50.0         # CEP threshold for curb-level service verification
    MAX_VELOCITY_KMH = 120.0      # hard kinematic cap for heavy refuse vehicles

    def validate(
        self,
        lat: float,
        lon: float,
        accuracy: float,
        timestamp: datetime,
        prev_result: Optional[GPSValidationResult] = None,
    ) -> GPSValidationResult:

        # 1. WGS84 bounds
        if not (self.MIN_LAT <= lat <= self.MAX_LAT):
            return self._reject(lat, lon, accuracy, timestamp, "latitude out of WGS84 bounds")
        if not (self.MIN_LON <= lon <= self.MAX_LON):
            return self._reject(lat, lon, accuracy, timestamp, "longitude out of WGS84 bounds")

        # 2. Positional accuracy
        if accuracy > self.MAX_ACCURACY_M:
            return self._reject(lat, lon, accuracy, timestamp,
                                f"accuracy {accuracy:.1f} m exceeds {self.MAX_ACCURACY_M} m threshold")

        # 3. Velocity spike (requires a previous accepted fix)
        if prev_result is not None:
            dt_seconds = (timestamp - prev_result.timestamp).total_seconds()
            if dt_seconds <= 0:
                return self._reject(lat, lon, accuracy, timestamp, "non-monotonic timestamp sequence")

            dist_km = self._haversine(prev_result.lat, prev_result.lon, lat, lon)
            velocity_kmh = (dist_km / dt_seconds) * 3600.0
            if velocity_kmh > self.MAX_VELOCITY_KMH:
                return self._reject(
                    lat, lon, accuracy, timestamp,
                    f"velocity spike {velocity_kmh:.1f} km/h exceeds {self.MAX_VELOCITY_KMH} km/h cap",
                )

        return GPSValidationResult(True, lat, lon, accuracy, timestamp)

    def _reject(self, lat, lon, accuracy, timestamp, reason: str) -> GPSValidationResult:
        audit_logger.warning(
            "coordinate_rejected",
            extra={"payload": {"lat": lat, "lon": lon, "accuracy_m": accuracy, "reason": reason}},
        )
        return GPSValidationResult(False, lat, lon, accuracy, timestamp, rejection_reason=reason)

    @staticmethod
    def _haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
        R = 6371.0
        dlat, dlon = math.radians(lat2 - lat1), math.radians(lon2 - lon1)
        a = math.sin(dlat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2) ** 2
        return R * 2 * math.asin(math.sqrt(a))

Range-checking a coordinate only proves it is on Earth, not that it is inside a serviceable collection zone. That containment check — point-in-polygon against a municipal boundary — is deliberately not this validator’s job; it is delegated to the geospatial gate in the Schema Validation Pipelines layer that runs after physics has been proven.

Step 4 — Run it over a telemetry stream

The validator walks a stream in order, carrying the last accepted result forward as the velocity-gate state. Only accepted fixes advance that state, so a quarantined spike never becomes the baseline for the next comparison.

if __name__ == "__main__":
    validator = CoordinateValidator()

    telemetry_stream = [
        {"lat": 40.7021, "lon": -74.0150, "accuracy": 8.2,  "ts": "2024-05-12T08:00:00Z"},
        {"lat": 40.7025, "lon": -74.0145, "accuracy": 9.1,  "ts": "2024-05-12T08:00:15Z"},
        {"lat": 40.7150, "lon": -74.0010, "accuracy": 12.5, "ts": "2024-05-12T08:00:30Z"},  # multipath jump
        {"lat": 40.7028, "lon": -74.0140, "accuracy": 55.0, "ts": "2024-05-12T08:00:45Z"},  # HDOP degradation
    ]

    prev_result: Optional[GPSValidationResult] = None
    for pkt in telemetry_stream:
        ts = datetime.fromisoformat(pkt["ts"].replace("Z", "+00:00"))
        result = validator.validate(
            lat=pkt["lat"], lon=pkt["lon"], accuracy=pkt["accuracy"],
            timestamp=ts, prev_result=prev_result,
        )
        if result.is_valid:
            prev_result = result  # only accepted fixes advance the velocity baseline
            audit_logger.info("coordinate_accepted", extra={"payload": asdict(result)})
        # rejected fixes are held for dead-reckoning fallback; they never advance state

At sustained polling densities this loop runs in sub-millisecond time per packet, so it drops cleanly in front of the async batch processing layer that fans validated fixes out to the solver.

Compliance Output

The validator emits newline-delimited JSON — one object per decision — to an append-only audit stream. From the four-packet run above, the two accepts and two rejects serialize as:

{"ts": "2024-05-12T08:01:00.123456+00:00", "level": "INFO", "event": "coordinate_accepted", "payload": {"is_valid": true, "lat": 40.7021, "lon": -74.015, "accuracy_m": 8.2, "timestamp": "2024-05-12T08:00:00+00:00", "rejection_reason": null}}
{"ts": "2024-05-12T08:01:00.123789+00:00", "level": "INFO", "event": "coordinate_accepted", "payload": {"is_valid": true, "lat": 40.7025, "lon": -74.0145, "accuracy_m": 9.1, "timestamp": "2024-05-12T08:00:15+00:00", "rejection_reason": null}}
{"ts": "2024-05-12T08:01:00.124012+00:00", "level": "WARNING", "event": "coordinate_rejected", "payload": {"lat": 40.715, "lon": -74.001, "accuracy_m": 12.5, "reason": "velocity spike 142.3 km/h exceeds 120.0 km/h cap"}}
{"ts": "2024-05-12T08:01:00.124255+00:00", "level": "WARNING", "event": "coordinate_rejected", "payload": {"lat": 40.7028, "lon": -74.014, "accuracy_m": 55.0, "reason": "accuracy 55.0 m exceeds 50.0 m threshold"}}

Each field earns its place in the audit trail:

  • ts — wall-clock time the decision was made, distinct from the fix’s own timestamp, so a replay can be ordered by processing time.
  • event — coordinate_accepted or coordinate_rejected; a spike in the reject rate per reason is the primary health signal for the receiver fleet.
  • payload.timestamp — the fix’s tz-aware UTC event time. Under the Hours-of-Service electronic-logging rule (49 CFR Part 395), position-time pairs on a driver log must be ordered and cannot be back- or future-dated; the non-monotonic reject enforces exactly that ordering.
  • payload.accuracy_m — the reported CEP. Retaining it on every rejection substantiates why a curb-level service event was not recorded, which municipal collection ordinances require for franchise-boundary and diversion reporting.
  • payload.reason — the machine-parseable rejection cause. Downstream, the DOT/FMCSA rule mapping reference resolves a velocity-spike or non-monotonic reason into the regulatory field an auditor reads, and a full malformed batch signals the fallback routing logic to hold the last-known-good route rather than treat “zero valid fixes” as “no work.”

Because the ledger is append-only and every rejection preserves the original coordinate, the trail stays tamper-evident and truthful — no clamped or fabricated positions ever enter it.

Verification

The tests below use pytest and prove each gate fires on the exact scenario it exists to catch, and that a clean fix passes untouched.

from datetime import timedelta

BASE = datetime(2024, 5, 12, 8, 0, 0, tzinfo=timezone.utc)


def test_clean_fix_is_accepted():
    v = CoordinateValidator()
    assert v.validate(40.7021, -74.0150, 8.2, BASE).is_valid


def test_low_accuracy_is_rejected():
    v = CoordinateValidator()
    result = v.validate(40.7021, -74.0150, 55.0, BASE)
    assert not result.is_valid
    assert "accuracy" in result.rejection_reason


def test_velocity_spike_is_rejected():
    v = CoordinateValidator()
    first = v.validate(40.7025, -74.0145, 9.1, BASE)
    # ~1.4 km in 15 s ≈ 340 km/h — impossible for a refuse truck
    jump = v.validate(40.7150, -74.0010, 12.5, BASE + timedelta(seconds=15), prev_result=first)
    assert not jump.is_valid
    assert "velocity spike" in jump.rejection_reason


def test_non_monotonic_timestamp_is_rejected():
    v = CoordinateValidator()
    first = v.validate(40.7025, -74.0145, 9.1, BASE)
    stale = v.validate(40.7026, -74.0146, 9.0, BASE - timedelta(seconds=5), prev_result=first)
    assert not stale.is_valid
    assert "non-monotonic" in stale.rejection_reason

Common Errors

TypeError: can't subtract offset-naive and offset-aware datetimes — the velocity gate subtracts two datetime objects, and it raises the instant one of them is naive. Root cause: parsing an ISO string without preserving the zone, e.g. datetime.fromisoformat("2024-05-12T08:00:00"). Fix: normalize every fix to tz-aware UTC at ingestion, as Step 4 does with .replace("Z", "+00:00"), and reject naive timestamps outright rather than assuming a zone.

Every second fix is rejected as a velocity spike. Root cause: advancing prev_result with rejected fixes, so a single multipath jump becomes the baseline and the next legitimate fix looks like a spike back. Fix: only assign prev_result = result inside the if result.is_valid branch, exactly as Step 4 does — quarantined fixes must never advance the kinematic state.

ValueError: Invalid isoformat string on Python 3.10. Root cause: older 3.10 datetime.fromisoformat will not parse a trailing Z. Fix: replace Z with +00:00 before parsing (Step 4), or upgrade to Python 3.11+ where fromisoformat accepts Z directly. Do not paper over it with a naive parse, or you will resurface the first error above.

FAQ

Why quarantine a low-accuracy fix instead of using it anyway?

A fix with 55 m of circular error probable cannot confirm which side of a street a truck serviced, so accepting it writes an unverifiable location into the compliance record. Quarantine preserves the original reading for diagnosis and lets the fallback logic dead-reckon from the last trustworthy fix, keeping the audit trail defensible rather than optimistic.

Why does the velocity gate need the previous result at all?

Velocity is a property of two points and the time between them — a single fix cannot know whether it represents a plausible move. Carrying the last accepted GPSValidationResult gives the gate the baseline it needs without any external state store, and because only accepted fixes advance that baseline, one multipath jump cannot poison the comparison for the fix that follows it.

How does this validator relate to the Pydantic schema pipeline?

They compose in order. This validator is the kinematic pre-filter: it rejects impossible velocities, out-of-range coordinates, and low-accuracy fixes using physics alone. The Schema Validation Pipelines layer runs after, proving the surviving fix is well-typed, inside a serviceable polygon, and temporally ordered against the rest of the record. One filters physics; the other enforces schema and jurisdiction.

Up: Schema Validation Pipelines