"""Production-grade route optimization using Google OR-Tools. ALGORITHM: TSP / VRP with Google OR-Tools - Industry-standard solver (same as used by major logistics companies) - Constraint-based optimization - Handles time windows (future proofing) - Guaranteed optimal or near-optimal solution FEATURES: - Automatic outlier detection and coordinate correction - Hybrid distance calculation (self-hosted Valhalla road matrix + Haversine fallback) - Robust error handling for invalid inputs """ import math import os import hashlib import logging import asyncio from typing import Dict, Any, List as _List, Optional, Tuple, Union from datetime import datetime, timedelta import httpx from app.services.routing.gps_smoother import smooth_order_coordinates import numpy as np from app.core.arrow_utils import calculate_haversine_matrix_vectorized from app.config.dynamic_config import get_config try: from ortools.constraint_solver import routing_enums_pb2 from ortools.constraint_solver import pywrapcp ORTOOLS_AVAILABLE = True except ImportError: ORTOOLS_AVAILABLE = False logging.warning("Google OR-Tools not found. Falling back to simple greedy solver.") logger = logging.getLogger(__name__) async def road_backend_status() -> Dict[str, Any]: """ Probe the Valhalla matrix backend. Module-level (not a RouteOptimizer method) so startup and health checks can call it without paying for a full optimizer construction, which pulls in the empirical ETA calculator and its history load. """ url = os.getenv("VALHALLA_URL", "").strip().rstrip("/") if not url: return { "configured": False, "reachable": False, "detail": "VALHALLA_URL not set - road sequencing disabled, aerial only", } try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.get(f"{url}/status") resp.raise_for_status() data = resp.json() return { "configured": True, "reachable": True, "url": url, "version": data.get("version"), "tileset_last_modified": data.get("tileset_last_modified"), } except Exception as e: return {"configured": True, "reachable": False, "url": url, "detail": str(e)} class RouteOptimizer: """Route optimization using Google OR-Tools (Async).""" def __init__(self): self.earth_radius = 6371 # Earth radius in km _cfg = get_config() # Initialize ETA Calculator - empirical (learned from actual delivery # times) with automatic fallback to the realistic formula. from app.services.routing.realistic_eta_calculator import ( get_time_of_day_category, ) from app.services.routing.empirical_eta_calculator import EmpiricalETACalculator self.eta_calculator = EmpiricalETACalculator() self.get_traffic_condition = get_time_of_day_category # Speed settings (ML-tuned via DynamicConfig) self.avg_speed_kmh = float(_cfg.get("avg_speed_kmh")) # Road factor (haversine -> road distance multiplier, ML-tuned) self.road_factor = float(_cfg.get("road_factor")) # Road travel-time matrix backend: self-hosted Valhalla (no per-request # cost, no rate limit). Unset VALHALLA_URL -> road sequencing stays off # and every caller falls back to aerial ordering. self.valhalla_url = os.getenv("VALHALLA_URL", "").strip().rstrip("/") self.use_road_matrix = bool(self.valhalla_url) # Solver time limit (ML-tuned) self.search_time_limit_seconds = int(_cfg.get("search_time_limit_seconds")) def haversine_distance( self, lat1: float, lon1: float, lat2: float, lon2: float ) -> float: """Calculate great circle distance between two points on Earth (in km).""" try: lat1, lon1, lat2, lon2 = map( math.radians, [float(lat1), float(lon1), float(lat2), float(lon2)] ) dlat = lat2 - lat1 dlon = lon2 - lon1 a = ( math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2 ) c = 2 * math.asin(math.sqrt(a)) return self.earth_radius * c except Exception: return 0.0 # ------------------------------------------------------------------ # ROAD-AWARE VISITING ORDER (Phase 2 - opt-in, cached) # ------------------------------------------------------------------ def _aerial_minutes( self, a: Tuple[float, float], b: Tuple[float, float] ) -> float: """Haversine travel-time estimate (minutes), used to patch unroutable pairs.""" km = self.haversine_distance(a[0], a[1], b[0], b[1]) * self.road_factor return (km / (self.avg_speed_kmh or 20.0)) * 60.0 async def _road_duration_matrix( self, coords: _List[Tuple[float, float]] ) -> Optional[_List[_List[float]]]: """ Full NxN road travel-TIME matrix (minutes) via self-hosted Valhalla. One /sources_to_targets call returns the whole matrix, so unlike the old Google Distance Matrix path there is no per-request element cap and no chunking. Costing defaults to `motorcycle`, which models the lane access and one-way behaviour our riders actually have — a car matrix systematically overstates their travel time. Valhalla reports an unroutable pair as time=null. Those are patched with an aerial estimate rather than left at 0, because a 0-cost edge would look free to the TSP and pull the whole sequence through it. If too many pairs are unroutable the tileset probably doesn't cover this region, so we bail to aerial entirely. Returns None on any failure so the caller falls back to aerial ordering. """ if not self.use_road_matrix: return None cfg = get_config() costing = str(cfg.get("routing_valhalla_costing", "motorcycle")) timeout = float(cfg.get("routing_matrix_timeout_seconds", 15.0)) max_unroutable = float(cfg.get("routing_matrix_max_unroutable_pct", 20.0)) n = len(coords) locations = [{"lat": float(la), "lon": float(lo)} for la, lo in coords] matrix = [[0.0] * n for _ in range(n)] try: async with httpx.AsyncClient(timeout=timeout) as client: resp = await client.post( f"{self.valhalla_url}/sources_to_targets", json={ "sources": locations, "targets": locations, "costing": costing, "units": "km", }, ) resp.raise_for_status() data = resp.json() rows = data.get("sources_to_targets") or [] if len(rows) != n: logger.debug( f"[RoadSeq] Valhalla returned {len(rows)} rows, expected {n}" ) return None unroutable = 0 for i, row in enumerate(rows): for el in row or []: j = el.get("to_index") if j is None or not (0 <= j < n): continue secs = el.get("time") if secs is None: unroutable += 1 matrix[i][j] = self._aerial_minutes(coords[i], coords[j]) else: matrix[i][j] = secs / 60.0 pct = 100.0 * unroutable / max(1, n * n) if pct > max_unroutable: logger.warning( f"[RoadSeq] {unroutable}/{n * n} pairs ({pct:.0f}%) unroutable via " f"Valhalla - tileset likely missing this region; using aerial" ) return None if unroutable: logger.debug( f"[RoadSeq] patched {unroutable}/{n * n} unroutable pairs with aerial" ) return matrix except Exception as e: logger.debug(f"[RoadSeq] matrix build failed: {e}") return None async def _road_optimal_order( self, start_lat: float, start_lon: float, points: _List[Tuple[float, float]], ) -> Optional[_List[int]]: """ Road-aware visiting order for `points`, starting from (start_lat, start_lon). Builds a real road travel-TIME matrix (Valhalla) and solves an OPEN TSP with OR-Tools (return-to-depot edge = 0), so the sequence respects real road geometry/one-ways instead of straight-line distance. Validated on live batches to cut real travel time ~5-13% vs aerial. Returns 0-based indices into `points` in optimal order, or None to signal the caller to fall back to the existing aerial greedy + 2-opt. Safe on the live path: * disabled unless `routing_use_road_distance` AND VALHALLA_URL is set * only for 3..`routing_road_max_stops` stops (fewer is trivial; more costs more solver time than the ordering gain is worth) * result cached in Redis (default 24h) keyed by the rounded coords in input order, so the returned indices always map back correctly """ cfg = get_config() if not cfg.get("routing_use_road_distance", False): return None if not self.use_road_matrix: return None n = len(points) if n < 3 or n > int(cfg.get("routing_road_max_stops", 25)): return None def _r(v) -> float: return round(float(v), 4) # ~11 m - enough to dedupe near-identical batches pts_key = "|".join(f"{_r(la)},{_r(lo)}" for la, lo in points) cache_key = "roadseq:" + hashlib.sha256( f"{_r(start_lat)},{_r(start_lon)}#{pts_key}".encode("utf-8") ).hexdigest() try: from app.services import cache as _cache cached = _cache.get_json(cache_key) if isinstance(cached, list) and len(cached) == n: logger.debug(f"[RoadSeq] cache hit ({n} stops)") return cached except Exception: pass locs = [(start_lat, start_lon)] + list(points) matrix = await self._road_duration_matrix(locs) if matrix is None: return None # matrix contains minutes, not km — use scale=100 (centiseconds) so # OR-Tools gets valid integer arc costs without the km→m * 1000 factor. route = self._solve_tsp_ortools(locs, matrix, matrix_scale=100) order = [i - 1 for i in route if i != 0] if len(order) != n: return None try: from app.services import cache as _cache _cache.set_json( cache_key, order, ttl_seconds=int(cfg.get("routing_road_cache_ttl_seconds", 86400)), ) except Exception: pass logger.info(f"[RoadSeq] road-optimal order for {n} stops -> {order}") return order # ------------------------------------------------------------------ # 2-OPT LOCAL SEARCH # ------------------------------------------------------------------ def _two_opt_improve( self, route: _List[int], dist_matrix, # numpy array or nested list ) -> _List[int]: """ 2-opt local search: repeatedly reverses route segments to find a shorter path. Algorithm: For each pair (i, j) where i < j, check whether reversing the sub-route between positions i and j reduces total distance. If yes, apply the reversal and restart scanning. Property: * Guaranteed to terminate (finitely many distinct routes). * Produces a *locally* 2-optimal solution - no 2-edge swap can improve it further. * Typical gain on delivery-scale instances: 815% shorter routes. * Works on OPEN routes (depot stays at position 0, free return). Complexity: O(n) per improvement pass, O(n) worst case total. Acceptable for n <= 50 orders per rider. """ if len(route) <= 3: return route def d(a: int, b: int) -> float: return float(dist_matrix[a][b]) best = list(route) n = len(best) improved = True while improved: improved = False for i in range(1, n - 1): for j in range(i + 1, n): # Cost of the TWO edges that would change in this swap: # Current : best[i-1] -> best[i] and best[j] -> best[j+1] # Reversed: best[i-1] -> best[j] and best[i] -> best[j+1] # (open TSP: no edge from last node back to depot) c_before = d(best[i - 1], best[i]) c_after = d(best[i - 1], best[j]) if j + 1 < n: c_before += d(best[j], best[j + 1]) c_after += d(best[i], best[j + 1]) # For asymmetric matrices (e.g., road duration), reversing the # sub-route changes the cost of internal edges. We must add # the exact difference. For symmetric matrices, this is 0. internal_diff = 0.0 for k in range(i, j): internal_diff += d(best[k + 1], best[k]) - d(best[k], best[k + 1]) if c_after + internal_diff < c_before - 1e-10: best[i : j + 1] = best[i : j + 1][::-1] improved = True return best # ------------------------------------------------------------------ # TSP SOLVER (with optional time windows + 2-opt post-processing) # ------------------------------------------------------------------ def _solve_tsp_ortools( self, locations: _List[Tuple[float, float]], dist_matrix: _List[_List[float]], time_windows: Optional[_List[Tuple[int, int]]] = None, matrix_scale: int = 1000, ) -> _List[int]: """ Solve Open TSP using Google OR-Tools + GLS + 2-opt refinement. Args: locations : list of (lat, lon) - index 0 is the depot (kitchen). dist_matrix : NxN cost matrix. Default unit is km (scale=1000 → metres). Pass matrix_scale=100 when the matrix contains minutes. time_windows: Optional list of (earliest_min, latest_min) per node. * Node 0 (depot): typically (0, horizon). * Delivery node i: (0, max_delivery_deadline). If None, no time-window constraints are applied. matrix_scale: Integer multiplier applied to matrix values before passing to OR-Tools (which requires integer arc costs). Returns: Route as a list of node indices starting with 0 (depot). """ if not ORTOOLS_AVAILABLE: route = self._solve_greedy(locations, dist_matrix) return self._two_opt_improve(route, dist_matrix) if not locations or len(locations) <= 1: return [0] n_nodes = len(locations) manager = pywrapcp.RoutingIndexManager(n_nodes, 1, 0) routing = pywrapcp.RoutingModel(manager) # -- DISTANCE CALLBACK -------------------------------------------- # Open TSP: return-to-depot edge always costs 0 so the solver # optimises the *path* from kitchen to last drop-off, not a closed loop. def distance_callback(from_index, to_index): from_node = manager.IndexToNode(from_index) to_node = manager.IndexToNode(to_index) if to_node == 0: return 0 return int(dist_matrix[from_node][to_node] * matrix_scale) transit_cb_idx = routing.RegisterTransitCallback(distance_callback) routing.SetArcCostEvaluatorOfAllVehicles(transit_cb_idx) # -- TIME-WINDOW DIMENSION (optional) ---------------------------- if time_windows and len(time_windows) == n_nodes: speed_km_per_min = max(0.1, self.avg_speed_kmh / 60.0) def time_callback(from_index, to_index): fn = manager.IndexToNode(from_index) tn = manager.IndexToNode(to_index) if tn == 0: return 0 travel_min = int(dist_matrix[fn][tn] / speed_km_per_min) return travel_min + 4 # +4 min avg door time per stop time_cb_idx = routing.RegisterTransitCallback(time_callback) max_horizon = 180 # 3-hour window routing.AddDimension( time_cb_idx, 30, # max waiting time at any node (30 min) max_horizon, False, # don't force start cumul to zero "Time", ) time_dim = routing.GetDimensionOrDie("Time") for node_idx, (earliest, latest) in enumerate(time_windows): if node_idx == 0: continue # depot has no hard window idx = manager.NodeToIndex(node_idx) time_dim.CumulVar(idx).SetRange(int(earliest), int(latest)) # Minimise time at end of route (encourages ASAP delivery) time_dim.SetGlobalSpanCostCoefficient(10) # -- SEARCH PARAMETERS -------------------------------------------- search_params = pywrapcp.DefaultRoutingSearchParameters() search_params.first_solution_strategy = ( routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC ) search_params.local_search_metaheuristic = ( routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH ) # TSP time budget, ceiling 2 seconds per kitchen. This cap applies to # per-rider TSP and the per-kitchen beatmap solves. # # GUIDED_LOCAL_SEARCH runs until its time limit EXPIRES - it does not # return early once it has the optimum. A flat cap therefore burned the # whole budget on trivial inputs, where PATH_CHEAPEST_ARC is done in # under 200ms. Scale with stop count; large inputs still get the ceiling. _cap_ms = min(self.search_time_limit_seconds, 2) * 1000 search_params.time_limit.FromMilliseconds( max(200, min(_cap_ms, 200 * len(locations))) ) solution = routing.SolveWithParameters(search_params) if solution: index = routing.Start(0) route = [] while not routing.IsEnd(index): route.append(manager.IndexToNode(index)) index = solution.Value(routing.NextVar(index)) # -- 2-OPT POST-PROCESSING --------------------------------- route = self._two_opt_improve(route, dist_matrix) return route else: route = self._solve_greedy(locations, dist_matrix) return self._two_opt_improve(route, dist_matrix) # ------------------------------------------------------------------ # TRUE MULTI-RIDER VRP (Capacitated Pickup-and-Delivery Problem) # ------------------------------------------------------------------ def solve_vrp_multi_rider( self, orders: _List[Dict[str, Any]], rider_data: _List[Dict[str, Any]], # [{"id": rid, "lat": lat, "lon": lon}, ...] max_orders_per_rider: int = 12, soft_prefs: Optional[Dict] = None, soft_home: Optional[Dict] = None, ) -> Optional[Dict[int, _List[Dict[str, Any]]]]: """ Capacitated VRP - assigns all orders to riders simultaneously in a single OR-Tools model, with strong load-balancing pressure. KEY FIXES over the previous PDPTW implementation: ------------------------------------------------- 1. CAPACITY MODEL (critical fix): Old model used +1 at pickup and -1 at delivery (PDPTW). Net demand per order = 0 -> solver could put ALL orders on 1 rider by interleaving pickup/delivery pairs (capacity never exceeds 1). New model: demand = 1 per delivery node only. Capacity dimension tracks TOTAL orders assigned per rider (monotonic). This correctly enforces max_orders_per_rider as a total limit. 2. MINIMUM VEHICLES + NATURAL BALANCE: SetFixedCostOfAllVehicles(500_000) penalises each extra vehicle by 500 km equivalent - far more than any real route saves. Solver uses ceil(orders/capacity) riders; distance cost balances loads naturally. 3. DETERMINISM: Riders sorted by ID so vehicle-0 always maps to the lowest-ID rider, regardless of API response ordering. SAVINGS first-solution strategy is deterministic for identical inputs. Node layout (total = 1 + N + M): +-----------------------------------------------------+ | 0 : global end-depot (free return) | | 1 .. N : rider GPS start positions | | N+1..N+M : order delivery nodes (demand = 1 each) | +-----------------------------------------------------+ Returns {rider_id: [ordered list of order dicts]} or None on failure. Falls back silently to None - caller must use 2-phase fallback. """ if not ORTOOLS_AVAILABLE: return None if not orders or not rider_data: return None try: M = len(orders) N = len(rider_data) # -- SORT RIDERS FOR DETERMINISM ------------------------------ # API response order can differ between calls. Sorting by rider ID # ensures vehicle-v always maps to the same rider on every call # with identical input data. rider_data = sorted(rider_data, key=lambda r: r["id"]) def _f(v): try: return float(v) except: return 0.0 # -- BUILD NODE COORDINATE LIST ------------------------------- # We use DELIVERY coordinates for VRP node locations. # Riders are scored and routed by where they actually *deliver*, # so the solver groups geographically close deliveries together. # Kitchen sequencing (visit Kitchen A -> deliver, Kitchen B -> deliver) # is handled downstream by beatmap_route / TSP. # Kitchen *affinity* is handled by the preference-discount matrix below: # preferred riders get a cost reduction on orders from their kitchen, # which steers the solver to assign them those orders even if a # non-preferred rider is marginally closer by delivery location. end_depot = (0.0, 0.0) # dummy; cost to return here = 0 rider_locs = [(r["lat"], r["lon"]) for r in rider_data] order_locs = [] for o in orders: dlat = _f(o.get("deliverylat") or o.get("droplat")) dlon = _f(o.get("deliverylong") or o.get("droplon")) if dlat == 0: # fallback to pickup coords if delivery missing dlat = _f(o.get("pickuplat")) dlon = _f(o.get("pickuplon") or o.get("pickuplong")) order_locs.append((dlat, dlon)) all_locs = [end_depot] + rider_locs + order_locs total_nodes = len(all_locs) # = 1 + N + M # -- DISTANCE MATRIX ------------------------------------------ lats = np.array([loc[0] for loc in all_locs]) lons = np.array([loc[1] for loc in all_locs]) dist_m = ( calculate_haversine_matrix_vectorized(lats, lons) * self.road_factor ) # -- PREFERENCE DISCOUNT MATRIX ------------------------------- # Build a per-rider discount (in metres) for orders from preferred # kitchens. A negative arc-cost delta is achieved by reducing that # rider's cost to reach preferred-kitchen delivery nodes. # # Discount = 3 000 m ~ 3 km equivalent, so a preferred rider # up to ~3 km farther than a non-preferred rider still wins the bid. # Home-zone bonus adds another 2 000 m for riders within 4 km of # the kitchen's home area. from app.config.rider_preferences import ( RIDER_PREFERRED_KITCHENS, RIDER_HOME_LOCATIONS, ) # SOFT steering uses learned-augmented affinity (config learned); # the HARD kitchen constraint below stays on curated config only. try: if soft_prefs is not None and soft_home is not None: _soft_prefs = soft_prefs _soft_home = soft_home else: from app.services.routing.rider_affinity_service import get_rider_affinity _aff = get_rider_affinity() _soft_prefs = _aff.get_preferred_kitchens() _soft_home = _aff.get_home_locations() except Exception: _soft_prefs = soft_prefs if soft_prefs is not None else RIDER_PREFERRED_KITCHENS _soft_home = soft_home if soft_home is not None else RIDER_HOME_LOCATIONS PREF_DISCOUNT_M = 3_000 # 3 km equivalent discount for preferred kitchen HOME_4KM_BONUS_M = 2_000 # extra 2 km for rider within home zone (4 km) HOME_2KM_BONUS_M = 4_000 # extra 4 km for rider very close to home zone def _haversine_m(la1, lo1, la2, lo2): import math la1,lo1,la2,lo2 = map(math.radians,[float(la1),float(lo1),float(la2),float(lo2)]) a = math.sin((la2-la1)/2)**2 + math.cos(la1)*math.cos(la2)*math.sin((lo2-lo1)/2)**2 return 6_371_000 * 2 * math.asin(min(1.0, math.sqrt(a))) # Precompute kitchen name for each order node. # Field priority (confirmed from real order payload): # pickupcustomer -> "Daily grubs(jayanthi kitchen)" primary # locationname -> "Daily grubs(jayanthi kitchen)" same value, backup # Legacy / alternate API versions also checked below. _KITCHEN_KEYS = [ "pickupcustomer", # confirmed in production order JSON "locationname", # confirmed in production order JSON "storename", "store_name", "restaurantname", "restaurant_name", "kitchenname", "kitchen_name", "partnername", "partner_name", "tenantname", # "Daily grubs" (without branch suffix) "brandname", "brand_name", "providername", "provider_name", "shopname", "shop_name", ] order_kitchens = [] for o in orders: kitchen = "" for _k in _KITCHEN_KEYS: _v = o.get(_k) if _v and str(_v).strip(): kitchen = str(_v).strip().lower() break order_kitchens.append(kitchen) # Log unique kitchen names found so user can verify they match prefs unique_kitchens = sorted(set(k for k in order_kitchens if k)) logger.info( f"[VRP] Kitchen names in orders ({len(unique_kitchens)} unique): {unique_kitchens}" ) # Build discount array: discount_m[v][order_i] # Node index for order_i in all_locs = 1 + N + order_i discount_m = [[0] * M for _ in range(N)] for v, rd in enumerate(rider_data): rid = rd["id"] prefs = [p.lower() for p in _soft_prefs.get(rid, [])] h_lat, h_lon = _soft_home.get(rid, (0.0, 0.0)) for i, kitchen in enumerate(order_kitchens): if not kitchen or not prefs: continue # Bidirectional substring match (mirrors AssignmentService logic) if any(p in kitchen or kitchen in p for p in prefs): disc = PREF_DISCOUNT_M # Tiebreak when two riders prefer the same kitchen: # give extra discount to the rider whose home zone # is closest to the delivery location. if h_lat != 0: dlat, dlon = order_locs[i] # delivery coords if dlat != 0: home_dist_m = _haversine_m(h_lat, h_lon, dlat, dlon) if home_dist_m <= 2_000: disc += HOME_2KM_BONUS_M elif home_dist_m <= 4_000: disc += HOME_4KM_BONUS_M discount_m[v][i] = disc # -- OR-TOOLS MODEL ------------------------------------------- starts = list(range(1, N + 1)) # each rider starts at its own node ends = [0] * N # all end at dummy depot (free) manager = pywrapcp.RoutingIndexManager(total_nodes, N, starts, ends) routing = pywrapcp.RoutingModel(manager) # Per-vehicle distance callbacks - each rider gets a preference discount # on delivery nodes belonging to their preferred kitchens. # OR-Tools requires one registered callback per vehicle when costs differ. def make_pref_cb(v_idx, disc_row): def cb(fi, ti): fn = manager.IndexToNode(fi) tn = manager.IndexToNode(ti) if tn == 0: return 0 base = int(dist_m[fn][tn] * 1000) # Apply discount if destination is a delivery node order_i = tn - (1 + N) if 0 <= order_i < M: base = max(0, base - disc_row[order_i]) return base return cb for v in range(N): cb_fn = make_pref_cb(v, discount_m[v]) cb_idx = routing.RegisterTransitCallback(cb_fn) routing.SetArcCostEvaluatorOfVehicle(cb_idx, v) # -- HARD KITCHEN CONSTRAINT ---------------------------------- # rider_preferences.py is the source of truth. # If a kitchen has at least one configured rider in the active # fleet, ONLY those riders are allowed to visit those order nodes. # Non-preferred riders are physically excluded by the solver. # # Fallback: if every preferred rider for a kitchen is absent from # the active fleet, the order is left unconstrained (any rider). # This prevents unsolvable models when a preferred rider is off-duty. # Build inverse map: kitchen_name_lower -> [vehicle_index, ...] kitchen_to_vehicles: dict = {} for v, rd in enumerate(rider_data): rid = rd["id"] for pref_k in RIDER_PREFERRED_KITCHENS.get(rid, []): kitchen_to_vehicles.setdefault(pref_k.lower(), []).append(v) _hard_applied = 0 for i, kitchen in enumerate(order_kitchens): if not kitchen: continue allowed_v: set = set() for cfg_k, vehicles in kitchen_to_vehicles.items(): if cfg_k in kitchen or kitchen in cfg_k: allowed_v.update(vehicles) if allowed_v: # OR-Tools hard constraint: solver cannot assign this order node_idx = manager.NodeToIndex(1 + N + i) try: routing.SetAllowedVehiclesForIndex(list(allowed_v), node_idx) except TypeError: # Fallback for OR-Tools wrapper bug with absl::Span routing.VehicleVar(node_idx).SetValues([int(x) for x in allowed_v]) _hard_applied += 1 logger.info( f"[VRP] Hard constraints: {_hard_applied}/{M} orders locked to " f"their configured riders. " f"({M - _hard_applied} orders unconstrained / no pref configured)" ) has_prefs = any(any(d > 0 for d in row) for row in discount_m) riders_with_prefs = sum(1 for row in discount_m if any(d > 0 for d in row)) logger.info( f"[VRP] Preference discounts applied: {has_prefs} | " f"riders with prefs: {riders_with_prefs}/{N}" ) if not has_prefs: active_ids = sorted(rd["id"] for rd in rider_data) missing_ids = [rid for rid in active_ids if rid not in RIDER_PREFERRED_KITCHENS] if missing_ids: # IDs are not registered at all logger.warning( f"[VRP] ! NO preference discounts - rider IDs not in config!\n" f" Active IDs not configured : {missing_ids}\n" f" -> Add them to RIDER_PREFERRED_KITCHENS in " f"app/config/rider_preferences.py" ) else: # IDs are fine but kitchen NAME in orders doesn't match configured prefs config_kitchens = sorted(set( k.lower() for prefs in RIDER_PREFERRED_KITCHENS.values() for k in prefs )) logger.warning( f"[VRP] ! NO preference discounts - kitchen NAMES don't match!\n" f" Order kitchen names : {unique_kitchens}\n" f" Configured prefs : {config_kitchens}\n" f" -> These must overlap (substring match). " f"Edit RIDER_PREFERRED_KITCHENS in rider_preferences.py " f"to use the exact names shown in 'Order kitchen names' above." ) # -- CAPACITY DIMENSION --------------------------------------- # demand[delivery_node] = 1 (each order = 1 unit). # Monotonically increases as rider picks up more orders. # At max_orders_per_rider the rider is full - solver must use another. demands = [0] * total_nodes for i in range(M): demands[1 + N + i] = 1 # delivery node i def demand_cb(fi): return demands[manager.IndexToNode(fi)] dem_idx = routing.RegisterUnaryTransitCallback(demand_cb) routing.AddDimensionWithVehicleCapacity( dem_idx, 0, # no slack [max_orders_per_rider] * N, # hard cap per rider True, # start cumul at 0 "Capacity", ) # -- LOAD BALANCING via Capacity span ------------------------- # SetGlobalSpanCostCoefficient on the CAPACITY dimension adds # (max_load - min_load) * coeff to the objective, pushing GLS # to redistribute orders from heavy riders to light riders. # # Why it is SAFE here (unlike on the distance dimension): # Fixed vehicle cost = 500,000 m # Max span possible = max_orders_per_rider = 12 orders # Max span saving = 12 x 3,000 = 36,000 m # 36,000 << 500,000 -> adding an empty rider ALWAYS costs # +464,000 m net -> solver will NEVER activate an extra rider # just to reduce span. Balance happens only among the # ceil(orders/cap) riders already chosen by the fixed cost. # # Effect: GLS will accept moving an order to a lighter rider # even if that rider is up to ~3 km farther, as long as doing # so reduces the max-min imbalance by at least 1 order. cap_dim = routing.GetDimensionOrDie("Capacity") cap_dim.SetGlobalSpanCostCoefficient(3_000) # -- MINIMUM VEHICLES (force fewest riders) ------------------- # Each vehicle activated adds 500 km to the objective. # 500 km >> any realistic multi-stop route (~ 60 km max), # so the solver always uses ceil(orders / capacity) riders. routing.SetFixedCostOfAllVehicles(500_000) # -- SEARCH PARAMETERS ---------------------------------------- sp = pywrapcp.DefaultRoutingSearchParameters() # PATH_CHEAPEST_ARC: works correctly for multi-depot VRP where # each vehicle starts at a different location (rider position). # SAVINGS (Clarke-Wright) requires a single shared depot and # returns None for multi-depot inputs - do NOT use SAVINGS here. sp.first_solution_strategy = ( routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC ) sp.local_search_metaheuristic = ( routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH ) # VRP time budget: ceiling of 3 seconds so the API response is never # blocked longer than that. The x 3 multiplier caused 15-second # responses with the default 5-second search_time_limit. # # GUIDED_LOCAL_SEARCH runs until its time limit EXPIRES rather than # stopping once the optimum is found, so a flat 3s spent ~2.9s of # pure waiting on a 1-order batch (measured on live traffic). # Scale with problem size; large batches still get the full ceiling. _cap_ms = min(self.search_time_limit_seconds, 3) * 1000 sp.time_limit.FromMilliseconds( max(250, min(_cap_ms, 250 * len(orders))) ) solution = routing.SolveWithParameters(sp) if not solution: logger.warning("[VRP] No solution found - falling back to 2-phase assignment") return None # -- EXTRACT & RETURN ASSIGNMENTS ----------------------------- rider_assignments: Dict[int, _List[Dict[str, Any]]] = {} for v in range(N): rider_id = rider_data[v]["id"] rider_route_orders: _List[Dict[str, Any]] = [] idx = routing.Start(v) while not routing.IsEnd(idx): node = manager.IndexToNode(idx) idx = solution.Value(routing.NextVar(idx)) # Delivery nodes are indices N+1 to N+M if node >= 1 + N: order_i = node - (1 + N) if 0 <= order_i < M: rider_route_orders.append(orders[order_i]) if rider_route_orders: rider_assignments[rider_id] = rider_route_orders assigned = sum(len(v) for v in rider_assignments.values()) load_counts = sorted([len(v) for v in rider_assignments.values()], reverse=True) logger.info( f"[VRP] Solution: {assigned}/{M} orders -> " f"{len(rider_assignments)}/{N} riders | " f"distribution: {load_counts}" ) return rider_assignments except Exception as exc: logger.error(f"[VRP] Solver error: {exc}", exc_info=True) return None def _solve_greedy(self, locations, dist_matrix): """Simple Greedy Nearest Neighbor fallback.""" unvisited = set(range(1, len(locations))) curr = 0 route = [0] while unvisited: nearest = min(unvisited, key=lambda x: dist_matrix[curr][x]) route.append(nearest) unvisited.remove(nearest) curr = nearest return route def _cleanup_coords( self, lat: Any, lon: Any, ref_lat: float, ref_lon: float ) -> Tuple[float, float]: """ Heuristic to fix bad coordinates. 1. Fixes lat==lon typo. 2. Fixes missing negative signs if needed (not needed for India). 3. Projects outlier > 500km to reference (centroid). """ try: lat = float(lat) lon = float(lon) except: return 0.0, 0.0 if lat == 0 or lon == 0: return lat, lon # 1. Check strict equality (typo) if abs(lat - lon) < 0.0001: if ref_lon != 0: # If reference is available, assume lat is correct and fix lon # (Common error: copy lat to lon field) return lat, ref_lon # 2. Check general outlier (e.g. 500km away) if ref_lat != 0 and ref_lon != 0: dist = self.haversine_distance(lat, lon, ref_lat, ref_lon) if dist > 500: # Returning reference prevents map explosion return ref_lat, ref_lon return lat, lon async def optimize_provider_payload( self, orders: _List[Dict[str, Any]], start_coords: Optional[tuple] = None ) -> _List[Dict[str, Any]]: """Optimize delivery route and add step metrics (OR-Tools).""" if not orders: return [] # Deep copy orders = [dict(order) for order in orders] # 0. KALMAN FILTER - Smooth noisy delivery GPS coordinates orders = smooth_order_coordinates(orders) # Helpers def _to_float(v: Any) -> float: try: return float(v) except: return 0.0 def _normalize_dt(val: Any) -> str: if val in (None, "", 0): return "" s = str(val).strip() for fmt in ("%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%d %H:%M:%S"): try: return datetime.strptime(s, fmt).strftime("%Y-%m-%d %H:%M:%S") except: pass return s # 1. PREPARE COORDINATES & CENTROID valid_lats = [] valid_lons = [] for o in orders: lat = _to_float(o.get("deliverylat")) lon = _to_float(o.get("deliverylong")) if lat != 0 and lon != 0: valid_lats.append(lat) valid_lons.append(lon) centroid_lat = sum(valid_lats) / len(valid_lats) if valid_lats else 0.0 centroid_lon = sum(valid_lons) / len(valid_lons) if valid_lons else 0.0 # 2. DETERMINE START LOCATION (With Fix) start_lat, start_lon = 0.0, 0.0 # Try explicit start_coords first if start_coords and len(start_coords) == 2: try: start_lat, start_lon = float(start_coords[0]), float(start_coords[1]) except: pass # Fallback to pickup location in orders if start_lat == 0: for o in orders: plat = _to_float(o.get("pickuplat")) plon = _to_float(o.get("pickuplon") or o.get("pickuplong")) if plat != 0: start_lat, start_lon = plat, plon break # Fallback to centroid if start_lat == 0: start_lat, start_lon = centroid_lat, centroid_lon # FIX BAD START COORDINATES start_lat, start_lon = self._cleanup_coords( start_lat, start_lon, centroid_lat, centroid_lon ) # 3. BUILD LOCATIONS LIST FOR SOLVER # Index 0 is Start (Depot), 1..N are orders locations = [(start_lat, start_lon)] points_map = [] # solver_idx-1 -> original order index order_to_delivery_loc = {} # original order index -> (lat, lon) for idx, order in enumerate(orders): lat = _to_float(order.get("deliverylat")) lon = _to_float(order.get("deliverylong")) # Project coordinates and ensure they are strings for Go compatibility lat, lon = self._cleanup_coords(lat, lon, centroid_lat, centroid_lon) order_str_lat, order_str_lon = str(lat), str(lon) order["deliverylat"] = order_str_lat order["deliverylong"] = order_str_lon if "droplat" in order: order["droplat"] = order_str_lat if "droplon" in order: order["droplon"] = order_str_lon locations.append((lat, lon)) points_map.append(idx) order_to_delivery_loc[idx] = (lat, lon) # 4. COMPUTE DISTANCE MATRICES lats = np.array([loc[0] for loc in locations]) lons = np.array([loc[1] for loc in locations]) # aerial_matrix - pure straight-line km; used for step ORDERING. # Road-distance ordering causes non-intuitive sequences: a delivery # that is physically 200 m away can appear "far" because the road # route loops around. Riders handle U-turns themselves - we just # tell them the nearest unvisited point by crow-flies distance. # dist_matrix - aerial x road_factor; used for ETA/cost analytics only. aerial_matrix = calculate_haversine_matrix_vectorized(lats, lons) dist_matrix = aerial_matrix * self.road_factor traffic = self.get_traffic_condition() # 5b. MULTI-KITCHEN BEATMAP DETECTION # If a rider's orders come from more than one kitchen, use kitchen-aware # sequencing: visit Kitchen A -> deliver A's orders -> Kitchen B -> deliver B's. # Only count real kitchens (exclude the no-pickup-coords bucket). kitchen_groups = self._group_by_kitchen(orders, _to_float) real_kitchen_count = sum(1 for k in kitchen_groups if k[0] != "?") if real_kitchen_count > 1: logger.info( f"Beatmap routing: {len(orders)} orders across {len(kitchen_groups)} kitchens" ) return await self._beatmap_route( orders, kitchen_groups, order_to_delivery_loc, start_lat, start_lon, _to_float, _normalize_dt ) # 6. EXTRACT TIME WINDOWS (one per node, index 0 = depot/kitchen) # -- Why here and not in the solver? ------------------------------- # pickupSlot / pickuptime is an ORDER-level field that the TSP solver # doesn't know about by itself. We convert them to minutes-from-now # so OR-Tools can enforce delivery deadlines without external API calls. # ------------------------------------------------------------------ time_windows: Optional[_List[Tuple[int, int]]] = None try: from dateutil.parser import parse as _parse_dt _now = datetime.now() tw = [(0, 180)] # depot (kitchen): always open within 3-hour horizon has_any_window = False for order in orders: slot = ( order.get("pickupSlot") or order.get("pickupslot") or order.get("pickuptime") or order.get("pickup_slot") ) if slot: try: t = _parse_dt(str(slot)) # Earliest: rider shouldn't deliver before kitchen is ready # Latest: food should be delivered within 45 min of ready ready_min = max(0, int((t - _now).total_seconds() / 60)) tw.append((0, ready_min + 45)) has_any_window = True except Exception: tw.append((0, 180)) else: tw.append((0, 180)) if has_any_window: time_windows = tw logger.debug(f"Time windows active for {sum(1 for w in tw if w[1] < 180)} orders") except Exception as _twe: logger.debug(f"Time window extraction skipped: {_twe}") # 6b. STEP ORDERING # Prefer a road-aware visiting order (Google optimize:true) when enabled; # it respects one-ways/turn restrictions that straight-line ordering can't. # Fall back to aerial greedy nearest-neighbour + 2-opt (the default): greedy # picks the closest unvisited stop, 2-opt removes crossing segments. Step # metrics + ETA below stay aerial-based regardless, so the empirical ETA # model (keyed on aerial buckets) remains valid. optimized_order_indices = None road_order = await self._road_optimal_order( start_lat, start_lon, [locations[i] for i in range(1, len(locations))] ) if road_order is not None: # road_order indexes into deliveries (0-based); solver index = wp + 1 optimized_order_indices = [wp + 1 for wp in road_order] else: route_indices = self._solve_greedy(locations, aerial_matrix) route_indices = self._two_opt_improve(route_indices, aerial_matrix) optimized_order_indices = [i for i in route_indices if i != 0] # 7. BUILD RESULT result = [] cumulative_dist = 0.0 cumulative_eta_min = 0 # total minutes from kitchen -> current delivery prev_idx = 0 # starts at depot (kitchen / start location) for step_num, solver_idx in enumerate(optimized_order_indices, start=1): order_idx = points_map[solver_idx - 1] order = dict(orders[order_idx]) # Clean routing fields (will be recalculated) for k in ("step", "previouskms", "cumulativekms", "eta", "actualkms", "ordertype"): order.pop(k, None) # Normalize dates for field in ["orderdate", "deliverytime", "created"]: if field in order: order[field] = _normalize_dt(order.get(field)) # Leg distance - aerial km (matches the ordering metric) step_dist = float(aerial_matrix[prev_idx][solver_idx]) cumulative_dist += step_dist # Step metadata order["step"] = int(step_num) order["previouskms"] = int(round(step_dist)) # Bug fix: was hardcoded 0 for step 1 order["cumulativekms"] = int(round(cumulative_dist)) # actualkms = direct pickup-to-door distance (for billing). # Always the order's own kitchen, never the routing depot: when # start_coords is the rider's current position (e.g. reconcile # mid-route) rather than the kitchen, start_lat/start_lon no # longer equals the pickup location. plat, plon = ( _to_float(order.get("pickuplat")), _to_float(order.get("pickuplon") or order.get("pickuplong")), ) if plat == 0: plat, plon = start_lat, start_lon dlat, dlon = locations[solver_idx] true_dist = self.haversine_distance(plat, plon, dlat, dlon) * 1.3 provided_kms = order.get("kms") if provided_kms not in (None, "", 0, "0"): try: true_dist = float(provided_kms) except: pass order["actualkms"] = str(round(true_dist, 2)) order["kms"] = str(provided_kms) if provided_kms else str(int(round(true_dist))) if "rider_charge" in order: order["rider_charge"] = round(float(order["rider_charge"]), 2) if "profit" in order: order["profit"] = round(float(order["profit"]), 2) order["ordertype"] = ( "Economy" if true_dist <= 5 else "Premium" if true_dist <= 12 else "Risky" ) leg_eta = self.eta_calculator.calculate_eta( distance_km=step_dist, is_first_order=(step_num == 1), order_type=order["ordertype"], time_of_day=traffic, kitchen=order.get("pickupcustomer") or order.get("locationname"), drop_coords=(dlat, dlon), rider_id=order.get("userid"), ) # -- CUMULATIVE ETA -------------------------------------------- # `eta` = leg time from previous stop (legacy field, kept) # `cumulative_eta` = total time from kitchen -> THIS delivery # Customers should use cumulative_eta for "when does my food arrive" # ------------------------------------------------------------- cumulative_eta_min += leg_eta order["eta"] = str(leg_eta) order["cumulative_eta"] = str(cumulative_eta_min) result.append(order) prev_idx = solver_idx return result # ------------------------------------------------------------------ # Multi-kitchen beatmap helpers # ------------------------------------------------------------------ def _group_by_kitchen(self, orders: _List[Dict], _to_float) -> Dict[tuple, _List[int]]: """ Group order indices by their kitchen (pickup) location. Returns {(rounded_lat, rounded_lon): [order_idx, ...]} Keys rounded to ~100 m so nearby pickup points merge into one kitchen. """ from collections import defaultdict groups: Dict[tuple, _List[int]] = defaultdict(list) for idx, order in enumerate(orders): plat = _to_float(order.get("pickuplat")) plon = _to_float(order.get("pickuplon") or order.get("pickuplong")) if plat == 0 or plon == 0: # No pickup coords - append to a special "no-kitchen" bucket groups[("?", "?")].append(idx) else: key = (round(plat, 3), round(plon, 3)) groups[key].append(idx) return dict(groups) async def _beatmap_route( self, orders: _List[Dict], kitchen_groups: Dict[tuple, _List[int]], order_to_delivery_loc: Dict[int, tuple], start_lat: float, start_lon: float, _to_float, _normalize_dt, ) -> _List[Dict]: """ Multi-kitchen beatmap: for each kitchen (nearest first) sequence the deliveries by aerial nearest-neighbour, then concatenate. Sequencing: greedy nearest unvisited stop by straight-line km. Riders handle U-turns; we just point them to the closest next drop. Flow: Kitchen A -> A1 -> A2 -> Kitchen B -> B1 -> B2 -> Steps are numbered continuously across all kitchens. """ result: _List[Dict] = [] global_step = 1 cumulative_dist = 0.0 cumulative_eta_min = 0 # total minutes from start -> current delivery traffic = self.get_traffic_condition() # Separate kitchens with real coords from the no-kitchen bucket real_kitchens = {k: v for k, v in kitchen_groups.items() if k[0] != "?"} no_kitchen_indices = kitchen_groups.get(("?", "?"), []) # Visit kitchens: nearest to start first, then nearest to last delivery remaining = list(real_kitchens.items()) current_pos = (start_lat, start_lon) while remaining: # Find nearest unvisited kitchen from current position best_i = min( range(len(remaining)), key=lambda i: self.haversine_distance( current_pos[0], current_pos[1], remaining[i][0][0], remaining[i][0][1] ), ) k_key, k_order_indices = remaining.pop(best_i) k_lat, k_lon = float(k_key[0]), float(k_key[1]) # Build mini location list: kitchen at index 0, deliveries at 1..N k_locs = [(k_lat, k_lon)] k_idx_to_order = [] # k_solver_idx-1 -> original order idx for order_idx in k_order_indices: dlat, dlon = order_to_delivery_loc.get(order_idx, (0.0, 0.0)) k_locs.append((dlat, dlon)) k_idx_to_order.append(order_idx) # Pure aerial distance matrix for this kitchen group k_lats = np.array([loc[0] for loc in k_locs]) k_lons = np.array([loc[1] for loc in k_locs]) k_aerial = calculate_haversine_matrix_vectorized(k_lats, k_lons) # Road-aware order for this kitchen's drops (opt-in, cached); # else aerial greedy nearest-neighbour + 2-opt to remove crossings. k_road = await self._road_optimal_order( k_lat, k_lon, [k_locs[i] for i in range(1, len(k_locs))] ) if k_road is not None: k_delivery_seq = [wp + 1 for wp in k_road] else: k_route = self._solve_greedy(k_locs, k_aerial) k_route = self._two_opt_improve(k_route, k_aerial) k_delivery_seq = [i for i in k_route if i != 0] k_prev_idx = 0 # start from kitchen for k_solver_idx in k_delivery_seq: order_idx = k_idx_to_order[k_solver_idx - 1] order = dict(orders[order_idx]) # Clean routing fields for fld in ("step", "previouskms", "cumulativekms", "eta", "actualkms", "ordertype"): order.pop(fld, None) for field in ["orderdate", "deliverytime", "created"]: if field in order: order[field] = _normalize_dt(order.get(field)) # Aerial leg distance (consistent with ordering metric) step_dist = float(k_aerial[k_prev_idx][k_solver_idx]) cumulative_dist += step_dist dlat, dlon = k_locs[k_solver_idx] true_dist = self.haversine_distance(k_lat, k_lon, dlat, dlon) * 1.3 provided_kms = order.get("kms") if provided_kms not in (None, "", 0, "0"): try: true_dist = float(provided_kms) except Exception: pass order["step"] = global_step order["previouskms"] = int(round(step_dist)) order["cumulativekms"] = int(round(cumulative_dist)) order["actualkms"] = str(round(true_dist, 2)) order["kms"] = str(provided_kms) if provided_kms else str(int(round(true_dist))) if "rider_charge" in order: order["rider_charge"] = round(float(order["rider_charge"]), 2) if "profit" in order: order["profit"] = round(float(order["profit"]), 2) order["ordertype"] = ( "Economy" if true_dist <= 5 else "Premium" if true_dist <= 12 else "Risky" ) leg_eta = self.eta_calculator.calculate_eta( distance_km=step_dist, is_first_order=(global_step == 1), order_type=order["ordertype"], time_of_day=traffic, kitchen=order.get("pickupcustomer") or order.get("locationname"), drop_coords=(dlat, dlon), rider_id=order.get("userid"), ) cumulative_eta_min += leg_eta order["eta"] = str(leg_eta) order["cumulative_eta"] = str(cumulative_eta_min) result.append(order) k_prev_idx = k_solver_idx global_step += 1 # Next kitchen search starts from the last delivery of this kitchen if k_delivery_seq: current_pos = k_locs[k_delivery_seq[-1]] # Append orders with no pickup coords at the end for order_idx in no_kitchen_indices: order = dict(orders[order_idx]) for fld in ("step", "previouskms", "cumulativekms", "eta", "actualkms", "ordertype", "cumulative_eta"): order.pop(fld, None) order["step"] = global_step order["previouskms"] = 0 order["cumulativekms"] = int(round(cumulative_dist)) order["actualkms"] = "0" # Bug fix: kms was missing for no-kitchen orders; set it consistently # with the normal path so downstream consumers always find the field. provided_kms = order.get("kms") order["kms"] = str(provided_kms) if provided_kms not in (None, "", 0, "0") else "0" order["ordertype"] = "Economy" order["eta"] = "15" cumulative_eta_min += 15 order["cumulative_eta"] = str(cumulative_eta_min) result.append(order) global_step += 1 return result