Setting Up Google OR-Tools for Waste Collection

Environment isolation, callback signatures, and dimension binding patterns for stable solver runs.

This page walks a municipal dispatch engineer from an empty virtualenv to a first reproducible OR-Tools solve that honours truck payload and service windows and emits a hashed compliance record — the exact setup task that fails silently when the C++ backend, the random seed, or the integer arithmetic is left unpinned.

It is the environment-bootstrap step that precedes the full model documented in the OR-Tools Implementation reference, which itself is the solver-integration sub-system of the wider VRP Route Optimization Algorithms pipeline. Everything below is complete and pasteable into a Python 3.10+ dispatch worker.

OR-Tools waste-routing setup: a six-stage pipeline with a deterministic-solve reproducibility gate and a capacity-relaxed fallback branch A pinned venv install flows left to right into telemetry ingestion, equirectangular distance and time matrices, a RoutingModel with a hard Payload dimension and a bounded Time dimension, a deterministic solve pinned to seed 42 and a 15-second limit (the accent-highlighted reproducibility gate), and finally a SHA-256 compliance manifest. When the solve returns None it drops to a dashed capacity-relaxed solve at 1000 kg — the safety gate — which on success rejoins the manifest step. Empty virtualenv → reproducible, hashed dispatch plan pinned venv ortools 9.11.4210 glibc C++ backend telemetry ServiceStop[] int kg · int min matrices equirectangular dist + travel time RoutingModel Payload · Time hard capacity, 0 slack solve seed 42 · 15 s deterministic plan SHA-256 sorted manifest tamper-evident OK None relaxed solve cap ×1.25 → 1000 kg relaxed OK reproducibility gate · pinned seed safety gate · capacity-relaxed fallback on None

Environment & Data Prerequisites

Deterministic routing needs a byte-for-byte reproducible solver, so pin every dependency and match the ortools wheel to your host architecture (manylinux2014_x86_64 on Linux CI, macosx_11_0_universal2 on Apple silicon). A mismatched wheel is the most common cause of an ImportError on the C++ backend.

python -m venv .venv
source .venv/bin/activate
pip install \
  ortools==9.11.4210 \
  numpy==2.1.2 \
  pydantic==2.9.2

Confirm the backend imports and print the build so the version lands in your deployment manifest:

python -c "from ortools.constraint_solver import pywrapcp; print('ortools backend OK')"

The solver consumes one row per service stop. Carry weight in whole kilograms and time in whole minutes since midnight as int, because OR-Tools arithmetic is integer-only and floating-point demand introduces drift across thousands of node insertions. The expected input shape:

Field Type Unit / range Purpose
stop_id str e.g. BIN_A1 audit key, joins back to the asset registry
lat float −90.0 … 90.0 matrix build
lon float −180.0 … 180.0 matrix build
demand_kg int 0 … 15 000 Payload dimension
window [int, int] minutes, start ≤ end, 0 … 1440 Time dimension

Coordinate and window validation happens upstream: raw feeds are normalised by the Telematics Sensor Data Ingestion pipeline, and the lat/lon bounds above mirror the checks in validating GPS coordinates in Python before a single index reaches the solver.

Step-by-Step Implementation

Step 1 — Define the typed input and load telemetry

A dataclass gives the pipeline a single typed contract and rejects the bool-as-int footgun that silently corrupts demand vectors. This is the mock payload the rest of the page solves against.

from __future__ import annotations
from dataclasses import dataclass


@dataclass(frozen=True)
class ServiceStop:
    stop_id: str
    lat: float
    lon: float
    demand_kg: int
    window: tuple[int, int]

    def __post_init__(self) -> None:
        if isinstance(self.demand_kg, bool):  # bool is an int subclass
            raise TypeError(f"{self.stop_id}: demand_kg must be an int, not bool")
        if self.window[1] < self.window[0]:
            raise ValueError(f"{self.stop_id}: window end precedes start")


TELEMETRY_PAYLOAD: list[ServiceStop] = [
    ServiceStop("DEPOT_01", 40.7128, -74.0060,   0, (0, 1440)),
    ServiceStop("BIN_A1",   40.7140, -74.0055, 185, (480, 720)),
    ServiceStop("BIN_A2",   40.7155, -74.0040, 210, (480, 720)),
    ServiceStop("BIN_B1",   40.7130, -74.0080, 195, (720, 960)),
    ServiceStop("BIN_B2",   40.7110, -74.0095, 220, (720, 960)),
]

TRUCK_CAPACITY_KG = 800
SPEED_M_PER_MIN = 500   # urban grid average, ~30 km/h
SERVICE_TIME_MIN = 3

Step 2 — Build the distance and time matrices

For a dense city grid an equirectangular approximation is accurate enough and far cheaper than Haversine. With mean latitude ϕˉ\bar\phi in radians and a scale of k=111,320k = 111{,}320 metres per degree, the inter-stop distance is:

dij=k(ϕiϕj)2+(cosϕˉ(λiλj))2d_{ij} = k\sqrt{(\phi_i-\phi_j)^2 + \big(\cos\bar\phi\,(\lambda_i-\lambda_j)\big)^2}

The cosine term corrects for meridian convergence; dropping it (as naïve grid code does) inflates east–west distances at higher latitudes. Travel time is distance over speed plus a fixed service dwell.

import math


def build_matrices(stops: list[ServiceStop]) -> tuple[list[list[int]], list[list[int]]]:
    n = len(stops)
    dist = [[0] * n for _ in range(n)]
    time = [[0] * n for _ in range(n)]
    for i in range(n):
        for j in range(n):
            if i == j:
                continue
            mean_lat = math.radians((stops[i].lat + stops[j].lat) / 2)
            dlat = stops[i].lat - stops[j].lat
            dlon = (stops[i].lon - stops[j].lon) * math.cos(mean_lat)
            metres = 111_320 * math.hypot(dlat, dlon)
            dist[i][j] = int(metres)
            time[i][j] = int(metres / SPEED_M_PER_MIN) + SERVICE_TIME_MIN
    return dist, time

Step 3 — Bind capacity and time windows as hard dimensions

Register the arc-cost callback first (that is the objective), then attach two dimensions. A slack_max of 0 on capacity makes the payload ceiling hard — an over-loaded plan is pruned from the search rather than penalised afterwards, the same zero-slack pattern the Capacity & Weight Limits sub-system documents in depth. Clamping each CumulVar to its window enforces the municipal collection hours detailed in Time Window Constraints.

from ortools.constraint_solver import pywrapcp


def build_model(stops, capacity_kg: int):
    dist, time = build_matrices(stops)
    manager = pywrapcp.RoutingIndexManager(len(stops), 1, 0)  # 1 truck, depot=0
    routing = pywrapcp.RoutingModel(manager)

    # Objective: minimise total arc distance
    transit = routing.RegisterTransitCallback(
        lambda a, b: dist[manager.IndexToNode(a)][manager.IndexToNode(b)]
    )
    routing.SetArcCostEvaluatorOfAllVehicles(transit)

    # Payload dimension — zero slack makes the capacity ceiling hard
    demands = [s.demand_kg for s in stops]
    demand_cb = routing.RegisterUnaryTransitCallback(
        lambda a: demands[manager.IndexToNode(a)]
    )
    routing.AddDimensionWithVehicleCapacity(
        demand_cb, 0, [capacity_kg], True, "Payload"
    )

    # Time dimension — bounded slack absorbs traffic variance
    time_cb = routing.RegisterTransitCallback(
        lambda a, b: time[manager.IndexToNode(a)][manager.IndexToNode(b)]
    )
    routing.AddDimension(time_cb, 30, 1440, False, "Time")
    time_dim = routing.GetDimensionOrDie("Time")
    for i, stop in enumerate(stops):
        if i == 0:  # depot has no service window
            continue
        time_dim.CumulVar(manager.NodeToIndex(i)).SetRange(*stop.window)

    return routing, manager

Step 4 — Configure a deterministic search and solve, with fallback

Two runs on identical input must return the identical plan or the audit trail is meaningless, so pin random_seed and a hard time_limit. When the model is unsatisfiable, SolveWithParameters returns None; the fallback rebuilds the model with a 25% capacity relaxation. It must rebuild, because RoutingDimension exposes no runtime SetCapacity — the vehicle-capacity list is registered once and cannot be mutated. Anything past the relaxation ceiling escalates to the Fallback Routing Logic state machine.

Capacity-relaxation fallback ladder: primary solve, relaxed solve, then hand-off — each transition annotated with its emitted structured event The primary solve at 800 kg logs route_optimized (OPTIMAL) on success or, on None, logs the warning primary_infeasible and drops to the capacity-relaxed solve at 1000 kg. That relaxed solve logs fallback_capacity_relaxed (FALLBACK_ACTIVE) on success or, on None, logs the critical escalate_to_fallback_router and hands off to the Fallback Routing Logic state machine, which raises a RuntimeError for human dispatch. Downward None branches are dashed danger edges; success branches exit right to hashed manifest chips. Every rung emits one structured audit event primary solve cap 800 kg · seed 42 · 15 s route_optimized OPTIMAL manifest fallback_triggered: false primary_infeasible None · log level WARNING capacity-relaxed solve cap ×1.25 → 1000 kg · model rebuilt fallback_capacity_relaxed FALLBACK_ACTIVE manifest fallback_triggered: true escalate_to_fallback_router None · log level CRITICAL Fallback Routing Logic raise RuntimeError · human dispatch state machine · no clean plan exists; the load is handed to a dispatcher for manual re-planning or a second vehicle.
from ortools.constraint_solver import routing_enums_pb2


def search_params() -> pywrapcp.DefaultRoutingSearchParameters:
    p = pywrapcp.DefaultRoutingSearchParameters()
    p.first_solution_strategy = (
        routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    )
    p.local_search_metaheuristic = (
        routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
    )
    p.time_limit.FromSeconds(15)
    p.random_seed = 42          # deterministic across environments
    return p


def solve(stops, capacity_kg: int):
    routing, manager = build_model(stops, capacity_kg)
    assignment = routing.SolveWithParameters(search_params())
    return routing, manager, assignment

Step 5 — Serialise a hashed compliance manifest

The site-wide JSONFormatter pattern turns every dispatch event into a machine-parseable audit record, and the SHA-256 over the sorted manifest is what lets an inspector prove yesterday’s plan was not edited after the fact.

import json
import hashlib
import logging


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


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


def build_manifest(routing, manager, assignment, capacity_kg, fallback=False) -> dict:
    routes, total_load = [], 0
    for v in range(manager.GetNumberOfVehicles()):
        idx, stops = routing.Start(v), []
        while not routing.IsEnd(idx):
            node = TELEMETRY_PAYLOAD[manager.IndexToNode(idx)]
            stops.append({"stop_id": node.stop_id, "demand_kg": node.demand_kg})
            total_load += node.demand_kg
            idx = assignment.Value(routing.NextVar(idx))
        routes.append({"vehicle_id": v, "stops": stops})

    manifest = {
        "timestamp": "2026-07-02T06:30:00Z",
        "solver_status": "FALLBACK_ACTIVE" if fallback else "OPTIMAL",
        "vehicle_capacity_kg": capacity_kg,
        "total_payload_kg": total_load,
        "epa_manifest_required": total_load > 4000,
        "fallback_triggered": fallback,
        "routes": routes,
    }
    body = json.dumps(manifest, sort_keys=True, separators=(",", ":"))
    manifest["sha256"] = hashlib.sha256(body.encode()).hexdigest()
    return manifest


def run() -> dict:
    routing, manager, assignment = solve(TELEMETRY_PAYLOAD, TRUCK_CAPACITY_KG)
    if assignment is not None:
        manifest = build_manifest(routing, manager, assignment, TRUCK_CAPACITY_KG)
        logger.info("route_optimized", extra={"capacity_kg": TRUCK_CAPACITY_KG,
                    "manifest_hash": manifest["sha256"]})
        return manifest

    relaxed = int(TRUCK_CAPACITY_KG * 1.25)
    logger.warning("primary_infeasible", extra={"capacity_kg": TRUCK_CAPACITY_KG})
    routing, manager, assignment = solve(TELEMETRY_PAYLOAD, relaxed)
    if assignment is not None:
        manifest = build_manifest(routing, manager, assignment, relaxed, fallback=True)
        logger.info("fallback_capacity_relaxed", extra={"capacity_kg": relaxed,
                    "fallback": True, "manifest_hash": manifest["sha256"]})
        return manifest

    logger.critical("escalate_to_fallback_router", extra={"fallback": True})
    raise RuntimeError("infeasible after relaxation; hand off to fallback state machine")


if __name__ == "__main__":
    print(json.dumps(run(), indent=2))

Compliance Output

A successful run prints one immutable manifest. Every field carries a regulatory or audit purpose:

{
  "timestamp": "2026-07-02T06:30:00Z",
  "solver_status": "OPTIMAL",
  "vehicle_capacity_kg": 800,
  "total_payload_kg": 810,
  "epa_manifest_required": false,
  "fallback_triggered": false,
  "routes": [
    {"vehicle_id": 0, "stops": [
      {"stop_id": "DEPOT_01", "demand_kg": 0},
      {"stop_id": "BIN_A1", "demand_kg": 185},
      {"stop_id": "BIN_A2", "demand_kg": 210},
      {"stop_id": "BIN_B1", "demand_kg": 195},
      {"stop_id": "BIN_B2", "demand_kg": 220}
    ]}
  ],
  "sha256": "…64 hex chars…"
}
  • vehicle_capacity_kg — the ceiling the solve ran under; must trace to the vehicle’s registered GVWR under 23 CFR 658.17. A fallback_triggered: true record with a raised value flags a plan that ran with relaxed capacity and needs review.
  • total_payload_kg — aggregate declared tonnage; drives the hazardous/solid-waste manifest trigger.
  • epa_manifest_required — set when tonnage crosses the reporting threshold that obliges an EPA e-Manifest under 40 CFR 262.20–262.23; the flag tells the compliance lake whether a chain-of-custody form must accompany the load.
  • fallback_triggered — records that the deterministic primary solve failed and a relaxed model produced the plan, so the route is never mistaken for a clean optimum.
  • sha256 — a cryptographic fingerprint of the sorted manifest; any later edit changes the hash, which is what makes the record tamper-evident for audit reconstruction.

Route timing in this manifest is reconciled against driver hours-of-service by the DOT/FMCSA Rule Mapping sub-system, and the capacity ceiling can be shifted from live density readings by Dynamic Threshold Tuning.

Verification

Prove the environment and the constraints actually bind with a single runnable check. On the example payload the total demand (810 kg) exceeds the 800 kg ceiling, so the primary solve is infeasible and the fallback must engage — asserting that behaviour confirms both the solve path and the capacity dimension fire.

def test_setup_produces_hashed_manifest() -> None:
    manifest = run()
    # 810 kg > 800 kg ceiling forces the capacity-relaxed fallback
    assert manifest["fallback_triggered"] is True
    assert manifest["vehicle_capacity_kg"] == 1000
    # No route may exceed the capacity it ran under
    served = sum(s["demand_kg"] for r in manifest["routes"] for s in r["stops"])
    assert served <= manifest["vehicle_capacity_kg"]
    # The audit fingerprint is a full SHA-256 digest
    assert len(manifest["sha256"]) == 64


if __name__ == "__main__":
    test_setup_produces_hashed_manifest()
    print("setup verified")

Common Errors

ImportError: libortools.so: cannot open shared object file — the installed ortools wheel does not match the host architecture or glibc. Root cause: a manylinux wheel pulled onto Apple silicon (or vice versa), or an Alpine base image with musl instead of glibc. Fix: reinstall on the target platform with pip install --force-reinstall --no-cache-dir ortools==9.11.4210, and build CI images FROM python:3.11-slim (glibc) rather than an Alpine base.

Solve returns None and run() raises infeasible after relaxation — the model is unsatisfiable even at relaxed capacity, usually because aggregate demand still exceeds the fleet ceiling or two windows are mutually unreachable. Root cause: too few vehicles or contradictory window ranges, not a solver bug. Fix: add a vehicle to the RoutingIndexManager count, widen the offending window, or attach routing.AddDisjunction([index], penalty) so the solver may drop one low-priority stop at a known cost instead of failing the whole plan.

AttributeError: 'RoutingDimension' object has no attribute 'SetCapacity' — an attempt to mutate the payload ceiling at runtime. Root cause: OR-Tools registers the vehicle-capacity vector once inside AddDimensionWithVehicleCapacity and never exposes a setter. Fix: rebuild the model with the new value, exactly as solve(stops, relaxed) does in Step 4 — never patch the dimension in place.

Up: OR-Tools Implementation for Waste Collection Routing