quantum/solvers/ILP_solver.py

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],
            },
        }