Compliance Reporting & Automation

Syncing ELD Hours-of-Service Logs in Python

Project a batch of ELD events into a per-driver duty-status log without dropping unassigned driving time.

This page solves one specific production failure: a batch of ELD events arrives from the telematics feed with a missing record in the sequence chain, and the naive importer projects a clean duty-status log that silently omits the driving that happened in the gap.

An electronic logging device emits a continuous stream of typed events — duty-status changes, engine power cycles, driver logins, and diagnostics — each stamped with a monotonic sequence counter. Syncing that stream means three things done in order: verify the sequence chain is contiguous, prove the timestamps are ordered and timezone-aware, and project the surviving events into a per-driver Record of Duty Status that keeps unassigned driving time as first-class evidence rather than discarding it. This synchronizer sits inside the FMCSA ELD Sync contract and feeds the wider Compliance Reporting & Automation framework; it consumes fixes that have already cleared the Telematics & Sensor Data Ingestion layer.

Environment & Data Prerequisites

The synchronizer runs on the Python 3.10+ standard library alone — json, logging, dataclasses, datetime, enum, and collections. The only third-party pin is pytest, used solely by the verification step:

python -m pip install pytest==8.2.0

Each inbound record is one decoded ELD event. The projector reasons over exactly these fields:

Field Unit / type Range / rule
sequence_id int (0x0000–0xFFFF) continuous counter, wraps modulo 65536
event_type int 1 = duty-status change, 5 = login/logout, 6 = engine power
event_code int for type 1: 1=OFF, 2=SB, 3=D, 4=ON
timestamp ISO 8601 string tz-aware UTC, non-decreasing per unit
engine_hours float, hours monotonic engine runtime
vehicle_miles float, miles monotonic odometer
driver_id str or null null means unidentified driving

Two rules govern the code before a line is written. The sequence counter is the integrity anchor: under 49 CFR 395.26 the ELD assigns every recorded event a continuous number, so a break in that number is a missing record, not a cosmetic anomaly. And a driver_id of null is not garbage — it is unidentified driving that 49 CFR 395.34© requires the carrier to surface as a data diagnostic and account for, so dropping it is itself a violation.

Step-by-Step Implementation

Step 1 — Structured audit logging

Every sync decision — a gap found, an event projected, an unidentified interval flagged — is emitted as single-line JSON so the whole batch is replayable from the log alone.

import json
import logging
from collections import defaultdict
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from enum import Enum
from typing import Optional


class JSONFormatter(logging.Formatter):
    """Emit each ELD sync decision as one JSON line for the audit stream."""

    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "event": record.getMessage(),
            "eld_unit": getattr(record, "eld_unit", None),
            "sequence_id": getattr(record, "sequence_id", None),
            "detail": getattr(record, "detail", None),
        }
        return json.dumps(entry)


sync_logger = logging.getLogger("waste_ops.eld_sync")
if not sync_logger.handlers:
    _h = logging.StreamHandler()
    _h.setFormatter(JSONFormatter())
    sync_logger.addHandler(_h)
sync_logger.setLevel(logging.INFO)

Step 2 — The event and interval contracts

An ELDEvent is frozen once decoded; a DutyInterval is what a duty-status change projects into. The duty-status codes come straight from the 49 CFR 395.26 event dictionary.

DUTY_STATUS = {1: "OFF", 2: "SB", 3: "D", 4: "ON"}  # 49 CFR 395.26 event_type 1 codes


@dataclass(frozen=True)
class ELDEvent:
    sequence_id: int
    event_type: int
    event_code: int
    timestamp: datetime          # tz-aware UTC
    engine_hours: float
    vehicle_miles: float
    driver_id: Optional[str]     # None => unidentified driving


@dataclass(frozen=True)
class DutyInterval:
    driver_id: str
    status: str
    start: datetime
    end: Optional[datetime]      # None => interval still open at batch end
    start_miles: float
    driving_seconds: int

Step 3 — Verify the sequence chain

The counter increments once per recorded event and wraps at 0xFFFF. The expected successor is therefore

snext=(sprev+1) mod 216s_{\text{next}} = (s_{\text{prev}} + 1) \bmod 2^{16}

Any deviation is a gap. Verifying it first means the projector never treats a truncated batch as complete.

def verify_sequence_chain(events: list[ELDEvent]) -> list[tuple[str, str]]:
    """Return [(prev_hex, cur_hex), ...] for every break in the counter."""
    ordered = sorted(events, key=lambda e: (e.timestamp, e.sequence_id))
    gaps: list[tuple[str, str]] = []
    for prev, cur in zip(ordered, ordered[1:]):
        expected = (prev.sequence_id + 1) & 0xFFFF
        if cur.sequence_id != expected:
            gaps.append((f"0x{prev.sequence_id:04X}", f"0x{cur.sequence_id:04X}"))
            sync_logger.warning(
                "sequence_gap",
                extra={"sequence_id": f"0x{cur.sequence_id:04X}",
                       "detail": {"expected": f"0x{expected:04X}"}},
            )
    return gaps

Step 4 — Project into a per-driver duty-status log

Duty-status change events are grouped by driver — with None mapped to the reserved UNIDENTIFIED profile so unassigned driving is preserved, not dropped. Each event closes the previous interval and opens the next; the final interval per driver stays open.

class ELDProjector:
    UNIDENTIFIED = "UNIDENTIFIED"

    def project(self, events: list[ELDEvent]) -> dict[str, list[DutyInterval]]:
        logs: dict[str, list[ELDEvent]] = defaultdict(list)
        for e in events:
            if e.event_type == 1:                       # duty-status change only
                logs[e.driver_id or self.UNIDENTIFIED].append(e)

        projected: dict[str, list[DutyInterval]] = {}
        for driver, evs in logs.items():
            evs.sort(key=lambda e: (e.timestamp, e.sequence_id))
            intervals: list[DutyInterval] = []
            for cur, nxt in zip(evs, evs[1:] + [None]):
                status = DUTY_STATUS.get(cur.event_code, f"code{cur.event_code}")
                end = nxt.timestamp if nxt is not None else None
                seconds = int((end - cur.timestamp).total_seconds()) if end else 0
                intervals.append(DutyInterval(
                    driver_id=driver, status=status, start=cur.timestamp, end=end,
                    start_miles=cur.vehicle_miles,
                    driving_seconds=seconds if status == "D" else 0,
                ))
                if driver == self.UNIDENTIFIED and status == "D":
                    sync_logger.warning(
                        "unidentified_driving",
                        extra={"sequence_id": f"0x{cur.sequence_id:04X}",
                               "detail": {"start_miles": cur.vehicle_miles}},
                    )
            projected[driver] = intervals
        return projected

Step 5 — Run the sync over a batch and serialize

The demo batch carries a real gap (0x0A25 never arrives) and a stretch of unidentified driving, so the output exercises both diagnostics.

def _parse(pkt: dict) -> ELDEvent:
    ts = datetime.fromisoformat(pkt["ts"].replace("Z", "+00:00"))
    if ts.tzinfo is None:
        raise ValueError(f"naive timestamp rejected: {pkt['ts']}")
    return ELDEvent(pkt["seq"], pkt["type"], pkt["code"], ts,
                    pkt["eh"], pkt["miles"], pkt.get("driver"))


if __name__ == "__main__":
    raw = [
        {"seq": 0x0A21, "type": 5, "code": 1, "ts": "2026-06-02T06:00:00Z", "eh": 1180.4, "miles": 45210.0, "driver": "D-4471"},
        {"seq": 0x0A22, "type": 1, "code": 4, "ts": "2026-06-02T06:05:00Z", "eh": 1180.5, "miles": 45210.0, "driver": "D-4471"},
        {"seq": 0x0A23, "type": 1, "code": 3, "ts": "2026-06-02T06:25:00Z", "eh": 1180.8, "miles": 45213.1, "driver": "D-4471"},
        {"seq": 0x0A24, "type": 1, "code": 4, "ts": "2026-06-02T08:10:00Z", "eh": 1182.5, "miles": 45261.7, "driver": "D-4471"},
        {"seq": 0x0A26, "type": 1, "code": 3, "ts": "2026-06-02T08:45:00Z", "eh": 1183.1, "miles": 45262.0, "driver": "D-4471"},  # 0x0A25 missing
        {"seq": 0x0A27, "type": 1, "code": 1, "ts": "2026-06-02T11:30:00Z", "eh": 1185.8, "miles": 45318.4, "driver": "D-4471"},
        {"seq": 0x0A28, "type": 1, "code": 3, "ts": "2026-06-02T11:45:00Z", "eh": 1186.0, "miles": 45319.2, "driver": None},     # unidentified
        {"seq": 0x0A29, "type": 1, "code": 1, "ts": "2026-06-02T11:52:00Z", "eh": 1186.1, "miles": 45321.0, "driver": None},
    ]
    events = [_parse(p) for p in raw]
    gaps = verify_sequence_chain(events)
    logs = ELDProjector().project(events)

    report = {
        "eld_unit": "3ESN-8842",
        "synced_at": datetime.now(timezone.utc).isoformat(),
        "sequence_check": {"contiguous": not gaps, "gaps": gaps},
        "duty_logs": {d: [asdict(i) for i in ivals] for d, ivals in logs.items()},
    }
    print(json.dumps(report, default=str, indent=2))

Compliance Output

The sync emits one canonical report per batch. Trimmed to its regulatory core, it reads:

{
  "eld_unit": "3ESN-8842",
  "synced_at": "2026-07-16T14:22:05.481920+00:00",
  "sequence_check": {"contiguous": false, "gaps": [["0x0A24", "0x0A26"]]},
  "duty_logs": {
    "D-4471": [
      {"driver_id": "D-4471", "status": "ON", "start": "2026-06-02T06:05:00+00:00", "end": "2026-06-02T06:25:00+00:00", "start_miles": 45210.0, "driving_seconds": 0},
      {"driver_id": "D-4471", "status": "D",  "start": "2026-06-02T06:25:00+00:00", "end": "2026-06-02T08:10:00+00:00", "start_miles": 45213.1, "driving_seconds": 6300},
      {"driver_id": "D-4471", "status": "ON", "start": "2026-06-02T08:10:00+00:00", "end": "2026-06-02T08:45:00+00:00", "start_miles": 45261.7, "driving_seconds": 0},
      {"driver_id": "D-4471", "status": "D",  "start": "2026-06-02T08:45:00+00:00", "end": "2026-06-02T11:30:00+00:00", "start_miles": 45262.0, "driving_seconds": 9900},
      {"driver_id": "D-4471", "status": "OFF","start": "2026-06-02T11:30:00+00:00", "end": null, "start_miles": 45318.4, "driving_seconds": 0}
    ],
    "UNIDENTIFIED": [
      {"driver_id": "UNIDENTIFIED", "status": "D",  "start": "2026-06-02T11:45:00+00:00", "end": "2026-06-02T11:52:00+00:00", "start_miles": 45319.2, "driving_seconds": 420},
      {"driver_id": "UNIDENTIFIED", "status": "OFF","start": "2026-06-02T11:52:00+00:00", "end": null, "start_miles": 45321.0, "driving_seconds": 0}
    ]
  }
}

Each field earns its place under the electronic-logging rule:

  • sequence_check.gaps — the pair of counter values bracketing every missing record. 49 CFR 395.26(b) requires each event to carry a continuous sequence identifier, so a recorded gap is the auditable proof that the log presented is known to be incomplete rather than falsely certified as whole.
  • duty_logs.<driver>.status — the reconstructed duty status (OFF, SB, D, ON) for each interval, exactly the four annotations the record of duty status is built from under 49 CFR 395.26(b)(4).
  • driving_seconds — elapsed time attributed to the D status, the raw material an auditor totals against the driving ceilings enforced in the DOT Hours-of-Service Logging reference.
  • start_miles — the odometer at each transition; 49 CFR 395.26(b)(6) requires accumulated vehicle miles on every duty-status-change record so distance can be cross-checked against position.
  • UNIDENTIFIED — the reserved profile that captures driving with no logged-in driver. 49 CFR 395.34© defines the unidentified-driving-records data diagnostic; keeping this interval preserved and separately labelled is precisely how the carrier discharges that obligation instead of erasing it.

Because the report states its own incompleteness and never folds unidentified driving into a named driver, the sync stays truthful under audit — the two failure modes that turn an ELD import into a falsified log are both surfaced, not smoothed over.

Verification

The tests prove the chain check catches the gap, the projector preserves unidentified driving, and driving time totals correctly.

BASE = "2026-06-02T{}Z"


def _ev(seq, code, hhmm, miles, driver="D-4471"):
    ts = datetime.fromisoformat(BASE.format(hhmm).replace("Z", "+00:00"))
    return ELDEvent(seq, 1, code, ts, 0.0, miles, driver)


def test_sequence_gap_is_detected():
    events = [_ev(0x10, 3, "06:00:00", 100.0), _ev(0x12, 1, "07:00:00", 140.0)]
    gaps = verify_sequence_chain(events)
    assert gaps == [("0x0010", "0x0012")]


def test_unidentified_driving_is_preserved():
    events = [_ev(0x20, 3, "09:00:00", 200.0, driver=None),
              _ev(0x21, 1, "09:30:00", 220.0, driver=None)]
    logs = ELDProjector().project(events)
    assert "UNIDENTIFIED" in logs
    assert logs["UNIDENTIFIED"][0].status == "D"


def test_driving_seconds_totals_correctly():
    events = [_ev(0x30, 3, "06:00:00", 300.0), _ev(0x31, 4, "07:45:00", 360.0)]
    logs = ELDProjector().project(events)
    driving = sum(i.driving_seconds for i in logs["D-4471"])
    assert driving == 105 * 60  # 06:00 -> 07:45

Common Errors

ValueError: naive timestamp rejected: 2026-06-02T06:00:00 — a feed sent a local wall-clock string with no offset. Root cause: subtracting a naive datetime from a tz-aware one to compute driving_seconds raises TypeError deep in the projector, so the guard in _parse fails fast instead. Fix: normalize every event to tz-aware UTC at decode time with .replace("Z", "+00:00") and reject anything still naive — never assume the carrier’s local zone.

Duty log looks complete but total driving is short. Root cause: the importer sorted events by sequence_id alone, and a counter that had wrapped past 0xFFFF reordered a post-rollover event ahead of its true predecessor, truncating an interval. Fix: sort by (timestamp, sequence_id) as both verify_sequence_chain and project do, and treat the counter purely as a contiguity check via (prev + 1) & 0xFFFF, not as the ordering key.

Unidentified driving silently vanishes from the report. Root cause: filtering events with if e.driver_id: before projection drops every None-driver record, which is the unassigned driving 49 CFR 395.34© says you must account for. Fix: map None to the reserved UNIDENTIFIED profile (Step 4) so the interval survives and raises its data diagnostic rather than disappearing.

FAQ

Why keep unidentified driving instead of assigning it to the last known driver?

Because assignment is a human review step the ELD rule reserves for the carrier, not something the importer may infer. 49 CFR 395.34© treats unidentified driving as a diagnostic to be resolved by review; guessing a driver would fabricate a certified log. The synchronizer’s job is to preserve the interval intact under the UNIDENTIFIED profile and flag it, leaving the assignment decision to the workflow that has the evidence to make it.

Can I project directly without running the sequence check first?

You can, but the log will lie. The projector builds intervals from whatever events it receives; if a driving-to-off-duty transition went missing in transit, the preceding driving interval extends across the gap and overstates or understates driving time with no signal. Running verify_sequence_chain first records the gap so the report declares its own incompleteness rather than presenting a plausible-looking fabrication.

Up: FMCSA ELD Sync