1"""Order -> van assignment as a CSP.2 3Variables X_o for every order o4Domains the vans {0, 1, 2}5Constraints6 unary window reachable from the depot by that van: leaving at shift start,7 max(shift_start + T(depot, o), e_o) <= l_o, and (if returning) the van can still8 get back to the depot before its shift ends.9 binary X_a != X_b ("different van") when the predicted times make it impossible to10 serve both windows in EITHER order (depot -> a -> b and depot -> b -> a), using the11 earliest shift start (a necessary condition for every van). These edges come from12 the NN time matrix. The integration loop adds more ("learned") edges.13 global capacity: sum of demands <= van capacity, number of stops <= max_stops.14 15Solvers (switchable parts)16 BT chronological backtracking, checks constraints against past assignments17 FC forward checking: after X_i = v, delete v from unassigned neighbours' domains and18 from any order that no longer fits in v's remaining capacity / stop budget19 AC-3 arc consistency on the "different van" constraints, run once before search and20 then after every assignment (MAC) on top of FC21Variable order: static, most-constrained (highest degree) first, ties by earliest window.22Value order: least-loaded van first (deterministic; identical for all three solvers, so they23return the same first solution and differ only in how much of the tree they explore).24 25Counters: assignments (values given to variables), backtracks (assignments undone because26their subtree had no solution), checks, fc_prunes, ac3_revisions, ac3_prunes.27"""28from __future__ import annotations29 30import time31from collections import deque32from dataclasses import dataclass, field33 34 35def fmt_t(m: float) -> str:36 m = int(round(m))37 return f"{m // 60:02d}:{m % 60:02d}"38 39 40@dataclass41class CSPResult:42 status: str # 'solved' | 'unsat' | 'limit'43 assignment: dict # order id -> van id44 stats: dict = field(default_factory=dict)45 46 47class AssignmentCSP:48 def __init__(self, instance, tm, learned_edges=()):49 self.inst, self.tm = instance, tm50 self.orders = list(instance.orders)51 self.ids = [o.id for o in self.orders]52 self.vans = list(instance.vans)53 self.unary_reasons: dict = {}54 self.domains0 = {o.id: self._unary(o) for o in self.orders}55 self.edges: dict = {} # frozenset({a,b}) -> 'nn' | 'learned'56 self.edge_reason: dict = {}57 self._pairwise()58 for a, b in learned_edges:59 self.add_edge(a, b, "learned")60 61 # ------------------------------------------------------------------ constraint building62 def _unary(self, o):63 dom, depot, tm = [], self.inst.depot, self.tm64 for v in self.vans:65 t = v.shift_start66 arr = t + tm.tt(depot, o.zone, t)67 beg = max(arr, o.earliest)68 back = beg + o.service + (tm.tt(o.zone, depot, beg + o.service) if self.inst.return_to_depot else 0)69 if beg > o.latest + 1e-9:70 self.unary_reasons.setdefault(o.id, []).append(71 f"van {v.id}: earliest arrival {fmt_t(arr)} is after the window closes {fmt_t(o.latest)}")72 elif back > v.shift_end + 1e-9:73 self.unary_reasons.setdefault(o.id, []).append(74 f"van {v.id}: cannot be back at the depot before shift end {fmt_t(v.shift_end)}")75 elif o.demand > v.capacity:76 self.unary_reasons.setdefault(o.id, []).append(77 f"van {v.id}: demand {o.demand} exceeds capacity {v.capacity}")78 else:79 dom.append(v.id)80 return dom81 82 def pair_feasible(self, a, b) -> bool:83 """Can one van serve a then b on time, starting from the depot at the earliest shift?"""84 tm, depot = self.tm, self.inst.depot85 s0 = min(v.shift_start for v in self.vans)86 ta = max(s0 + tm.tt(depot, a.zone, s0), a.earliest)87 if ta > a.latest + 1e-9:88 return False89 d = ta + a.service90 tb = max(d + tm.tt(a.zone, b.zone, d), b.earliest)91 return tb <= b.latest + 1e-992 93 def _pairwise(self):94 for i, a in enumerate(self.orders):95 for b in self.orders[i + 1:]:96 if not self.pair_feasible(a, b) and not self.pair_feasible(b, a):97 self.add_edge(a.id, b.id, "nn")98 99 def add_edge(self, a, b, source):100 key = frozenset((a, b))101 if a != b and key not in self.edges:102 self.edges[key] = source103 104 def neighbors(self):105 nb = {i: set() for i in self.ids}106 for e in self.edges:107 a, b = tuple(e)108 nb[a].add(b); nb[b].add(a)109 return nb110 111 # ------------------------------------------------------------------ solving112 def solve(self, fc: bool = False, ac3: bool = False, max_assignments: int = 300_000,113 time_limit: float = 60.0, var_order: str = "degree") -> CSPResult:114 """var_order: 'degree' (most constrained first) or 'input' (order ids as given)."""115 t0 = time.perf_counter()116 nb = self.neighbors()117 demand = {o.id: o.demand for o in self.orders}118 cap = {v.id: v.capacity for v in self.vans}119 maxs = {v.id: v.max_stops for v in self.vans}120 D = {i: set(self.domains0[i]) for i in self.ids}121 st = {"solver": "BT" + ("+FC" if fc else "") + ("+AC3" if ac3 else ""), "assignments": 0,122 "backtracks": 0, "checks": 0, "fc_prunes": 0, "ac3_revisions": 0, "ac3_prunes": 0}123 if var_order == "degree":124 order = sorted(self.ids, key=lambda i: (-len(nb[i]), self.inst.order(i).earliest, i))125 else:126 order = list(self.ids)127 A: dict = {}128 load = {v: 0 for v in cap}129 count = {v: 0 for v in cap}130 trail: list = [] # (var, value) removed from domains131 132 def remove(var, val):133 D[var].discard(val)134 trail.append((var, val))135 136 def undo(mark):137 while len(trail) > mark:138 var, val = trail.pop()139 D[var].add(val)140 141 def ac3_run(queue) -> bool:142 q = deque(queue)143 while q:144 x, y = q.popleft()145 st["ac3_revisions"] += 1146 if len(D[y]) == 1: # x != y : only a singleton domain can prune147 (b,) = tuple(D[y])148 if b in D[x]:149 remove(x, b); st["ac3_prunes"] += 1150 if not D[x]:151 return False152 q.extend((z, x) for z in nb[x] if z != y and z not in A)153 return True154 155 if any(not D[i] for i in self.ids):156 st["time_s"] = time.perf_counter() - t0157 return CSPResult("unsat", {}, st)158 if ac3 and not ac3_run([(x, y) for x in self.ids for y in nb[x]]):159 st["time_s"] = time.perf_counter() - t0160 return CSPResult("unsat", {}, st)161 162 status = {"limit": False}163 164 def consistent(i, v) -> bool:165 st["checks"] += 1166 if load[v] + demand[i] > cap[v] or count[v] + 1 > maxs[v]:167 return False168 return all(A.get(j) != v for j in nb[i])169 170 def bt(k) -> bool:171 if k == len(order):172 return True173 if st["assignments"] >= max_assignments or time.perf_counter() - t0 > time_limit:174 status["limit"] = True175 return False176 i = order[k]177 for v in sorted(D[i], key=lambda v: (count[v], v)):178 if not fc and not consistent(i, v):179 continue180 A[i] = v; load[v] += demand[i]; count[v] += 1181 st["assignments"] += 1182 mark = len(trail)183 ok = True184 if fc:185 changed = []186 for u in order[k + 1:]:187 if v in D[u] and (u in nb[i] or load[v] + demand[u] > cap[v] or count[v] >= maxs[v]):188 remove(u, v); st["fc_prunes"] += 1; changed.append(u)189 if not D[u]:190 ok = False191 break192 if ok and ac3:193 ok = ac3_run([(z, u) for u in changed for z in nb[u] if z not in A])194 if ok and bt(k + 1):195 return True196 undo(mark)197 del A[i]; load[v] -= demand[i]; count[v] -= 1198 st["backtracks"] += 1199 if status["limit"]:200 return False201 return False202 203 found = bt(0)204 st["time_s"] = time.perf_counter() - t0205 if found:206 return CSPResult("solved", dict(A), st)207 return CSPResult("limit" if status["limit"] else "unsat", {}, st)208 209 # ------------------------------------------------------------------ diagnostics210 def max_clique(self):211 nb = self.neighbors()212 best: list = []213 214 def bk(R, P, X):215 nonlocal best216 if not P and not X:217 if len(R) > len(best):218 best = list(R)219 return220 if len(R) + len(P) <= len(best):221 return222 u = max(P | X, key=lambda z: len(nb[z] & P))223 for w in list(P - nb[u]):224 bk(R | {w}, P & nb[w], X & nb[w])225 P = P - {w}; X = X | {w}226 227 bk(set(), set(self.ids), set())228 return sorted(best)229 230 231def check_assignment(csp: AssignmentCSP, assignment: dict) -> list:232 """Returns a list of violated constraints (empty = valid). Used by the tests."""233 bad = []234 for i in csp.ids:235 if assignment.get(i) not in csp.domains0[i]:236 bad.append(f"order {i}: van {assignment.get(i)} not in domain {csp.domains0[i]}")237 for e in csp.edges:238 a, b = tuple(e)239 if assignment.get(a) == assignment.get(b):240 bad.append(f"orders {a},{b} share van {assignment.get(a)} but must differ")241 for v in csp.vans:242 mine = [csp.inst.order(i) for i in csp.ids if assignment.get(i) == v.id]243 if sum(o.demand for o in mine) > v.capacity:244 bad.append(f"van {v.id}: capacity exceeded")245 if len(mine) > v.max_stops:246 bad.append(f"van {v.id}: too many stops")247 return bad