Parsing IoT Fill-Level Sensor Payloads in Python

Raw smart-bin telemetry arrives as fragmented, out-of-order JSON frames whose ultrasonic distance readings mean nothing until they are decoded to a fill percentage, validated against a contract, and cleaned of vibration noise — this page builds the parser that turns those bytes into a routable, audit-ready record.

Fill-level frames are the first thing the Telematics & Sensor Data Ingestion pipeline touches, and everything downstream trusts their output blindly. If a condensation spike leaks through as a genuine 90% reading, a truck is dispatched to an empty container; if a valid overflow is silently dropped, a full bin is skipped and the municipality eats an ordinance violation. The decoder here is the edge stage the Bin Sensor API Sync layer consumes: it converts a single vendor’s ultrasonic or infrared frame into the canonical SensorReading shape that sync then deduplicates and reconciles.

Environment & Data Prerequisites

The parser targets Python 3.10+ and treats every inbound frame as untrusted bytes until a Pydantic v2 model proves it. Pin the toolchain:

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

A single decoded ultrasonic frame carries a device identity, a sampling timestamp, the raw time-of-flight distance the transducer measured, and a diagnostics block. The expected wire shape and valid ranges are:

The transducer reports the distance from the sensor face (mounted in the lid) down to the surface of the waste, not the fill directly. Below the ranges lie the noise sources that make raw frames unroutable: temperature drift changes the speed of sound, condensation on the transducer scatters the return echo, and mechanical vibration during compaction throws transient distance spikes. Steps 1 and 4 exist specifically to neutralize those.

Step-by-Step Implementation

The decode runs in five deterministic stages: convert distance to fill, bind the contract, stream-parse a batch without accumulating heap, condition the signal, then serialize the canonical record.

Step 1 — Convert ultrasonic distance to a fill percentage

An ultrasonic sensor measures the distance distance_mm from the lid to the top of the waste. For a bin of internal depth depth_mm, the fill fraction is the proportion of the depth that is no longer empty air. Compensate the raw echo for the temperature-dependent speed of sound before the ratio, because an uncorrected reading drifts by roughly 0.17% per degree Celsius. For a measured distance $d$ in a bin of depth $D$, the fill percentage is:

$$\text{fill}_\% = \left(1 - \frac{d}{D}\right) \times 100, \qquad 0 \le d \le D$$

def distance_to_fill_pct(distance_mm: float, depth_mm: float,
                         temperature_c: float | None) -> float:
    """Convert a temperature-compensated ultrasonic distance to a fill percentage."""
    if depth_mm <= 0:
        raise ValueError("depth_mm must be positive")
    # Speed of sound rises ~0.6 m/s per °C; normalise the echo to a 20°C baseline.
    if temperature_c is not None:
        drift = 1.0 + 0.0006 * (temperature_c - 20.0) / 343.0 * 343.0
        distance_mm = distance_mm / drift
    clamped = min(max(distance_mm, 0.0), depth_mm)
    return round((1.0 - clamped / depth_mm) * 100.0, 2)

Clamping distance_mm to [0, depth_mm] guarantees the output stays in [0, 100] even when a spurious echo returns a distance longer than the bin is deep.

Step 2 — Bind the payload to a strict contract

Validation happens at the ingress boundary so malformed frames never propagate into routing. Pydantic v2 enforces the pattern, boundary checks, and required fields, and clamps the timestamp to UTC. This is the same defensive posture documented across the Schema Validation Pipelines reference and mirrors the coordinate-bound checks in Validating GPS Coordinates in Python.

from datetime import datetime, timezone
from pydantic import BaseModel, ConfigDict, Field, field_validator


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

    device_id: str = Field(min_length=8, max_length=32, pattern=r"^[A-Z0-9-]+$")
    sampled_at: datetime
    distance_mm: float = Field(ge=0.0)
    depth_mm: float = Field(gt=0.0)
    signal_strength_dbm: int = Field(ge=-120, le=-30)
    temperature_c: float | None = Field(default=None, ge=-40.0, le=85.0)
    diagnostics: dict = Field(default_factory=dict)

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

Step 3 — Stream-parse the batch with a generator

Municipal reporting windows arrive as bursts, and naive list accumulation fragments the heap under sustained load. A generator paired with orjson yields one validated record at a time, keeping memory flat regardless of batch size. Malformed frames are never dropped silently — they are yielded as an explicit rejection so the caller can quarantine them.

import orjson
from collections.abc import Iterator
from typing import Any
from pydantic import ValidationError


def stream_parse_frames(raw_frames: list[bytes]) -> Iterator[dict[str, Any]]:
    """Yield one decoded reading (or rejection) per frame with O(1) heap per frame."""
    for frame in raw_frames:
        try:
            decoded = orjson.loads(frame)
            valid = RawFrame.model_validate(decoded)
        except (orjson.JSONDecodeError, ValidationError) as exc:
            yield {
                "status": "rejected",
                "error_code": type(exc).__name__,
                "raw_frame_size": len(frame),
                "compliance_code": "EPA-40CFR-262-NONCONFORM",
            }
            continue
        yield {
            "status": "decoded",
            "device_id": valid.device_id,
            "sampled_at": valid.sampled_at.isoformat(),
            "raw_fill_pct": distance_to_fill_pct(
                valid.distance_mm, valid.depth_mm, valid.temperature_c),
            "signal_strength_dbm": valid.signal_strength_dbm,
            "diagnostics": valid.diagnostics,
        }

Step 4 — Condition the signal with a rolling median

Ultrasonic fill readings carry non-Gaussian noise, so a mean is the wrong central estimator — one vibration spike drags it. A rolling median suppresses transient artifacts while preserving the genuine step change of a fresh waste deposit, and any reading more than three standard deviations off the window baseline is quarantined for manual review rather than routed.

import statistics
from collections import deque
from typing import Any


class RollingMedianFilter:
    def __init__(self, window_size: int = 7, sigma_threshold: float = 3.0):
        self.window: deque[float] = deque(maxlen=window_size)
        self.sigma_threshold = sigma_threshold

    def process(self, raw_fill: float) -> dict[str, Any]:
        self.window.append(raw_fill)
        if len(self.window) < 3:
            return {"filtered_fill": raw_fill, "status": "warming_up"}

        median = statistics.median(self.window)
        stdev = statistics.pstdev(self.window)
        if stdev == 0:
            return {"filtered_fill": median, "status": "stable"}

        deviation = abs(raw_fill - median) / stdev
        if deviation > self.sigma_threshold:
            return {
                "filtered_fill": median,
                "status": "quarantined",
                "deviation_sigma": round(deviation, 2),
                "action": "manual_review_required",
            }
        return {"filtered_fill": round(median, 2), "status": "validated"}

Step 5 — Serialize the canonical record and log compliance

The final stage binds the filtered fill to a SHA-256 hash of the exact decoded payload and a compliance tier, then emits a structured audit line. This reuses the site-wide JSONFormatter pattern so log aggregators parse compliance metrics without regex.

import hashlib
import json
import logging
from datetime import datetime, timezone


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_obj = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        for attr in ("device_id", "compliance_tier", "regulatory_code",
                     "payload_hash", "fill_pct_filtered", "ingress_status"):
            if hasattr(record, attr):
                log_obj[attr] = getattr(record, attr)
        return json.dumps(log_obj, separators=(",", ":"))


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

COMPLIANCE_TIERS = {
    "TIER_1_STANDARD": "EPA_RCRA_SUBTITLE_D",
    "TIER_2_DIVERSION": "STATE_75PCT_DIVERSION",
    "TIER_3_CONTAMINATED": "EPA_40CFR262_HAZ_FLAG",
}


def emit_compliance_record(record: dict, filtered_fill: float, status: str) -> dict:
    canonical = json.dumps(record, sort_keys=True, separators=(",", ":")).encode()
    payload_hash = hashlib.sha256(canonical).hexdigest()

    tier = "TIER_1_STANDARD"
    if filtered_fill > 80.0:
        tier = "TIER_2_DIVERSION"
    if record.get("diagnostics", {}).get("contamination_flag") is True:
        tier = "TIER_3_CONTAMINATED"

    audit = {
        "device_id": record["device_id"],
        "compliance_tier": tier,
        "regulatory_code": COMPLIANCE_TIERS[tier],
        "payload_hash": payload_hash,
        "fill_pct_filtered": round(filtered_fill, 2),
        "ingress_status": status,
    }
    logger.info("fill_level_ingested", extra=audit)
    return audit

The full frame-to-record loop wires the five steps together. A rejected or quarantined reading suppresses its dispatch trigger; a validated reading over the 80% threshold is what promotes a bin’s service priority in the Capacity & Weight Limits solver.

def run_pipeline(raw_stream: list[bytes]) -> list[dict]:
    filt = RollingMedianFilter(window_size=7, sigma_threshold=3.0)
    emitted: list[dict] = []
    for record in stream_parse_frames(raw_stream):
        if record["status"] == "rejected":
            logger.warning("schema_rejection", extra={
                "device_id": "unknown", "ingress_status": "rejected"})
            continue
        cond = filt.process(record["raw_fill_pct"])
        emitted.append(emit_compliance_record(record, cond["filtered_fill"], cond["status"]))
    return emitted

Compliance Output

A single validated frame produces one immutable, newline-delimited JSON audit line. Every field carries regulatory weight:

{"timestamp":"2026-07-02 06:00:11","level":"INFO","logger":"fill_level_parser","message":"fill_level_ingested","device_id":"BIN-A1B2C3D4","compliance_tier":"TIER_2_DIVERSION","regulatory_code":"STATE_75PCT_DIVERSION","payload_hash":"9f2c...e41a","fill_pct_filtered":84.0,"ingress_status":"validated"}

• device_id — the chain-of-custody anchor tying a reading to a physical container, required for the positional evidence the FMCSA Electronic Logging Device rule expects under 49 CFR 395.8.
• compliance_tier / regulatory_code — maps the reading to a reporting bucket: standard RCRA Subtitle D collection (40 CFR Part 258), a state diversion-rate credit, or a contamination flag that gates the container out of a standard route under the EPA hazardous-waste generator standards in 40 CFR Part 262 (Uniform Manifest EPA Form 8700-22).
• payload_hash — a SHA-256 over the canonical decoded bytes; a single changed byte yields a different hash, so retroactive tampering is detectable, satisfying RCRA record-integrity expectations. The append-only record shape is specified in Route Schema Design.
• fill_pct_filtered — the conditioned value dispatch actually acts on, never the raw echo, so a quarantined spike can never trigger a phantom collection.
• ingress_status — validated, quarantined, or warming_up; a quarantined reading is retained as evidence but suppressed from dispatch, deferring to the Fallback Routing Logic when a sensor degrades mid-window.

Verification

Assert both the decode arithmetic and the noise-suppression contract on known inputs. A half-empty bin must decode to 50%, a vibration spike must be quarantined rather than filtered through, and a malformed frame must be rejected with its compliance code intact.

import orjson


def test_distance_decodes_to_fill():
    # Sensor 600mm from waste in a 1200mm bin => 50% full.
    assert distance_to_fill_pct(600.0, 1200.0, temperature_c=20.0) == 50.0


def test_vibration_spike_is_quarantined():
    filt = RollingMedianFilter(window_size=5, sigma_threshold=3.0)
    for stable in (60.0, 61.0, 59.0, 60.5):
        filt.process(stable)
    result = filt.process(5.0)  # transient compaction spike
    assert result["status"] == "quarantined"
    assert result["filtered_fill"] != 5.0  # median, not the spike, is routed


def test_malformed_frame_is_rejected():
    bad = orjson.dumps({"device_id": "x", "sampled_at": "2026-07-02T06:00:00Z"})
    out = list(stream_parse_frames([bad]))
    assert out[0]["status"] == "rejected"
    assert out[0]["compliance_code"] == "EPA-40CFR-262-NONCONFORM"

Common Errors

pydantic_core._pydantic_core.ValidationError: sampled_at Input should have timezone info. The vendor sent a naive timestamp (2026-07-02T06:00:00 with no Z or offset). The enforce_utc validator refuses it on purpose — a naive timestamp cannot be ordered against another vendor’s clock. Fix it at decode time by requiring the sensor firmware to emit ISO 8601 with an explicit offset, or reject to quarantine; never assume UTC on the reader’s behalf, because a silent assumption corrupts the reconciliation the Bin Sensor API Sync layer performs downstream.

orjson.JSONDecodeError: unexpected character on a truncated frame. A cellular handoff cut the payload mid-transmission, so the buffer holds partial bytes. The generator already yields a rejected record instead of crashing the batch — the fix is not to catch it harder but to confirm your intake reassembles fragmented frames before handing complete byte objects to stream_parse_frames; the bounded-buffer pattern lives in Async Batch Processing.

Fill percentage pinned at 0% for a bin you know is full. The distance_mm and depth_mm are swapped, or depth_mm was set to the external bin height rather than the internal cavity. Because distance_to_fill_pct clamps to [0, depth_mm], an inverted reading silently floors at zero instead of raising. Verify depth_mm against the bin model’s internal spec and assert distance_mm <= depth_mm in staging before trusting the conversion in production.

Up: Bin Sensor API Sync

Related

• Bin Sensor API Sync — deduplicates and clock-reconciles the canonical records this parser emits.
• Schema Validation Pipelines — edge-side contract enforcement and deterministic quarantine ledgers.
• Validating GPS Coordinates in Python — the sibling boundary-check pattern for spatial payloads.
• Async Batch Processing — bounded queues that reassemble fragmented frames before parsing.
• Fallback Routing Logic — deterministic degradation when a reading is quarantined or a sensor goes dark.