solve.py

Go to the documentation of this file.

#!/usr/bin/env python3
# Copyright 2010-2025 Google LLC
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
 
"""Solve optimization problems, as defined by Model in model.py."""
import types
from typing import Callable, Optional
 
from ortools.math_opt import parameters_pb2
from ortools.math_opt import rpc_pb2
from ortools.math_opt.core.python import solver
from ortools.math_opt.python import callback
from ortools.math_opt.python import compute_infeasible_subsystem_result
from ortools.math_opt.python import errors
from ortools.math_opt.python import init_arguments
from ortools.math_opt.python import message_callback
from ortools.math_opt.python import model
from ortools.math_opt.python import model_parameters
from ortools.math_opt.python import parameters
from ortools.math_opt.python import result
from ortools.util.python import solve_interrupter
from pybind11_abseil.status import StatusNotOk
 
SolveCallback = Callable[[callback.CallbackData], callback.CallbackResult]
 
 
def solve(
    opt_model: model.Model,
    solver_type: parameters.SolverType,
    *,
    params: Optional[parameters.SolveParameters] = None,
    model_params: Optional[model_parameters.ModelSolveParameters] = None,
    msg_cb: Optional[message_callback.SolveMessageCallback] = None,
    callback_reg: Optional[callback.CallbackRegistration] = None,
    cb: Optional[SolveCallback] = None,
    streamable_init_args: Optional[init_arguments.StreamableSolverInitArguments] = None,
    remove_names: bool = False,
    interrupter: Optional[solve_interrupter.SolveInterrupter] = None,
) -> result.SolveResult:
    """Solves an optimization model.
 
    Thread-safety: this function must not be called while modifying the Model
    (adding variables...). Some solvers may add more restriction regarding
    threading. Please see SolverType::XXX documentation for details.
 
    Args:
      opt_model: The optimization model.
      solver_type: The underlying solver to use.
      params: Configuration of the underlying solver.
      model_params: Configuration of the solver that is model specific.
      msg_cb: A callback that gives back the underlying solver's logs by the line.
      callback_reg: Configures when the callback will be invoked (if provided) and
        what data will be collected to access in the callback.
      cb: A callback that will be called periodically as the solver runs.
      streamable_init_args: Configuration for initializing the underlying solver.
      remove_names: When true, remove all names for the ModelProto.
      interrupter: An optional interrupter that the solver can use to interrupt
        the solve early.
 
    Returns:
      A SolveResult containing the termination reason, solution(s) and stats.
 
    Raises:
      RuntimeError: On a solve error.
    """
    # First, initialize optional arguments that were not set to default values.
    # Note that in python, default arguments must be immutable, and these are not.
    params = params or parameters.SolveParameters()
    model_params = model_params or model_parameters.ModelSolveParameters()
    callback_reg = callback_reg or callback.CallbackRegistration()
    streamable_init_args = (
        streamable_init_args or init_arguments.StreamableSolverInitArguments()
    )
    model_proto = opt_model.export_model(remove_names=remove_names)
    proto_cb = None
    if cb is not None:
        proto_cb = lambda x: cb(  # pylint: disable=g-long-lambda
            callback.parse_callback_data(x, opt_model)
        ).to_proto()
    # Solve
    try:
        proto_result = solver.solve(
            model_proto,
            solver_type.value,
            streamable_init_args.to_proto(),
            params.to_proto(),
            model_params.to_proto(),
            msg_cb,
            callback_reg.to_proto(),
            proto_cb,
            interrupter.pybind_interrupter if interrupter is not None else None,
        )
    except StatusNotOk as e:
        raise _status_not_ok_to_exception(e) from None
    return result.parse_solve_result(proto_result, opt_model, validate=False)
 
 
def compute_infeasible_subsystem(
    opt_model: model.Model,
    solver_type: parameters.SolverType,
    *,
    params: Optional[parameters.SolveParameters] = None,
    msg_cb: Optional[message_callback.SolveMessageCallback] = None,
    streamable_init_args: Optional[init_arguments.StreamableSolverInitArguments] = None,
    remove_names: bool = False,
    interrupter: Optional[solve_interrupter.SolveInterrupter] = None,
) -> compute_infeasible_subsystem_result.ComputeInfeasibleSubsystemResult:
    """Computes an infeasible subsystem of the input model.
 
    Args:
      opt_model: The optimization model to check for infeasibility.
      solver_type: Which solver to use to compute the infeasible subsystem. As of
        August 2023, the only supported solver is Gurobi.
      params: Configuration of the underlying solver.
      msg_cb: A callback that gives back the underlying solver's logs by the line.
      streamable_init_args: Configuration for initializing the underlying solver.
      remove_names: When true, remove all names for the ModelProto.
      interrupter: An optional interrupter that the solver can use to interrupt
        the solve early.
 
    Returns:
      An `ComputeInfeasibleSubsystemResult` where `feasibility` indicates if the
      problem was proven infeasible.
 
    Throws:
      RuntimeError: on invalid inputs or an internal solver error.
    """
    params = params or parameters.SolveParameters()
    streamable_init_args = (
        streamable_init_args or init_arguments.StreamableSolverInitArguments()
    )
    model_proto = opt_model.export_model(remove_names=remove_names)
    # Solve
    try:
        proto_result = solver.compute_infeasible_subsystem(
            model_proto,
            solver_type.value,
            streamable_init_args.to_proto(),
            params.to_proto(),
            msg_cb,
            interrupter.pybind_interrupter if interrupter is not None else None,
        )
    except StatusNotOk as e:
        raise _status_not_ok_to_exception(e) from None
    return (
        compute_infeasible_subsystem_result.parse_compute_infeasible_subsystem_result(
            proto_result, opt_model
        )
    )
 
 
class IncrementalSolver:
    """Solve an optimization multiple times, with modifications between solves.
 
    Prefer calling simply solve() above in most cases when incrementalism is not
    needed.
 
    Thread-safety: The __init__(), solve() methods must not be called while
    modifying the Model (adding variables...). The user is expected to use proper
    synchronization primitives to serialize changes to the model and the use of
    this object. Note though that it is safe to call methods from different
    IncrementalSolver instances on the same Model concurrently. The solve() method
    must not be called concurrently on different threads for the same
    IncrementalSolver. Some solvers may add more restriction regarding
    threading. Please see to SolverType::XXX documentation for details.
 
    This class references some resources that are freed when it is garbage
    collected (which should usually happen when the last reference is lost). In
    particular, it references some C++ objects. Although it is not mandatory, it
    is recommended to free those as soon as possible. To do so it is possible to
    use this class in the `with` statement:
 
    with IncrementalSolver(model, SolverType.GLOP) as solver:
      ...
 
    When it is not possible to use `with`, the close() method can be called.
    """
 
    def __init__(
        self,
        opt_model: model.Model,
        solver_type: parameters.SolverType,
        *,
        streamable_init_args: Optional[
            init_arguments.StreamableSolverInitArguments
        ] = None,
        remove_names: bool = False,
    ):
        streamable_init_args = (
            streamable_init_args or init_arguments.StreamableSolverInitArguments()
        )
        self._model = opt_model
        self._solver_type = solver_type
        self._update_tracker = self._model.add_update_tracker()
        try:
            self._proto_solver = solver.new(
                solver_type.value,
                self._model.export_model(remove_names=remove_names),
                streamable_init_args.to_proto(),
            )
        except StatusNotOk as e:
            raise _status_not_ok_to_exception(e) from None
        self._closed = False
 
    def solve(
        self,
        *,
        params: Optional[parameters.SolveParameters] = None,
        model_params: Optional[model_parameters.ModelSolveParameters] = None,
        msg_cb: Optional[message_callback.SolveMessageCallback] = None,
        callback_reg: Optional[callback.CallbackRegistration] = None,
        cb: Optional[SolveCallback] = None,
        interrupter: Optional[solve_interrupter.SolveInterrupter] = None,
    ) -> result.SolveResult:
        """Solves the current optimization model.
 
        Args:
          params: The non-model specific solve parameters.
          model_params: The model specific solve parameters.
          msg_cb: An optional callback for solver messages.
          callback_reg: The parameters controlling when cb is called.
          cb: An optional callback for LP/MIP events.
          interrupter: An optional interrupter that the solver can use to interrupt
            the solve early.
 
        Returns:
          The result of the solve.
 
        Raises:
          RuntimeError: If called after being closed, or on a solve error.
        """
        if self._closed:
            raise RuntimeError("the solver is closed")
 
        update = self._update_tracker.export_update()
        if update is not None:
            try:
                if not self._proto_solver.update(update):
                    self._proto_solver = solver.new(
                        self._solver_type.value,
                        self._model.export_model(),
                        parameters_pb2.SolverInitializerProto(),
                    )
            except StatusNotOk as e:
                raise _status_not_ok_to_exception(e) from None
            self._update_tracker.advance_checkpoint()
        params = params or parameters.SolveParameters()
        model_params = model_params or model_parameters.ModelSolveParameters()
        callback_reg = callback_reg or callback.CallbackRegistration()
        proto_cb = None
        if cb is not None:
            proto_cb = lambda x: cb(  # pylint: disable=g-long-lambda
                callback.parse_callback_data(x, self._model)
            ).to_proto()
        try:
            result_proto = self._proto_solver.solve(
                params.to_proto(),
                model_params.to_proto(),
                msg_cb,
                callback_reg.to_proto(),
                proto_cb,
                interrupter.pybind_interrupter if interrupter is not None else None,
            )
        except StatusNotOk as e:
            raise _status_not_ok_to_exception(e) from None
        return result.parse_solve_result(result_proto, self._model, validate=False)
 
    def close(self) -> None:
        """Closes this solver, freeing all its resources.
 
        This is optional, the code is correct without calling this function. See the
        class documentation for details.
 
        After a solver has been closed, it can't be used anymore. Prefer using the
        context manager API when possible instead of calling close() directly:
 
        with IncrementalSolver(model, SolverType.GLOP) as solver:
          ...
        """
        if self._closed:
            return
        self._closed = True
 
        del self._model
        del self._solver_type
        del self._update_tracker
        del self._proto_solver
 
    def __enter__(self) -> "IncrementalSolver":
        """Returns the solver itself."""
        return self
 
    def __exit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[types.TracebackType],
    ) -> None:
        """Closes the solver."""
        self.close()
 
 
def _status_not_ok_to_exception(err: StatusNotOk) -> Exception:
    """Converts a StatusNotOk to the best matching Python exception.
 
    Args:
      err: The input errors.
 
    Returns:
      The corresponding exception.
    """
    ret = errors.status_proto_to_exception(
        rpc_pb2.StatusProto(code=err.canonical_code, message=err.message)
    )
    # We never expect StatusNotOk to be OK.
    assert ret is not None, err
    return ret