Compliance Reporting & Automation

Logging Driver Duty-Status Transitions in Python

Capture each duty-status change as an ordered, tamper-evident event to reconstruct a Record of Duty Status.

This page solves one specific production failure: two duty-status changes land in the same second, the log keeps them in arrival order, and the reconstructed record of duty status shows a driver in two statuses at once — enough for an auditor to reject the whole day.

A record of duty status is a reconstruction, not a raw feed. It is rebuilt from an ordered series of transitions — off-duty to driving, driving to on-duty-not-driving, on-duty to sleeper — each carrying the moment, the odometer, and the position where it happened. To be defensible that series has to be strictly ordered even when timestamps collide, tamper-evident so a back-dated edit is visible, and splittable at midnight so each calendar day’s grid totals correctly. This transition log is the event source for the DOT Hours-of-Service Logging contract; it is the human-facing counterpart to the device stream handled in Syncing ELD Hours-of-Service Logs in Python, and the driving ceilings it checks trace back to the route model in Mapping DOT Hours of Service to Waste Routes.

Environment & Data Prerequisites

The logger uses only the Python 3.10+ standard library — hashlib, json, logging, dataclasses, datetime, and enum. pytest is pinned solely for verification:

python -m pip install pytest==8.2.0

Each event is one duty-status transition. The reconstruction reasons over exactly these fields:

Field Unit / type Range / rule
from_status enum OFF, SB, D, ON — the status being left
to_status enum OFF, SB, D, ON — the status being entered
timestamp ISO 8601 string tz-aware UTC, non-decreasing
seq int tiebreaker for same-second transitions, strictly increasing
odometer_km int, kilometres monotonic; integer to stay hash-stable
lat, lon float, degrees WGS84 position of the change

Two rules shape the code. Timestamp alone is not a total order — CMV clocks record to the second, so two changes can share one, and only a strictly increasing seq companion breaks the tie deterministically. And the four duty statuses are exactly the grid rows the record of duty status is drawn on under 49 CFR 395.8(b), so the enum is the regulation, not a convenience.

Step-by-Step Implementation

Step 1 — Structured audit logging

Each accepted transition and each reconstruction result is emitted as one JSON line, so a day can be replayed from the log alone.

import hashlib
import json
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone, timedelta, date
from enum import Enum
from typing import Optional


class JSONFormatter(logging.Formatter):
    """Emit each duty-status logging decision as one JSON line."""

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


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

Step 2 — The duty status enum and the hash-linked transition

The status enum is the regulatory grid; the transition is frozen and carries a prev_hash, so the log is an append-only chain where any edit to an earlier entry invalidates every hash after it.

class DutyStatus(str, Enum):
    OFF = "off_duty"                # 49 CFR 395.8(b) grid line 1
    SB = "sleeper_berth"           # line 2
    D = "driving"                  # line 3
    ON = "on_duty_not_driving"     # line 4


@dataclass(frozen=True)
class DutyTransition:
    driver_id: str
    from_status: DutyStatus
    to_status: DutyStatus
    timestamp: datetime            # tz-aware UTC
    seq: int                       # same-second tiebreaker
    odometer_km: int
    lat: float
    lon: float
    prev_hash: str
    entry_hash: str = ""

    def compute_hash(self) -> str:
        payload = json.dumps({
            "driver_id": self.driver_id,
            "from": self.from_status.value,
            "to": self.to_status.value,
            "timestamp": self.timestamp.isoformat(),
            "seq": self.seq,
            "odometer_km": self.odometer_km,
            "prev_hash": self.prev_hash,
        }, sort_keys=True, separators=(",", ":")).encode("utf-8")
        return hashlib.sha256(payload).hexdigest()

Step 3 — Append with strict ordering and continuity checks

Appending enforces three invariants: the new event is not older than the last, a same-second event has a higher seq, and its from_status matches the last to_status so the chain of statuses is continuous.

GENESIS_HASH = "0" * 64


class DutyStatusLog:
    def __init__(self, driver_id: str) -> None:
        self.driver_id = driver_id
        self.entries: list[DutyTransition] = []

    def append(self, from_status, to_status, timestamp, seq,
               odometer_km, lat, lon) -> DutyTransition:
        if timestamp.tzinfo is None:
            raise ValueError("naive timestamp rejected; transitions must be tz-aware UTC")
        if self.entries:
            last = self.entries[-1]
            if (timestamp, seq) <= (last.timestamp, last.seq):
                raise ValueError(
                    f"non-increasing (timestamp, seq): {(timestamp.isoformat(), seq)} "
                    f"<= {(last.timestamp.isoformat(), last.seq)}")
            if from_status != last.to_status:
                raise ValueError(
                    f"discontinuous status: from={from_status.value} "
                    f"but last to={last.to_status.value}")
            prev_hash = last.entry_hash
        else:
            prev_hash = GENESIS_HASH

        draft = DutyTransition(self.driver_id, from_status, to_status, timestamp,
                               seq, odometer_km, lat, lon, prev_hash)
        sealed = DutyTransition(self.driver_id, from_status, to_status, timestamp,
                                seq, odometer_km, lat, lon, prev_hash,
                                entry_hash=draft.compute_hash())
        self.entries.append(sealed)
        rods_logger.info("transition_appended", extra={
            "driver_id": self.driver_id,
            "transition": f"{from_status.value}->{to_status.value}"})
        return sealed

Step 4 — Reconstruct driving time, splitting at midnight

The record of duty status is per calendar day, so a driving interval that crosses midnight is split and its time attributed to each day. Cumulative driving is the sum of driving-status intervals,

Tdrive=∑i(ti+1−ti)⋅1 ⁣[si=D]T_{\text{drive}} = \sum_{i} (t_{i+1} - t_i)\cdot \mathbb{1}\!\left[s_i = \text{D}\right]

where sis_i is the status held during interval ii and 1\mathbb{1} is 1 when the driver was driving.

DRIVING_CEILING = timedelta(hours=11)   # 49 CFR 395.3(a)(3)(i)


def reconstruct_daily_driving(log: DutyStatusLog,
                              now: Optional[datetime] = None) -> dict[date, timedelta]:
    """Sum driving time per calendar day, splitting intervals at midnight UTC."""
    if not log.entries:
        return {}
    now = now or datetime.now(timezone.utc)
    per_day: dict[date, timedelta] = {}

    # each transition holds `from_status` until the next transition's timestamp
    points = log.entries
    for cur, nxt in zip(points, list(points[1:]) + [None]):
        status = cur.to_status
        start = cur.timestamp
        end = nxt.timestamp if nxt is not None else now  # terminal interval runs to now
        if status != DutyStatus.D:
            continue
        # split [start, end) across UTC calendar-day boundaries
        cursor = start
        while cursor < end:
            midnight = datetime.combine(cursor.date() + timedelta(days=1),
                                        datetime.min.time(), tzinfo=timezone.utc)
            segment_end = min(end, midnight)
            per_day[cursor.date()] = per_day.get(cursor.date(), timedelta()) + (segment_end - cursor)
            cursor = segment_end
    return per_day

Step 5 — Run it over a shift and flag the ceiling

The demo shift starts before midnight and runs past it, so the reconstruction exercises the split, then checks each day against the 11-hour driving ceiling.

def _ts(s: str) -> datetime:
    return datetime.fromisoformat(s.replace("Z", "+00:00"))


if __name__ == "__main__":
    log = DutyStatusLog("D-4471")
    D, ON, OFF = DutyStatus.D, DutyStatus.ON, DutyStatus.OFF
    log.append(OFF, ON, _ts("2026-06-02T21:00:00Z"), 1, 45210, 40.702, -74.015)
    log.append(ON, D,  _ts("2026-06-02T21:20:00Z"), 2, 45210, 40.703, -74.014)
    log.append(D,  ON, _ts("2026-06-03T02:40:00Z"), 3, 45498, 40.688, -74.121)  # crosses midnight
    log.append(ON, D,  _ts("2026-06-03T03:10:00Z"), 4, 45498, 40.688, -74.121)
    log.append(D,  OFF,_ts("2026-06-03T09:30:00Z"), 5, 45790, 40.702, -74.015)

    daily = reconstruct_daily_driving(log, now=_ts("2026-06-03T09:30:00Z"))
    for day, driven in sorted(daily.items()):
        over = driven > DRIVING_CEILING
        rods_logger.info("daily_driving", extra={
            "driver_id": "D-4471",
            "cumulative_drive_h": round(driven.total_seconds() / 3600, 2)})
        print(f"{day}: {driven} driving"
              f"{'  <-- EXCEEDS 11h CEILING' if over else ''}")

Compliance Output

The reconstruction emits a per-day summary alongside the append-only transition chain. Serialized for the audit stream, one day reads:

{
  "driver_id": "D-4471",
  "rods_date": "2026-06-03",
  "driving": "07:00:00",
  "driving_ceiling": "11:00:00",
  "ceiling_exceeded": false,
  "transitions": [
    {"from": "driving", "to": "on_duty_not_driving", "timestamp": "2026-06-03T02:40:00+00:00",
     "seq": 3, "odometer_km": 45498, "lat": 40.688, "lon": -74.121,
     "prev_hash": "5c2f...a1", "entry_hash": "e8d4...77"},
    {"from": "on_duty_not_driving", "to": "driving", "timestamp": "2026-06-03T03:10:00+00:00",
     "seq": 4, "odometer_km": 45498, "lat": 40.688, "lon": -74.121,
     "prev_hash": "e8d4...77", "entry_hash": "1b90...c3"}
  ]
}

Each field carries a specific meaning under the record-of-duty-status rule:

  • from / to — the grid lines the driver moved between. 49 CFR 395.8(b) requires the record of duty status to show all four duty statuses across the day; storing both endpoints of each change is what lets the grid be redrawn exactly.
  • timestamp and seq — the time of the change plus the tiebreaker that guarantees a total order when two changes share a second, so 49 CFR 395.8©'s requirement that the record be a truthful sequence cannot be defeated by clock granularity.
  • odometer_km — the vehicle miles at the change, required alongside duty status under 49 CFR 395.8(d) so distance can be reconciled with the driving intervals.
  • lat / lon — the location of the status change; 49 CFR 395.8(d) requires the name of the city, town, or village with state abbreviation for each change of duty status, and the coordinate is what resolves to that place name.
  • prev_hash / entry_hash — the append-only linkage. Because each entry hashes the one before it, a back-dated or altered transition breaks every hash downstream, giving the record the tamper-evidence an auditor relies on and mirroring the chain built in Generating Tamper-Evident Compliance Reports.
  • driving / ceiling_exceeded — the reconstructed per-day driving total against the 11-hour ceiling of 49 CFR 395.3(a)(3)(i), the headline number an auditor checks first.

Because the log is append-only and hash-linked, and because midnight is split rather than smeared, each day’s grid totals independently and any retroactive edit is visible — the two properties that separate a defensible record of duty status from a reconstructed guess.

Verification

The tests prove same-second ordering is enforced, the hash chain links, and cumulative driving reconstructs correctly across midnight.

def _log() -> DutyStatusLog:
    return DutyStatusLog("D-TEST")


def test_same_second_requires_increasing_seq():
    log = _log()
    log.append(DutyStatus.OFF, DutyStatus.D, _ts("2026-06-02T08:00:00Z"), 1, 0, 0.0, 0.0)
    try:
        log.append(DutyStatus.D, DutyStatus.ON, _ts("2026-06-02T08:00:00Z"), 1, 5, 0.0, 0.0)
        assert False, "expected non-increasing (timestamp, seq) rejection"
    except ValueError as e:
        assert "non-increasing" in str(e)


def test_hash_chain_links_entries():
    log = _log()
    a = log.append(DutyStatus.OFF, DutyStatus.D, _ts("2026-06-02T08:00:00Z"), 1, 0, 0.0, 0.0)
    b = log.append(DutyStatus.D, DutyStatus.OFF, _ts("2026-06-02T10:00:00Z"), 2, 90, 0.0, 0.0)
    assert b.prev_hash == a.entry_hash
    assert a.entry_hash != GENESIS_HASH


def test_driving_reconstructs_across_midnight():
    log = _log()
    log.append(DutyStatus.OFF, DutyStatus.D, _ts("2026-06-02T22:00:00Z"), 1, 0, 0.0, 0.0)
    log.append(DutyStatus.D, DutyStatus.OFF, _ts("2026-06-03T02:00:00Z"), 2, 200, 0.0, 0.0)
    daily = reconstruct_daily_driving(log, now=_ts("2026-06-03T02:00:00Z"))
    assert daily[date(2026, 6, 2)] == timedelta(hours=2)   # 22:00 -> 24:00
    assert daily[date(2026, 6, 3)] == timedelta(hours=2)   # 00:00 -> 02:00

Common Errors

ValueError: non-increasing (timestamp, seq). Root cause: two duty-status changes were recorded in the same second and appended with the same seq, so no total order exists and the reconstructed grid would place the driver in two statuses at once. Fix: assign a strictly increasing seq to every transition — the device or entry UI should emit it — so same-second events still sort deterministically, exactly as Step 3 enforces.

ValueError: discontinuous status: from=driving but last to=on_duty_not_driving. Root cause: a transition was dropped or reordered, so the new event’s from_status does not match the previous to_status and the status chain has a hole — the classic missing terminal status symptom where a day never returns to off-duty. Fix: append transitions in order and never skip one; if a change is genuinely missing, insert the reconstructed intermediate transition rather than letting the chain break silently.

Driving hours look correct daily but the total for a cross-midnight shift is wrong. Root cause: a driving interval spanning midnight was attributed entirely to its start date, so one calendar day was overstated and the next understated against its 11-hour ceiling. Fix: split every interval at the UTC day boundary before summing, as reconstruct_daily_driving does with its while cursor < end loop, so each day’s grid totals independently.

FAQ

Why store both from_status and to_status when the previous entry already implies it?

Redundancy is the point. Storing from_status lets the append step verify continuity — that this change actually starts where the last one ended — and catch a dropped or reordered transition immediately instead of at audit time. It also means a single entry is self-describing when pulled out of context, so a reviewer reading one line knows the full move without walking the whole chain.

Should midnight be split in UTC or the driver's local time?

The grid is a local-time artifact, but the log stores UTC, so the honest answer is: split at whatever boundary the record of duty status is drawn on, and store the offset used. This implementation splits at UTC midnight for determinism; a production system reconstructing the printed grid applies the driver’s home-terminal time zone under 49 CFR 395.8, splitting at local midnight instead. The mechanism is identical — only the boundary moves.

Up: DOT Hours-of-Service Logging