File size: 6,287 Bytes
beeea66 | 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 | import inspect
import pyomo.environ as pyo
from typing import Any, Dict
from .base_solver import BaseSolver
class ILPSolver(BaseSolver):
"""
Exact MAPF solver via Pyomo ILP (hard constraints, no penalties). Works
with either GridILPBuilder or GraphILPBuilder — this class never branches
on grid vs. graph itself, it just asks the builder for vars_per_time and
local_index(v) to translate a solved Pyomo variable into a flat index.
Solves the whole time horizon in a single shot — no windowing, no
sampling — then packs the result into the same flat (robot, time,
position) variable index QUBO solvers use, so decode_path() and every
other BaseSolver post-processing helper work unchanged.
"""
def __init__(
self,
pyomo_solver_name="appsi_highs",
normalize_scale=0,
num_reads=1,
max_corrections=0,
verbose_level=2,
time_limit=30,
**kwargs,
):
"""
Args:
time_limit: Max solve time in seconds (default 30). Nothing else
in this codebase bounds ILP's runtime — unlike QAOA's fixed
opt_steps budget, branch-and-bound can otherwise run
arbitrarily long as problems scale up. If the limit is hit
before HiGHS proves optimality, it returns its best
incumbent so far (termination_condition reflects this —
check it before trusting `energy` as the true optimum).
"""
super().__init__(
solver="ilp",
normalize_scale=normalize_scale,
num_reads=num_reads,
max_corrections=max_corrections,
verbose_level=verbose_level,
pyomo_solver_name=pyomo_solver_name,
time_limit=time_limit, # forwarded so to_dict()/the manifest records it too
**kwargs,
)
self.pyomo_solver_name = pyomo_solver_name
self.time_limit = time_limit
def solve(self, builder, optimization=False, preprocess=True) -> Dict[str, Any]:
"""
Solve the ILP model held by builder.
Args:
builder: GridILPBuilder or GraphILPBuilder instance
optimization: Accepted for interface compatibility; unused (no
variational step for an exact ILP solve).
preprocess: When True (default), builder.build() fixes x[a, v, t]
to 0 for every BFS-from-start-unreachable (v, t) — see
BaseILPBuilder.build(). When False, only the per-robot active
time window is fixed; the full free-cell set stays open.
Always rebuilds (even if builder.model already exists) so
this flag reliably takes effect regardless of any prior
build() call — no windowing/sampling either way, ILP solves
the whole horizon exactly in one shot.
Returns:
Dictionary containing solution, energy, and raw response
"""
builder.build(preprocess=preprocess)
pyomo_solver = pyo.SolverFactory(self.pyomo_solver_name)
if "timelimit" in inspect.signature(pyomo_solver.solve).parameters:
# APPSI-backed solvers (appsi_highs, appsi_cbc, ...) — Pyomo's
# LegacySolverInterface.solve() replaces self.config with a fresh
# default config on every call and only reads time_limit from
# this kwarg, so setting pyomo_solver.config.time_limit ahead of
# time (the obvious-looking approach) is silently discarded.
results = pyomo_solver.solve(builder.model, timelimit=self.time_limit)
else:
# Plain legacy plugin backends (cbc, glpk, ...) — no shared
# kwarg; option name varies by solver (cbc wants 'seconds',
# glpk wants 'tmlim'). 'time_limit' covers the common case but
# isn't universal for every possible backend.
pyomo_solver.options["time_limit"] = self.time_limit
results = pyomo_solver.solve(builder.model)
problem = builder.problem
robot_nums = problem.get_robot_nums()
T = problem.T
vars_per_time = builder.vars_per_time
total_vars = vars_per_time * T * problem.num_robots
solution = {idx: 0 for idx in range(total_vars)}
for a in builder.model.A:
robot_num = robot_nums[a]
robot_offset = robot_num * (vars_per_time * T)
for t in builder.model.T:
for v in builder.model.V:
if pyo.value(builder.model.x[a, v, t]) > 0.5:
solution[robot_offset + t * vars_per_time + builder.local_index(v)] = 1
break
self.logger.standard(
f"ILP solve complete: {results.solver.termination_condition}, "
f"energy={pyo.value(builder.model.obj)}"
)
# Write the solved paths back onto problem.robots — QUBO solvers do
# this as a side effect of their windowed loop (builder.update_problem());
# ILP has no windowing loop, so it has to happen explicitly here. Callers
# (BenchmarkRunner, qubo_cli.py's output printing) read robot.path directly.
path = self.decode_path(solution, problem)
num_to_id = {num: rid for rid, num in robot_nums.items()}
for robot_num, coords in self.get_robot_paths(path).items():
robot = problem.robots[num_to_id[robot_num]]
robot.path = coords
robot.current_position = coords[-1][:2]
robot.active = False
return {
"solution": solution,
"energy": pyo.value(builder.model.obj),
"raw_response": results,
"metadata": {
"termination_condition": str(results.solver.termination_condition),
"solver_config": self.to_dict(),
# Single-entry list, not a per-window loop like QUBO's — ILP
# solves the whole horizon in one shot. Wrapped in a list so
# BenchmarkRunner.run_build()'s existing window_stats
# aggregation (benchmark.py) picks it up unchanged.
"window_stats": [builder.bfs_stats],
},
}
|