Solving VRPTW with Strict Pickup Windows

Municipal waste collection operates under rigid statutory pickup windows. Commercial corridors require service completion before 07:00 to avoid municipal traffic penalties and loading dock congestion. Residential zones mandate noise abatement compliance, restricting heavy compactor operations between 18:00 and 06:00. The Vehicle Routing Problem with Time Windows (VRPTW) must absorb real-world telemetry drift while guaranteeing deterministic execution paths for daily dispatch cycles. This reference details a production-grade implementation aligning algorithmic routing with municipal compliance mandates.

Telemetry Ingestion & State Estimation

Raw GPS streams from heavy-duty compactors introduce positional jitter averaging ±15 meters under urban canyon multipath conditions. Onboard scale sensors report payload mass with ±2.5% variance during dynamic loading cycles. Before constraint evaluation, we apply a discrete Kalman filter to smooth coordinate trajectories and stabilize velocity estimates. Memory profiling during batch ingestion requires strict buffer allocation to prevent swap thrashing on edge dispatch servers.

import numpy as np
import logging
import json
from typing import List, Tuple

# Pre-allocate fixed-width arrays to eliminate dynamic memory fragmentation
class TelemetryBuffer:
    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:
        """Returns False if 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

Regulatory Constraint Translation

Municipal ordinances map directly to solver-ready bounds. Hard constraints enforce absolute arrival cutoffs with zero tolerance. Soft constraints penalize early arrivals that trigger idle engine emissions or violate residential quiet hours. The Time Window Constraints framework dictates penalty weighting, violation thresholds, and slack allocation. Integration with broader VRP Route Optimization Algorithms ensures scalable dispatch architectures across multi-depot fleets.

Regulatory compliance requires explicit tracking of window adherence at the node level. We implement a structured compliance logger that records actual arrival, window boundaries, penalty application, and payload verification for audit trails.

Production OR-Tools Implementation

The routing engine uses Google OR-Tools with custom dimension callbacks. The callback evaluates cumulative travel time against strict window boundaries. Capacity tracking integrates axle weight limits and volumetric fill rates. Deterministic seed initialization ensures reproducible route assignments across daily dispatch cycles.

Window data is stored alongside the solver in a windows list so the compliance log can extract the declared bounds for each node by index — RoutingIndexManager does not expose a GetNodeData method; node metadata must be maintained in the application layer.

from ortools.constraint_solver import routing_enums_pb2, pywrapcp
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(message)s',
    handlers=[logging.StreamHandler()]
)


class WasteRouteSolver:
    def __init__(
        self,
        time_matrix: List[List[int]],
        windows: List[Tuple[int, int]],
        capacities: List[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 app layer for compliance logging

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

        self.search_params = pywrapcp.DefaultRoutingSearchParameters()
        self.search_params.first_solution_strategy = routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
        self.search_params.local_search_metaheuristic = routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
        self.search_params.time_limit.FromSeconds(30)
        self.search_params.log_search = True

    def _register_time_dimension(self, time_matrix: List[List[int]], windows: List[Tuple[int, int]]):
        def time_callback(from_index, to_index):
            from_node = self.manager.IndexToNode(from_index)
            to_node = self.manager.IndexToNode(to_index)
            return time_matrix[from_node][to_node]

        transit_idx = self.routing.RegisterTransitCallback(time_callback)
        self.routing.AddDimension(
            transit_idx,
            300,    # Max waiting time (slack) in seconds — absorbs traffic variance
            86400,  # Max route duration in seconds (24 h)
            False,
            "Time"
        )
        time_dim = self.routing.GetDimensionOrDie("Time")
        for node_idx, (start, end) in enumerate(windows):
            if node_idx > 0:  # Skip depot (index 0)
                routing_index = self.manager.NodeToIndex(node_idx)
                time_dim.CumulVar(routing_index).SetRange(start, end)

    def _register_capacity_dimension(self, capacities: List[int]):
        def capacity_callback(from_index):
            node = self.manager.IndexToNode(from_index)
            return capacities[node]

        cap_idx = self.routing.RegisterUnaryTransitCallback(capacity_callback)
        self.routing.AddDimensionWithVehicleCapacity(
            cap_idx,
            0,
            [15000],  # Vehicle max payload (kg)
            True,
            "Payload"
        )

    def solve(self) -> dict:
        solution = self.routing.SolveWithParameters(self.search_params)
        if not solution:
            raise RuntimeError("No feasible route found under strict window constraints.")
        return self._extract_compliance_log(solution)

    def _extract_compliance_log(self, solution) -> dict:
        time_dim = self.routing.GetDimensionOrDie("Time")
        cap_dim = self.routing.GetDimensionOrDie("Payload")
        route_log = []
        index = self.routing.Start(0)

        while not self.routing.IsEnd(index):
            node = self.manager.IndexToNode(index)
            arrival = solution.Min(time_dim.CumulVar(index))
            payload = solution.Min(cap_dim.CumulVar(index))
            # Window data is held in the application layer, not in RoutingIndexManager
            window_start, window_end = self.windows[node] if node < len(self.windows) else (0, 86400)
            route_log.append({
                "node_id": node,
                "arrival_epoch": arrival,
                "payload_kg": payload,
                "window_start": window_start,
                "window_end": window_end,
            })
            index = solution.Value(self.routing.NextVar(index))
        return {"route": route_log, "status": "COMPLIANT"}

Workflow: Strict Commercial Zone Cutoff & Compliance Logging

The following workflow demonstrates enforcement of a hard commercial pickup window (04:30–06:45 UTC, expressed in seconds since midnight) with structured compliance logging. Early arrivals trigger soft penalties; late arrivals violate hard constraints and abort dispatch.

Mock Payload:

{
  "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
}

Execution & Structured Log Output:

import json
import logging

payload = {
    "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(
    time_matrix=payload["time_matrix"],
    windows=payload["windows"],
    capacities=payload["capacities"],
)
result = solver.solve()

for stop in result["route"]:
    in_window = stop["window_start"] <= stop["arrival_epoch"] <= stop["window_end"]
    log_entry = {
        "event": "PICKUP_COMPLIANCE_CHECK",
        "node_id": stop["node_id"],
        "actual_arrival": stop["arrival_epoch"],
        "window": [stop["window_start"], stop["window_end"]],
        "payload_kg": stop["payload_kg"],
        "status": "HARD_PASS" if in_window else "VIOLATION",
    }
    logging.info(json.dumps(log_entry))

Structured Log Output (newline-delimited JSON):

{"event": "PICKUP_COMPLIANCE_CHECK", "node_id": 1, "actual_arrival": 16800, "window": [16200, 24300], "payload_kg": 2500, "status": "HARD_PASS"}
{"event": "PICKUP_COMPLIANCE_CHECK", "node_id": 3, "actual_arrival": 18900, "window": [14400, 19800], "payload_kg": 5300, "status": "HARD_PASS"}
{"event": "PICKUP_COMPLIANCE_CHECK", "node_id": 2, "actual_arrival": 20100, "window": [18000, 21600], "payload_kg": 8400, "status": "HARD_PASS"}

The solver enforces strict window adherence via CumulVar bounds. Slack allocation absorbs traffic variance without violating municipal cutoffs. Payload accumulation respects axle weight limits, preventing overweight citations. Deterministic seeding via search_params.random_seed = 42 guarantees identical route generation across identical input states, satisfying audit reproducibility requirements.

Deterministic Dispatch & Validation

Production deployment requires pre-flight validation of time matrices against historical GPS drift. Integrate real-time traffic APIs to adjust time_matrix baselines hourly. Compliance logs must be serialized to immutable storage for municipal audit retrieval. The routing engine integrates directly with fleet telematics via MQTT or REST endpoints, pushing optimized sequences to onboard dispatch tablets before 03:00 daily.

For advanced constraint tuning, reference the official OR-Tools Routing Guide and Python’s structured logging documentation to implement custom penalty functions and audit pipelines.