File size: 9,376 Bytes
7a3d380 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | """Third exact backend: Google OR-Tools CP-SAT via AddCircuit.
Same objective and semantics as every other backend (Section 3.3 of
the technical documentation), but modeled WITHOUT big-M rows: CP-SAT
states each conditional constraint as a reified implication ("if this
arc is driven then T_j >= T_i + service + travel"), and per-adjuster
tours as AddCircuit constraints over eligibility-pruned arc sets. The
logic-heavy constructs that are the compact MILP's bottleneck (Section
3.5) - and that exposed two wrong Gurobi certificates (seeds 3003 and
404) - are native here.
Like the MILP backend it is an anytime exact solver: at the time limit
it returns the best incumbent plus a bound; given enough time it
returns a proof. Unlike Gurobi it has no license size limits (Apache
2.0, ships inside ortools), so it delivers proofs on the Hugging Face
Space where HiGHS runs out of steam - the seed-505 referee instance
(Gurobi ~90 s, HiGHS: no proof) is proven in ~3.5 s. All data is
integer minutes, which is CP-SAT's native currency.
Both optional toggles are modeled: the lunch break as the same
pre/post/on-arc placement disjunction the MILP uses (reified, no
big-M), workload balance as the linear beta * span term.
"""
from ortools.sat.python import cp_model
import config
from data_gen import Adjuster, Claim, arrival_ranges
from solver import Route, Solution, Stop
W = config.MAX_WAIT_MINUTES
LUNCH = config.LUNCH_BREAK
BETA = config.BALANCE_COEFFICIENT
def solve(adjusters: list[Adjuster], claims: list[Claim],
miles: list[list[float]], travel_min: list[list[int]],
time_limit_s: int = 60, lunch_break: bool = False,
balance: bool = False, workers: int = 8):
"""Returns (Solution | None, info dict). info carries the CP-SAT
status name, wall time, proven flag, and the final lower bound."""
n_adj = len(adjusters)
m = cp_model.CpModel()
def node(i): # claim index -> travel-matrix node
return n_adj + i
elig = {k: [i for i, c in enumerate(claims)
if config.is_eligible(a, c, miles[k][node(i)])]
for k, a in enumerate(adjusters)}
unservable = [c for i, c in enumerate(claims)
if not any(i in e for e in elig.values())]
# Service-start time per claim: inside the window AND finishing by
# its end - identical to the OR-Tools model's tightened ranges.
T = {}
for i, c in enumerate(claims):
dom = cp_model.Domain.from_intervals(
[list(r) for r in arrival_ranges(c)])
T[i] = m.new_int_var_from_domain(dom, f"T_{i}")
drop = {i: m.new_bool_var(f"drop_{i}") for i in range(len(claims))}
visit, arcs_by_k, L, E, used = {}, {}, {}, {}, {}
for k, a in enumerate(adjusters):
L[k] = m.new_int_var(a.shift_start, a.shift_end, f"L_{k}")
E[k] = m.new_int_var(a.shift_start, a.shift_end, f"E_{k}")
m.add(E[k] >= L[k])
arcs, lits = [], {}
unused = m.new_bool_var(f"unused_{k}")
used[k] = unused.Not()
arcs.append((0, 0, unused))
for li, i in enumerate(elig[k], start=1):
skip = m.new_bool_var(f"skip_{k}_{i}")
arcs.append((li, li, skip))
visit[k, i] = skip.Not()
m.add_implication(unused, skip)
for li, i in enumerate(elig[k], start=1):
t_hi = travel_min[k][node(i)]
lit = m.new_bool_var(f"x_{k}_h_{i}")
arcs.append((0, li, lit))
lits["h", i] = lit
m.add(T[i] >= L[k] + t_hi).only_enforce_if(lit)
m.add(T[i] <= L[k] + t_hi + W).only_enforce_if(lit)
t_ih = travel_min[node(i)][k]
lit2 = m.new_bool_var(f"x_{k}_{i}_h")
arcs.append((li, 0, lit2))
lits[i, "h"] = lit2
m.add(E[k] >= T[i] + claims[i].service_minutes + t_ih
).only_enforce_if(lit2)
for lj, j in enumerate(elig[k], start=1):
if i == j:
continue
t_ij = travel_min[node(i)][node(j)]
lit3 = m.new_bool_var(f"x_{k}_{i}_{j}")
arcs.append((li, lj, lit3))
lits[i, j] = lit3
dep = T[i] + claims[i].service_minutes
m.add(T[j] >= dep + t_ij).only_enforce_if(lit3)
m.add(T[j] <= dep + t_ij + W).only_enforce_if(lit3)
m.add_circuit(arcs)
m.add(E[k] == L[k]).only_enforce_if(unused)
arcs_by_k[k] = lits
for i in range(len(claims)):
m.add(sum(visit[k, i] for k in range(n_adj) if (k, i) in visit)
+ drop[i] == 1)
if lunch_break:
# Same placement disjunction as the exact MILP, anchored to
# the ROUTE span: the break ends before the route leaves home
# (pre), starts after it returns (post), or rides on exactly
# one traversed arc - INCLUDING the home legs - where travel
# and break must both fit between departure and arrival. An
# earlier draft anchored pre/post to service times and skipped
# the home arcs; adversarial review produced an instance where
# that relaxation certified a break overlapping the drive home.
for k in range(n_adj):
B = m.new_int_var(LUNCH["earliest_start"],
LUNCH["latest_start"], f"B_{k}")
pre = m.new_bool_var(f"pre_{k}")
post = m.new_bool_var(f"post_{k}")
placements = [pre, post]
dur = LUNCH["duration"]
m.add(B + dur <= L[k]).only_enforce_if([pre, used[k]])
m.add(B >= E[k]).only_enforce_if([post, used[k]])
for (i, j), lit in arcs_by_k[k].items():
on = m.new_bool_var(f"on_{k}_{i}_{j}")
m.add_implication(on, lit)
if i == "h":
dep = L[k]
t_ij = travel_min[k][node(j)]
arr = T[j]
elif j == "h":
dep = T[i] + claims[i].service_minutes
t_ij = travel_min[node(i)][k]
arr = E[k]
else:
dep = T[i] + claims[i].service_minutes
t_ij = travel_min[node(i)][node(j)]
arr = T[j]
m.add(B >= dep).only_enforce_if(on)
m.add(B + dur <= arr).only_enforce_if(on)
m.add(arr >= dep + t_ij + dur).only_enforce_if(on)
placements.append(on)
m.add(sum(placements) == 1).only_enforce_if(used[k])
travel_terms = []
for k in range(n_adj):
for (i, j), lit in arcs_by_k[k].items():
fr = k if i == "h" else node(i)
to = k if j == "h" else node(j)
travel_terms.append(travel_min[fr][to] * lit)
pen_terms = [config.effective_penalty(c.priority, c.age_days) * drop[i]
for i, c in enumerate(claims)]
obj = sum(travel_terms) + sum(pen_terms)
if balance:
obj += BETA * sum(E[k] - L[k] for k in range(n_adj))
m.minimize(obj)
s = cp_model.CpSolver()
s.parameters.max_time_in_seconds = float(time_limit_s)
s.parameters.num_workers = workers
status = s.solve(m)
info = {"status": s.status_name(status),
"proven": status == cp_model.OPTIMAL,
"wall_s": round(s.wall_time, 1)}
if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE):
return None, info
info["bound"] = int(s.best_objective_bound)
routes = []
for k, a in enumerate(adjusters):
succ = {i: j for (i, j), lit in arcs_by_k[k].items()
if s.value(lit)}
stops, cur, prev_node = [], succ.get("h"), k
total_mi, total_tm, total_sv = 0.0, 0, 0
while cur is not None and cur != "h":
c = claims[cur]
mi = miles[prev_node][node(cur)]
tm = travel_min[prev_node][node(cur)]
stops.append(Stop(c, s.value(T[cur]),
s.value(T[cur]) + c.service_minutes, mi, tm))
total_mi += mi
total_tm += tm
total_sv += c.service_minutes
prev_node = node(cur)
cur = succ.get(cur)
if stops:
total_mi += miles[prev_node][k]
total_tm += travel_min[prev_node][k]
# Route start/end: L/E carry no objective pressure without the
# balance term, so CP-SAT may leave them anywhere in the wait
# slack. Restamp the tight OR-Tools convention (leave as late
# as possible, return as early as possible) unless a toggle
# makes the model's own values load-bearing (break anchoring /
# span cost).
if not stops:
start = end = a.shift_start
elif lunch_break or balance:
start, end = s.value(L[k]), s.value(E[k])
else:
start = stops[0].arrival_min - stops[0].travel_min_from_prev
end = stops[-1].departure_min + travel_min[prev_node][k]
routes.append(Route(a, stops, start, end,
total_mi, total_tm, total_sv))
dropped = [claims[i] for i in range(len(claims)) if s.value(drop[i])]
sol = Solution(routes, dropped, unservable, int(s.objective_value),
sum(r.total_miles for r in routes),
sum(r.total_travel_min for r in routes))
return sol, info
|