How to Map DOT Hours of Service to Waste Routes

This page shows how to turn FMCSA Hours-of-Service (HOS) limits into a deterministic Python state machine that consumes ELD telemetry from a packer truck and returns a hard False for any collection segment that would push a driver past the 11-hour drive ceiling, the 14-hour on-duty window, or the mandatory 30-minute break — before that segment ever reaches the routing solver.

A municipal packer truck over 10,001 lb operated in commerce is a commercial motor vehicle under 49 CFR 390.5, so every shift it runs is governed by the same driving-time limits as an interstate carrier. This how-to is the operational companion to the DOT/FMCSA Rule Mapping reference: that page binds HOS into an OR-Tools Time dimension at model-build time; this page builds the per-vehicle clock that decides, segment by segment, whether the current shift still has legal driving left. Get the clock wrong and a mileage-minimizing solver will happily plan a shift that ends with a citation on the curb at 6 a.m.

Environment & Data Prerequisites

This page targets Python 3.10+ — the code uses X | None union syntax that older interpreters reject. HOS tracking needs no third-party solver; the standard library is enough. Pin only what you use for the structured audit log:

python -m pip install python-dateutil==2.9.0

The input is a stream of ELD-derived route segments. Each segment is one telemetry window with the shape below. Units matter: velocities are miles per hour, durations are integer seconds, and timestamps are timezone-aware UTC (datetime). Feeding naive local timestamps here is the single most common cause of a break window that never closes.

The reason the raw feed cannot be trusted directly is that ELDs log idle compaction time as motion. A packer sitting at a stop running its hydraulics registers engine load and jittery sub-3-mph GPS, which many ELDs bucket as driving. Classifying true movement with a velocity gate is the same discipline the Telematics & Sensor Data Ingestion pipeline applies upstream — and the zero-velocity drift problem it solves is documented in full on Validating GPS Coordinates in Python.

Step-by-Step Implementation

Step 1 — Model the immutable segment

Segments are frozen dataclasses: once a telemetry window is recorded it is never mutated, which keeps the audit trail tamper-evident.

from dataclasses import dataclass
from datetime import datetime


@dataclass(frozen=True)
class RouteSegment:
    segment_id: str
    timestamp: datetime          # tz-aware UTC
    velocity_mph: float
    drive_seconds: int
    on_duty_seconds: int
    is_compaction_cycle: bool

Step 2 — Configure the site-standard JSON audit logger

Every HOS decision must land in a machine-readable audit line. The site uses a custom logging.Formatter subclass (not a third-party logger) so the same record shape appears across the whole compliance stack:

import json
import logging
from datetime import datetime, timezone


class JSONLogFormatter(logging.Formatter):
    """Emit one newline-delimited JSON object per audit event."""

    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "event": record.getMessage(),
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
        # Structured fields ride in on `extra={...}`.
        for key, value in getattr(record, "hos", {}).items():
            entry[key] = value
        return json.dumps(entry)


_handler = logging.StreamHandler()
_handler.setFormatter(JSONLogFormatter())
logger = logging.getLogger("hos")
logger.setLevel(logging.INFO)
logger.handlers = [_handler]

Step 3 — Bind the HOS limits into a state machine

The engine holds the federal ceilings as class constants with their CFR citation, applies the velocity gate, and accumulates drive and on-duty time. The break predicate is the rule that trips a shift into a blocked state after eight cumulative driving hours:

$$\text{break\_required} \iff \frac{\sum \text{drive\_seconds}}{3600} \ge 8 \quad\text{(49 CFR 395.3(a)(3)(ii))}$$

from dataclasses import dataclass, field
from datetime import timedelta


@dataclass
class HOSComplianceEngine:
    shift_start: datetime
    cumulative_drive_sec: int = 0
    cumulative_on_duty_sec: int = 0
    last_break_end: datetime | None = None
    break_required: bool = False
    violation_log: list[str] = field(default_factory=list)

    MAX_DRIVE_HOURS: float = 11.0        # 49 CFR 395.3(a)(3)(i)
    MAX_ON_DUTY_HOURS: float = 14.0      # 49 CFR 395.3(a)(2)
    BREAK_THRESHOLD_HOURS: float = 8.0   # 49 CFR 395.3(a)(3)(ii)
    MIN_BREAK_DURATION_MIN: int = 30     # 49 CFR 395.3(a)(3)(ii)
    VELOCITY_THRESHOLD_MPH: float = 3.5

    def _apply_velocity_filter(self, seg: RouteSegment) -> dict[str, int]:
        """Reclassify sub-threshold windows (idle compaction) as non-driving."""
        if seg.velocity_mph < self.VELOCITY_THRESHOLD_MPH:
            return {"drive_sec": 0, "on_duty_sec": seg.on_duty_seconds}
        return {"drive_sec": seg.drive_seconds, "on_duty_sec": seg.on_duty_seconds}

The velocity gate is deliberately conservative: below 3.5 mph the window contributes zero driving seconds but its full on-duty seconds, because a driver packing at a stop is still on duty even though they are not driving.

Step 4 — Solve: process each segment against the clock

process_segment is the decision function the router calls. It returns True when the segment is legal to drive and False when routing must stop and insert a break or terminate the shift. Hard limits are checked first and can never be relaxed:

    def process_segment(self, seg: RouteSegment) -> bool:
        filtered = self._apply_velocity_filter(seg)
        self.cumulative_drive_sec += filtered["drive_sec"]
        self.cumulative_on_duty_sec += filtered["on_duty_sec"]

        drive_h = self.cumulative_drive_sec / 3600.0
        on_duty_h = self.cumulative_on_duty_sec / 3600.0

        if on_duty_h > self.MAX_ON_DUTY_HOURS:
            self.violation_log.append("14-hour on-duty window exceeded")
            logger.warning("hos_violation", extra={"hos": {
                "segment_id": seg.segment_id, "rule": "49 CFR 395.3(a)(2)"}})
            return False

        if drive_h > self.MAX_DRIVE_HOURS:
            self.violation_log.append("11-hour drive limit exceeded")
            logger.warning("hos_violation", extra={"hos": {
                "segment_id": seg.segment_id, "rule": "49 CFR 395.3(a)(3)(i)"}})
            return False

        # A completed 30-minute break clears the flag before we re-test it.
        if (self.break_required and self.last_break_end
                and seg.timestamp >= self.last_break_end
                + timedelta(minutes=self.MIN_BREAK_DURATION_MIN)):
            self.break_required = False
            logger.info("break_compliance_restored", extra={"hos": {
                "segment_id": seg.segment_id}})

        if drive_h >= self.BREAK_THRESHOLD_HOURS and not self.break_required \
                and self.last_break_end is None:
            self.break_required = True
            logger.info("break_threshold_met", extra={"hos": {
                "segment_id": seg.segment_id,
                "cumulative_drive_h": round(drive_h, 2),
                "rule": "49 CFR 395.3(a)(3)(ii)"}})

        if self.break_required:
            logger.warning("routing_blocked", extra={"hos": {
                "segment_id": seg.segment_id,
                "reason": "30-min break required after 8h driving",
                "action": "inject_break_node"}})
            return False

        return True

When the engine returns False because a break is due, the caller inserts a rest node and sets last_break_end to the break start, then resumes. If no compliant continuation exists at all, control passes to the Fallback Routing Logic tier — split the shift, assign a relief driver with a fresh clock, or defer stops — but it never relaxes an HOS limit to force a solution.

Step 5 — Drive the machine with a mock ELD payload

if __name__ == "__main__":
    shift_start = datetime(2024, 6, 15, 5, 0, 0, tzinfo=timezone.utc)
    engine = HOSComplianceEngine(shift_start=shift_start)

    payload = [
        RouteSegment("S-101", shift_start + timedelta(hours=4), 12.4, 14400, 14400, False),
        RouteSegment("S-102", shift_start + timedelta(hours=6), 8.1, 7200, 7200, False),
        RouteSegment("S-103", shift_start + timedelta(hours=8, minutes=15), 5.2, 3600, 3600, True),
        RouteSegment("S-104", shift_start + timedelta(hours=8, minutes=20), 15.3, 1800, 1800, False),
    ]

    for seg in payload:
        ok = engine.process_segment(seg)
        # After S-103 crosses 8h, S-104 is blocked until a 30-min break is logged.
        if not ok and engine.break_required and engine.last_break_end is None:
            engine.last_break_end = seg.timestamp  # scheduler injects the rest node

Compliance Output

The machine emits one newline-delimited JSON object per HOS decision. This is the exact audit record a municipal contract or an FMCSA review expects — every line carries the CFR rule it enforces:

{"event": "break_threshold_met", "timestamp": "2024-06-15T13:15:00+00:00", "segment_id": "S-103", "cumulative_drive_h": 8.0, "rule": "49 CFR 395.3(a)(3)(ii)"}
{"event": "routing_blocked", "timestamp": "2024-06-15T13:20:00+00:00", "segment_id": "S-104", "reason": "30-min break required after 8h driving", "action": "inject_break_node"}

Each field maps to a regulatory purpose:

• event — the decision class an auditor filters on; hos_violation lines are the ones that must never appear on a clean shift.
• rule — the CFR citation the decision enforces (395.3(a)(2) for the 14-hour window, 395.3(a)(3)(i) for the 11-hour drive ceiling, 395.3(a)(3)(ii) for the break), so the log is self-documenting under audit.
• cumulative_drive_h — the driving total at the moment the break flag tripped; it is the evidence that the 8-hour threshold was honored, not guessed.
• action — inject_break_node tells the downstream router what structural change the compliance layer demanded, tying the schedule change back to its cause.
• timestamp — a UTC, ISO-8601 instant. FMCSA’s ELD record-retention rule (49 CFR 395.8(k)) requires these supporting records be kept for six months, and signing/retention is handled by Security & Access Boundaries.

Verification

Confirm the clock blocks the post-threshold segment and that no false violation is recorded. This runs against the mock payload with no external services:

def test_break_blocks_segment_after_eight_hours():
    start = datetime(2024, 6, 15, 5, 0, 0, tzinfo=timezone.utc)
    eng = HOSComplianceEngine(shift_start=start)

    assert eng.process_segment(
        RouteSegment("S-101", start + timedelta(hours=4), 12.4, 14400, 14400, False)) is True
    assert eng.process_segment(
        RouteSegment("S-102", start + timedelta(hours=6), 8.1, 7200, 7200, False)) is True
    # Crosses 8h cumulative driving -> break flag set, segment still legal to finish.
    eng.process_segment(
        RouteSegment("S-103", start + timedelta(hours=8, minutes=15), 5.2, 3600, 3600, True))
    assert eng.break_required is True
    # Next segment is blocked until a 30-min break is logged.
    assert eng.process_segment(
        RouteSegment("S-104", start + timedelta(hours=8, minutes=20), 15.3, 1800, 1800, False)) is False
    assert eng.violation_log == []   # a blocked break is NOT a violation

The key assertion is the last one: a segment blocked for a break is a prevented violation, not a recorded one — violation_log stays empty because the machine stopped the illegal drive before it happened.

Common Errors

TypeError: can't compare offset-naive and offset-aware datetimes — the break-elapsed check compares seg.timestamp against last_break_end. It raises the moment a naive datetime sneaks into the stream. Root cause: an ELD payload parsed without a timezone. Fix: normalize every timestamp to UTC at ingestion (ts.replace(tzinfo=timezone.utc) or parse with dateutil.parser.isoparse) before constructing a RouteSegment.

The break flag never clears and every later segment returns False — symptom of a scheduler that blocks on the break but forgets to set last_break_end. Root cause: the rest node was inserted in the route but the engine’s clock was never told the break started. Fix: whenever the router honors an inject_break_node action, set engine.last_break_end to the break’s start instant; the next segment at least 30 minutes later then re-enables driving.

Idle compaction time inflates cumulative_drive_sec and trips a false 11-hour violation — root cause: bypassing _apply_velocity_filter and adding seg.drive_seconds directly, so sub-3.5-mph hopper-packing windows count as driving. Fix: always route segments through process_segment; never accumulate raw ELD driving seconds without the velocity gate.

Related

• DOT/FMCSA Rule Mapping for Waste Route Optimization — how the same HOS limits bind into an OR-Tools Time dimension at model-build time.
• Fallback Routing Logic for Municipal Waste Operations — the degradation tier invoked when no HOS-compliant continuation exists.
• Route Schema Design for Waste Management Logistics — the data contract the segment fields above conform to.
• Security & Access Boundaries for Waste Route Optimization — signing and six-month retention for the audit lines this machine emits.
• Validating GPS Coordinates in Python — the upstream velocity and drift checks that make the ELD feed trustworthy.

Up: DOT/FMCSA Rule Mapping