Telematics & Sensor Data Ingestion

Municipal waste operations fail quietly at the ingestion boundary. A dropped GPS burst becomes a missing proof-of-service; a duplicated fill-level packet inflates a route that never needed running; an unordered timestamp corrupts an hours-of-service log that an auditor later rejects. This is the layer that turns raw device chatter into structured, verifiable operational records — and it is written for the people who own that failure: waste ops managers who answer to municipal contracts, logistics engineers who feed a solver, and Python developers who have to make the pipeline idempotent, memory-bounded, and audit-safe.

The architecture on this page prioritizes explicit state management over speculative processing. Every payload receives a UTC-normalized timestamp, monotonic sequence validation, and an immutable ledger reference before any downstream system — routing solver, dispatch dashboard, or compliance report — is allowed to consume it. Sensor gaps are reconciled here, at ingest, so that the VRP Route Optimization Algorithms that build the distance matrix never see a coordinate that a truck could not physically have produced.

Ingestion Constraint Architecture

Ingestion constraints are not stylistic preferences; they are the invariants that keep the compliance trail defensible and the solver feasible. Treat each as a hard boundary that rejects or quarantines a payload rather than a soft heuristic that silently repairs it.

• Timestamp monotonicity and UTC awareness. A collection timeline reconstructed from out-of-order or timezone-naive timestamps is unusable as evidence. Device clocks drift, and cellular reconnects replay buffered packets out of sequence. Every record must carry a UTC-aware, RFC 3339 timestamp, and per-asset sequence numbers must be strictly increasing before a record is committed. A regression is a signal of clock skew or replay, not a value to average away.
• Coordinate plausibility. Latitude and longitude are bounded to [-90, 90] and [-180, 180], but the real constraint is displacement rate: a municipal collection vehicle cannot jump 900 metres between two one-second samples. Urban-canyon multipath and tunnel handoffs produce exactly these phantom jumps, and accepting them poisons the routing matrix. The bounding-and-displacement logic is developed in depth under Validating GPS Coordinates in Python.
• Contract completeness. Field types, payload size, and required identifiers are verified at the edge. A record missing an asset_id or carrying a malformed signature cannot be mapped to a chain of custody, so it is routed to a quarantine ledger — never dropped, never guessed.
• Regulatory retention windows. Telemetry that backs a compliance claim inherits that claim’s retention obligation. Records tied to hours-of-service or e-manifest reporting must be written to append-only storage before acknowledgment, because a payload that was accepted but not durably logged is a compliance gap waiting to surface in an audit.

These are hard constraints because each one, if softened, converts a recoverable ingest error into an irreversible records error. The rest of this page shows how they are enforced in code.

Core Ingestion Pattern

The ingestion core is a bounded, asynchronous worker that pulls validated telemetry off a queue, enforces the sequence and drift invariants, and writes a cryptographically referenced audit record before advancing state. Contract enforcement is delegated to a Pydantic model — the same defensive posture documented across the Schema Validation Pipelines — and structured logging follows the site-wide JSONFormatter pattern, because Python’s standard logging module ships no JSON formatter of its own.

The model uses Pydantic v2’s field_validator (the v1 @validator decorator is deprecated) and clamps every timestamp to UTC at the boundary, so no downstream code has to reason about timezones.

import asyncio
import hashlib
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Optional

from pydantic import BaseModel, Field, field_validator
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type


class JSONFormatter(logging.Formatter):
    """Custom JSON formatter — the standard logging module has no built-in JSONFormatter."""
    def format(self, record: logging.LogRecord) -> str:
        log_obj = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
        }
        log_obj.update({k: v for k, v in record.__dict__.items()
                        if k not in logging.LogRecord.__dict__ and not k.startswith("_")})
        return json.dumps(log_obj)


logger = logging.getLogger("telemetry_ingestion")
logger.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
logger.addHandler(_handler)


class TelemetryPayload(BaseModel):
    asset_id: str = Field(..., min_length=3, max_length=32)
    timestamp: datetime
    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)
    sequence_id: int
    signature: str

    @field_validator("timestamp")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("Timestamp must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)


@dataclass
class IngestionState:
    last_sequence: int = -1
    last_position: Optional[tuple[float, float]] = None
    max_drift_meters: float = 150.0
    processed_count: int = 0


def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Great-circle drift check without an external geospatial dependency."""
    import math
    R = 6_371_000.0
    dlat = math.radians(lat2 - lat1)
    dlon = 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))


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=8),
    retry=retry_if_exception_type((ConnectionError, asyncio.TimeoutError)),
    reraise=True,
)
async def persist_to_ledger(record: dict[str, Any]) -> str:
    """Deterministic append-only ledger write with cryptographic hashing."""
    payload_bytes = json.dumps(record, sort_keys=True).encode("utf-8")
    ledger_hash = hashlib.sha256(payload_bytes).hexdigest()
    await asyncio.sleep(0.01)  # stand-in for durable I/O latency
    return ledger_hash


async def process_batch(batch: list[TelemetryPayload], state: IngestionState) -> list[str]:
    ledger_refs: list[str] = []
    for payload in batch:
        if payload.sequence_id <= state.last_sequence:
            logger.warning("Sequence regression rejected",
                           extra={"asset_id": payload.asset_id, "seq": payload.sequence_id})
            continue

        if state.last_position is not None:
            drift = haversine_distance(state.last_position[0], state.last_position[1],
                                       payload.latitude, payload.longitude)
            if drift > state.max_drift_meters:
                logger.warning("Positional drift exceeds threshold",
                               extra={"drift_m": round(drift, 1), "asset_id": payload.asset_id})

        audit_record = {
            "asset_id": payload.asset_id,
            "ingested_at": datetime.now(timezone.utc).isoformat(),
            "device_ts": payload.timestamp.isoformat(),
            "lat": payload.latitude,
            "lon": payload.longitude,
            "fill_pct": payload.fill_level_pct,
            "seq": payload.sequence_id,
            "sig": payload.signature,
        }

        try:
            ref = await persist_to_ledger(audit_record)
            ledger_refs.append(ref)
            state.last_sequence = payload.sequence_id
            state.last_position = (payload.latitude, payload.longitude)
            state.processed_count += 1
        except Exception as exc:  # ledger write exhausted its retries
            logger.error("Ledger persistence failed",
                         extra={"error": str(exc), "asset_id": payload.asset_id})

    return ledger_refs


async def ingestion_worker(queue: "asyncio.Queue[TelemetryPayload]", batch_size: int = 50) -> None:
    state = IngestionState()
    while True:
        batch: list[TelemetryPayload] = []
        try:
            for _ in range(batch_size):
                batch.append(await asyncio.wait_for(queue.get(), timeout=2.0))
        except asyncio.TimeoutError:
            pass  # flush whatever accumulated before the window closed

        if not batch:
            continue

        logger.info("Processing telemetry batch",
                    extra={"batch_size": len(batch),
                           "asset_count": len({p.asset_id for p in batch})})
        refs = await process_batch(batch, state)
        logger.info("Batch committed",
                    extra={"ledger_refs": len(refs), "total_processed": state.processed_count})

The pattern is deliberately boring: no speculative repair, no in-place mutation of a payload, and a single place — process_batch — where sequence and drift invariants are decided before state advances. That is what makes the ledger reproducible when an auditor replays it months later.

Fleet Device Topology & Gateway Distribution

Ingestion rarely faces one device type. A single collection zone mixes cellular fleet telematics units on the trucks, LoRaWAN fill-level sensors in the bins, and onboard scale controllers on the CAN bus — each with its own firmware cadence, power budget, and clock discipline. Treating them as one homogeneous stream is the fastest way to a corrupted timeline.

Fleet telematics units transmit coordinate, heading, and engine telemetry at variable intervals dictated by OEM firmware and network conditions. Container networks report fill levels, tilt angles, and temperature across fragmented LoRaWAN and cellular gateways, operating under duty-cycle limits that make bursty, delayed delivery the norm rather than the exception. The ingestion layer normalizes these disparate payloads into the single TelemetryPayload contract above, reconciling per-vendor clock skew before any record is keyed to a canonical asset identifier.

Gateway distribution mirrors depot distribution on the routing side. Each yard fronts a set of gateways, and a device binds to a primary gateway with explicit fallbacks so a single congested LoRaWAN concentrator does not strand an entire route’s telemetry. When a bin controller reconnects after a multi-hour outage it typically bursts its queued payloads all at once; the ingestion layer must absorb that burst under bounded memory rather than letting it saturate the queue. The multi-vendor normalization and burst-reconciliation routines live under Bin Sensor API Sync, with the low-level decoding covered in Parsing IoT Fill-Level Sensor Payloads.

Dynamic Calibration & Telemetry Integration

Live telemetry does more than record what happened — it recalibrates the pipeline that decides what happens next. Fill-level readings raise or lower a bin’s service priority; GPS velocity tightens or loosens polling cadence; scale readings validate that a truck actually collected what its route predicted. The ingestion layer is where those feedback signals are cleaned before they reach a solver parameter.

Adaptive cadence is the first control. When a vehicle holds a steady corridor velocity, polling intervals expand to conserve bandwidth and vendor quota; during service stops, sharp turns, or geofence breaches, the layer requests higher-frequency updates. The full backoff and event-driven envelope is documented under GPS Polling Strategies, including how to sustain sub-minute resolution during missed-stop reconciliation in Polling GPS Telematics Without Rate Limiting.

Drift smoothing is the second. Rather than discard a suspect coordinate outright — which would tear a hole in the service timeline — the layer flags displacement that exceeds a plausible rate and applies a lightweight exponential moving average, preserving continuity while suppressing phantom detours. The great-circle distance behind the max_drift_meters check is the Haversine formula:

$$d = 2R \arcsin\!\sqrt{\sin^{2}\!\frac{\varphi_2 - \varphi_1}{2} + \cos\varphi_1 \cos\varphi_2 \sin^{2}\!\frac{\lambda_2 - \lambda_1}{2}}$$

where $\varphi$ is latitude, $\lambda$ is longitude, and $R \approx 6{,}371{,}000$ m is the Earth’s mean radius.

Fallback logic is the third and most important control, because telemetry gaps are guaranteed. When a sensor goes dark — dead battery, cellular dead zone, gateway contention — the pipeline must not feed the solver a stale or null reading. It substitutes a historical fill-rate baseline for the affected bin and marks the record as estimated, so the Fallback Routing Logic downstream can degrade deterministically instead of diverging. This is the same contract the routing layer expects when it reconciles sensor gaps before matrix construction.

Immutable Compliance Logging

Regulatory frameworks require verifiable proof of service and equipment utilization, and volatile caches cannot supply it. Every ingested record receives a sequential ledger identifier and a SHA-256 hash of its canonical byte representation; state transitions generate append-only entries that survive restarts, network partitions, and deployment rollbacks. A compliance officer queries the deterministic trail directly, without reconstructing historical state from memory.

The audit record carries the fields that specific mandates consume. Driver-time and positional evidence maps to the FMCSA Electronic Logging Device rule under 49 CFR Part 395 (with the ELD data-element requirements in 49 CFR 395.8), which the routing side turns into feasibility constraints via the DOT/FMCSA Rule Mapping reference. Hazardous-waste chain-of-custody maps to the EPA e-Manifest program under 40 CFR Part 262, whose Uniform Hazardous Waste Manifest is EPA Form 8700-22 — the manifest tracking number and generator ID become required fields on any record that touches a hazmat pickup. Retention and record-integrity expectations align with the RCRA recordkeeping provisions those parts reference.

Because each entry hashes its own payload and the writes are append-only, retroactive modification is detectable: a changed byte produces a different hash and breaks the chain. That property is what lets a municipality answer a public-records request or an environmental audit with a trail an auditor can independently verify. The canonical shape of these records is specified in Route Schema Design, and the broader immutability model lives in the Core Architecture & Compliance Mapping section.

Failure Modes & Production Gotchas

Every constraint above corresponds to a production failure that will occur. Each needs an explicit mitigation, not a hope.

Burst reconnect saturation. A bin that reconnects after hours offline replays its entire buffer. Unbounded, that burst exhausts the queue and stalls live traffic. Bound the intake and shed with backpressure rather than crashing:

async def enqueue(queue: "asyncio.Queue[TelemetryPayload]", payload: TelemetryPayload) -> bool:
    try:
        queue.put_nowait(payload)   # bounded queue: asyncio.Queue(maxsize=10_000)
        return True
    except asyncio.QueueFull:
        logger.warning("Backpressure: queue full, spilling to disk buffer",
                       extra={"asset_id": payload.asset_id})
        return False  # caller writes to local disk buffer for later replay

GPS drift and sequence replay. The haversine_distance gate flags implausible jumps and the sequence_id <= last_sequence check rejects replays, but a device that resets its sequence counter after a firmware update looks like a permanent regression. Detect a monotonic reset explicitly (a sequence that drops to near zero and then climbs) and rebind the asset’s baseline rather than silently dropping every new packet.

Clock skew across vendors. Two devices agree on wall-clock but disagree by seconds. Enforcing UTC awareness in the model catches naive timestamps; reconciling skew requires stamping each record with both device_ts and server ingested_at, so a later audit can prove ordering even when a device clock lied.

Memory bottlenecks under peak load. Morning dispatch and end-of-day depot returns generate synchronized bursts. Cap batch size, use bounded queues, and write in chunks so a broker slowdown never balloons the heap on a constrained edge gateway. When broker latency exceeds a threshold, shift to local disk buffering with an explicit size limit instead of accumulating in RAM.

Silent quarantine growth. Malformed payloads routed to the quarantine ledger are invisible until someone counts them. Alert on quarantine rate, categorize failures deterministically (structural parse error, semantic constraint violation, signature mismatch), and retain the original bytes so a firmware misconfiguration can be diagnosed and replayed without halting live ingestion.

Performance & Scalability

Throughput targets follow fleet size, and the numbers that matter are the ones measured at the ingest boundary, not the broker. A mid-size municipal fleet of a few hundred vehicles at sub-minute GPS resolution plus a few thousand bins reporting hourly produces a steady stream that a single bounded worker handles comfortably; the scaling pressure is bursty reconnect, not average rate.

Three levers keep it linear. First, decouple ingestion from optimization with an event-driven broker so a solver cycle never blocks a payload write — Async Batch Processing aggregates discrete events into route-ready snapshots using bounded queues and chunked database writes. Second, size batches and queue depth explicitly: batch_size bounds per-cycle memory, and asyncio.Queue(maxsize=...) bounds the intake so peak-hour bursts spill to disk instead of the heap. Third, monitor queue depth and quarantine rate as first-class metrics and let queue depth drive horizontal scaling in a containerized deployment, so the pipeline adds workers before it drops packets.

Circuit breakers close the loop: after repeated ledger or broker timeouts, isolate the failing dependency, switch to disk buffering, and replay on recovery. By enforcing strict contract validation, isolating ingestion from routing computation, and maintaining cryptographic audit trails, municipal operators get deterministic telemetry processing that scales with the fleet without ever compromising the compliance record.

Related

• GPS Polling Strategies — adaptive cadence that balances bandwidth, accuracy, and urban-canyon drift.
• Bin Sensor API Sync — normalize fragmented LoRaWAN/cellular payloads under clock skew and bursty reconnects.
• Schema Validation Pipelines — edge-side contract enforcement with deterministic quarantine ledgers.
• Async Batch Processing — bounded queues and chunked writes that decouple ingestion latency from solver cycles.
• Validating GPS Coordinates in Python — bounding-box and Haversine checks that catch drift and spoofed coordinates at ingest.

Up: Route Optimization Home