Revise CplexOptimizer (#139)

* * Adds CplexOptimizer.cplex_parameters

* Adds a check whether any solution is found or not

* Simplifies CplexOptimizer

* update

Author: t-imamichi

qiskit_optimization/algorithms/cplex_optimizer.py

@@ -13,16 +13,22 @@
 """The CPLEX optimizer wrapped to be used within Qiskit's optimization module."""
 
 import logging
+from typing import Any, Dict, Optional
+from warnings import warn
 
 from qiskit.exceptions import MissingOptionalLibraryError
-from .optimization_algorithm import OptimizationAlgorithm, OptimizationResult
-from ..exceptions import QiskitOptimizationError
+
+from .optimization_algorithm import (
+    OptimizationAlgorithm,
+    OptimizationResult,
+    OptimizationResultStatus,
+)
 from ..problems.quadratic_program import QuadraticProgram
 
 logger = logging.getLogger(__name__)
 
 try:
-    from cplex.exceptions import CplexSolverError
+    from cplex import Cplex  # pylint: disable=unused-import
 
     _HAS_CPLEX = True
 except ImportError:
@@ -44,11 +50,14 @@ class CplexOptimizer(OptimizationAlgorithm):
         >>> if optimizer: result = optimizer.solve(problem)
     """
 
-    def __init__(self, disp: bool = False) -> None:
+    def __init__(
+        self, disp: bool = False, cplex_parameters: Optional[Dict[str, Any]] = None
+    ) -> None:
         """Initializes the CplexOptimizer.
 
         Args:
             disp: Whether to print CPLEX output or not.
+            cplex_parameters: The parameters for CPLEX.
 
         Raises:
             MissingOptionalLibraryError: CPLEX is not installed.
@@ -61,6 +70,7 @@ def __init__(self, disp: bool = False) -> None:
             )
 
         self._disp = disp
+        self._cplex_parameters = cplex_parameters
 
     @staticmethod
     def is_cplex_installed():
@@ -84,6 +94,19 @@ def disp(self, disp: bool):
         """
         self._disp = disp
 
+    @property
+    def cplex_parameters(self) -> Optional[Dict[str, Any]]:
+        """Returns parameters for CPLEX"""
+        return self._cplex_parameters
+
+    @cplex_parameters.setter
+    def cplex_parameters(self, parameters: Optional[Dict[str, Any]]):
+        """Set parameters for CPLEX
+        Args:
+            parameters: The parameters for CPLEX
+        """
+        self._cplex_parameters = parameters
+
     # pylint:disable=unused-argument
     def get_compatibility_msg(self, problem: QuadraticProgram) -> str:
         """Checks whether a given problem can be solved with this optimizer.
@@ -116,33 +139,26 @@ def solve(self, problem: QuadraticProgram) -> OptimizationResult:
             QiskitOptimizationError: If the problem is incompatible with the optimizer.
         """
 
-        # convert to CPLEX problem
-        cplex = problem.to_docplex().get_cplex()
-
-        # set display setting
-        if not self.disp:
-            cplex.set_log_stream(None)
-            cplex.set_error_stream(None)
-            cplex.set_warning_stream(None)
-            cplex.set_results_stream(None)
-
-        # solve problem
-        try:
-            cplex.solve()
-        except CplexSolverError as ex:
-            raise QiskitOptimizationError(str(ex)) from ex
-
-        # process results
-        sol = cplex.solution
-
-        # create results
-        result = OptimizationResult(
-            x=sol.get_values(),
-            fval=sol.get_objective_value(),
-            variables=problem.variables,
-            status=self._get_feasibility_status(problem, sol.get_values()),
-            raw_results=sol,
-        )
-
-        # return solution
-        return result
+        mod = problem.to_docplex()
+        sol = mod.solve(log_output=self._disp, cplex_parameters=self._cplex_parameters)
+        if sol is None:
+            # no solution is found
+            warn("CPLEX cannot solve the model")
+            x = [0.0] * mod.number_of_variables
+            return OptimizationResult(
+                x=x,
+                fval=problem.objective.evaluate(x),
+                variables=problem.variables,
+                status=OptimizationResultStatus.FAILURE,
+                raw_results=None,
+            )
+        else:
+            # a solution is found
+            x = sol.get_values(mod.iter_variables())
+            return OptimizationResult(
+                x=x,
+                fval=sol.get_objective_value(),
+                variables=problem.variables,
+                status=self._get_feasibility_status(problem, x),
+                raw_results=sol,
+            )

releasenotes/notes/add-cplex-params-2b7bd7bde01400c7.yaml

@@ -0,0 +1,11 @@
+---
+features:
+  - |
+    Simplifies :class`qiskit_optimization.algorithms.CplexOptimizer` by
+    calling CPLEX from `docplex.mp.model.Model.solve` directly.
+    Also adds a fallback code if no solution is found by CPLEX.
+  - |
+    Adds ``cplex_parameters`` as a dictionary to
+    :class`qiskit_optimization.algorithms.CplexOptimizer`
+    so that users can set parameters for CPLEX such as time limit
+    and number of threads.

test/algorithms/test_cplex_optimizer.py

@@ -13,12 +13,12 @@
 """ Test Cplex Optimizer """
 
 import unittest
-from test.optimization_test_case import (
-    QiskitOptimizationTestCase,
-    requires_extra_library,
-)
-from ddt import ddt, data
-from qiskit_optimization.algorithms import CplexOptimizer
+from test.optimization_test_case import QiskitOptimizationTestCase, requires_extra_library
+
+import numpy as np
+from ddt import data, ddt
+
+from qiskit_optimization.algorithms import CplexOptimizer, OptimizationResultStatus
 from qiskit_optimization.problems import QuadraticProgram
 
 
@@ -51,6 +51,29 @@ def test_cplex_optimizer(self, config):
         for i in range(problem.get_num_vars()):
             self.assertAlmostEqual(result.x[i], x[i])
 
+    @data(
+        ("op_ip1.lp", [0, 2], 6),
+        ("op_mip1.lp", [1, 1, 0], 6),
+        ("op_lp1.lp", [0.25, 1.75], 5.8750),
+    )
+    @requires_extra_library
+    def test_cplex_optimizer_no_solution(self, config):
+        """Cplex Optimizer Test if no solution is found"""
+        cplex_optimizer = CplexOptimizer(disp=False, cplex_parameters={"timelimit": 0})
+        # unpack configuration
+        filename, _, _ = config
+
+        # load optimization problem
+        problem = QuadraticProgram()
+        lp_file = self.get_resource_path(filename, "algorithms/resources")
+        problem.read_from_lp_file(lp_file)
+
+        # solve problem with cplex
+        result = cplex_optimizer.solve(problem)
+        np.testing.assert_array_almost_equal(result.x, np.zeros(problem.get_num_vars()))
+        self.assertEqual(result.status, OptimizationResultStatus.FAILURE)
+        self.assertEqual(result.raw_results, None)
+
 
 if __name__ == "__main__":
     unittest.main()