"""Alternative solver backend: PyVRP (iterated local search). Drop-in replacement for solver.solve(): same inputs (adjusters, claims, mileage and travel-minute matrices), same typed Solution/Route/Stop output, so main.py and the report/visualization layers work unchanged. Model mapping (PyVRP >= 0.13): - one depot + one single-vehicle VehicleType per adjuster (their home), with the shift as the vehicle's time window; - one routing *profile* per adjuster: edges exist only between the adjuster's home and the claims they are ELIGIBLE for (skill match AND inside their service territory, config.is_eligible) - other arcs are simply absent, which PyVRP treats as effectively infinite (skills/territory become structurally impossible, not penalized); - claims are optional clients (required=False) carrying their drop penalty as a prize - identical economics to OR-Tools disjunctions; - edge distance = edge duration = integer travel minutes, plus service_duration at each client, so the objective (travel minutes + penalties of dropped claims) matches solver.py exactly. One semantic difference vs. the OR-Tools model: PyVRP has no per-stop waiting cap (OR-Tools caps slack at config.MAX_WAIT_MINUTES). Solutions are validated after extraction; any wait beyond the cap is reported by compare_solvers.py rather than silently accepted. """ from __future__ import annotations import itertools from pyvrp import Model from pyvrp.stop import MaxRuntime import config from data_gen import Adjuster, Claim from solver import Route, Solution, Stop def solve(adjusters: list[Adjuster], claims: list[Claim], miles: list[list[float]], travel_min: list[list[int]], time_limit_s: int = config.SOLVER_TIME_LIMIT_SECONDS, seed: int = 0, warm_start: Solution | None = None, balance: bool = False) -> Solution | None: """Solve with PyVRP. warm_start, if given, is a Solution from another backend (typically OR-Tools) used as PyVRP's initial solution - this removes ILS's random-start warm-up penalty on large instances.""" if any(c.extra_windows for c in claims): raise ValueError( "split-availability claims (multiple time windows) are not " "supported by this backend yet - use ortools, cpsat, or " "milp (exact)") n_adj = len(adjusters) m = Model() profiles = [m.add_profile(name=a.adjuster_id) for a in adjusters] depots = [m.add_depot(x=a.home_lon, y=a.home_lat, name=a.adjuster_id) for a in adjusters] for a, d, p in zip(adjusters, depots, profiles): m.add_vehicle_type(num_available=1, start_depot=d, end_depot=d, tw_early=a.shift_start, tw_late=a.shift_end, unit_distance_cost=1, profile=p, unit_duration_cost=(config.BALANCE_COEFFICIENT if balance else 0), name=a.adjuster_id) clients = [] unservable: list[Claim] = [] qualified: list[list[int]] = [] for c in claims: latest_start = max(c.window_start, c.window_end - c.service_minutes) clients.append(m.add_client( x=c.lon, y=c.lat, service_duration=c.service_minutes, tw_early=c.window_start, tw_late=latest_start, prize=config.effective_penalty(c.priority, c.age_days), required=False, name=c.claim_id)) i = len(qualified) q = [k for k, a in enumerate(adjusters) if config.is_eligible(a, c, miles[k][n_adj + i])] qualified.append(q) if not q: unservable.append(c) # Location index convention matches the matrices: depots are created # first (0 .. n_adj-1), clients follow in claim order. for k, (prof, depot) in enumerate(zip(profiles, depots)): mine = [i for i, q in enumerate(qualified) if k in q] for i in mine: m.add_edge(depot, clients[i], distance=travel_min[k][n_adj + i], duration=travel_min[k][n_adj + i], profile=prof) m.add_edge(clients[i], depot, distance=travel_min[n_adj + i][k], duration=travel_min[n_adj + i][k], profile=prof) for i, j in itertools.permutations(mine, 2): m.add_edge(clients[i], clients[j], distance=travel_min[n_adj + i][n_adj + j], duration=travel_min[n_adj + i][n_adj + j], profile=prof) initial = None if warm_start is not None: from pyvrp import Route as PyRoute, Solution as PySolution data = m.data() node = {c.claim_id: n_adj + i for i, c in enumerate(claims)} adj_index = {a.adjuster_id: k for k, a in enumerate(adjusters)} py_routes = [] for r in warm_start.routes: if r.stops: visits = [node[s.claim.claim_id] for s in r.stops] py_routes.append( PyRoute(data, visits, adj_index[r.adjuster.adjuster_id])) initial = PySolution(data, py_routes) result = m.solve(stop=MaxRuntime(time_limit_s), seed=seed, display=False, collect_stats=False, initial_solution=initial) if result.best is None or not result.is_feasible(): return None routes: list[Route] = [] served_ids: set[str] = set() best_routes = {r.vehicle_type(): r for r in result.best.routes()} for k, adj in enumerate(adjusters): route = Route(adjuster=adj) r = best_routes.get(k) if r is not None and r.visits(): route.start_min = r.start_time() route.end_min = r.end_time() prev = k for visit in r.schedule(): loc = visit.location if loc < n_adj: # depot entries continue claim = claims[loc - n_adj] route.stops.append(Stop( claim=claim, arrival_min=visit.start_service, departure_min=visit.end_service, travel_miles_from_prev=miles[prev][loc], travel_min_from_prev=travel_min[prev][loc], )) route.total_miles += miles[prev][loc] route.total_travel_min += travel_min[prev][loc] route.total_service_min += claim.service_minutes served_ids.add(claim.claim_id) # sanity: skills/territory are structural in this model assert config.is_eligible(adj, claim, miles[k][loc]), ( f"PyVRP assigned {claim.claim_id} to ineligible " f"{adj.adjuster_id}") prev = loc route.total_miles += miles[prev][k] route.total_travel_min += travel_min[prev][k] else: route.start_min = adj.shift_start route.end_min = adj.shift_start routes.append(route) dropped = [c for c in claims if c.claim_id not in served_ids] total_travel = sum(r.total_travel_min for r in routes) objective = total_travel + sum( config.effective_penalty(c.priority, c.age_days) for c in dropped) if balance: objective += config.BALANCE_COEFFICIENT * sum( r.end_min - r.start_min for r in routes if r.stops) return Solution( routes=routes, dropped=dropped, unservable=unservable, objective=objective, total_miles=sum(r.total_miles for r in routes), total_travel_min=total_travel, ) def _retime_with_lunch(order, k, adj, node, by_id, travel_min): """Exact check + retiming: can a 30-minute lunch (starting inside the config window) be inserted into this fixed stop order? Tries every gap with interval propagation under the same slack semantics as solver.py - the lunch consumes part of the gap's 240-minute wait budget, exactly like an OR-Tools break consuming slack. Returns (times, pos) for the first feasible gap - pos is the index of the stop the lunch precedes (len(order) = after the last stop) - or None. The concrete times are chosen so the STAMPED schedule still hosts the break: the backward compression pass never squeezes the hosting gap below lunch size or past the window. """ L_lo = config.LUNCH_BREAK["earliest_start"] L_hi = config.LUNCH_BREAK["latest_start"] dur = config.LUNCH_BREAK["duration"] cap = config.MAX_WAIT_MINUTES def latest(c): return max(c.window_start, c.window_end - c.service_minutes) n = len(order) for pos in range(n + 1): # lunch in the gap before stop `pos` dep_lo, dep_hi, prev = adj.shift_start, adj.shift_end, k lo_hi, ok = [], True for i, cid in enumerate(order): c = by_id[cid] leg = travel_min[prev][node[cid]] if i == pos: l1 = max(L_lo, dep_lo + leg) l2 = min(L_hi, dep_hi + leg + cap - dur) if l1 > l2: ok = False break t_lo = max(c.window_start, l1 + dur) # hostability ceiling: with lunch <= L_hi and the gap # wait capped, service here cannot start later than # L_hi + cap - else the stamped times could not host # the break. t_hi = min(latest(c), dep_hi + leg + cap, L_hi + cap) else: t_lo = max(c.window_start, dep_lo + leg) t_hi = min(latest(c), dep_hi + leg + cap) if t_lo > t_hi: ok = False break lo_hi.append((t_lo, t_hi)) dep_lo, dep_hi = (t_lo + c.service_minutes, t_hi + c.service_minutes) prev = node[cid] if not ok: continue leg_home = 0 if prev == k else travel_min[prev][k] if pos == n: # lunch between last stop and home l1 = max(L_lo, dep_lo) l2 = min(L_hi, dep_hi + cap - dur) if l1 > l2 or l1 + dur + leg_home > adj.shift_end: continue elif dep_lo + leg_home > adj.shift_end: continue times = [0] * n for i in range(n - 1, -1, -1): if i == n - 1: times[i] = lo_hi[i][0] else: c = by_id[order[i]] leg = travel_min[node[order[i]]][node[order[i + 1]]] need = times[i + 1] - cap - leg times[i] = max(lo_hi[i][0], need - c.service_minutes) if i + 1 == pos: # do not compress the hosting gap: this stop must # depart early enough that a 30-min break fits in # the window before the next service starts. ceil = (min(L_hi, times[i + 1] - dur) - leg - c.service_minutes) times[i] = min(times[i], ceil) return times, pos return None def _apply_lunch(sol, adjusters, claims, travel_min): """Prove a lunch fits every driven route and stamp retimed stop times AND lunch-aware route endpoints; returns True, or None if any route cannot host a break (caller falls back to the lunch-honoring construction).""" L_lo = config.LUNCH_BREAK["earliest_start"] L_hi = config.LUNCH_BREAK["latest_start"] dur = config.LUNCH_BREAK["duration"] n_adj = len(adjusters) node = {c.claim_id: n_adj + i for i, c in enumerate(claims)} by_id = {c.claim_id: c for c in claims} for k, route in enumerate(sol.routes): if not route.stops: continue order = [s.claim.claim_id for s in route.stops] res = _retime_with_lunch(order, k, adjusters[k], node, by_id, travel_min) if res is None: return None times, pos = res for stop, t in zip(route.stops, times): stop.arrival_min = t stop.departure_min = t + stop.claim.service_minutes leg1 = route.stops[0].travel_min_from_prev route.start_min = times[0] - leg1 if pos == 0 and route.start_min < L_lo + dur: # the break cannot finish before departure (OR-Tools would # then take it at zero span cost) - intrude minimally: the # latest departure that still hosts a full break en route. route.start_min = max(adjusters[k].shift_start, min(L_hi, times[0] - dur) - leg1) last_dep = route.stops[-1].departure_min leg_home = travel_min[node[order[-1]]][k] route.end_min = last_dep + leg_home if pos == len(order) and route.end_min > L_hi: # break cannot start after returning home - host it in the # last gap, extending the span by exactly the break. route.end_min = max(L_lo, last_dep) + dur + leg_home return True def solve_hybrid(adjusters: list[Adjuster], claims: list[Claim], miles: list[list[float]], travel_min: list[list[int]], time_limit_s: int = config.SOLVER_TIME_LIMIT_SECONDS, seed: int = 0, lunch_break: bool = False, balance: bool = False) -> Solution | None: """Best of both backends: a short OR-Tools run supplies a sensible constructive start, then PyVRP's ILS spends the remaining budget improving it. Falls back gracefully if either stage fails. Toggles: balance is passed natively to both stages (PyVRP via its unit duration cost). The lunch break has no PyVRP construct, so hybrid guarantees it differently: the construction honors it, the polish runs break-unaware, and every polished route must then PASS an exact retiming proof that a 30-minute break fits (trying every insertion position under the full slack semantics). Routes get the retimed schedule stamped in; if any route cannot host a break, the polish is discarded and the lunch-honoring construction returned - the result is never worse than the construction and never carries a fake lunch.""" if any(c.extra_windows for c in claims): raise ValueError( "split-availability claims (multiple time windows) are not " "supported by this backend yet - use ortools, cpsat, or " "milp (exact)") import solver as ortools_solver construct_s = max(2, min(5, time_limit_s // 3)) improve_s = max(1, time_limit_s - construct_s) start = ortools_solver.solve(adjusters, claims, miles, travel_min, time_limit_s=construct_s, lunch_break=lunch_break, balance=balance) result = solve(adjusters, claims, miles, travel_min, time_limit_s=improve_s, seed=seed, warm_start=start, balance=balance) if result is not None and lunch_break: if _apply_lunch(result, adjusters, claims, travel_min) is None: result = None # no room for lunch: fall back elif balance: travel = result.total_travel_min pen = sum(config.effective_penalty(c.priority, c.age_days) for c in result.dropped) result.objective = pen + travel + ( config.BALANCE_COEFFICIENT * sum( r.end_min - r.start_min for r in result.routes if r.stops)) if result is None: return start if start is not None and start.objective < result.objective: return start return result