OR-Tools Implementation for Waste Collection Routing

A waste-routing engine is only as trustworthy as the solver call at its centre. Google OR-Tools will happily return a plan from a distance matrix that contains a stray NaN, from a demand vector that silently coerced a null to zero, or from a search that used a fresh random seed on every run — and each of those plans looks authoritative right up until an inspector asks why yesterday’s route cannot be reproduced. Without a disciplined integration, the failure is not a crash; it is a confident, unauditable answer that puts an overloaded truck on a posted street or misses a contracted service window.

This page is the solver-integration sub-system of the VRP Route Optimization Algorithms pipeline. It shows how to turn raw service data into a validated OR-Tools model, bind capacity and time as hard dimensions, configure a deterministic search, and emit a compliance record the audit database can trust. Each numeric bound traces to a federal regulation, and every code block is complete and pasteable into a Python 3.10+ dispatch worker. For the one-command environment bootstrap that precedes it, follow setting up Google OR-Tools for waste collection.

Prerequisites

Pin the solver and validation stack so a dispatch run is byte-for-byte reproducible during an audit. These versions are what the code below is tested against:

python --version   # 3.10 or newer
pip install \
  ortools==9.11.4210 \
  pydantic==2.9.2 \
  numpy==2.1.2

Before a single index reaches the solver, every service stop must pass a typed schema. An under-specified node — a None coordinate, a negative demand, a service window whose end precedes its start — is the most common cause of silent solver corruption, so it must be rejected at ingestion rather than discovered mid-solve. The ServiceNode model below is the shared contract this whole pipeline consumes; it mirrors the stricter route contract defined in Route Schema Design and validates units and ordering up front.

from __future__ import annotations
from pydantic import BaseModel, Field, field_validator, model_validator

# Solver arithmetic is integer-only. Carry weight in kilograms and time in
# whole minutes as ints so no floating-point drift accumulates across
# thousands of node insertions during the search.
KG_PER_TONNE = 1000


class ServiceNode(BaseModel):
    node_id: int = Field(ge=0)
    lat: float = Field(ge=-90.0, le=90.0)
    lon: float = Field(ge=-180.0, le=180.0)
    demand_kg: int = Field(ge=0, le=15_000)      # single-stop physical ceiling
    service_start_min: int = Field(ge=0, le=1_440)
    service_end_min: int = Field(ge=0, le=1_440)
    bin_type: str

    @field_validator("demand_kg")
    @classmethod
    def reject_bool_footgun(cls, v: int) -> int:
        if isinstance(v, bool):  # bool is an int subclass; catch it early
            raise ValueError("demand_kg must be an integer count of kilograms")
        return v

    @model_validator(mode="after")
    def window_must_be_ordered(self) -> "ServiceNode":
        if self.service_end_min < self.service_start_min:
            raise ValueError(
                f"node {self.node_id}: service_end_min precedes service_start_min"
            )
        return self

Validating coordinates and window ordering at this boundary is what lets every downstream stage assume its inputs are already sane. The same discipline governs the upstream feed: sensor rows are normalised by the Telematics Sensor Data Ingestion pipeline before they are ever shaped into a ServiceNode.

Core Implementation

The integration is four bindings in a fixed order: validate the matrices, construct the index manager and model, register the constraint dimensions, then run a deterministic search. The solver minimises total arc cost across every vehicle route $r_v$,

$$\min \sum_{v \in V} \sum_{(i,j)\, \in\, r_v} c_{ij}$$

subject to the capacity and time dimensions bound below. The objective is cost; the dimensions are what keep the cheapest plan legal.

Step 1 — Validate the precomputed matrix

Distance and duration matrices are precomputed with OSRM or Valhalla to remove network latency from the solve. A corrupted matrix does not raise inside OR-Tools — it propagates as undefined behaviour through constraint propagation — so validate it before the manager is ever constructed.

import numpy as np


def validate_matrix(matrix: np.ndarray, node_count: int) -> None:
    """Reject a matrix that would produce undefined solver behaviour."""
    if matrix.shape != (node_count, node_count):
        raise ValueError(
            f"matrix shape {matrix.shape} != expected ({node_count}, {node_count})"
        )
    if np.any(np.isnan(matrix)) or np.any(np.isinf(matrix)):
        raise RuntimeError("matrix contains NaN/inf; abort solver initialisation")
    if np.any(matrix < 0):
        raise ValueError("negative travel cost detected; verify routing topology")

Step 2 — Build the index manager and model

The RoutingIndexManager maps solver-internal indices to your node ids and pins the depot; the RoutingModel owns the constraints and the search. Register the arc-cost callback first so the objective above is defined before any dimension is attached.

from ortools.constraint_solver import pywrapcp, routing_enums_pb2


def build_model(
    nodes: list[ServiceNode],
    distance_matrix: np.ndarray,
    num_vehicles: int,
    depot_index: int = 0,
) -> tuple[pywrapcp.RoutingModel, pywrapcp.RoutingIndexManager]:
    validate_matrix(distance_matrix, len(nodes))
    if num_vehicles < 1:
        raise ValueError("fleet must contain at least one vehicle")

    manager = pywrapcp.RoutingIndexManager(len(nodes), num_vehicles, depot_index)
    routing = pywrapcp.RoutingModel(manager)

    def arc_cost(from_index: int, to_index: int) -> int:
        i = manager.IndexToNode(from_index)
        j = manager.IndexToNode(to_index)
        return int(distance_matrix[i][j])

    transit_idx = routing.RegisterTransitCallback(arc_cost)
    routing.SetArcCostEvaluatorOfAllVehicles(transit_idx)
    return routing, manager

Step 3 — Bind capacity and time as hard dimensions

Capacity is a dimension that accumulates demand along a route and caps it per vehicle; a slack_max of 0 makes the ceiling hard, so an over-payload plan is pruned from the search rather than penalised after the fact. This is the same dimension pattern the Capacity & Weight Limits sub-system documents in depth.

def add_capacity_dimension(
    routing: pywrapcp.RoutingModel,
    manager: pywrapcp.RoutingIndexManager,
    nodes: list[ServiceNode],
    vehicle_capacities: list[int],
) -> None:
    def demand_callback(from_index: int) -> int:
        return nodes[manager.IndexToNode(from_index)].demand_kg

    demand_idx = routing.RegisterUnaryTransitCallback(demand_callback)
    routing.AddDimensionWithVehicleCapacity(
        demand_idx,
        0,                    # zero slack: no phantom headroom
        vehicle_capacities,   # hard per-vehicle payload ceiling (kg)
        True,                 # start cumul at zero (empty hopper at depot)
        "Payload",
    )

Time is a second dimension. Registering it with bounded slack lets the solver absorb traffic variance, while CumulVar(index).SetRange(...) clamps each stop to its permitted service hours — the mechanism the Time Window Constraints sub-system builds on. Attaching a disjunction penalty lets an unreachable stop be dropped at a known cost instead of forcing the whole plan infeasible.

def add_time_dimension(
    routing: pywrapcp.RoutingModel,
    manager: pywrapcp.RoutingIndexManager,
    nodes: list[ServiceNode],
    time_matrix: np.ndarray,
    slack_max: int = 60,      # minutes of buffer for traffic variance
    horizon: int = 1_440,     # 24h planning window, in minutes
    drop_penalty: int = 10_000,
) -> None:
    def time_callback(from_index: int, to_index: int) -> int:
        i = manager.IndexToNode(from_index)
        j = manager.IndexToNode(to_index)
        return int(time_matrix[i][j])

    time_idx = routing.RegisterTransitCallback(time_callback)
    routing.AddDimension(time_idx, slack_max, horizon, False, "Time")
    time_dim = routing.GetDimensionOrDie("Time")

    for node in nodes:
        if node.node_id == 0:          # depot has no service window
            continue
        index = manager.NodeToIndex(node.node_id)
        time_dim.CumulVar(index).SetRange(
            node.service_start_min, node.service_end_min
        )
        routing.AddDisjunction([index], drop_penalty)

Step 4 — Configure a deterministic search and solve

Production dispatch must be reproducible: two runs on identical input must return the identical plan, or the audit trail is meaningless. A fixed random_seed plus a hard time_limit is what guarantees that. Wrap the solve so a None return — an unsatisfiable model — is surfaced as an explicit signal rather than an opaque failure.

import logging

logger = logging.getLogger("ortools.solve")


def configure_search() -> pywrapcp.DefaultRoutingSearchParameters:
    params = pywrapcp.DefaultRoutingSearchParameters()
    params.first_solution_strategy = (
        routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    )
    params.local_search_metaheuristic = (
        routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
    )
    params.time_limit.FromSeconds(120)
    params.random_seed = 42            # deterministic across environments
    params.log_search = False
    return params


def solve(routing, params):
    try:
        assignment = routing.SolveWithParameters(params)
    except MemoryError:
        logger.critical("solver ran out of memory during search")
        raise
    if assignment is None:
        raise RuntimeError(
            "solver returned no feasible plan; check dimension feasibility "
            "and matrix connectivity"
        )
    return assignment

Regulatory Mapping

Every numeric bound above must trace to a rule an inspector can cite. Do not assert limits in prose — pin them to the code of federal regulations so the compliance registry can version them.

The Time dimension’s horizon is not an arbitrary 1,440 minutes; it is the wall inside which a driver’s route must fit under the FMCSA hours-of-service limit at 49 CFR 395.3. Reconciling each route’s on-duty span against a driver’s remaining hours is handled by the DOT/FMCSA Rule Mapping sub-system, which consumes the same Time cumulative values this model emits. Municipal codes routinely impose stricter local ceilings — posted street weights, frost-law bans, tighter collection windows — that override the federal baseline; resolve the effective bound as the minimum of the federal envelope and any active local restriction.

Validation & Verification

A dimension that silently never binds is worse than none: it gives false assurance. Prove the model fires with an end-to-end test whose demand and windows force an observable outcome, then assert against the cumulative values the solver reports.

def test_capacity_and_window_bind() -> None:
    nodes = [
        ServiceNode(node_id=0, lat=40.0, lon=-83.0, demand_kg=0,
                    service_start_min=0, service_end_min=1_440, bin_type="depot"),
        ServiceNode(node_id=1, lat=40.1, lon=-83.1, demand_kg=9_000,
                    service_start_min=480, service_end_min=600, bin_type="rear"),
        ServiceNode(node_id=2, lat=40.2, lon=-83.2, demand_kg=9_000,
                    service_start_min=480, service_end_min=600, bin_type="rear"),
    ]
    dist = np.array([[0, 10, 12], [10, 0, 8], [12, 8, 0]], dtype=int)

    routing, manager = build_model(nodes, dist, num_vehicles=2)
    add_capacity_dimension(routing, manager, nodes, [10_000, 10_000])
    add_time_dimension(routing, manager, nodes, dist)
    assignment = solve(routing, configure_search())

    payload = routing.GetDimensionOrDie("Payload")
    time_dim = routing.GetDimensionOrDie("Time")
    for v in range(routing.vehicles()):
        idx = routing.Start(v)
        while not routing.IsEnd(idx):
            assert assignment.Value(payload.CumulVar(idx)) <= 10_000
            start = assignment.Min(time_dim.CumulVar(idx))
            assert 0 <= start <= 1_440
            idx = assignment.Value(routing.NextVar(idx))

In production, the same cumulative values become a structured compliance record. Follow the site-wide JSONFormatter pattern so events land in the audit database as machine-parseable records rather than free text:

import json


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "event": record.getMessage(),
            "vehicle": getattr(record, "vehicle", None),
            "node_id": getattr(record, "node_id", None),
            "arrival_min": getattr(record, "arrival_min", None),
            "cumulative_kg": getattr(record, "cumulative_kg", None),
        }
        return json.dumps(payload, separators=(",", ":"))


def emit_audit_records(routing, manager, assignment, nodes) -> list[dict]:
    """Serialise one immutable record per stop for the audit database."""
    payload = routing.GetDimensionOrDie("Payload")
    time_dim = routing.GetDimensionOrDie("Time")
    records: list[dict] = []
    for v in range(routing.vehicles()):
        idx = routing.Start(v)
        while not routing.IsEnd(idx):
            node = manager.IndexToNode(idx)
            records.append({
                "vehicle": v,
                "node_id": nodes[node].node_id,
                "arrival_min": assignment.Min(time_dim.CumulVar(idx)),
                "cumulative_kg": assignment.Value(payload.CumulVar(idx)),
            })
            idx = assignment.Value(routing.NextVar(idx))
    return records

The two fields an auditor checks are cumulative_kg — which must never exceed the vehicle’s capacity on any stop — and arrival_min, which must fall inside the node’s declared window. Every dispatched plan should carry one such record per stop.

Failure Modes & Edge Cases

When a model is unsatisfiable, SolveWithParameters returns None. That is the solver correctly reporting that no plan honours every hard constraint — not a bug to be worked around by loosening a bound in place. Degrade along a defined ladder instead.

1. Infeasible constraint set. If aggregate demand exceeds fleet payload, or windows cannot all be met, provision reserve vehicles or drop the lowest-priority stops through the disjunction penalty already attached in Step 3 — never raise vehicle_capacities past a GVWR-derived bound.

def solve_with_relaxation(routing, manager, nodes, base_params, max_rounds: int = 3):
    """Widen the search budget, then fall back to structured degradation."""
    for attempt in range(1, max_rounds + 1):
        params = base_params
        params.time_limit.FromSeconds(120 * attempt)  # more search, same seed
        assignment = routing.SolveWithParameters(params)
        if assignment is not None:
            return assignment
        logger.warning("solve_infeasible", extra={"node_id": None,
                       "cumulative_kg": sum(n.demand_kg for n in nodes)})
    raise RuntimeError("infeasible after relaxation; escalate to fallback router")

2. Solver timeout with no incumbent. A hard time_limit guarantees termination but not a solution on a large matrix. Guided local search yields a feasible-but-suboptimal incumbent well before the limit; a compliant sub-optimal route always beats a timeout, so capture the best assignment seen rather than demanding optimality. When even that fails, hand off to the Fallback Routing Logic state machine, which preserves service continuity and hard compliance boundaries when the primary solver diverges.

3. Telemetry gaps. A disconnected scale or GPS feed yields stale or missing demand. Do not solve on a None; substitute a last-known-good estimate and flag the route for manual review — reconciliation is the job of the ingestion pipeline, and the runtime ceiling can be shifted by the Dynamic Threshold Tuning layer when live readings show density swings.

4. Memory pressure on large matrices. A dense $n \times n$ matrix for a metro fleet can exhaust the worker. Build the matrix in chunks, use int32 rather than float64, and construct the model inside a try/except MemoryError so an out-of-memory event is logged and retried at a smaller batch rather than killing the dispatch process silently.

Integration Checklist

Complete every item before this solver reaches production dispatch:

• Every service node passes the ServiceNode schema; coordinates, demand, and window ordering are validated at ingestion.
• The distance and duration matrices pass validate_matrix (shape, no NaN/inf, no negatives) before the manager is built.
• The capacity dimension is registered with slack_max = 0 so the payload ceiling is hard, using a per-vehicle capacity vector.
• The time dimension clamps each stop with SetRange, and unreachable stops carry an explicit disjunction drop penalty.
• RoutingSearchParameters sets a fixed random_seed and a hard time_limit so runs are reproducible.
• The solve is wrapped so a None return raises an explicit infeasibility signal, and MemoryError is caught and logged.
• The end-to-end capacity-and-window test runs in CI and both assertions pass.
• Every dispatched route emits one JSON audit record per stop with cumulative_kg and arrival_min.
• Infeasible or timed-out solves escalate to the fallback state machine — never by loosening a hard bound in place.

FAQ

Why does the solver return None, and is that an error?

None means the model is unsatisfiable under its hard constraints — it is a correct signal, not a crash. Aggregate demand may exceed fleet payload, or time windows may be mutually unreachable. Escalate through the relaxation ladder (more search budget, reserve vehicles, disjunction drops) rather than widening a hard bound, and log the event so the infeasibility is auditable.

How do I make two OR-Tools runs return identical routes?

Set params.random_seed to a fixed value and a hard params.time_limit, and carry all quantities as integers (kilograms, whole minutes). Guided local search is deterministic given a fixed seed and identical input; floating-point demand or an unset seed is what makes borderline routes flip non-deterministically between runs.

Should capacity or time windows ever be soft constraints?

Capacity maps to statutory GVWR and axle ceilings, so it stays hard with zero slack. Time windows are usually hard too, but a bounded disjunction penalty lets the solver drop a single unreachable stop at a known cost instead of failing the whole plan — which is safer than silently missing a compliance window with no record.

Why validate the distance matrix separately instead of trusting OR-Tools?

OR-Tools does not raise on a NaN, infinite, or negative arc cost; the corruption propagates through constraint propagation as undefined behaviour and yields a confidently wrong plan. A cheap up-front validate_matrix check turns a silent bad route into a loud, catchable error before the manager is ever constructed.

Do I compute the distance matrix inside OR-Tools?

No. Precompute it with OSRM or Valhalla and pass it in, so network latency and routing-engine variability are out of the solve path. Precomputation also lets you validate and version the matrix, which is what makes a dispatch reproducible for audit.

Up: VRP Route Optimization Algorithms

Related

• Setting up Google OR-Tools for Waste Collection — the environment bootstrap and dependency pinning that precede this model
• Capacity & Weight Limits — the payload dimension in full, with CFR-mapped axle thresholds
• Time Window Constraints — hard service-window boundaries and slack tuning
• Dynamic Threshold Tuning — shifting solver ceilings from live telemetry
• DOT/FMCSA Rule Mapping — reconciling route timing against hours-of-service records