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.
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:
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
Related
- Logging Driver Duty-Status Transitions — capturing the raw ELD transitions at the event level
- FMCSA ELD Sync — reconciling the RODS against the carrier’s device records
- EPA e-Manifest Integration — the manifests filed for the regulated stops on the same run
- Audit Report Generation — sealing the daily RODS into a tamper-evident report
- DOT/FMCSA Rule Mapping — encoding the same ceilings as solver constraints rather than logs