Bin Sensor API Sync

A smart-bin network never speaks with one voice. One vendor pushes ultrasonic fill readings over MQTT, another exposes a REST batch endpoint polled on a cadence, a third dumps a LoRaWAN concentrator’s buffer all at once when a controller reconnects after hours in a cellular dead zone. Without a synchronization layer that reconciles those streams into a single, idempotent, timestamp-ordered view of every bin, the routing engine optimizes over contradictions: a container gets dispatched twice because a replayed packet re-triggered its overflow threshold, or a genuinely full bin is skipped because a stale reading overwrote a fresh one after a clock-skewed timestamp sorted it to the back of the queue. This reference builds the sync layer that prevents both failures — a deterministic bridge between heterogeneous sensor APIs and the dispatch systems that consume them.

It sits directly under the Telematics & Sensor Data Ingestion framework, which requires every payload to carry a UTC-normalized timestamp, a monotonic sequence check, and an immutable ledger reference before any downstream system reads it. Sync is where multi-vendor payloads are normalized and deduplicated so that the distance matrix built by the VRP Route Optimization Algorithms never sees a fill level a bin could not physically have reported.

Prerequisites

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

python -m pip install \
  pydantic==2.7.1 \
  requests==2.32.3 \
  tenacity==8.4.1 \
  orjson==3.10.6

The sync layer reasons about one canonical record shape. Vendor-specific wire formats are decoded into it at the edge — the low-level frame decoding for ultrasonic and infrared payloads is covered in Parsing IoT Fill-Level Sensor Payloads — and everything downstream reads only the canonical model. The contract carries the fields the routing and compliance sides both depend on:

Two timestamps are non-negotiable: device_ts records when the sensor sampled, ingested_at records when the sync layer received it. Keeping both is what lets an auditor prove ordering later even when a device clock lied — the reconciliation discipline documented across the Schema Validation Pipelines reference.

Core Implementation

Sync is built in four steps: define the canonical model with contract validators, fetch batches over a pooled session with deterministic retry, deduplicate by content hash so replays are free, then reconcile clock skew and project time-to-full before handing the record to the ledger.

Step 1 — The canonical model with contract validators

The model clamps every timestamp to UTC at the boundary using Pydantic v2’s field_validator (the v1 @validator decorator is deprecated), so no downstream code has to reason about timezones. It also binds the DOT axle-weight ceiling directly into the type, rejecting a reading that could not correspond to a routable, legally loadable vehicle.

import hashlib
import orjson
from datetime import datetime, timezone
from typing import Optional
from pydantic import BaseModel, ConfigDict, field_validator


class SensorReading(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")

    bin_id: str
    fill_level_pct: float
    device_ts: datetime
    ingested_at: datetime
    sequence_id: int
    weight_kg: Optional[float] = None

    @field_validator("device_ts", "ingested_at")
    @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)

    @field_validator("fill_level_pct")
    @classmethod
    def validate_fill_range(cls, v: float) -> float:
        if not 0.0 <= v <= 100.0:
            raise ValueError("fill_level_pct must be between 0 and 100")
        return v

    @field_validator("weight_kg")
    @classmethod
    def validate_axle_weight(cls, v: Optional[float]) -> Optional[float]:
        if v is not None and not 0.0 <= v <= 15000.0:
            raise ValueError("weight_kg outside DOT/FMCSA single-axle routing bound")
        return v

    def payload_hash(self) -> str:
        """Deterministic SHA-256 over identity fields — the dedup key."""
        canonical = {
            "bin_id": self.bin_id,
            "sequence_id": self.sequence_id,
            "device_ts": self.device_ts.isoformat(),
            "fill_level_pct": self.fill_level_pct,
        }
        blob = orjson.dumps(canonical, option=orjson.OPT_SORT_KEYS)
        return hashlib.sha256(blob).hexdigest()

Step 2 — Pooled fetch with deterministic retry

Cellular backhaul in municipal service zones degrades in bursts. A persistent requests.Session reuses TCP connections, and tenacity applies bounded exponential backoff so a transient fault never trips a false overflow alert. Retries are restricted to connection-level errors — an HTTP 4xx is a contract problem, not a network blip, and retrying it only hides the bug.

import logging
import requests
from tenacity import (
    retry, stop_after_attempt, wait_exponential, retry_if_exception_type,
)

logger = logging.getLogger("bin_sync")


class SensorSyncError(Exception):
    """Raised when an upstream sensor API fails non-transiently."""


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout)),
)
def fetch_sensor_batch(session: requests.Session, endpoint: str, token: str) -> list[dict]:
    request_id = hashlib.sha256(f"{endpoint}:{token}".encode()).hexdigest()[:16]
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json",
        "X-Request-Id": request_id,
    }
    try:
        resp = session.get(endpoint, headers=headers, timeout=(3.0, 10.0))
        resp.raise_for_status()
    except requests.HTTPError as exc:
        logger.error("upstream_http_error",
                     extra={"status": exc.response.status_code, "endpoint": endpoint})
        raise SensorSyncError(f"HTTP {exc.response.status_code} during sync") from exc
    payload = resp.json()
    logger.info("batch_retrieved",
                extra={"request_id": request_id, "count": len(payload)})
    return payload

Step 3 — Idempotent normalization and content-hash dedup

Every vendor batch is normalized into SensorReading objects and keyed by payload_hash. Because the hash is derived only from identity fields, a replayed LoRaWAN burst produces hashes the layer has already seen, so duplicates are dropped for free — no downstream recalculation is triggered. Malformed payloads are never silently discarded; they are routed to a quarantine list with the exact violation attached, following the same defensive posture as Validating GPS Coordinates in Python.

from pydantic import ValidationError


def normalize_and_dedup(
    raw_batch: list[dict], seen_hashes: set[str],
) -> tuple[list[SensorReading], list[dict]]:
    accepted: list[SensorReading] = []
    quarantined: list[dict] = []
    for item in raw_batch:
        try:
            reading = SensorReading.model_validate(item)
        except ValidationError as exc:
            quarantined.append({
                "raw": item,
                "bin_id": item.get("bin_id", "unknown"),
                "violation": exc.errors(),
                "compliance_code": "EPA-40CFR-262-NONCONFORM",
            })
            logger.warning("payload_quarantined",
                           extra={"bin_id": item.get("bin_id"), "errors": len(exc.errors())})
            continue

        h = reading.payload_hash()
        if h in seen_hashes:
            logger.debug("duplicate_dropped", extra={"bin_id": reading.bin_id, "hash": h})
            continue
        seen_hashes.add(h)
        accepted.append(reading)
    return accepted, quarantined

Step 4 — Clock-skew reconciliation and time-to-full projection

Two vendors can disagree on wall-clock by seconds, so ordering by device_ts alone is unsafe. The layer keeps the latest reading per bin by sequence_id first and device_ts only as a tiebreaker, and it flags any record whose device clock diverges from the server clock beyond a tolerance. Once a stable current reading exists, a linear fill-rate model projects when the bin will reach the overflow threshold, which is what lets dispatch escalate a bin’s service priority before it actually overflows. For a bin at fill level $f$ percent rising at rate $r$ percent per hour toward threshold $T$, the projected time to full is:

$$t_{\text{full}} = \frac{T - f}{r}, \qquad r > 0$$

from dataclasses import dataclass

MAX_SKEW_SECONDS = 90.0
OVERFLOW_THRESHOLD_PCT = 80.0


@dataclass
class BinState:
    bin_id: str
    fill_level_pct: float
    sequence_id: int
    device_ts: datetime
    skew_flagged: bool
    hours_to_full: float | None


def reconcile(readings: list[SensorReading], fill_rate_pct_per_hr: dict[str, float]) -> dict[str, BinState]:
    latest: dict[str, SensorReading] = {}
    for r in readings:
        cur = latest.get(r.bin_id)
        if cur is None or (r.sequence_id, r.device_ts) > (cur.sequence_id, cur.device_ts):
            latest[r.bin_id] = r

    states: dict[str, BinState] = {}
    for bin_id, r in latest.items():
        skew = abs((r.ingested_at - r.device_ts).total_seconds())
        rate = fill_rate_pct_per_hr.get(bin_id, 0.0)
        remaining = OVERFLOW_THRESHOLD_PCT - r.fill_level_pct
        hours_to_full = (remaining / rate) if rate > 0 and remaining > 0 else None
        states[bin_id] = BinState(
            bin_id=bin_id,
            fill_level_pct=r.fill_level_pct,
            sequence_id=r.sequence_id,
            device_ts=r.device_ts,
            skew_flagged=skew > MAX_SKEW_SECONDS,
            hours_to_full=hours_to_full,
        )
        if skew > MAX_SKEW_SECONDS:
            logger.warning("clock_skew", extra={"bin_id": bin_id, "skew_seconds": skew})
    return states

The reconciled BinState map is the only view the dispatch layer reads. A bin whose hours_to_full falls inside the next service window is promoted; a skew_flagged bin is still usable for routing but its timestamp is not trusted as compliance evidence until reconciled against ingested_at.

Regulatory Mapping

Fill-level telemetry becomes evidence the moment it justifies — or fails to justify — a collection, so each field maps to a specific rule. These citations belong in the sync contract itself, not in prose commentary:

• weight_kg is bounded at 15,000 kg because a synced reading that implies an overweight collection would seed an illegal route. The federal single-axle limit is set by the bridge formula under 23 CFR 658.17, and the gross vehicle rating traces to the Federal Motor Vehicle Safety Standards under 49 CFR 571. The solver-side binding of this limit lives in Capacity & Weight Limits.
• device_ts / ingested_at supply the positional-and-time evidence that maps to the FMCSA Electronic Logging Device rule under 49 CFR Part 395 (data-element requirements in 49 CFR 395.8). The field-by-field translation into solver feasibility constraints is documented in the DOT/FMCSA Rule Mapping reference.
• Quarantined hazardous-stream payloads are tagged EPA-40CFR-262-NONCONFORM against the EPA hazardous-waste generator standards under 40 CFR Part 262, whose Uniform Hazardous Waste Manifest is EPA Form 8700-22. A bin carrying a hazardous designation cannot enter a standard collection route without a manifest tracking number attached.
• payload_hash and the append-only ledger satisfy the RCRA recordkeeping and record-integrity expectations those parts reference: a changed byte produces a different hash and breaks the chain, so retroactive modification is detectable. The canonical audit-record shape is specified in Route Schema Design.

Validation & Verification

Confirm the sync layer behaves by asserting both the happy path and the failure paths, and by checking the structured log fields it emits. A replayed batch must add zero new records; a malformed payload must land in quarantine, not the accepted list; a clock-skewed reading must be flagged but still routed.

import pytest
from datetime import datetime, timedelta, timezone

UTC = timezone.utc


def _raw(bin_id: str, seq: int, fill: float, skew_s: float = 0.0) -> dict:
    dev = datetime(2026, 7, 2, 6, 0, tzinfo=UTC)
    return {
        "bin_id": bin_id,
        "fill_level_pct": fill,
        "device_ts": dev.isoformat(),
        "ingested_at": (dev + timedelta(seconds=skew_s)).isoformat(),
        "sequence_id": seq,
        "weight_kg": 4200.0,
    }


def test_replayed_batch_is_deduplicated():
    seen: set[str] = set()
    batch = [_raw("bin-A", 1, 55.0)]
    first, _ = normalize_and_dedup(batch, seen)
    second, _ = normalize_and_dedup(batch, seen)  # identical replay
    assert len(first) == 1 and len(second) == 0


def test_out_of_range_fill_is_quarantined():
    seen: set[str] = set()
    accepted, quarantined = normalize_and_dedup([_raw("bin-B", 1, 140.0)], seen)
    assert accepted == [] and len(quarantined) == 1
    assert quarantined[0]["compliance_code"] == "EPA-40CFR-262-NONCONFORM"


def test_clock_skew_is_flagged_but_routed():
    seen: set[str] = set()
    readings, _ = normalize_and_dedup([_raw("bin-C", 1, 60.0, skew_s=300.0)], seen)
    state = reconcile(readings, {"bin-C": 10.0})["bin-C"]
    assert state.skew_flagged is True
    assert state.hours_to_full == pytest.approx(2.0)  # (80 - 60) / 10

In production, alert on the payload_quarantined event grouped by bin_id and on clock_skew rate: a spike in either is an early signal that a firmware rollout or an NTP failure on an edge gateway has drifted an entire vendor fleet off the contract.

Failure Modes & Edge Cases

Every guarantee above corresponds to a production failure that will occur. Each needs an explicit, logged mitigation — never a silent repair, because a repaired reading is an unauditable reading.

Sequence counter reset after a firmware update. A device that resets sequence_id to zero after an update looks like a permanent regression, so reconcile would keep discarding every new packet as older than the last seen. Detect a monotonic reset explicitly — a sequence that drops near zero and then climbs — and rebind the bin’s baseline instead of dropping traffic:

def is_sequence_reset(prev_seq: int, new_seq: int, threshold: int = 8) -> bool:
    """A large drop followed by a fresh low counter is a firmware reset, not a replay."""
    return new_seq < prev_seq and new_seq <= threshold

Sensor goes dark mid-window. A dead battery or cellular dead zone leaves a bin with no fresh reading. The sync layer must not feed dispatch a stale or null fill level; it substitutes a historical fill-rate baseline and marks the record estimated, so the Fallback Routing Logic downstream degrades deterministically instead of skipping a bin that is actually full:

def estimated_state(bin_id: str, last: BinState, hours_elapsed: float,
                    fill_rate_pct_per_hr: float) -> BinState:
    projected = min(100.0, last.fill_level_pct + fill_rate_pct_per_hr * hours_elapsed)
    logger.info("estimated_fill_substituted",
                extra={"bin_id": bin_id, "projected_pct": projected})
    return BinState(bin_id, projected, last.sequence_id, last.device_ts,
                    skew_flagged=last.skew_flagged, hours_to_full=None)

Burst reconnect saturation. A concentrator that reconnects after hours offline replays its entire buffer at once. Decouple the intake with a bounded queue so the burst spills to a disk buffer under backpressure rather than exhausting the heap on a constrained gateway — the full pattern lives in Async Batch Processing, which aggregates discrete sync events into route-ready snapshots without blocking a solver cycle.

Two vendors, same bin, conflicting fill. A dual-instrumented container can report from two devices. Because reconcile keys on (sequence_id, device_ts) per bin_id, the vendor namespaces must be merged into a single bin_id at decode time; otherwise the two streams never contend and dispatch sees whichever wrote last. Normalize the identifier before sync, not after.

Integration Checklist

• Pydantic pinned to 2.7.1; SensorReading runs strict=True, extra="forbid".
• Vendor identifiers normalized to a single bin_id namespace before sync.
• requests.Session created once outside the fetch loop to reuse TCP connections.
• Retry restricted to ConnectionError/Timeout; HTTP 4xx never retried.
• seen_hashes backed by a durable store (Redis/DB), not an in-process set, so dedup survives restarts.
• Both device_ts and ingested_at written to every ledger record for skew forensics.
• payload_quarantined and clock_skew events alerted on, grouped by bin_id.
• Sequence-reset detection wired in before deploying any firmware update.
• Estimated-fill substitution capped at 100% and flagged so it is never treated as measured evidence.
• Intake queue bounded with a disk-spill path for burst reconnects.

FAQ

Why hash on identity fields instead of the whole payload?

Hashing only bin_id, sequence_id, device_ts, and fill_level_pct makes the dedup key stable against vendor-added metadata that varies between retransmissions — signal strength, battery voltage, transport headers. If the whole payload were hashed, a replayed burst with a jittered RSSI field would look novel and trigger a duplicate route recalculation, which is exactly the failure the sync layer exists to prevent.

Should `seen_hashes` live in memory?

Only for a single-process prototype. In production the dedup set must be a durable, shared store — Redis with a TTL, or a unique constraint on payload_hash in the ledger table — so that a worker restart or a horizontally scaled second worker does not re-admit a burst it has already processed. An in-process set silently loses its dedup guarantee on every deploy.

How is bin sync different from GPS polling?

They solve mirror-image problems. GPS Polling Strategies pull vehicle position on an adaptive cadence the client controls, so the challenge is rate-limiting and drift. Bin sync mostly receives pushed or buffered fill events the client cannot pace, so the challenge is idempotency and clock-skew reconciliation across bursty, out-of-order arrivals. The two views are joined downstream to correlate a truck’s proximity with a bin’s state.

What happens to a reading that fails validation mid-collection?

It never reaches the accepted list. normalize_and_dedup routes it to the quarantine list with the exact Pydantic errors and a compliance code, and logs a payload_quarantined event; nothing corrupted enters the dispatch view. The original bytes are retained so a firmware misconfiguration can be diagnosed and replayed without halting live ingestion.

Why keep two timestamps per reading?

device_ts is when the sensor sampled; ingested_at is when the sync layer received it. Keeping both lets an auditor prove ordering even when a device clock is wrong, and lets reconcile flag skew instead of trusting a lie. A single timestamp forces a choice between trusting the device clock (which drifts) or the server clock (which loses sample time), and either choice corrupts the compliance trail.

Up: Telematics & Sensor Data Ingestion

Related

• Parsing IoT Fill-Level Sensor Payloads — decoding ultrasonic and infrared frames into the canonical model this layer syncs.
• Schema Validation Pipelines — edge-side contract enforcement with deterministic quarantine ledgers.
• GPS Polling Strategies — adaptive cadence that pairs vehicle position with bin state for chain-of-custody.
• Async Batch Processing — bounded queues and chunked writes that absorb burst reconnects.
• Fallback Routing Logic — deterministic degradation when a sensor goes dark mid-window.