DOT Hours-of-Service Logging for Waste Collection Fleets

Record driver duty status and prove compliance with 49 CFR 395 driving and on-duty ceilings.

An hours-of-service violation is rarely discovered on the day it happens — it surfaces weeks later at a roadside inspection or during a compliance review, when an officer reconstructs a driver’s day from the log and finds the truck was still driving past the eleventh hour. The defense is not a better route; it is a log that proves the ceilings held, reconstructable to the minute and reproducible on demand. This is the subsystem that produces that log. It records every duty-status transition a waste-collection driver makes, accumulates those transitions into a daily Record of Duty Status, computes the driving and on-duty totals, and flags any run at the 49 CFR 395 limits before it becomes a citation. What breaks without it is provability: the operation may well have been compliant, but with no defensible record, an inspection that asks “prove it” has no answer.

This is the hours-of-service logging sub-system of the Compliance Reporting & Automation layer, and its scope is deliberately narrow: it is about the log and the report, not the solver constraint. Encoding hours-of-service as a routing bound the optimizer cannot cross — mandatory rest nodes, per-vehicle time ceilings — is a different job, handled under DOT/FMCSA rule mapping. Here the truck has already run; the task is to record what the driver did and prove it stayed legal. Logging the raw transitions themselves at the event level is detailed in logging driver duty-status transitions. Every code block is complete and pasteable into a Python 3.10+ compliance worker.

A driver's Record of Duty Status drawn as the classic ELD four-row grid across a 24-hour day Four horizontal duty-status rows — off duty, sleeper berth, driving, and on duty not driving — span a 24-hour horizontal axis marked at midnight, 6, noon, 18, and midnight. A continuous step line traces the day: off duty until 6, on duty not driving during the pre-trip, then alternating driving and on-duty collection segments through midday, and back to off duty in the evening. A shaded band from the start of driving marks the 14-hour on-duty window, and a caption notes the 11-hour driving cap within it. OFF SB D ON 00:00 06:00 12:00 18:00 24:00 14-hour on-duty window (11-hour driving cap within) accumulated driving checked against the 11-hour ceiling
A day's Record of Duty Status on the classic ELD grid: the step line moves between off duty, sleeper berth, driving, and on-duty-not-driving, while the shaded band marks the 14-hour window and the accumulated driving time is checked against the 11-hour cap.

Prerequisites

Pin the validation stack so a driver-day is reconstructable byte-for-byte during an inspection. This is the version the code below is tested against:

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

Every duty-status change is a typed event before it becomes a log line. The DutyStatusEvent model carries the four regulatory statuses, a UTC timestamp, the odometer reading, and the position — the fields an ELD records at each transition and the ones an inspector cross-checks. Validating them at ingestion means a malformed event (a naive timestamp, an impossible odometer regression) is rejected at the edge rather than corrupting the day’s totals mid-computation.

from __future__ import annotations
from datetime import datetime, timezone
from enum import Enum
from pydantic import BaseModel, Field, field_validator


class DutyStatus(str, Enum):
    """The four duty statuses of 49 CFR 395.8."""
    OFF_DUTY = "off_duty"
    SLEEPER_BERTH = "sleeper_berth"
    DRIVING = "driving"
    ON_DUTY_NOT_DRIVING = "on_duty_not_driving"


class DutyStatusEvent(BaseModel):
    driver_id: str = Field(min_length=1)
    status: DutyStatus
    ts_utc: datetime                       # transition instant, UTC
    odometer_km: float = Field(ge=0)
    lat: float = Field(ge=-90.0, le=90.0)
    lon: float = Field(ge=-180.0, le=180.0)

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

The insistence on a timezone-aware UTC instant is not pedantry. Hours-of-service is computed over intervals, and an interval built from one local timestamp and one UTC timestamp is silently wrong by the offset — which near a midnight boundary can invent or erase an entire duty period. Normalizing at the model boundary is the cheapest place to prevent the clock-skew failure that otherwise surfaces as a phantom violation.

Core Implementation

A Record of Duty Status is the ordered sequence of a driver’s status changes over one day, and the compliance question is answered by accumulating the time spent in each status between consecutive events. Driving time and on-duty time each roll up against their own ceiling; the on-duty window is measured from the first on-duty event, not from the accumulated on-duty seconds. The class below builds the RODS, computes the totals, and flags violations of the three ceilings that govern a property-carrying driver.

Note that Python’s standard logging module has no built-in JSON formatter — the structured logging here uses a small custom class with hours-of-service-specific fields.

import json
import logging
from dataclasses import dataclass, field
from datetime import timedelta
from typing import Optional

logger = logging.getLogger("hos")
logger.setLevel(logging.INFO)

# 49 CFR 395.3 ceilings for a property-carrying driver.
DRIVING_CAP = timedelta(hours=11)      # max driving in a duty period
ON_DUTY_WINDOW = timedelta(hours=14)   # window in which driving must occur
WEEKLY_CAP_7DAY = timedelta(hours=60)  # 60 hours / 7 days
WEEKLY_CAP_8DAY = timedelta(hours=70)  # 70 hours / 8 days


@dataclass
class Violation:
    rule: str          # "driving_11h" | "window_14h" | "weekly_60_70h"
    citation: str      # the CFR paragraph
    limit_seconds: int
    actual_seconds: int


@dataclass
class DailyRODS:
    driver_id: str
    date: str
    driving: timedelta = timedelta()
    on_duty: timedelta = timedelta()          # driving + on_duty_not_driving
    window_span: timedelta = timedelta()      # first on-duty -> last on-duty
    violations: list[Violation] = field(default_factory=list)


def build_daily_rods(driver_id: str, date: str,
                     events: list[DutyStatusEvent],
                     rolling_7day: timedelta = timedelta(),
                     use_8day: bool = False) -> DailyRODS:
    """Accumulate one driver-day into a RODS and flag 49 CFR 395.3 violations."""
    ordered = sorted(events, key=lambda e: e.ts_utc)
    rods = DailyRODS(driver_id=driver_id, date=date)

    first_on_duty: Optional[datetime] = None
    last_on_duty_end: Optional[datetime] = None

    for cur, nxt in zip(ordered, ordered[1:]):
        span = nxt.ts_utc - cur.ts_utc
        if cur.status == DutyStatus.DRIVING:
            rods.driving += span
        if cur.status in (DutyStatus.DRIVING, DutyStatus.ON_DUTY_NOT_DRIVING):
            rods.on_duty += span
            if first_on_duty is None:
                first_on_duty = cur.ts_utc
            last_on_duty_end = nxt.ts_utc

    if first_on_duty and last_on_duty_end:
        rods.window_span = last_on_duty_end - first_on_duty

    _flag_violations(rods, rolling_7day, use_8day)
    logger.info("rods_built", extra={
        "driver_id": driver_id, "date": date,
        "driving_s": int(rods.driving.total_seconds()),
        "on_duty_s": int(rods.on_duty.total_seconds()),
        "window_s": int(rods.window_span.total_seconds()),
        "violations": [v.rule for v in rods.violations]})
    return rods


def _flag_violations(rods: DailyRODS, rolling_7day: timedelta,
                     use_8day: bool) -> None:
    if rods.driving > DRIVING_CAP:
        rods.violations.append(Violation(
            "driving_11h", "49 CFR 395.3(a)(3)(i)",
            int(DRIVING_CAP.total_seconds()), int(rods.driving.total_seconds())))
    if rods.window_span > ON_DUTY_WINDOW:
        rods.violations.append(Violation(
            "window_14h", "49 CFR 395.3(a)(2)",
            int(ON_DUTY_WINDOW.total_seconds()), int(rods.window_span.total_seconds())))
    weekly_cap = WEEKLY_CAP_8DAY if use_8day else WEEKLY_CAP_7DAY
    weekly_total = rolling_7day + rods.on_duty
    if weekly_total > weekly_cap:
        rods.violations.append(Violation(
            "weekly_60_70h", "49 CFR 395.3(b)",
            int(weekly_cap.total_seconds()), int(weekly_total.total_seconds())))

Two subtleties are worth calling out. First, the 14-hour rule constrains a window, not a sum: once a driver goes on duty, all driving must finish within 14 elapsed hours regardless of breaks taken inside it, which is why window_span is measured from the first to the last on-duty instant rather than by adding on-duty segments. Second, the 60-/70-hour ceiling is a rolling figure across the prior week, so this day’s on-duty total is checked against the carried rolling sum — the reporting layer keeps that rolling state per driver across days.

The short-haul exception

Municipal collection routes frequently qualify for the short-haul exception at 49 CFR 395.1(e), which relieves a driver who starts and returns to the same work-reporting location within a bounded time and distance from keeping a full paper-style RODS. The exception does not remove the underlying ceilings — the driver still may not exceed the driving and window limits — but it changes what must be logged, so the subsystem records whether each driver-day claimed it and why it qualified.

SHORT_HAUL_RADIUS_KM = 241.0        # ~150 air-miles, 49 CFR 395.1(e)(1)
SHORT_HAUL_WINDOW = timedelta(hours=14)


def qualifies_short_haul(events: list[DutyStatusEvent],
                         home_lat: float, home_lon: float) -> bool:
    """Record whether a driver-day meets the 49 CFR 395.1(e)(1) short-haul terms.
    Ceilings still apply; only the RODS record-keeping burden changes."""
    from math import radians, sin, cos, asin, sqrt
    ordered = sorted(events, key=lambda e: e.ts_utc)
    if not ordered:
        return False
    span = ordered[-1].ts_utc - ordered[0].ts_utc
    if span > SHORT_HAUL_WINDOW:
        return False
    for e in ordered:
        dlat, dlon = radians(e.lat - home_lat), radians(e.lon - home_lon)
        a = (sin(dlat / 2) ** 2 + cos(radians(home_lat)) *
             cos(radians(e.lat)) * sin(dlon / 2) ** 2)
        if 2 * 6371.0 * asin(sqrt(a)) > SHORT_HAUL_RADIUS_KM:
            return False
    return True

The air-mile radius that defines short-haul eligibility is a great-circle distance from the work-reporting location, the Haversine formula:

d=2rarcsin ⁣sin2 ⁣(Δφ2)+cosφ1cosφ2sin2 ⁣(Δλ2)d = 2r\,\arcsin\!\sqrt{\sin^2\!\left(\tfrac{\Delta\varphi}{2}\right) + \cos\varphi_1\cos\varphi_2\sin^2\!\left(\tfrac{\Delta\lambda}{2}\right)}

The constraint side of these same rules — turning the ceilings into solver bounds and mandatory rest nodes so a route is never planned to violate them — is handled under DOT/FMCSA rule mapping, and the resulting logs are reconciled against the carrier’s device records by the FMCSA ELD Sync subsystem. A day’s HOS record and the manifests filed for its regulated stops are sealed together, tying this log to the EPA e-Manifest Integration output for the same run.

Regulatory Mapping

Each element of the model and the violation checks traces to a paragraph of 49 CFR Part 395, so the compliance registry can version exactly the rule an inspector cites.

Model / check Regulatory source What it governs
DailyRODS, four DutyStatus values 49 CFR 395.8 The Record of Duty Status and its duty-status categories
driving_11h violation 49 CFR 395.3(a)(3)(i) 11-hour driving limit within a duty period
window_14h violation 49 CFR 395.3(a)(2) 14-hour on-duty window in which driving must occur
weekly_60_70h violation 49 CFR 395.3(b) 60-hour/7-day and 70-hour/8-day on-duty ceilings
qualifies_short_haul 49 CFR 395.1(e)(1) Short-haul exception common to municipal collection
Device retention & transfer 49 CFR 395.20–395.38 ELD mandate, record retention, and roadside data transfer

The ELD-mandate paragraphs at 49 CFR 395.20–395.38 are why this log cannot be a private spreadsheet: the records must be retained (six months) and transferable on demand at a roadside inspection, which is the retention and availability obligation the parent Compliance Reporting & Automation layer enforces through its WORM store.

Validation & Verification

The one test that matters is that a violation actually fires — a checker that silently never flags is worse than none, because it manufactures false assurance. The test below constructs a day that drives twelve hours and asserts the 11-hour driving violation is raised with the correct citation.

import pytest
from datetime import datetime, timezone, timedelta


def _ev(driver: str, status: DutyStatus, hour: float,
        odo: float) -> DutyStatusEvent:
    base = datetime(2026, 7, 16, tzinfo=timezone.utc)
    return DutyStatusEvent(driver_id=driver, status=status,
                           ts_utc=base + timedelta(hours=hour),
                           odometer_km=odo, lat=40.0, lon=-83.0)


def test_driving_violation_fires() -> None:
    # 06:00 start driving, straight through to 18:00 -> 12 hours driving.
    events = [
        _ev("D-100", DutyStatus.ON_DUTY_NOT_DRIVING, 5.5, 1000),
        _ev("D-100", DutyStatus.DRIVING, 6.0, 1000),
        _ev("D-100", DutyStatus.OFF_DUTY, 18.0, 1620),
    ]
    rods = build_daily_rods("D-100", "2026-07-16", events)
    rules = {v.rule: v for v in rods.violations}
    assert "driving_11h" in rules, "12h of driving must breach the 11h cap"
    assert rules["driving_11h"].citation == "49 CFR 395.3(a)(3)(i)"
    assert rules["driving_11h"].actual_seconds == 12 * 3600


def test_compliant_day_has_no_violations() -> None:
    events = [
        _ev("D-101", DutyStatus.ON_DUTY_NOT_DRIVING, 6.0, 2000),
        _ev("D-101", DutyStatus.DRIVING, 6.5, 2000),
        _ev("D-101", DutyStatus.OFF_DUTY, 14.0, 2300),  # 7.5h driving, 8h window
    ]
    rods = build_daily_rods("D-101", "2026-07-16", events)
    assert rods.violations == []

In production, the rods_built structured log carries driving_s, on_duty_s, window_s, and the list of violation rules — the four fields an auditor reads to confirm the day stayed under the ceilings, and the record that proves the checker ran.

Failure Modes & Edge Cases

Overlapping or duplicate events. Two events with the same timestamp, or a device that re-sends a transition, would double-count an interval. Sort by timestamp and collapse exact duplicates before accumulation; a zero-length interval between identical statuses contributes nothing, but a duplicated transition must not restart the window.

def dedupe_events(events: list[DutyStatusEvent]) -> list[DutyStatusEvent]:
    """Drop exact-duplicate transitions before RODS accumulation."""
    seen: set[tuple] = set()
    out: list[DutyStatusEvent] = []
    for e in sorted(events, key=lambda x: x.ts_utc):
        key = (e.driver_id, e.status, e.ts_utc)
        if key not in seen:
            seen.add(key)
            out.append(e)
    return out

Missing status change. A device drops the transition that ends a driving segment, so the interval runs to the next event and overstates one status. Detect an implausibly long single interval and flag the driver-day for manual review rather than silently attributing hours the driver may not have worked.

Clock skew. A transition stamped in device-local time on a truck whose clock drifted pushes an interval across a boundary it never crossed. Because the model rejects naive timestamps and normalizes to UTC, a skewed-but-aware event is at least comparable; a monotonicity check catches the rest — a later event whose timestamp precedes an earlier one is rejected before it can invent a negative interval.

Midnight boundary. A duty period that spans midnight belongs partly to two calendar days, and splitting it wrong either double-counts or drops the crossing interval. Attribute each interval to the day of its start event consistently, and let the rolling weekly sum span days so the 60-/70-hour ceiling is measured across the boundary rather than reset by it.

Integration Checklist

Complete every item before an HOS logging subsystem reaches production:

FAQ

Is this the same as the hours-of-service routing constraint?

No. This subsystem records what a driver actually did and proves the ceilings held after the fact — it is the log and the report. Preventing a route from being planned to violate the ceilings, by encoding rest nodes and per-vehicle time bounds the solver cannot cross, is a separate concern handled under DOT/FMCSA rule mapping. One proves compliance; the other enforces it in the plan.

Why is the 14-hour rule a window instead of a running total?

Because 49 CFR 395.3(a)(2) caps the elapsed time from going on duty to the last permitted driving, not the sum of on-duty segments. Breaks taken inside the window do not extend it — the clock keeps running. That is why the code measures the span from the first to the last on-duty instant rather than adding up on-duty intervals.

Does the short-haul exception mean the driving limits do not apply?

No. The 49 CFR 395.1(e) exception relieves qualifying drivers of some record-keeping burden — a full graph-grid RODS — but the 11-hour driving and 14-hour window ceilings still apply. The subsystem records that a day claimed the exception and why it qualified, and it keeps checking the ceilings regardless.

How should a duty period that crosses midnight be logged?

Attribute each interval to the day its start event falls on, and let the rolling weekly on-duty sum span days rather than reset at midnight. That keeps the 60-/70-hour ceiling measured across the boundary and avoids either double-counting the crossing interval or dropping it, which is a common source of phantom or missed violations.

Up: Compliance Reporting & Automation