Add callback and parameter configuration support to CPLEX_PY solver (#850)

* Update CPLEX_PYTest to use CPLEX_PY instead of CPLEX_CMD

* Add support for custom CPLEX solver parameters via arguments

Extended the CPLEX solver API to accept additional parameters through `solverParams`, enabling more granular configuration of the solver. Added utility methods for setting, getting, and managing solver parameters. Included corresponding unit tests to ensure functionality and correctness.

* Add support for registering CPLEX callbacks during solve

Introduced an optional `callback` parameter to allow registering CPLEX callback classes in the solving process. Updated the relevant methods in `cplex_api.py` and added a test case to validate the functionality.

* Add docstrings for parameter handling methods and tests

* Update type ignore comments for cplex imports

Extended type ignore comments to address additional linter issues, including untyped imports and unused ignores.

* Applying the black linter

---------

Co-authored-by: Franco Peschiera <[email protected]>

Author: G-Carneiro

pulp/apis/cplex_api.py

@@ -1,5 +1,6 @@
 import os
 import warnings
+from typing import Iterable, Optional
 
 from .. import constants
 from .core import LpSolver, LpSolver_CMD, PulpSolverError, clock, log, subprocess
@@ -252,7 +253,7 @@ class CPLEX_PY(LpSolver):
     name = "CPLEX_PY"
     try:
         global cplex
-        import cplex  # type: ignore[import-not-found]
+        import cplex  # type: ignore[import-not-found, import-untyped, unused-ignore]
     except Exception as e:
         err = e
         """The CPLEX LP/MIP solver from python. Something went wrong!!!!"""
@@ -276,6 +277,7 @@ def __init__(
             warmStart=False,
             logPath=None,
             threads=None,
+            **solverParams,
         ):
             """
             :param bool mip: if False, assume LP even if integer variables
@@ -285,6 +287,15 @@ def __init__(
             :param bool warmStart: if True, the solver will use the current value of variables as a start
             :param str logPath: path to the log file
             :param int threads: number of threads to be used by CPLEX to solve a problem (default None uses all available)
+
+            :param dict solverParams: Additional parameters to set in the CPLEX solver.
+
+            Parameters should use dot notation as specified in the CPLEX documentation.
+            The 'parameters.' prefix is optional. For example:
+
+                  * parameters.advance (or advance)
+                  * parameters.barrier.algorithm (or barrier.algorithm)
+                  * parameters.mip.strategy.probe (or mip.strategy.probe)
             """
 
             LpSolver.__init__(
@@ -297,22 +308,25 @@ def __init__(
                 logPath=logPath,
                 threads=threads,
             )
+            self.solverParams = solverParams
 
         def available(self):
             """True if the solver is available"""
             return True
 
-        def actualSolve(self, lp, callback=None):  # type: ignore[misc]
+        def actualSolve(self, lp, callback: Optional[Iterable[type[cplex.callbacks.Callback]]] = None):  # type: ignore[misc]
             """
             Solve a well formulated lp problem
 
             creates a cplex model, variables and constraints and attaches
             them to the lp model which it then solves
+
+            :param callback: Optional list of CPLEX callback classes to register during solve
             """
             self.buildSolverModel(lp)
             # set the initial solution
             log.debug("Solve the Model using cplex")
-            self.callSolver(lp)
+            self.callSolver(lp, callback=callback)
             # get the solution information
             solutionStatus = self.findSolutionValues(lp)
             for var in lp._variables:
@@ -430,6 +444,47 @@ def cplex_var_types(var):
                 self.solverModel.MIP_starts.add(
                     cplex.SparsePair(ind=ind, val=val), effort, "1"
                 )
+            for param, value in self.solverParams.items():
+                self.set_param(param, value)
+
+        def set_param(self, name: str, value):
+            """
+            Sets a parameter value using its name.
+            """
+            param = self.search_param(name=name)
+            param.set(value)
+
+        def get_param(self, name: str):
+            """
+            Returns the value of a named parameter by searching within the instance's parameters.
+            """
+            param = self.search_param(name=name)
+            return param.get()
+
+        def search_param(self, name: str):
+            """
+            Searches for a solver model parameter by its name and returns the corresponding attribute.
+
+            The method takes a parameter name string, processes it to remove the "parameters." prefix
+            and splits it by periods to traverse the attribute hierarchy of the solver model's parameters.
+            """
+            name = name.replace("parameters.", "")
+            param = self.solverModel.parameters
+            for attr in name.split("."):
+                param = getattr(param, attr)
+            return param
+
+        def get_all_params(self):
+            """
+            Returns all parameters from the solver model.
+            """
+            return self.solverModel.parameters.get_all()
+
+        def get_changed_params(self):
+            """
+            Returns the parameters that have been changed in the solver model.
+            """
+            return self.solverModel.parameters.get_changed()
 
         def setlogfile(self, fileobj):
             """
@@ -458,9 +513,21 @@ def setTimeLimit(self, timeLimit=0.0):
             """
             self.solverModel.parameters.timelimit.set(timeLimit)
 
-        def callSolver(self, isMIP):
-            """Solves the problem with cplex"""
+        def callSolver(
+            self,
+            isMIP,
+            callback: Optional[Iterable[type[cplex.callbacks.Callback]]] = None,
+        ):
+            """
+            Solves the problem with cplex
+
+
+            :param callback: Optional list of CPLEX callback classes to register during solve
+            """
             # solve the problem
+            if callback is not None:
+                for call in callback:
+                    self.solverModel.register_callback(call)
             self.solveTime = -clock()
             self.solverModel.solve()
             self.solveTime += clock()

pulp/tests/test_pulp.py

@@ -2073,6 +2073,72 @@ class CPLEX_CMDTest(BaseSolverTest.PuLPTest):
 class CPLEX_PYTest(BaseSolverTest.PuLPTest):
     solveInst = solvers.CPLEX_CMD
 
+    def _build(self, **kwargs):
+        """
+        Builds and returns a solver instance after creating and initializing a bin packing problem.
+        """
+        problem = create_bin_packing_problem(bins=40, seed=99)
+        solver = self.solveInst(**kwargs)
+        solver.buildSolverModel(lp=problem)
+        return solver
+
+    def test_get_param(self):
+        """
+        Tests the `get_param` method of the solver instance to ensure the correct
+        value is returned for a given parameter key.
+        """
+        solver = self._build()
+        self.assertEqual(solver.get_param("barrier.algorithm"), 0)
+
+    def test_get_param_with_full_path(self):
+        """
+        Test case for accessing a solver's parameter by its full hierarchical path.
+        """
+        solver = self._build()
+        self.assertEqual(solver.get_param("parameters.barrier.algorithm"), 0)
+
+    def test_set_param(self):
+        """
+        Tests the functionality for setting a parameter in the solver.
+        """
+        param = "barrier.limits.iteration"
+        solver = self._build(**{param: 100})
+        self.assertEqual(solver.get_param(name=param), 100)
+
+    def test_set_param_with_full_path(self):
+        """
+        Tests the functionality for setting a parameter using its full hierarchical path in the solver.
+        """
+        param = "parameters.barrier.limits.iteration"
+        solver = self._build(**{param: 100})
+        self.assertEqual(solver.get_param(name=param), 100)
+
+    def test_changed_param(self):
+        param = "parameters.barrier.limits.iteration"
+        solver = self._build(**{param: 100})
+        self.assertEqual(len(solver.get_changed_params()), 1)
+
+    def test_set_all_params(self):
+        solver = self._build()
+        parameters = solver.get_all_params()
+        for param, value in parameters:
+            solver.set_param(name=str(param), value=value)
+        self.assertEqual(solver.get_changed_params(), [])
+
+    def test_callback(self):
+        from cplex.callbacks import IncumbentCallback  # type: ignore[import-not-found, import-untyped, unused-ignore]
+
+        counter = 0
+
+        class Callback(IncumbentCallback):
+            def __call__(self):
+                nonlocal counter
+                counter += 1
+
+        problem = create_bin_packing_problem(bins=5, seed=55)
+        pulpTestCheck(problem, self.solver, [LpStatusOptimal], callback=[Callback])
+        self.assertGreaterEqual(counter, 1)
+
 
 class XPRESS_CMDTest(BaseSolverTest.PuLPTest):
     solveInst = solvers.XPRESS_CMD