"""Exact per-adjuster sequencing for pre-assigned days (--solver sequence). When an upstream system has already decided WHICH adjuster handles WHICH claims (an assigned_to column in claims.csv), the remaining problem per adjuster is a prize-collecting Traveling Salesman Problem with Time Windows: pick the stop order - and, when not everything fits, the served subset - minimizing driving minutes plus the usual priority-scaled drop penalties (config.effective_penalty: 10^6 must-today, 3,000 high, 600 normal, all escalating 25% per day of age). At real per-adjuster sizes (a dozen claims or fewer) this is solved EXACTLY by branch and bound over stop orders - no heuristic, a provable optimum per adjuster, and the day's total is the sum of independent optima. The timing rules mirror solver.py precisely, including its slack semantics: feasibility is checked by propagating the INTERVAL of possible departure times (not a single earliest time), so service may be deliberately delayed inside an earlier window to keep a later wait under the 240-minute cap - exactly what the OR-Tools time dimension's slack variables and free route start allow. Service must start and finish inside the window, and the route must return home by shift end. Claims whose assignment can never work (missing/unknown adjuster, wrong skill, outside the service territory) and claims that simply do not fit the day are dropped, penalized, and reported with a reason so the dispatcher can send them back upstream for rescheduling. """ from __future__ import annotations import time import config from data_gen import Adjuster, Claim from setpartition import _latest_start from solver import Route, Solution, Stop def _step(dep_lo, dep_hi, prev, cid, k, by_id, node, travel_min): """One interval-propagation step: given the feasible departure-time range [dep_lo, dep_hi] at the previous node, return the feasible service-start range at claim cid, or None if the arc is infeasible. Mirrors the OR-Tools time dimension exactly: the wait at a stop (service start minus physical arrival) may not exceed the slack cap, but the departure from the PREVIOUS node is free within its range - so service can be delayed upstream to absorb a later wait.""" c = by_id[cid] leg = travel_min[prev][node[cid]] t_lo = max(c.window_start, dep_lo + leg) t_hi = min(_latest_start(c), dep_hi + leg + config.MAX_WAIT_MINUTES) return (t_lo, t_hi) if t_lo <= t_hi else None def _best_day(cids, k, adj, node, by_id, travel_min, budget_s: float = 20.0, lunch_break: bool = False, balance: bool = False): """Exact prize-collecting TSPTW for one adjuster's claim list. Branch and bound over stop orders where any suffix of claims may be dropped at its effective penalty. Feasibility uses interval propagation (see _step), so every order OR-Tools could drive is accepted and nothing else. Returns (cost, served_order, truncated); cost = travel minutes + penalties of dropped claims. """ deadline = time.time() + budget_s pen = {c: config.effective_penalty(by_id[c].priority, by_id[c].age_days) for c in cids} pts = [k] + [node[c] for c in cids] min_in = {c: min(travel_min[p][node[c]] for p in pts if p != node[c]) for c in cids} min_home = {c: travel_min[node[c]][k] for c in cids} # Incumbent: serve nothing, pay every penalty, never leave home. best = [sum(pen.values()), []] truncated = [False] seq = [] def dfs(prev, dep_lo, dep_hi, remaining, travel): # Option A: stop here - return home, drop whatever remains. ret = 0 if prev == k else travel_min[prev][k] if prev == k or dep_lo + ret <= adj.shift_end: total = travel + ret + sum(pen[c] for c in remaining) if lunch_break or balance: timed = _min_span_times(list(seq), k, adj, node, by_id, travel_min, lunch_break=lunch_break) if timed is None: total = None # cannot host the break elif balance: total += config.BALANCE_COEFFICIENT * ( timed[2] - timed[1]) if total is not None and total < best[0]: best[0], best[1] = total, list(seq) if time.time() > deadline: truncated[0] = True return if remaining: ret_lb = min([0 if prev == k else travel_min[prev][k]] + [min_home[c] for c in remaining]) lb = travel + ret_lb + sum(min(min_in[c], pen[c]) for c in remaining) if balance: lb += config.BALANCE_COEFFICIENT * ( travel + sum(by_id[c].service_minutes for c in seq)) if lb >= best[0]: return # Option B: serve one more claim next (all choices explored). for cid in sorted(remaining, key=lambda c: -pen[c]): rng = _step(dep_lo, dep_hi, prev, cid, k, by_id, node, travel_min) if rng is None: continue s = by_id[cid].service_minutes remaining.remove(cid) seq.append(cid) dfs(node[cid], rng[0] + s, rng[1] + s, remaining, travel + travel_min[prev][node[cid]]) seq.pop() remaining.add(cid) dfs(k, adj.shift_start, adj.shift_end, set(cids), 0) return best[0], best[1], truncated[0] def _schedule(order, k, adj, node, by_id, travel_min): """Concrete, validator-clean service times for a feasible order: forward interval pass, then a backward pass choosing each departure as early as the NEXT stop's wait cap allows.""" lo_hi = [] dep_lo, dep_hi, prev = adj.shift_start, adj.shift_end, k for cid in order: t_lo, t_hi = _step(dep_lo, dep_hi, prev, cid, k, by_id, node, travel_min) lo_hi.append((t_lo, t_hi)) s = by_id[cid].service_minutes dep_lo, dep_hi, prev = t_lo + s, t_hi + s, node[cid] times = [0] * len(order) for i in range(len(order) - 1, -1, -1): if i == len(order) - 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] - config.MAX_WAIT_MINUTES - leg times[i] = max(lo_hi[i][0], need - c.service_minutes) return times def _min_span_times(order, k, adj, node, by_id, travel_min, lunch_break=False): """Concrete times for a feasible order under OR-Tools semantics, chosen like the OR-Tools finalizers: earliest end, then latest start (minimal span). With lunch_break, tries every placement the MILP's break model allows - before departure, after return, or inside one gap (consuming part of that gap's wait budget) - and returns the minimal-span hosting. Returns (times, start, end) or None if no placement hosts the break.""" L_lo = config.LUNCH_BREAK["earliest_start"] L_hi = config.LUNCH_BREAK["latest_start"] dur = config.LUNCH_BREAK["duration"] cap = config.MAX_WAIT_MINUTES n = len(order) if n == 0: s = adj.shift_start return [], s, s def base_forward(pos=None, start_floor=None, end_cap=None): lo_hi = [] dlo = adj.shift_start if start_floor is None else max( adj.shift_start, start_floor) dhi, prev = adj.shift_end, k for i, cid in enumerate(order): c = by_id[cid] leg = travel_min[prev][node[cid]] if i == pos: if dlo > L_hi: return None t_lo = max(c.window_start, dlo + leg + dur, L_lo + dur) t_hi = min(_latest_start(c), dhi + leg + cap, L_hi + leg + cap) else: t_lo = max(c.window_start, dlo + leg) t_hi = min(_latest_start(c), dhi + leg + cap) if i == n - 1 and end_cap is not None: leg_h = travel_min[node[cid]][k] t_hi = min(t_hi, end_cap - c.service_minutes - leg_h) if t_lo > t_hi: return None lo_hi.append((t_lo, t_hi)) dlo, dhi = t_lo + c.service_minutes, t_hi + c.service_minutes prev = node[cid] return lo_hi def backward(lo_hi, pos=None): times = [0] * n for i in range(n - 1, -1, -1): c = by_id[order[i]] if i == n - 1: times[i] = lo_hi[i][0] if pos == n and times[i] + c.service_minutes > L_hi: return None continue leg = travel_min[node[order[i]]][node[order[i + 1]]] t = min(lo_hi[i][1], times[i + 1] - leg - c.service_minutes) if i + 1 == pos: t = min(t, times[i + 1] - leg - dur - c.service_minutes, L_hi - c.service_minutes) t = max(t, times[i + 1] - cap - leg - c.service_minutes) if t < lo_hi[i][0]: return None times[i] = t return times def endpoints(times, pos): leg1 = travel_min[k][node[order[0]]] leg_h = travel_min[node[order[-1]]][k] last_dep = times[-1] + by_id[order[-1]].service_minutes if pos == 0: start = max(adj.shift_start, min(times[0] - leg1 - dur, L_hi)) start = max(start, times[0] - cap - leg1) else: start = times[0] - leg1 if pos == n: end = max(last_dep + leg_h + dur, L_lo + dur) else: end = last_dep + leg_h if end > adj.shift_end or start < adj.shift_start: return None return start, end # base timing (no lunch, or lunch outside the span) lo_hi = base_forward() if lo_hi is None: return None times = backward(lo_hi) if times is None: return None ep = endpoints(times, None) if ep is None: return None if not lunch_break: return times, ep[0], ep[1] best = None if ep[0] >= L_lo + dur or ep[1] <= L_hi: # break outside the span best = (times, ep[0], ep[1]) if best is None: # a delayed departure (same order) may host the break BEFORE # the route: retime with the departure floored at L_lo + dur lh = base_forward(start_floor=L_lo + dur) if lh is not None: t2 = backward(lh) if t2 is not None: ep2 = endpoints(t2, None) if ep2 is not None and ep2[0] >= L_lo + dur: best = (t2, ep2[0], ep2[1]) if best is None: # or an earlier finish may host it AFTER the route lh = base_forward(end_cap=L_hi) if lh is not None: t2 = backward(lh) if t2 is not None: ep2 = endpoints(t2, None) if ep2 is not None and ep2[1] <= L_hi: best = (t2, ep2[0], ep2[1]) for pos in range(n + 1): lh = base_forward(pos if pos < n else None) if lh is None: continue if pos == n: dep = lh[-1][0] + by_id[order[-1]].service_minutes if dep > L_hi: continue t = backward(lh, pos) if t is None: continue ep2 = endpoints(t, pos) if ep2 is None: continue if best is None or (ep2[1] - ep2[0]) < (best[2] - best[1]): best = (t, ep2[0], ep2[1]) if best is None: return None return best def solve_sequenced(adjusters: list[Adjuster], claims: list[Claim], miles: list[list[float]], travel_min: list[list[int]], time_limit_s: float | None = None, method: str = "enumeration", engine: str = "auto", lunch_break: bool = False, balance: bool = False, ) -> tuple[Solution, dict]: """Sequence a pre-assigned day. Returns (Solution, info). time_limit_s is the TOTAL budget; it is split evenly across the adjusters that have assigned claims (searches finish in milliseconds at realistic sizes, so the guard exists for pathological inputs). method="enumeration" (default): exact branch and bound only. method="milp": the enumeration still runs first (it is essentially free and provides the incumbent), then each adjuster's day is ALSO solved as a tiny single-vehicle MILP - the Section 3.3 model with the assignment fixed - warm-started with the enumeration's answer and certified by Gurobi (engine="auto"/"gurobi"; the free restricted license suffices at these sizes) or HiGHS. The two independent proofs are compared; info["milp"] records per-adjuster status, engine, and agreement. info["drop_reasons"] maps every dropped claim id to a plain-English reason; info["proven_optimal"] is True when every per-adjuster search ran to completion. """ 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) node = {c.claim_id: n_adj + i for i, c in enumerate(claims)} by_id = {c.claim_id: c for c in claims} by_adj = {a.adjuster_id: (k, a) for k, a in enumerate(adjusters)} reasons: dict[str, str] = {} lists: dict[str, list[str]] = {a.adjuster_id: [] for a in adjusters} for c in claims: if not c.assigned_to: reasons[c.claim_id] = ("no assignment - fill the " "assigned_to column in claims.csv") elif c.assigned_to not in by_adj: reasons[c.claim_id] = (f"assigned to unknown adjuster " f"'{c.assigned_to}'") else: k, a = by_adj[c.assigned_to] if c.peril not in a.skills: reasons[c.claim_id] = (f"{a.adjuster_id} lacks the " f"{c.peril} skill") elif (a.max_radius_miles is not None and miles[k][node[c.claim_id]] > a.max_radius_miles): reasons[c.claim_id] = ( f"outside {a.adjuster_id}'s " f"{a.max_radius_miles:.0f}-mile territory") else: lists[a.adjuster_id].append(c.claim_id) active = sum(1 for cids in lists.values() if cids) budget = (max(2.0, time_limit_s / max(1, active)) if time_limit_s else 20.0) chosen: dict[str, list[str]] = {} proven = True milp_info: dict[str, dict] = {} for a in adjusters: cids = lists[a.adjuster_id] if not cids: continue k, _ = by_adj[a.adjuster_id] cost, order, truncated = _best_day(cids, k, a, node, by_id, travel_min, budget_s=budget, lunch_break=lunch_break, balance=balance) proven = proven and not truncated if method == "milp": m_order, exact = _milp_day(a, k, cids, order, node, by_id, miles, travel_min, time_limit_s=max(5, budget), engine=engine, lunch_break=lunch_break, balance=balance) agree = (exact.status == "Optimal" and exact.objective is not None and abs(exact.objective - cost) < 1e-6) milp_info[a.adjuster_id] = { "status": exact.status, "engine": exact.engine, "objective": exact.objective, "agrees_with_enumeration": agree, } if agree: order = m_order # identical cost; take the certified order else: proven = proven and exact.status == "Optimal" chosen[a.adjuster_id] = order for cid in cids: if cid not in order: reasons[cid] = ("does not fit the assigned day " "(windows/shift) - reschedule") routes, served = [], set() for k, a in enumerate(adjusters): seq = chosen.get(a.adjuster_id, []) route = Route(adjuster=a) if lunch_break or balance: timed = _min_span_times(seq, k, a, node, by_id, travel_min, lunch_break=lunch_break) times, t_start, t_end = timed else: times = _schedule(seq, k, a, node, by_id, travel_min) t_start = t_end = None prev = k for cid, t in zip(seq, times): c = by_id[cid] route.stops.append(Stop( claim=c, arrival_min=t, departure_min=t + c.service_minutes, travel_miles_from_prev=miles[prev][node[cid]], travel_min_from_prev=travel_min[prev][node[cid]])) route.total_miles += miles[prev][node[cid]] route.total_travel_min += travel_min[prev][node[cid]] route.total_service_min += c.service_minutes served.add(cid) prev = node[cid] if seq: route.start_min = (t_start if t_start is not None else route.stops[0].arrival_min - route.stops[0].travel_min_from_prev) route.total_miles += miles[prev][k] route.total_travel_min += travel_min[prev][k] route.end_min = (t_end if t_end is not None else route.stops[-1].departure_min + travel_min[prev][k]) else: route.start_min = route.end_min = a.shift_start routes.append(route) dropped = [c for c in claims if c.claim_id not in served] unservable = [c for i, c in enumerate(claims) if not any(config.is_eligible(a, c, miles[k][n_adj + i]) for k, a in enumerate(adjusters))] travel_total = sum(r.total_travel_min for r in routes) objective = travel_total + 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) sol = Solution( routes=routes, dropped=dropped, unservable=unservable, objective=objective, total_miles=sum(r.total_miles for r in routes), total_travel_min=travel_total) info = { "mode": "sequence", "method": method, "proven_optimal": proven, "drop_reasons": reasons, "assigned": {aid: len(cids) for aid, cids in lists.items()}, "served": {aid: len(order) for aid, order in chosen.items()}, } if method == "milp": info["milp"] = milp_info info["milp_all_certified"] = all( v["agrees_with_enumeration"] for v in milp_info.values()) engines = {v["engine"] for v in milp_info.values()} info["engine"] = engines.pop() if len(engines) == 1 else "mixed" return sol, info def _sub_instance(a, k, cids, node, by_id, miles, travel_min): """Slice the full matrices down to one adjuster's day: index 0 is the home, 1..m the assigned claims in list order.""" idx = [k] + [node[c] for c in cids] sub_t = [[travel_min[i][j] for j in idx] for i in idx] sub_m = [[miles[i][j] for j in idx] for i in idx] return [by_id[c] for c in cids], sub_m, sub_t def _warm_solution(a, order, k, cids, node, by_id, miles, travel_min): """Package the enumeration's answer as a Solution so the MILP can start from the (already optimal) incumbent and spend its time on the proof.""" times = _schedule(order, k, a, node, by_id, travel_min) route = Route(adjuster=a) prev = k for cid, t in zip(order, times): c = by_id[cid] route.stops.append(Stop( claim=c, arrival_min=t, departure_min=t + c.service_minutes, travel_miles_from_prev=miles[prev][node[cid]], travel_min_from_prev=travel_min[prev][node[cid]])) prev = node[cid] if order: route.start_min = (route.stops[0].arrival_min - route.stops[0].travel_min_from_prev) route.end_min = (route.stops[-1].departure_min + travel_min[prev][k]) dropped = [by_id[c] for c in cids if c not in order] return Solution(routes=[route], dropped=dropped, unservable=[], objective=0) def _milp_day(a, k, cids, order, node, by_id, miles, travel_min, time_limit_s, engine, lunch_break=False, balance=False): """Solve one adjuster's day as a tiny prize-collecting TSPTW MILP - the Section 3.3 model restricted to a single vehicle - via milp_solver.solve_exact (Gurobi when licensed, HiGHS otherwise). The model is small enough (~180 variables for a 12-claim day) for Gurobi's free restricted license. Returns (order, ExactResult).""" import milp_solver sub_claims, sub_m, sub_t = _sub_instance(a, k, cids, node, by_id, miles, travel_min) # Rebase the warm start onto the sub-instance's indexing. sub_node = {c: 1 + i for i, c in enumerate(cids)} warm = _warm_solution(a, order, 0, cids, {c: sub_node[c] for c in cids}, by_id, sub_m, sub_t) exact = milp_solver.solve_exact([a], sub_claims, sub_t, time_limit_s=time_limit_s, warm_start=warm, miles=sub_m, engine=engine, lunch_break=lunch_break, balance=balance) return list(exact.routes.get(a.adjuster_id, [])), exact