🔨 Add branching priorities to CPLEXSHELL

- Include `branch_priority` and `branch_direction` attributes of variable data
- Read off these attributes as part of the CPLEXSHELL solve and write them to a CPLEX .ord file
- Update the CPLEX OPL script to read these .ord file should it exist

pyomo/core/base/var.py

@@ -11,6 +11,7 @@
 __all__ = ['Var', '_VarData', '_GeneralVarData', 'VarList', 'SimpleVar']
 
 import logging
+from typing import Optional
 from weakref import ref as weakref_ref
 
 from pyomo.common.modeling import NoArgumentGiven
@@ -278,6 +279,10 @@ def to_string(self, verbose=None, labeler=None, smap=None, compute_values=False)
         return self.name
 
 
+PriorityType = int
+BranchDirectionType = int
+
+
 class _GeneralVarData(_VarData):
     """
     This class defines the data for a single variable.
@@ -308,7 +313,7 @@ class _GeneralVarData(_VarData):
     these attributes in certain cases.
     """
 
-    __slots__ = ('_value', '_lb', '_ub', '_domain', 'fixed', 'stale')
+    __slots__ = ('_value', '_lb', '_ub', '_domain', 'fixed', 'stale', 'branch_priority', 'branch_direction')
 
     def __init__(self, domain=Reals, component=None):
         #
@@ -341,6 +346,9 @@ def __init__(self, domain=Reals, component=None):
                 "for bounds (like a Pyomo Set). Examples: NonNegativeReals, "
                 "Integers, Binary" % (domain, (RealSet, IntegerSet, BooleanSet)))
 
+        self.branch_priority = None  # type: Optional[PriorityType]
+        self.branch_direction = None  # type: Optional[BranchDirectionType]
+
     def __getstate__(self):
         state = super(_GeneralVarData, self).__getstate__()
         for i in _GeneralVarData.__slots__:

pyomo/core/tests/unit/test_var.py

@@ -440,6 +440,24 @@ def test_value(self):
         self.assertEqual( type(tmp), int)
         self.assertEqual( tmp, 3 )
 
+    def test_branch_priority_attr(self):
+        """Test branch_priority attribute"""
+        self.model.x = Var()
+        self.instance = self.model.create_instance()
+
+        self.assertIsNone(self.instance.x.branch_priority)
+        self.instance.x.branch_priority = 1
+        self.assertEqual(self.instance.x.branch_priority, 1)
+
+    def test_branch_direction_attr(self):
+        """Test branch_direction attribute"""
+        self.model.x = Var()
+        self.instance = self.model.create_instance()
+
+        self.assertIsNone(self.instance.x.branch_direction)
+        self.instance.x.branch_direction = -1
+        self.assertEqual(self.instance.x.branch_direction, -1)
+
 
 class TestArrayVar(TestSimpleVar):
 

pyomo/solvers/plugins/solvers/CPLEX.py

@@ -112,6 +112,33 @@ def __new__(cls, *args, **kwds):
         return opt
 
 
+class CPLEXBranchDirection:
+    default = 0
+    down = -1
+    up = 1
+
+    ALL = {default, down, up}
+
+    @staticmethod
+    def to_str(branch_direction):
+        try:
+            return {
+                CPLEXBranchDirection.down: "DN",
+                CPLEXBranchDirection.up: "UP",
+            }[branch_direction]
+        except KeyError:
+            return ""
+
+
+class ORDFileSchema:
+    HEADER = "* ENCODING=ISO-8859-1\nNAME             Priority Order\n"
+    FOOTER = "ENDATA\n"
+
+    @classmethod
+    def ROW(cls, name, priority, branch_direction=None):
+        return " %s %s %s\n" % (CPLEXBranchDirection.to_str(branch_direction), name, priority)
+
+
 @SolverFactory.register('_cplex_shell', doc='Shell interface to the CPLEX LP/MIP solver')
 class CPLEXSHELL(ILMLicensedSystemCallSolver):
     """Shell interface to the CPLEX LP/MIP solver
@@ -201,6 +228,35 @@ def _warm_start(self, instance):
             mst_file.write("</variables>\n")
             mst_file.write("</CPLEXSolution>\n")
 
+    def _write_priorities_file(self, instance) -> None:
+        """ Write a variable priorities file in the CPLEX ORD format. """
+        from pyomo.core.base import Var
+
+        if isinstance(instance, IBlock):
+            smap = getattr(instance, "._symbol_maps")[self._smap_id]
+        else:
+            smap = instance.solutions.symbol_map[self._smap_id]
+        byObject = smap.byObject
+
+        with open(self._priorities_file_name, "w") as ord_file:
+            ord_file.write(ORDFileSchema.HEADER)
+            for var in instance.component_data_objects(Var):
+                priority = var.branch_priority
+                if priority is None:
+                    continue
+
+                if not (0 <= priority == int(priority)):
+                    raise ValueError("`priority` must be a non-negative integer")
+
+                if id(var) not in byObject or not var.active:
+                    continue
+
+                ord_file.write(
+                    ORDFileSchema.ROW(byObject[id(var)], priority, var.branch_direction)
+                )
+
+            ord_file.write(ORDFileSchema.FOOTER)
+
     # over-ride presolve to extract the warm-start keyword, if specified.
     def _presolve(self, *args, **kwds):
 
@@ -234,6 +290,14 @@ def _presolve(self, *args, **kwds):
                 self._warm_start_file_name = pyutilib.services.TempfileManager.\
                                              create_tempfile(suffix = '.cplex.mst')
 
+        self._branching_priorities_solve = kwds.pop('priorities', False)
+
+        self._priorities_file_name = None
+        if self._branching_priorities_solve:
+            self._priorities_file_name = pyutilib.services.TempfileManager.create_tempfile(
+                suffix=".cplex.ord"
+            )
+
         # let the base class handle any remaining keywords/actions.
         ILMLicensedSystemCallSolver._presolve(self, *args, **kwds)
 
@@ -259,6 +323,16 @@ def _presolve(self, *args, **kwds):
                     print("Warm start write time= %.2f seconds"
                           % (end_time-start_time))
 
+            if self._branching_priorities_solve:
+                start_time = time.time()
+                self._write_priorities_file(args[0])
+                end_time = time.time()
+                if self._report_timing:
+                    print(
+                        "Branching priorities write time= %.2f seconds"
+                        % (end_time - start_time)
+                    )
+
     def _default_executable(self):
         executable = pyomo.common.Executable("cplex")
         if not executable:
@@ -328,6 +402,9 @@ def create_command_line(self, executable, problem_files):
            (self._warm_start_file_name is not None):
             script += 'read %s\n' % (self._warm_start_file_name,)
 
+        if self._priorities_file_name is not None:
+            script += "read %s\n" % (self._priorities_file_name,)
+
         if 'relax_integrality' in self.options:
             script += 'change problem lp\n'
 
@@ -351,6 +428,9 @@ def create_command_line(self, executable, problem_files):
                 print("Solver warm-start file="
                       +self._warm_start_file_name)
 
+            if self._priorities_file_name is not None:
+                print("Solver priorities file=" + self._priorities_file_name)
+
         #
         # Define command line
         #

pyomo/solvers/tests/checks/test_cplex.py

@@ -9,9 +9,15 @@
 #  ___________________________________________________________________________
 
 import os
+
+import pyutilib
 import pyutilib.th as unittest
 
-from pyomo.solvers.plugins.solvers.CPLEX import _validate_file_name
+from pyomo.core import Binary, ConcreteModel, Constraint, Objective, Var
+from pyomo.opt import ProblemFormat, convert_problem
+from pyomo.solvers.plugins.solvers.CPLEX import (CPLEXSHELL, MockCPLEX,
+                                                 _validate_file_name)
+
 
 class _mock_cplex_128(object):
     def version(self):
@@ -54,5 +60,99 @@ def test_validate_file_name(self):
         with self.assertRaisesRegexp(ValueError, msg):
             _validate_file_name(_128, fname, 'xxx')
 
+
+class CPLEXShellPrioritiesFile(unittest.TestCase):
+    def setUp(self):
+        from pyomo.solvers.plugins.converter.model import PyomoMIPConverter  # register the `ProblemConverterFactory`
+        from pyomo.repn.plugins.cpxlp import ProblemWriter_cpxlp  # register the `WriterFactory`
+
+        self.mock_model = self.get_mock_model()
+        self.mock_cplex_shell = self.get_mock_cplex_shell(self.mock_model)
+        self.mock_cplex_shell._priorities_file_name = pyutilib.services.TempfileManager.create_tempfile(
+            suffix=".cplex.ord"
+        )
+
+    def tearDown(self):
+        pyutilib.services.TempfileManager.clear_tempfiles()
+
+    def get_mock_model(self):
+        model = ConcreteModel()
+        model.x = Var(within=Binary)
+        model.con = Constraint(expr=model.x >= 1)
+        model.obj = Objective(expr=model.x)
+        return model
+
+    def get_mock_cplex_shell(self, mock_model):
+        solver = MockCPLEX()
+        solver._problem_files, solver._problem_format, solver._smap_id = convert_problem(
+            (mock_model,),
+            ProblemFormat.cpxlp,
+            [ProblemFormat.cpxlp],
+            has_capability=lambda x: True,
+        )
+        return solver
+
+    def get_priorities_file_as_string(self, mock_cplex_shell):
+        with open(mock_cplex_shell._priorities_file_name, "r") as ord_file:
+            priorities_file = ord_file.read()
+        return priorities_file
+
+    def test_write_empty_priorities_file(self):
+        CPLEXSHELL._write_priorities_file(self.mock_cplex_shell, self.mock_model)
+
+        with open(self.mock_cplex_shell._priorities_file_name, "r") as ord_file:
+            priorities_file = ord_file.read()
+
+        self.assertEqual(
+            priorities_file,
+            "* ENCODING=ISO-8859-1\nNAME             Priority Order\nENDATA\n",
+        )
+
+    def test_write_priority_to_priorities_file(self):
+        self.mock_model.x.branch_priority = 10
+
+        CPLEXSHELL._write_priorities_file(self.mock_cplex_shell, self.mock_model)
+        priorities_file = self.get_priorities_file_as_string(self.mock_cplex_shell)
+
+        self.assertEqual(
+            priorities_file,
+            "* ENCODING=ISO-8859-1\nNAME             Priority Order\n  x1 10\nENDATA\n",
+        )
+        self.assertIsNone(self.mock_model.x.branch_direction)
+
+    def test_write_priority_and_direction_to_priorities_file(self):
+        self.mock_model.x.branch_priority = 10
+        self.mock_model.x.branch_direction = -1
+
+        CPLEXSHELL._write_priorities_file(self.mock_cplex_shell, self.mock_model)
+        priorities_file = self.get_priorities_file_as_string(self.mock_cplex_shell)
+
+        self.assertEqual(
+            priorities_file,
+            "* ENCODING=ISO-8859-1\nNAME             Priority Order\n DN x1 10\nENDATA\n",
+        )
+
+    def test_raise_due_to_invalid_priority(self):
+        self.mock_model.x.branch_priority = -1
+        with self.assertRaises(ValueError):
+            CPLEXSHELL._write_priorities_file(self.mock_cplex_shell, self.mock_model)
+
+        self.mock_model.x.branch_priority = 1.1
+        with self.assertRaises(ValueError):
+            CPLEXSHELL._write_priorities_file(self.mock_cplex_shell, self.mock_model)
+
+    def test_use_default_due_to_invalid_direction(self):
+        self.mock_model.x.branch_priority = 10
+        self.mock_model.x.branch_direction = "invalid_branching_direction"
+
+        CPLEXSHELL._write_priorities_file(self.mock_cplex_shell, self.mock_model)
+        priorities_file = self.get_priorities_file_as_string(self.mock_cplex_shell)
+
+        self.assertEqual(
+            priorities_file,
+            "* ENCODING=ISO-8859-1\nNAME             Priority Order\n  x1 10\nENDATA\n",
+        )
+
+
 if __name__ == "__main__":
     unittest.main()