Your dispatch solver returns a plan that looks optimal but puts a 40,000 lb packer on a commercial dock after its 06:45 gate closes — this page shows how to bind those pickup windows as hard bounds the optimizer cannot cross, absorb telemetry drift with slack, and log every arrival so the timestamps reconcile during a municipal audit.

The parent guide, Time Window Constraints, explains why service windows in the VRP Route Optimization Algorithms pipeline are hard boundaries rather than advisory preferences: a commercial corridor requires service completion before 07:00 to clear loading-dock congestion, and residential noise ordinances forbid compactor operation between 18:00 and 06:00. This deep dive is the runnable Vehicle Routing Problem with Time Windows (VRPTW) implementation of that constraint — from ingesting jittery GPS telemetry through to serializing the audit record. Every block below is complete and pasteable into a Python 3.10+ dispatch worker.

Environment & Data Prerequisites

Pin the solver and validation stack so a dispatch run is byte-for-byte reproducible when an auditor replays it:

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

The solver consumes one pre-built dispatch payload per depot. Validate its shape before any constraint binding runs — a ragged time matrix or a window whose close precedes its open is the most common cause of a silently infeasible solve. The payload has this exact shape:

Field	Unit	Expected range	Meaning
`time_matrix`	seconds	0 – 86,400	Square `N×N` inter-node travel times, row/col 0 = depot
`windows`	`[open, close]` seconds since midnight	0 – 86,400	Hard pickup window per node; depot is `[0, 86400]`
`capacities`	kg	0 – 30,000	Payload picked up at each node (depot = 0)
`vehicle_capacity`	kg	0 – 30,000	GVWR-derived per-truck payload ceiling

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


class DispatchPayload(BaseModel):
    """A single-depot VRPTW problem, validated before it reaches the solver."""
    time_matrix: list[list[int]]
    windows: list[tuple[int, int]]
    capacities: list[int] = Field(min_length=1)
    vehicle_capacity: int = Field(gt=0, le=30_000)

    @model_validator(mode="after")
    def check_shape(self) -> "DispatchPayload":
        n = len(self.time_matrix)
        if any(len(row) != n for row in self.time_matrix):
            raise ValueError("time_matrix must be square")
        if not (len(self.windows) == len(self.capacities) == n):
            raise ValueError("windows, capacities, and matrix must share node count")
        for open_s, close_s in self.windows:
            if close_s < open_s:
                raise ValueError(f"window close {close_s} precedes open {open_s}")
        return self

Rejecting a malformed problem here means the constraint logic downstream never has to defend against an 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: stabilize the raw telemetry that seeds the matrix, translate ordinances into window bounds, bind those bounds to the OR-Tools time dimension, then solve and serialize the audit log.

Step 1 — Stabilize the telemetry that seeds the matrix

Raw GPS streams from heavy compactors carry positional jitter averaging ±15 m under urban-canyon multipath, and on-board scale sensors report payload mass with ±2.5% variance during dynamic loading. Feeding that noise straight into travel-time estimation destabilizes the matrix from one dispatch cycle to the next. A discrete Kalman filter smooths the coordinate track before distances are computed; the update for a scalar state estimate $\hat{x}_t$ with Kalman gain $K_t$ is:

$\hat{x}_t = \hat{x}_t^- + K_t\left(z_t - \hat{x}_t^-\right), \qquad K_t = \frac{P_t^-}{P_t^- + R}$

where $z_t$ is the raw measurement, $\hat{x}_t^-$ the prior prediction, $P_t^-$ the predicted error covariance, and $R$ the measurement-noise covariance. Pre-allocating fixed-width arrays for the ingest buffer keeps RAM flat and prevents swap thrashing on edge dispatch servers during a large batch. The coordinate-level smoothing itself is covered in Validating GPS Coordinates in Python; here we only need the buffer that holds the settled readings.

import numpy as np


class TelemetryBuffer:
    """Fixed-width ingest buffer for settled GPS/scale readings."""

    def __init__(self, max_nodes: int):
        self.gps_lat = np.empty(max_nodes, dtype=np.float64)
        self.gps_lon = np.empty(max_nodes, dtype=np.float64)
        self.payload_kg = np.empty(max_nodes, dtype=np.float32)
        self.timestamp = np.empty(max_nodes, dtype=np.int64)
        self._ptr = 0

    def push(self, lat: float, lon: float, mass: float, ts: int) -> bool:
        """Append one reading; returns False when the buffer is full."""
        if self._ptr >= len(self.gps_lat):
            return False
        self.gps_lat[self._ptr] = lat
        self.gps_lon[self._ptr] = lon
        self.payload_kg[self._ptr] = mass
        self.timestamp[self._ptr] = ts
        self._ptr += 1
        return True

Step 2 — Translate ordinances into window bounds

Every window originates in a rule: a commercial cutoff, a residential quiet-hours band, or a transfer-station gate schedule. Express each as [open, close] seconds since midnight so the solver works in a single integer time base. Hard constraints — the ones that carry a citation — become the SetRange bounds in Step 3. The declared bounds are also kept in the application layer, because RoutingIndexManager exposes no GetNodeData method; node metadata must be maintained by your own code for the compliance log to reproduce them.

Step 3 — Bind the windows to the OR-Tools time dimension

The routing engine advances a cumulative time variable along the route. Arrival at node $j$ from node $i$ is $t_j = \max\left(a_j,\; t_i + \tau_{ij}\right)$ , clamped to the window $a_j \le t_j \le b_j$ ; the max inserts wait time (slack) when the truck reaches a stop before the window opens, and the upper clamp is the hard rejection that makes a late plan infeasible. SetRange on each node’s CumulVar installs exactly that clamp. Capacity runs as a second dimension in parallel so an overweight leg is rejected on the same solve. A fixed random_seed makes the assignment reproducible across identical inputs — the property an audit replay depends on.

from ortools.constraint_solver import routing_enums_pb2, pywrapcp


class WasteRouteSolver:
    """Single-depot VRPTW solver with hard pickup windows and payload cap."""

    def __init__(
        self,
        time_matrix: list[list[int]],
        windows: list[tuple[int, int]],
        capacities: list[int],
        vehicle_capacity: int,
    ):
        num_nodes = len(time_matrix)
        self.manager = pywrapcp.RoutingIndexManager(num_nodes, 1, 0)
        self.routing = pywrapcp.RoutingModel(self.manager)
        self.windows = windows  # kept in the app layer for compliance logging

        self._register_time_dimension(time_matrix, windows)
        self._register_capacity_dimension(capacities, vehicle_capacity)

        self.params = pywrapcp.DefaultRoutingSearchParameters()
        self.params.first_solution_strategy = (
            routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
        )
        self.params.local_search_metaheuristic = (
            routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
        )
        self.params.time_limit.FromSeconds(30)
        self.params.random_seed = 42  # reproducible routes across dispatch cycles

    def _register_time_dimension(self, time_matrix, windows):
        def time_callback(from_index, to_index):
            i = self.manager.IndexToNode(from_index)
            j = self.manager.IndexToNode(to_index)
            return time_matrix[i][j]

        transit_idx = self.routing.RegisterTransitCallback(time_callback)
        self.routing.AddDimension(
            transit_idx,
            300,     # max slack (wait) in seconds — absorbs traffic/GPS drift
            86_400,  # max route duration in seconds (24 h)
            False,   # do not force cumulative time to start at zero
            "Time",
        )
        time_dim = self.routing.GetDimensionOrDie("Time")
        for node_idx, (open_s, close_s) in enumerate(windows):
            if node_idx == 0:
                continue  # depot has no service window
            index = self.manager.NodeToIndex(node_idx)
            time_dim.CumulVar(index).SetRange(open_s, close_s)

    def _register_capacity_dimension(self, capacities, vehicle_capacity):
        def demand_callback(from_index):
            return capacities[self.manager.IndexToNode(from_index)]

        cap_idx = self.routing.RegisterUnaryTransitCallback(demand_callback)
        self.routing.AddDimensionWithVehicleCapacity(
            cap_idx, 0, [vehicle_capacity], True, "Payload"
        )

Step 4 — Solve and serialize the compliance log

Solving walks the returned assignment and reads the settled arrival and payload off each CumulVar. Every stop is emitted 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 no plan satisfies the windows, the solve raises so the caller can hand control to the Fallback Routing Logic rather than dispatching a non-compliant route.

import json
import logging


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": self.formatTime(record),
            "level": record.levelname,
            "event": record.getMessage(),
            "node_id": getattr(record, "node_id", None),
            "actual_arrival": getattr(record, "actual_arrival", None),
            "window": getattr(record, "window", None),
            "payload_kg": getattr(record, "payload_kg", None),
            "status": getattr(record, "status", None),
        }
        return json.dumps(payload, separators=(",", ":"))


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


def solve_and_log(solver: WasteRouteSolver) -> dict:
    solution = solver.routing.SolveWithParameters(solver.params)
    if not solution:
        raise RuntimeError("No feasible route under the declared pickup windows.")

    time_dim = solver.routing.GetDimensionOrDie("Time")
    cap_dim = solver.routing.GetDimensionOrDie("Payload")
    route = []
    index = solver.routing.Start(0)

    while not solver.routing.IsEnd(index):
        node = solver.manager.IndexToNode(index)
        arrival = solution.Min(time_dim.CumulVar(index))
        payload = solution.Min(cap_dim.CumulVar(index))
        open_s, close_s = solver.windows[node]
        in_window = open_s <= arrival <= close_s
        status = "HARD_PASS" if in_window else "VIOLATION"
        if node != 0:  # depot start carries no pickup window
            logger.info(
                "PICKUP_COMPLIANCE_CHECK",
                extra={
                    "node_id": node,
                    "actual_arrival": arrival,
                    "window": [open_s, close_s],
                    "payload_kg": payload,
                    "status": status,
                },
            )
        route.append({"node_id": node, "arrival": arrival,
                      "payload_kg": payload, "status": status})
        index = solution.Value(solver.routing.NextVar(index))

    return {"route": route, "status": "COMPLIANT"}


if __name__ == "__main__":
    payload = DispatchPayload(
        time_matrix=[[0, 1200, 1800, 2400], [1200, 0, 900, 1500],
                     [1800, 900, 0, 1100], [2400, 1500, 1100, 0]],
        windows=[(0, 86400), (16200, 24300), (18000, 21600), (14400, 19800)],
        capacities=[0, 2500, 3100, 2800],
        vehicle_capacity=15000,
    )
    solver = WasteRouteSolver(
        payload.time_matrix, payload.windows,
        payload.capacities, payload.vehicle_capacity,
    )
    print(json.dumps(solve_and_log(solver), indent=2))

Compliance Output

The mock execution emits one newline-delimited JSON line per serviced stop. Each field carries a regulatory purpose an auditor can trace:

{"ts":"2026-07-02 03:01:41","level":"INFO","event":"PICKUP_COMPLIANCE_CHECK","node_id":1,"actual_arrival":16800,"window":[16200,24300],"payload_kg":2500,"status":"HARD_PASS"}
{"ts":"2026-07-02 03:01:41","level":"INFO","event":"PICKUP_COMPLIANCE_CHECK","node_id":3,"actual_arrival":18900,"window":[14400,19800],"payload_kg":5300,"status":"HARD_PASS"}
{"ts":"2026-07-02 03:01:41","level":"INFO","event":"PICKUP_COMPLIANCE_CHECK","node_id":2,"actual_arrival":20100,"window":[18000,21600],"payload_kg":8400,"status":"HARD_PASS"}

node_id — ties the stop to a specific service location for the chain-of-custody record that reconciles pickup against the facility receipt.
actual_arrival — the solved arrival in seconds since midnight; it must fall inside window for the plan to be legal, and it is the timestamp an ordinance complaint is checked against.
window — the declared [open, close] bound the stop was authorized for, reproduced from the application layer so the audit does not depend on solver internals.
payload_kg — running load at the stop, verifying the truck never crossed its GVWR-derived ceiling en route.
status — HARD_PASS means arrival landed inside the window; a VIOLATION would signal the plan should never have dispatched and must be re-routed.

Verification

Prove both the pass path and the hard-rejection path against the example data with one pytest module:

import pytest


def _base_kwargs():
    return dict(
        time_matrix=[[0, 1200, 1800, 2400], [1200, 0, 900, 1500],
                     [1800, 900, 0, 1100], [2400, 1500, 1100, 0]],
        capacities=[0, 2500, 3100, 2800],
        vehicle_capacity=15000,
    )


def test_strict_windows_pass():
    solver = WasteRouteSolver(windows=[(0, 86400), (16200, 24300),
                                       (18000, 21600), (14400, 19800)],
                              **_base_kwargs())
    result = solve_and_log(solver)
    assert result["status"] == "COMPLIANT"
    serviced = [s for s in result["route"] if s["node_id"] != 0]
    assert all(s["status"] == "HARD_PASS" for s in serviced)


def test_infeasible_window_raises():
    # Node 2 must be served before node 1's travel time can ever elapse.
    solver = WasteRouteSolver(windows=[(0, 86400), (16200, 24300),
                                       (0, 100), (14400, 19800)],
                              **_base_kwargs())
    with pytest.raises(RuntimeError):
        solve_and_log(solver)

The first test confirms every serviced stop lands inside its window; the second confirms an unsatisfiable window raises instead of dispatching a bad plan. Wire the module into CI so a regression in the dimension binding fails the build before it reaches dispatch.

Common Errors

ModuleNotFoundError: No module named 'ortools' — the worker image was built without the pinned solver. Root cause: OR-Tools ships as a manylinux wheel that a slim base image often omits. Fix: pip install ortools==9.11.4210; pin the exact version, because the routing_enums_pb2 enum names have shifted between major releases.

RuntimeError: No feasible route under the declared pickup windows. — the solver found no assignment that satisfies every SetRange. Root cause: overlapping or physically unreachable windows, or a slack ceiling too tight to absorb travel time between two stops. Fix: confirm each window’s close exceeds its open in the DispatchPayload validator, then either widen the affected band or raise the AddDimension slack argument; if the problem is genuinely over-constrained, route the overflow through Fallback Routing Logic.

Windows silently ignored — every stop is served in matrix order — the plan looks valid but arrivals fall outside the declared bands. Root cause: SetRange was applied to the raw node index instead of the routing index, so the clamp never bound the live variable. Fix: always resolve the variable through self.manager.NodeToIndex(node_idx) before calling CumulVar, and confirm your window units are seconds since midnight — mixing a minutes value into a seconds base understates the window by 60× and defeats the constraint.

Up: Time Window Constraints

Time Window Constraints — encoding service windows as hard OR-Tools time dimensions
OR-Tools Implementation — dimension registration and solver configuration
Handling Truck Capacity Constraints in Python — the parallel payload dimension and axle apportionment
Validating GPS Coordinates in Python — smoothing the coordinate track that seeds the time matrix
Fallback Routing Logic — re-assigning stops when the windows make a route infeasible