FMCSA ELD Sync for Waste Fleet Compliance

An Electronic Logging Device streams events all day, but until those events are decoded, sequence-checked, and reconciled against the routes the trucks actually ran, they are just opaque telematics traffic — and an inspector standing at a landfill scale house will find the gaps you did not. Municipal refuse fleets fall squarely inside the federal ELD mandate: a rear-loader crossing state lines for regional disposal, or any truck over the weight threshold operated on hours-of-service records, must carry a registered ELD and be able to produce an ordered, complete event log on demand. When the sync layer silently drops a malfunction event, misreads a duty-status code, or accepts an out-of-order sequence, the driver’s Record of Duty Status stops matching physical reality, and every downstream compliance artifact inherits the error.

This is the ELD ingestion sub-system of the Compliance Reporting & Automation pipeline. It shows how to pull ELD event records off a telematics feed, prove the event sequence_id chain is monotonic and unbroken, decode the FMCSA event type and event code numbers into duty-status meaning, and branch malfunction (M) and data-diagnostic (D) events into their own handling path so they are never confused with normal driving. What breaks without it is concrete: unaudited ELD data no one can replay, unhandled malfunction and diagnostic events that a roadside inspection surfaces before you do, and unassigned driving time that reconciles to no driver. Every code block is complete and pasteable into a Python 3.10+ compliance worker.

Prerequisites

Pin the validation stack so an ELD decode is byte-for-byte reproducible when an auditor replays it months later. The sync layer needs only Pydantic and the standard library — event decoding is deterministic arithmetic and dictionary lookups, not a solver:

python --version   # 3.10 or newer
pip install \
  pydantic==2.9.2 \
  pytest==8.2.0

Every event arriving from the telematics gateway must pass a typed schema before the chain logic touches it. An event with a sequence_id outside the 16-bit space, a naive (timezone-unaware) timestamp, or an unknown event_type is the most common cause of a silently corrupt Record of Duty Status, so reject it at ingestion. The ELDEvent model carries exactly the FMCSA-defined data elements the decode and projection stages read:

from __future__ import annotations
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, field_validator

# The FMCSA event sequence ID is a continuously increasing 16-bit value that
# wraps at 0xFFFF back to 0x0000 (Appendix A to Subpart B of Part 395, §4.5.4).
SEQUENCE_MODULUS = 0x1_0000  # 65_536


class ELDEvent(BaseModel):
    sequence_id: int = Field(ge=0, le=0xFFFF)   # continuous 16-bit event counter
    event_type: int = Field(ge=1, le=7)         # FMCSA event type number
    event_code: int = Field(ge=0, le=9)         # meaning is type-dependent
    # For type 7 (malfunction/diagnostic) events, the single-character indicator:
    #   malfunction letters P/E/T/L/R/S/O, or data-diagnostic digits 1..6.
    indicator: Optional[str] = None
    driver_id: str = Field(min_length=1)
    timestamp: datetime                         # tz-aware UTC event time
    engine_hours: float = Field(ge=0.0)         # total engine hours at the event
    vehicle_miles: int = Field(ge=0)            # total vehicle distance (miles)
    lat: float = Field(ge=-90.0, le=90.0)
    lon: float = Field(ge=-180.0, le=180.0)

    @field_validator("timestamp")
    @classmethod
    def require_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None or v.utcoffset() is None:
            raise ValueError("ELD event timestamp must be timezone-aware UTC")
        return v

The sequence ID is not a database primary key — it is a per-ELD counter defined by the FMCSA technical specification to be continuous and increasing, wrapping from 0xFFFF back to 0x0000. That wraparound is the single most misread detail in ELD ingestion, so the chain check below is written to expect it. Timestamps must be timezone-aware UTC for the same reason the Telematics & Sensor Data Ingestion framework insists on it: the moment you subtract two events to measure duty duration, a naive datetime either raises or silently computes a wrong interval.

Core Implementation

Four steps turn raw events into a synced, audit-ready duty timeline: pull the events, verify the sequence chain, decode the type and code numbers, and branch out the malfunction and diagnostic events. Each step is independently testable.

Verify the monotonic sequence chain

The chain check walks consecutive events and asserts that each sequence_id is exactly one greater than the last, modulo the 16-bit space. Formally, for an ordered event stream the expected identifier is

$$s_{i+1} \equiv (s_i + 1) \pmod{2^{16}}$$

and the forward distance between two identifiers, which distinguishes a small forward gap from a backward (out-of-order) event, is

$$\Delta_i = (s_{i+1} - s_i) \bmod 2^{16}.$$

A $\Delta$ of exactly 1 is a clean chain; a small $\Delta$ greater than 1 is a gap (events were lost between them); a $\Delta$ in the upper half of the space is really a backward step — an out-of-order or replayed event.

from dataclasses import dataclass


@dataclass(frozen=True)
class SequenceBreak:
    prev_seq: int
    this_seq: int
    delta: int
    kind: str          # "gap" | "out_of_order"


def verify_sequence_chain(events: list[ELDEvent]) -> list[SequenceBreak]:
    """Return every discontinuity in the ELD event sequence_id chain."""
    breaks: list[SequenceBreak] = []
    for prev, cur in zip(events, events[1:]):
        delta = (cur.sequence_id - prev.sequence_id) % SEQUENCE_MODULUS
        if delta == 1:
            continue  # clean link, including the 0xFFFF -> 0x0000 wrap
        # A delta in the upper half of the space is a backward step, not a
        # forward gap: the event arrived out of order or was replayed.
        kind = "out_of_order" if delta > SEQUENCE_MODULUS // 2 else "gap"
        breaks.append(SequenceBreak(prev.sequence_id, cur.sequence_id, delta, kind))
    return breaks

Note the wrap is handled for free: when prev.sequence_id is 0xFFFF and cur.sequence_id is 0x0000, the modular subtraction yields 1, so the last event before a rollover links cleanly to the first after it. A naive cur == prev + 1 check would flag that legitimate rollover as a break on every ELD once per 65,536 events.

Decode event type and code numbers

The FMCSA specification encodes meaning as two small integers. The type says what kind of event this is; the code refines it. The tables below map the numbers this pipeline handles into machine-readable labels, keeping malfunction and diagnostic decoding in their own dictionaries because they are keyed by an indicator character, not the numeric code.

EVENT_TYPE = {
    1: "duty_status_change",
    2: "intermediate_log",
    3: "login_logout",
    4: "engine_power",
    7: "malfunction_or_diagnostic",
}

# Type 1 duty-status change: event_code selects the driver's status.
DUTY_STATUS = {
    1: "off_duty",
    2: "sleeper_berth",
    3: "driving",
    4: "on_duty_not_driving",
}

# Type 7 malfunction indicators (letters) and data-diagnostic indicators (digits).
MALFUNCTION_CODE = {
    "P": "power_compliance",
    "E": "engine_sync_compliance",
    "T": "timing_compliance",
    "L": "positioning_compliance",
    "R": "data_recording_compliance",
    "S": "data_transfer_compliance",
    "O": "other_detected",
}
DATA_DIAGNOSTIC_CODE = {
    "1": "power_data",
    "2": "engine_sync_data",
    "3": "missing_required_elements",
    "4": "data_transfer_data",
    "5": "unidentified_driving",
    "6": "other_identified",
}


@dataclass(frozen=True)
class DecodedEvent:
    sequence_id: int
    kind: str                 # from EVENT_TYPE
    detail: str               # duty status, malfunction, or diagnostic label
    is_malfunction: bool      # an (M) event under 395.34
    is_diagnostic: bool       # a (D) event under 395.34
    driver_id: str
    timestamp: datetime


def decode_event(ev: ELDEvent) -> DecodedEvent:
    kind = EVENT_TYPE.get(ev.event_type, "unknown")
    detail, is_m, is_d = kind, False, False

    if ev.event_type == 1:
        detail = DUTY_STATUS.get(ev.event_code, "unknown_duty_status")
    elif ev.event_type == 7:
        code = (ev.indicator or "").strip().upper()
        if code in MALFUNCTION_CODE:
            detail, is_m = MALFUNCTION_CODE[code], True
        elif code in DATA_DIAGNOSTIC_CODE:
            detail, is_d = DATA_DIAGNOSTIC_CODE[code], True
        else:
            detail = "unknown_indicator"

    return DecodedEvent(
        sequence_id=ev.sequence_id,
        kind=kind,
        detail=detail,
        is_malfunction=is_m,
        is_diagnostic=is_d,
        driver_id=ev.driver_id,
        timestamp=ev.timestamp,
    )

Project the duty-status timeline and branch malfunctions

With events decoded, the sync layer projects the driver’s Record of Duty Status — the ordered sequence of status intervals — while diverting any malfunction (M) or data-diagnostic (D) event into a separate list so it is escalated rather than folded into duty time. This is the step that ties ELD data back to routing: the driving intervals it produces are what the DOT Hours-of-Service Logging subsystem reconciles against planned route legs, and any driving interval with no assigned driver is exactly the unidentified-driving diagnostic an inspector looks for.

def sync_eld_stream(events: list[ELDEvent]) -> dict:
    """Sync one ordered ELD event stream into a duty timeline + exceptions."""
    breaks = verify_sequence_chain(events)
    decoded = [decode_event(ev) for ev in events]

    duty_intervals: list[dict] = []
    malfunctions: list[DecodedEvent] = []
    open_status: Optional[DecodedEvent] = None

    for d in decoded:
        if d.is_malfunction or d.is_diagnostic:
            malfunctions.append(d)          # 395.34 branch — never duty time
            continue
        if d.kind == "duty_status_change":
            if open_status is not None:
                duty_intervals.append({
                    "status": open_status.detail,
                    "driver_id": open_status.driver_id,
                    "start": open_status.timestamp.isoformat(),
                    "end": d.timestamp.isoformat(),
                })
            open_status = d

    return {
        "sequence_breaks": [b.__dict__ for b in breaks],
        "duty_intervals": duty_intervals,
        "malfunctions": [m.__dict__ for m in malfunctions],
        "synced": not breaks and not malfunctions,
    }

The open_status accumulator closes each duty interval when the next status change arrives, which is how a stream of point events becomes a set of [start, end) intervals a compliance report can total. Positions on each event (lat, lon) let the projection be cross-checked against the GPS Polling Strategies fixes that fed the same trip, so a “driving” status with no corresponding movement — or movement with no status — surfaces immediately.

Regulatory Mapping

Every field the sync layer reads or asserts must trace to the rule that defines it. Do not describe ELD behavior in prose alone — pin it to the code of federal regulations so the compliance registry can version the mapping.

The event_type and event_code numbers, the indicator letters and digits, and the continuous 16-bit sequence_id are all defined by Appendix A to Subpart B of Part 395 — the ELD technical specification — which is why decoding them is a lookup against fixed tables rather than a heuristic. Malfunction and data-diagnostic handling is governed by 49 CFR 395.34: an unresolved malfunction obliges the driver to reconstruct paper logs, so a sync layer that swallows a type-7 event does not merely lose data, it hides a legally material condition. The reconciliation between ELD driving intervals and the routes actually dispatched is where this subsystem meets Audit Report Generation, which bundles the synced timeline into the tamper-evident record a municipality produces on a records request.

Validation & Verification

An ELD sync that never rejects a broken chain or flags a malfunction gives false assurance. Prove both fire with tests that construct a deliberately broken sequence and a malfunction event, then assert the pipeline catches them.

from datetime import timezone

BASE = datetime(2026, 3, 2, 6, 0, 0, tzinfo=timezone.utc)


def _ev(seq: int, etype: int, code: int, minute: int, **kw) -> ELDEvent:
    return ELDEvent(
        sequence_id=seq, event_type=etype, event_code=code,
        driver_id=kw.get("driver_id", "D-4417"),
        timestamp=BASE.replace(minute=minute),
        engine_hours=kw.get("engine_hours", 1200.0),
        vehicle_miles=kw.get("vehicle_miles", 88_000),
        lat=40.44, lon=-79.99, indicator=kw.get("indicator"),
    )


def test_sequence_gap_is_detected() -> None:
    events = [_ev(0x00A0, 1, 3, 0), _ev(0x00A3, 1, 4, 10)]  # jumped A0 -> A3
    breaks = verify_sequence_chain(events)
    assert len(breaks) == 1
    assert breaks[0].kind == "gap" and breaks[0].delta == 3


def test_sequence_wrap_is_not_a_break() -> None:
    events = [_ev(0xFFFF, 1, 3, 0), _ev(0x0000, 1, 4, 10)]
    assert verify_sequence_chain(events) == []   # legitimate 16-bit rollover


def test_out_of_order_event_is_flagged() -> None:
    events = [_ev(0x00A5, 1, 3, 0), _ev(0x00A2, 1, 4, 10)]  # backward step
    breaks = verify_sequence_chain(events)
    assert breaks[0].kind == "out_of_order"


def test_malfunction_event_branches_out_of_duty_time() -> None:
    events = [
        _ev(0x0010, 1, 3, 0),                       # driving
        _ev(0x0011, 7, 1, 5, indicator="E"),        # engine-sync malfunction (M)
        _ev(0x0012, 1, 4, 10),                       # on-duty not driving
    ]
    result = sync_eld_stream(events)
    assert result["synced"] is False
    assert len(result["malfunctions"]) == 1
    assert result["malfunctions"][0]["detail"] == "engine_sync_compliance"
    # the malfunction never became a duty interval
    assert all(iv["status"] != "engine_sync_compliance" for iv in result["duty_intervals"])

In production, emit each sync outcome as a structured record so the assertions above have a logged counterpart an auditor can query. Follow the site-wide JSONFormatter pattern, with fields distinct to ELD sync:

import json
import logging


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "event": record.getMessage(),
            "driver_id": getattr(record, "driver_id", None),
            "sequence_break_count": getattr(record, "sequence_break_count", None),
            "malfunction_count": getattr(record, "malfunction_count", None),
            "duty_interval_count": getattr(record, "duty_interval_count", None),
            "synced": getattr(record, "synced", None),
        }
        return json.dumps(payload, separators=(",", ":"))


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


def log_sync(result: dict, driver_id: str) -> None:
    logger.info(
        "eld_stream_synced",
        extra={
            "driver_id": driver_id,
            "sequence_break_count": len(result["sequence_breaks"]),
            "malfunction_count": len(result["malfunctions"]),
            "duty_interval_count": len(result["duty_intervals"]),
            "synced": result["synced"],
        },
    )

The two fields an auditor reads first are sequence_break_count and malfunction_count: both must be zero for a stream to be clean, and any non-zero value must resolve to a recorded reconstruction or repair action.

Failure Modes & Edge Cases

ELD data fails in ways that look benign until an inspection. Handle each along a defined path rather than papering over it.

1. Sequence gap. Events lost in transit leave a forward jump in the sequence_id chain. Never renumber to close the hole — that destroys the evidence that data is missing. Instead, record the gap, request a re-pull of the missing range from the ELD, and hold the affected duty intervals as provisional until the gap is filled or explained.

def reconcile_gaps(events: list[ELDEvent]) -> list[tuple[int, int]]:
    """Return (start, end) sequence ranges to re-request from the ELD."""
    missing: list[tuple[int, int]] = []
    for b in verify_sequence_chain(events):
        if b.kind == "gap":
            lo = (b.prev_seq + 1) % SEQUENCE_MODULUS
            hi = (b.this_seq - 1) % SEQUENCE_MODULUS
            missing.append((lo, hi))
    return missing

2. Out-of-order events. A backward sequence_id (a delta in the upper half of the 16-bit space) means an event arrived late or was replayed. Sort by sequence_id only within a known non-wrapping window; across a suspected rollover, sort by tz-aware timestamp and re-verify. Do not silently reorder — log every reordering as an audit event.

3. Unassigned driving time. A “driving” event with no authenticated driver_id is the unidentified-driving data diagnostic (indicator 5) under 395.34. It must be surfaced for a driver to claim or the carrier to annotate, never dropped. Route these to the malfunction branch so they cannot become anonymous duty time.

4. Clock drift. When an event’s timestamp moves backward while its sequence_id moves forward, the ELD clock has drifted or been tampered with. Trust the sequence order for chain integrity but flag the timestamp inversion, because duty-duration math on an inverted clock silently understates driving time. This is the same non-monotonic-timestamp defense the ingestion layer applies to raw fixes; here it protects the hours-of-service total.

Integration Checklist

Complete every item before an ELD sync reaches production compliance storage:

• Every event passes the ELDEvent schema; sequence_id is within the 16-bit space and timestamp is timezone-aware UTC.
• The chain check treats the 0xFFFF to 0x0000 rollover as a clean link, not a break.
• Gaps and out-of-order events are distinguished by forward delta and logged separately, never silently renumbered.
• Event type and code numbers decode against the fixed Appendix A tables, with unknown handled explicitly.
• Malfunction (M) and data-diagnostic (D) events branch out of the duty timeline and are escalated under 395.34.
• Unassigned driving time is routed to the diagnostic branch for driver claim or carrier annotation.
• Timestamp inversions against a forward-moving sequence are flagged as clock drift.
• Each synced stream emits an eld_stream_synced record with sequence_break_count and malfunction_count.
• Synced driving intervals reconcile against dispatched route legs before the timeline is treated as final.

FAQ

Why does the sequence check use modular arithmetic instead of a plain increment?

Because the FMCSA event sequence ID is a 16-bit counter that wraps from 0xFFFF back to 0x0000 by design. A plain cur == prev + 1 check would flag that legitimate rollover as a break once every 65,536 events on every device. Computing the forward distance modulo 2 to the 16th treats the wrap as the clean single-step link it is, while still catching real gaps and backward jumps.

Should a malfunction event ever be counted as duty time?

No. A type-7 malfunction or data-diagnostic event under 49 CFR 395.34 describes the state of the device, not the driver’s activity. Folding it into the duty timeline both corrupts the hours-of-service total and hides a condition the driver may be legally required to respond to by reconstructing paper logs. Branch it to a separate handler and escalate it.

How do I tell a dropped-event gap from an out-of-order event?

Compute the forward distance delta = (this_seq − prev_seq) mod 2^16. A small delta greater than one is a forward gap — events were lost between the two. A delta in the upper half of the space is really a backward step, meaning the event arrived late or was replayed. The two demand different responses: a gap triggers a re-pull, an out-of-order event triggers a re-sort and re-verify.

What is unassigned driving time and why does it matter?

It is a driving-status interval the ELD recorded with no authenticated driver attached — the unidentified-driving data diagnostic (indicator 5) under 395.34. Left unhandled it is untraceable movement an inspector will question. The sync layer must surface it for a driver to claim or the carrier to annotate; it must never become anonymous duty time or be dropped.

Do municipal waste trucks actually fall under the ELD mandate?

Many do. The mandate turns on operating on hours-of-service records with a commercial vehicle above the weight threshold, which regional-hauling and transfer runs routinely exceed. Short-haul exceptions exist, but they are conditional and must be proven, so the defensible posture is to sync and audit ELD data for the fleet and let the exemption analysis sit on top of a complete record rather than in place of one.

Up: Compliance Reporting & Automation

Related

• Syncing ELD Hours-of-Service Logs in Python — the end-to-end pull, decode, and duty-total walkthrough for one driver’s day
• DOT Hours-of-Service Logging — reconciling synced driving intervals against duty-status limits
• Audit Report Generation — bundling the synced timeline into a tamper-evident compliance artifact
• EPA e-Manifest Integration — the sibling reporting stream for hazardous-waste shipments
• GPS Polling Strategies — the position fixes each ELD event is cross-checked against