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.
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 in radians and a scale of metres per degree, the inter-stop distance is:
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.
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. Afallback_triggered: truerecord 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
Related
- OR-Tools Implementation for Waste Collection Routing — the full validated model this setup feeds into
- Handling Truck Capacity Constraints in Python — load-cell filtering and axle-load compliance for the Payload dimension
- Solving VRPTW with Strict Pickup Windows — hard service-window binding in depth
- Validating GPS Coordinates in Python — the coordinate checks that must pass before matrix construction
- Fallback Routing Logic — where a solve that fails even after relaxation escalates