VRP Route Optimization Algorithms

Handling Truck Capacity Constraints in Python

Production patterns for binding capacity dimensions in OR-Tools without solver divergence.

Your solver keeps aborting a route mid-shift because a single noisy scale reading spikes 900 kg over the ceiling — this page shows how to filter that telemetry, enforce payload and per-axle limits as hard boundaries, and log every check so the numbers reconcile at the scale house.

The parent guide, Capacity & Weight Limits, covers how to bind payload as a hard OR-Tools capacity dimension inside the VRP Route Optimization Algorithms pipeline. This deep dive handles the part that sits upstream of the solver: turning raw, jittery load-cell readings into a stable weight the constraint model can trust, and splitting that weight across axle groups the way federal law requires. Every block below is complete and pasteable into a Python 3.10+ dispatch worker.

Environment & Data Prerequisites

Pin the ingestion and validation stack so a capacity check is byte-for-byte reproducible during an audit:

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

The raw payload arriving from an on-board weighing system is one reading per collection stop. Validate its shape before any filtering runs — a None mass or a negative axle load is the most common cause of a silently corrupt capacity decision. The incoming record has this exact shape:

Field Unit Expected range Meaning
truck_id string fleet asset tag Vehicle emitting the reading
raw_kg kg 0 – 30,000 Unfiltered gross payload from the load cell
front_axle_kg kg 0 – 12,000 Steer-axle share of the load
rear_axle_kg kg 0 – 20,000 Tandem drive-axle share
stop_id string stop identifier Service node just completed
timestamp_ms int (epoch ms) > 0 Reading time for audit ordering 
from __future__ import annotations
from pydantic import BaseModel, Field


class LoadCellReading(BaseModel):
    """A single on-board weighing payload, validated at ingestion."""
    truck_id: str = Field(min_length=1)
    raw_kg: float = Field(ge=0.0, le=30_000.0)
    front_axle_kg: float = Field(ge=0.0, le=12_000.0)
    rear_axle_kg: float = Field(ge=0.0, le=20_000.0)
    stop_id: str = Field(min_length=1)
    timestamp_ms: int = Field(gt=0)

Rejecting out-of-range values here means the filter and constraint logic downstream never has to defend against physically impossible input — the same discipline the Telematics Sensor Data Ingestion pipeline applies before any reading reaches the routing matrix.

Step-by-Step Implementation

The flow is four stages: validate the reading, stabilize it, project it against the hard ceiling, then apportion it across axles. The stages compose into one enforcement cycle at the end.

Per-stop capacity enforcement cycle A horizontal pipeline processes one on-board weighing reading. The raw load-cell reading passes to Pydantic validation, which rejects out-of-range values. Valid readings enter a recursive EMA filter: while fewer than min_samples readings have accumulated the stage branches downward to a WARMUP DEFER outcome, otherwise the steady-state smoothed value continues to axle apportionment. Axle apportionment checks the steer, tandem, and gross loads against 23 CFR 658.17 limits — a steer axle at most 9,072 kg and a tandem group at most 15,422 kg; any breach branches down to a DEPOT_REDIRECT outcome. Compliant loads reach threshold projection, which lazily sums remaining stop weights against the GVWR-derived ceiling times a 0.98 buffer. If the projected load reaches or exceeds that ceiling the route branches down to a FORCED_DEPOT_RETURN that hands off the remaining stops; otherwise it resolves to CONTINUE_ROUTE. Per-stop capacity enforcement cycle 23 CFR 658.17 · axle group limits GVWR ceiling · buffer_pct 0.98 record valid steady Sₜ compliant Load-cell reading on-board weighing raw_kg · axle_kg Pydantic validate LoadCellReading reject out-of-range EMA stabilize recursive smoothing O(1) per truck Axle apportion steer · tandem · gross GVWR Threshold project lazy stop sum × buffer 0.98 count < min_samples axle > CFR limit load ≥ 0.98·max load < ceiling WARMUP action: DEFER DEPOT_REDIRECT axle violation FORCED_DEPOT_RETURN hand off remaining stops CONTINUE_ROUTE constraint valid

Step 1 — Stabilize the load-cell stream

Municipal scale sensors produce noisy weight streams with calibration drift and compaction-induced spikes. Feeding that raw signal to a hard ceiling causes premature route termination. A recursive exponential moving average (EMA) suppresses transient noise with O(1) memory per asset and no phase lag. The steady-state update for smoothing factor α\alpha is:

St=α xt+(1−α) St−1S_t = \alpha\,x_t + (1 - \alpha)\,S_{t-1}

where xtx_t is the raw reading and St−1S_{t-1} the prior smoothed value. A short warm-up window uses a simple running mean until enough samples accumulate to trust the recursion.

from typing import Dict, Optional


class LoadCellFilter:
    """Stateful EMA filter for streaming load-cell telemetry."""

    def __init__(self, alpha: float = 0.35, min_samples: int = 5):
        self.alpha = alpha
        self.min_samples = min_samples
        self._state: Dict[str, Dict[str, float]] = {}

    def update(self, truck_id: str, raw_kg: float) -> Optional[float]:
        if truck_id not in self._state:
            self._state[truck_id] = {"ema": raw_kg, "count": 0}

        state = self._state[truck_id]
        state["count"] += 1

        if state["count"] < self.min_samples:
            # Warm-up: running mean until enough samples are collected.
            state["ema"] = (state["ema"] * (state["count"] - 1) + raw_kg) / state["count"]
            return None

        state["ema"] = self.alpha * raw_kg + (1 - self.alpha) * state["ema"]
        return round(state["ema"], 1)

    def reset(self, truck_id: str) -> None:
        self._state.pop(truck_id, None)

The filter returns None during warm-up so callers explicitly defer a capacity decision rather than acting on an unsettled estimate. Call reset when a truck empties at a transfer station so the next load starts a fresh average. Tuning alpha at runtime against seasonal density swings is the job of the Dynamic Threshold Tuning layer.

Step 2 — Project remaining stops against the hard ceiling

Capacity is an immutable state transition along the route: once the projected running load crosses the ceiling, the plan is invalid. Building the projection with a generator keeps RAM flat during solver initialization on municipal-scale node graphs, because stop weights are consumed lazily instead of materialized into a list.

from typing import Iterator, Tuple


def validate_remaining_capacity(
    remaining_stops: Iterator[float],
    current_load_kg: float,
    max_capacity_kg: float,
    buffer_pct: float = 0.98,
) -> Tuple[bool, str, float]:
    """Lazily evaluate a stop sequence against the hard payload ceiling."""
    projected_load = current_load_kg
    threshold = max_capacity_kg * buffer_pct

    for stop_weight in remaining_stops:
        projected_load += stop_weight
        if projected_load > threshold:
            return False, "CAPACITY_THRESHOLD_EXCEEDED", round(projected_load, 1)

    return True, "VALID", round(projected_load, 1)

The buffer_pct of 0.98 reserves a two-percent margin so a truck never leaves a stop sitting exactly on its GVWR-derived limit — a scale-house re-weigh with a slightly heavier calibration would otherwise flip a legal load into a citation.

Step 3 — Apportion the load across axle groups

Regulatory frameworks govern per-axle distribution, not just gross vehicle weight, so the engine maintains separate counters for the steer and tandem drive groups. Under 23 U.S.C. § 127(a) and 23 CFR 658.17, a single steer axle is capped at 20,000 lb (9,072 kg) and a tandem drive-axle group at 34,000 lb (15,422 kg); the aggregate is bounded by the vehicle’s GVWR. Municipal ordinances routinely post lower limits on residential streets, so treat these federal numbers as the upper envelope and resolve the effective ceiling against local overrides. A breach must branch the route immediately.

def validate_axle_compliance(
    front_load_kg: float,
    rear_load_kg: float,
    config: dict,
) -> Tuple[bool, str]:
    """Check steer, tandem, and gross ceilings against a jurisdiction config."""
    max_single = config.get("axle_limit_kg", 9072)     # 20,000 lb, 23 CFR 658.17(b)
    max_tandem = config.get("tandem_limit_kg", 15422)  # 34,000 lb, 23 CFR 658.17(c)
    gross = front_load_kg + rear_load_kg

    if front_load_kg > max_single:
        return False, "FRONT_AXLE_EXCEEDED"
    if rear_load_kg > max_tandem:
        return False, "REAR_AXLE_EXCEEDED"
    if gross > config.get("gvwr_kg", 26308):
        return False, "GVWR_EXCEEDED"
    return True, "COMPLIANT"

Step 4 — Wire the enforcement cycle with structured logging

The cycle stitches the three stages together and serializes each decision through the site-wide JSONFormatter pattern — a custom formatter class, not a stdlib config call — so compliance events land in the audit database as machine-parseable records. When a check fails, the cycle hands control to the depot-redirect path documented in Fallback Routing Logic.

import json
import logging
from dataclasses import dataclass, asdict
from typing import Any, Dict


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_record = {
            "ts": self.formatTime(record),
            "level": record.levelname,
            "event": record.getMessage(),
            "truck_id": getattr(record, "truck_id", None),
            "event_type": getattr(record, "event_type", None),
            "compliance": getattr(record, "compliance", None),
        }
        return json.dumps(log_record, separators=(",", ":"))


logger = logging.getLogger("capacity_engine")
_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
logger.addHandler(_handler)
logger.setLevel(logging.INFO)


@dataclass
class StopPayload:
    truck_id: str
    raw_kg: float
    front_axle_kg: float
    rear_axle_kg: float
    stop_id: str
    timestamp_ms: int


def process_collection_cycle(
    payload: StopPayload,
    filter_engine: LoadCellFilter,
    route_state: Dict[str, Any],
    compliance_config: Dict[str, int],
) -> Dict[str, Any]:
    """One end-to-end capacity decision: stabilize, apportion, enforce."""

    # 1. Stabilize telemetry; defer until the filter warms up.
    filtered_kg = filter_engine.update(payload.truck_id, payload.raw_kg)
    if filtered_kg is None:
        return {"status": "WARMUP", "action": "DEFER"}

    route_state["current_load"] = filtered_kg

    # 2. Axle apportionment check.
    axle_ok, axle_msg = validate_axle_compliance(
        payload.front_axle_kg, payload.rear_axle_kg, compliance_config
    )

    # 3. Hard payload threshold.
    threshold_exceeded = (
        route_state["current_load"] >= route_state["max_capacity"] * 0.98
    )

    extra = {
        "truck_id": payload.truck_id,
        "event_type": "CAPACITY_CHECK",
        "compliance": {"axle_status": axle_msg, "threshold_hit": threshold_exceeded},
    }

    if not axle_ok:
        logger.warning("axle_violation", extra=extra)
        return {"status": "COMPLIANCE_FAIL", "action": "DEPOT_REDIRECT", "reason": axle_msg}

    if threshold_exceeded:
        logger.info("capacity_threshold_reached", extra=extra)
        route_state["fallback_triggered"] = True
        return {
            "status": "THRESHOLD_EXCEEDED",
            "action": "FORCED_DEPOT_RETURN",
            "remaining_stops": route_state.get("unassigned_stops", []),
        }

    logger.info("constraint_validated", extra=extra)
    return {"status": "VALID", "action": "CONTINUE_ROUTE"}


if __name__ == "__main__":
    raw = LoadCellReading(
        truck_id="WM-8842", raw_kg=14250.0, front_axle_kg=4100.0,
        rear_axle_kg=10200.0, stop_id="STOP-7741", timestamp_ms=1709482100000,
    )
    mock_payload = StopPayload(**raw.model_dump())

    engine = LoadCellFilter(alpha=0.35, min_samples=1)  # min_samples=1 for a single-shot demo
    state = {"current_load": 13800.0, "max_capacity": 14500.0,
             "unassigned_stops": ["STOP-7742", "STOP-7743"]}
    config = {"axle_limit_kg": 9072, "tandem_limit_kg": 15422, "gvwr_kg": 26308}

    result = process_collection_cycle(mock_payload, engine, state, config)
    print(json.dumps(result, indent=2))

Compliance Output

The mock execution above emits one structured audit line plus the returned decision. Each field carries a regulatory purpose an auditor can trace:

{
  "ts": "2026-07-02 06:31:41,004",
  "level": "INFO",
  "event": "capacity_threshold_reached",
  "truck_id": "WM-8842",
  "event_type": "CAPACITY_CHECK",
  "compliance": {"axle_status": "COMPLIANT", "threshold_hit": true}
}
  • truck_id — ties the reading to a specific asset for the DOT/FMCSA vehicle record and the weighbridge reconciliation.
  • event_type — classifies the audit entry so a compliance query can isolate every CAPACITY_CHECK across a shift.
  • compliance.axle_status — the per-axle verdict against 23 CFR 658.17; COMPLIANT means neither steer nor tandem group breached its statutory ceiling.
  • compliance.threshold_hit — records that the truck reached its GVWR-derived payload limit, justifying the forced disposal return in the decision payload.

The returned decision (FORCED_DEPOT_RETURN with the remaining stops) is what the dispatcher hands to the re-routing layer, and the remaining_stops list is the exact hand-off the Fallback Routing Logic consumes to re-assign uncollected work.

Verification

Prove all three behaviors — warm-up deferral, axle rejection, and threshold firing — with one pytest module against the example data:

def test_capacity_cycle_behaviors():
    config = {"axle_limit_kg": 9072, "tandem_limit_kg": 15422, "gvwr_kg": 26308}

    # Warm-up defers before min_samples is reached.
    warm = LoadCellFilter(alpha=0.35, min_samples=5)
    p = StopPayload("WM-1", 5000.0, 2000.0, 3000.0, "S1", 1)
    assert process_collection_cycle(p, warm, {"max_capacity": 20000.0}, config)["action"] == "DEFER"

    # Axle overload redirects to the depot regardless of gross headroom.
    hot = LoadCellFilter(min_samples=1)
    over = StopPayload("WM-2", 9500.0, 9500.0, 0.0, "S2", 2)
    r = process_collection_cycle(over, hot, {"max_capacity": 26000.0}, config)
    assert r["action"] == "DEPOT_REDIRECT" and r["reason"] == "FRONT_AXLE_EXCEEDED"

    # A full truck forces a disposal return.
    full = LoadCellFilter(min_samples=1)
    p3 = StopPayload("WM-3", 14400.0, 4200.0, 10200.0, "S3", 3)
    r3 = process_collection_cycle(p3, full, {"max_capacity": 14500.0}, config)
    assert r3["action"] == "FORCED_DEPOT_RETURN"

All three assertions pass on the code above; wire the module into CI so a regression in the filter or the axle logic fails the build before it reaches dispatch.

Common Errors

ModuleNotFoundError: No module named 'pydantic' — the validation schema in Step 0 imports Pydantic v2. Root cause: the worker image was built without the pinned dependency. Fix: pip install pydantic==2.9.2; a v1 install will also break because model_dump replaced dict(), so pin the major version explicitly.

TypeError: '>=' not supported between instances of 'NoneType' and 'float' — a caller passed the filter’s warm-up return straight into a comparison. Root cause: LoadCellFilter.update returns None until min_samples readings accumulate, and the calling code skipped the if filtered_kg is None guard. Fix: honor the WARMUP/DEFER branch and never compare an unsettled estimate against the ceiling.

A truck passes the gross check but is still overloaded on one axle — the plan looked legal because only raw_kg was validated. Root cause: gross weight can sit under GVWR while the steer axle exceeds 9,072 kg, which validate_axle_compliance catches but a gross-only check misses. Fix: always run the axle apportionment step, and confirm your load cells report kilograms — mixing a pounds reading into a kilogram limit understates the load by a factor of 2.2 and silently defeats the ceiling.

Up: Capacity & Weight Limits