Telematics & Sensor Data Ingestion

Quarantining Malformed Telemetry Batches in Python

Split bad rows into an append-only dead-letter ledger so one corrupt payload never blocks a solve.

This page solves one specific production failure: a single corrupt record inside a telemetry batch raises on validation, the whole batch is rejected, and a route solve that needed ninety-nine good readings is blocked by one bad one.

The all-or-nothing reflex — validate the batch, and if anything fails throw the batch away — is the wrong default for waste-fleet ingestion, where a truck’s shift produces hundreds of readings and losing all of them because one frame was truncated means losing the shift’s service record. The correct pattern is per-record triage: validate each row independently, route the good rows downstream, and divert only the bad rows into a durable, append-only dead-letter ledger with a machine-readable reason code, so nothing is lost and nothing is mutated. This quarantine stage sits inside the Schema Validation Pipelines layer, downstream of the kinematic checks in Validating GPS Coordinates in Python — physics-plausible fixes still have to survive schema triage before a solve trusts them.

Per-record triage routes valid rows downstream and diverts malformed rows to an append-only quarantine ledger A mixed batch is validated one record at a time; valid records flow downstream to the solver, invalid records are appended to a dead-letter ledger with a reason code and the original preserved unmutated, and a batch summary reports total, valid, quarantined, and the reason-code breakdown. mixed batch ok fill_pct 140 ok missing ts per-record validate Pydantic, one row at a time valid downstream → solver original payload preserved invalid quarantine ledger append-only · reason code original, never mutated reason code range / missing / type field + message batch summary total · valid · quarantined reason_codes

Environment & Data Prerequisites

The quarantine pipeline uses Pydantic for per-record validation and the Python 3.10+ standard library — json, logging, datetime, dataclasses, and enum — for the ledger and summary:

python -m pip install pydantic==2.9.2 pytest==8.2.0

Each batch is a list of raw dictionaries decoded upstream, each intended to be one telemetry record. The pipeline reasons over exactly these fields:

Field Unit / type Validity rule
device_id str required, non-empty
ts ISO 8601 string required, parseable, tz-aware UTC
lat float, degrees −90.0 … 90.0
lon float, degrees −180.0 … 180.0
fill_pct float, percent 0.0 … 100.0

Two rules matter before any code. First, validation is per-record, never per-batch: a batch of one hundred with three bad rows must yield ninety-seven downstream and three quarantined, not zero. Second, the original raw record is preserved verbatim in the ledger — the pipeline never coerces, clamps, or repairs a bad row, because a mutated record destroys the evidence an auditor needs to see what actually arrived.

Step-by-Step Implementation

Step 1 — The record schema and reason codes

The valid-record shape is a Pydantic model, and every way a record can fail maps to a small, closed set of machine-readable reason codes. A closed enum keeps the ledger queryable — an operator can count OUT_OF_RANGE versus MISSING_FIELD without parsing free text.

import json
import logging
from enum import Enum
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Any
from pydantic import BaseModel, Field, ValidationError


class TelemetryRecord(BaseModel):
    device_id: str = Field(min_length=1)
    ts: datetime
    lat: float = Field(ge=-90.0, le=90.0)
    lon: float = Field(ge=-180.0, le=180.0)
    fill_pct: float = Field(ge=0.0, le=100.0)


class ReasonCode(str, Enum):
    MISSING_FIELD = "missing_field"
    TYPE_ERROR = "type_error"
    OUT_OF_RANGE = "out_of_range"
    BAD_TIMESTAMP = "bad_timestamp"
    UNKNOWN = "unknown"


def classify(error_type: str, loc: tuple[Any, ...]) -> ReasonCode:
    """Map a Pydantic error type + location to a closed reason code."""
    if error_type in {"missing"}:
        return ReasonCode.MISSING_FIELD
    if error_type.startswith(("greater_than", "less_than")):
        return ReasonCode.OUT_OF_RANGE
    if "datetime" in error_type or loc[-1:] == ("ts",):
        return ReasonCode.BAD_TIMESTAMP
    if error_type.endswith(("_type", "_parsing")):
        return ReasonCode.TYPE_ERROR
    return ReasonCode.UNKNOWN

Step 2 — An append-only quarantine ledger

The ledger is append-only by construction: it exposes append and read access, but no update or delete. Each entry stores the original raw record verbatim, the reason code, the offending field, and the human-readable validator message. Because the original is stored as-is, the entry is both evidence and a replay candidate once the upstream bug is fixed.

@dataclass(frozen=True)
class QuarantineEntry:
    original: dict[str, Any]        # stored verbatim — never coerced or clamped
    reason_code: ReasonCode
    field_path: str
    message: str
    quarantined_at: str


class QuarantineLedger:
    def __init__(self) -> None:
        self._entries: list[QuarantineEntry] = []

    def append(self, entry: QuarantineEntry) -> None:
        self._entries.append(entry)     # only ever grows; no update, no delete

    def __len__(self) -> int:
        return len(self._entries)

    def entries(self) -> tuple[QuarantineEntry, ...]:
        return tuple(self._entries)     # read-only view

Step 3 — Per-record triage into valid vs. quarantined

The triage function validates each raw record independently. A record that validates is added to the downstream list; one that raises ValidationError is appended to the ledger with a reason code derived from the first error, and the loop moves on — one bad row never aborts the pass.

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

    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "logged_at": 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.quarantine")
if not audit_logger.handlers:
    _h = logging.StreamHandler()
    _h.setFormatter(JSONFormatter())
    audit_logger.addHandler(_h)
audit_logger.setLevel(logging.INFO)


def triage_batch(
    raw_batch: list[dict[str, Any]],
    ledger: QuarantineLedger,
) -> tuple[list[TelemetryRecord], dict[str, int]]:
    valid: list[TelemetryRecord] = []
    reason_counts: dict[str, int] = {}
    now = datetime.now(timezone.utc).isoformat()

    for raw in raw_batch:
        try:
            valid.append(TelemetryRecord.model_validate(raw))
        except ValidationError as exc:
            first = exc.errors()[0]                      # classify on the first failure
            code = classify(first["type"], first["loc"])
            ledger.append(QuarantineEntry(
                original=dict(raw),                       # copy so the source dict is untouched
                reason_code=code,
                field_path=".".join(str(p) for p in first["loc"]),
                message=first["msg"],
                quarantined_at=now,
            ))
            reason_counts[code.value] = reason_counts.get(code.value, 0) + 1

    audit_logger.info(
        "batch_triaged",
        extra={"payload": {
            "total": len(raw_batch),
            "valid": len(valid),
            "quarantined": len(ledger),
            "reason_codes": reason_counts,
        }},
    )
    return valid, reason_counts

Step 4 — Run a mixed batch

The example batch mixes two good records with three malformed ones: a fill level above 100, a missing timestamp, and a non-numeric latitude. The pipeline routes the two good rows downstream and quarantines the three bad ones without ever failing the batch.

if __name__ == "__main__":
    ledger = QuarantineLedger()
    raw_batch = [
        {"device_id": "bin-A17", "ts": "2026-07-16T09:00:00+00:00", "lat": 40.70, "lon": -74.01, "fill_pct": 62.5},
        {"device_id": "bin-A17", "ts": "2026-07-16T09:00:15+00:00", "lat": 40.70, "lon": -74.01, "fill_pct": 140.0},   # out of range
        {"device_id": "bin-B04", "lat": 40.71, "lon": -74.00, "fill_pct": 15.0},                                        # missing ts
        {"device_id": "bin-B04", "ts": "2026-07-16T09:00:45+00:00", "lat": "north", "lon": -74.00, "fill_pct": 20.0},   # type error
        {"device_id": "bin-C22", "ts": "2026-07-16T09:01:00+00:00", "lat": 40.69, "lon": -74.02, "fill_pct": 48.0},
    ]
    valid, reasons = triage_batch(raw_batch, ledger)
    print(f"total {len(raw_batch)}, valid {len(valid)}, quarantined {len(ledger)}")  # total 5, valid 2, quarantined 3

Compliance Output

Each batch emits one JSON summary line, and each quarantined row becomes one durable ledger entry. The run above produces the summary:

{"logged_at": "2026-07-16T09:01:11.882043+00:00", "level": "INFO", "event": "batch_triaged", "payload": {"total": 5, "valid": 2, "quarantined": 3, "reason_codes": {"out_of_range": 1, "missing_field": 1, "type_error": 1}}}

and three ledger entries, the first of which serializes as:

{"original": {"device_id": "bin-A17", "ts": "2026-07-16T09:00:15+00:00", "lat": 40.7, "lon": -74.01, "fill_pct": 140.0}, "reason_code": "out_of_range", "field_path": "fill_pct", "message": "Input should be less than or equal to 100", "quarantined_at": "2026-07-16T09:01:11.882043+00:00"}

Each field earns its place in the audit trail:

  • total, valid, quarantined — the reconciliation triple for the batch. valid + quarantined must equal total; a batch where it does not is proof a record vanished, which municipal collection reporting cannot tolerate.
  • reason_codes — the closed-enum breakdown of why rows were rejected. A spike in one code is a targeted signal about an upstream decoder bug, far more actionable than a raw reject count.
  • original — the raw record stored verbatim in the ledger, never coerced or clamped. Preserving the exact bytes that arrived is what makes the ledger admissible evidence and lets the row be replayed once the upstream defect is fixed.
  • field_path and message — the offending field and the validator’s own message. Together they let an operator resolve the failure without re-running the batch, and the DOT/FMCSA rule mapping reference resolves a persistent reason code into the completeness field an auditor reads.

Because the ledger is append-only and every original is preserved unmutated, a full-batch failure never silently discards a shift’s work; when a batch quarantines abnormally many rows the fallback routing logic holds the last-known-good route rather than treating a thin valid set as “no work.”

Verification

The tests prove the two properties that matter: a mixed batch splits into exactly the right valid and quarantined sets without the batch failing, and the ledger is append-only with the original preserved unmutated.

import pytest


def _mixed_batch() -> list[dict]:
    return [
        {"device_id": "bin-A17", "ts": "2026-07-16T09:00:00+00:00", "lat": 40.70, "lon": -74.01, "fill_pct": 62.5},
        {"device_id": "bin-A17", "ts": "2026-07-16T09:00:15+00:00", "lat": 40.70, "lon": -74.01, "fill_pct": 140.0},
        {"device_id": "bin-B04", "lat": 40.71, "lon": -74.00, "fill_pct": 15.0},
    ]


def test_mixed_batch_splits_and_never_fails_whole_batch():
    ledger = QuarantineLedger()
    valid, reasons = triage_batch(_mixed_batch(), ledger)
    assert len(valid) == 1                       # the one good row survived
    assert len(ledger) == 2                      # both bad rows quarantined, not the whole batch
    assert reasons == {"out_of_range": 1, "missing_field": 1}


def test_original_record_is_preserved_unmutated():
    ledger = QuarantineLedger()
    batch = _mixed_batch()
    before = [dict(r) for r in batch]            # snapshot
    triage_batch(batch, ledger)
    assert batch == before                       # triage did not mutate the source rows
    entry = next(e for e in ledger.entries() if e.reason_code is ReasonCode.OUT_OF_RANGE)
    assert entry.original["fill_pct"] == 140.0   # stored verbatim, not clamped to 100


def test_ledger_has_no_update_or_delete():
    ledger = QuarantineLedger()
    assert not hasattr(ledger, "update")
    assert not hasattr(ledger, "delete")
    assert not hasattr(ledger, "pop")            # append-only surface only

Common Errors

One bad frame throws away an entire shift’s telemetry. Root cause: the batch was validated as a whole — for example TypeAdapter(list[TelemetryRecord]).validate_python(raw_batch) — so the first ValidationError aborts every record. Fix: validate per record inside a try/except as triage_batch does in Step 3, so a failure diverts one row to the ledger and the loop continues with the rest.

The quarantine ledger loses the reason a row was rejected. Root cause: the exception was swallowed with a bare except: continue, so the row disappeared with no reason code, no field, and no message. Fix: capture exc.errors()[0], classify it into a closed ReasonCode (Step 1), and store the field path and validator message on the entry — without the reason, the ledger is a list of mysteries no operator can act on.

The stored “original” no longer matches what arrived. Root cause: the same dict object was mutated after being handed to the ledger — for example clamping fill_pct to 100 in place, or reusing one dict across loop iterations — so the ledger’s original reflects the repaired value, not the malformed one. Fix: store a copy (dict(raw) in Step 3) and never coerce a quarantined record; the ledger’s value is precisely that it holds the untouched evidence.

FAQ

Why keep the malformed record at all instead of just dropping it?

A dropped record is an unexplained gap: an auditor sees a bin that reported nothing and cannot tell whether the sensor was offline or the frame was corrupt. The quarantine ledger converts that gap into an accounted-for rejection with a reason code and the original payload, so the batch reconciles exactly (valid + quarantined == total) and the row can be replayed once the upstream decoder is fixed.

Should a very high quarantine rate stop the solve?

Yes — treat the quarantine rate as a circuit-breaker signal. A batch where most rows fail usually means an upstream schema drift, not scattered bad sensors, and solving on the thin valid remainder would produce a route that ignores most of the fleet. Route that condition to the fallback logic to hold the last-known-good plan while the decoder is investigated, rather than letting a mostly-empty valid set look like a completed ingest.

Up: Schema Validation Pipelines