PyROS Add caching for computed uncertain parameter bounds

pyomo/contrib/pyros/CHANGELOG.txt

@@ -3,6 +3,12 @@ PyROS CHANGELOG
 ===============
 
 
+-------------------------------------------------------------------------------
+PyROS 1.3.14    20 Mar 2026
+-------------------------------------------------------------------------------
+- Add caching for uncertainty set parameter bounds
+
+
 -------------------------------------------------------------------------------
 PyROS 1.3.13    16 Jan 2026
 -------------------------------------------------------------------------------

pyomo/contrib/pyros/pyros.py

@@ -386,10 +386,13 @@ def solve(
         )
 
         model_data = ModelData(original_model=model, timing=TimingData(), config=None)
-        with time_code(
-            timing_data_obj=model_data.timing,
-            code_block_name="main",
-            is_main_timer=True,
+        with (
+            uncertainty_set._cache,
+            time_code(
+                timing_data_obj=model_data.timing,
+                code_block_name="main",
+                is_main_timer=True,
+            ),
         ):
             kwds.update(
                 dict(

pyomo/contrib/pyros/tests/test_grcs.py

@@ -3985,7 +3985,7 @@ def solve(self, model, *args, **kwargs):
 
 class TestPyROSUnavailableSubsolvers(unittest.TestCase):
     """
-    Check that appropriate exceptionsa are raised if
+    Check that appropriate exceptions are raised if
     PyROS is invoked with unavailable subsolvers.
     """
 
@@ -5232,5 +5232,252 @@ def test_discrete_set_subsolver_error_recovery(self, name, sec_con_UB):
             )
 
 
+# @SolverFactory.register("slow_solver")
+class SlowSolver:
+    """
+    Solver which sleeps for a specified time before solving.
+    """
+
+    def __init__(self, sleep_time, sub_solver):
+        self.sleep_time = sleep_time
+        self.sub_solver = sub_solver
+
+        self.options = Bunch()
+
+    def available(self, exception_flag=True):
+        return True
+
+    def license_is_valid(self):
+        return True
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, et, ev, tb):
+        pass
+
+    def solve(self, model, **kwargs):
+        """
+        Sleep, then solve a model.
+
+        Parameters
+        ----------
+        model : ConcreteModel
+            Model of interest.
+
+        Returns
+        -------
+        results : SolverResults
+            Solver results.
+        """
+
+        # ensure only one active objective
+        active_objs = [
+            obj for obj in model.component_data_objects(Objective, active=True)
+        ]
+        assert len(active_objs) == 1
+
+        # sleep for specified time
+        time.sleep(self.sleep_time)
+
+        print("I slept. Now, I will solve.")
+        # invoke subsolver
+        results = self.sub_solver.solve(model, **kwargs)
+
+        return results
+
+
+class CustomExactBoundsUncertaintySet(BoxSet):
+    """
+    Custom uncertainty set that always solves optimization bounding problems.
+    """
+
+    def __init__(self, bounds, sleep_time, cache):
+        super().__init__(bounds)
+        self.sleep_time = sleep_time
+        self.cache = cache
+
+    @property
+    def parameter_bounds(self):
+        """
+        Solve bounding problems to calculate exact parameter bounds.
+        """
+        solver = SlowSolver(
+            sub_solver=SolverFactory("baron"), sleep_time=self.sleep_time
+        )
+        bounds = self._compute_exact_parameter_bounds(solver=solver)
+
+        if not self.cache:
+            self._cache.clear()
+
+        return bounds
+
+
+@unittest.skipUnless(ipopt_available, "IPOPT is not available.")
+class TestPyROSCache(unittest.TestCase):
+    """
+    Test PyROS cache creation and clearing.
+    """
+
+    def test_pyros_cache_creation(self):
+        """
+        Check that PyROS creates a cache for storing computed exact parameter bounds.
+        """
+        m = build_leyffer_two_cons()
+
+        # Define the uncertainty set
+        interval = CustomExactBoundsUncertaintySet(
+            bounds=[(0.25, 2)], sleep_time=0, cache=True
+        )
+
+        # Instantiate the PyROS solver
+        pyros_solver = SolverFactory("pyros")
+
+        # Define subsolvers utilized in the algorithm
+        local_subsolver = SolverFactory("ipopt")
+        global_subsolver = SolverFactory("ipopt")
+
+        # check cache exists
+        self.assertTrue(hasattr(interval, "_cache"))
+
+        # Call the PyROS solver
+        results = pyros_solver.solve(
+            model=m,
+            first_stage_variables=[m.x1],
+            second_stage_variables=[m.x2],
+            uncertain_params=[m.u],
+            uncertainty_set=interval,
+            local_solver=local_subsolver,
+            global_solver=global_subsolver,
+            options={
+                "objective_focus": ObjectiveType.worst_case,
+                "solve_master_globally": True,
+            },
+        )
+
+        # check cache has been cleared after solve
+        self.assertTrue(hasattr(interval, "_cache"))
+        self.assertEqual(
+            interval._cache, {}, msg="Did not clear uncertainty set cache after solve."
+        )
+
+    def test_pyros_cache_time(self):
+        """
+        Check that caching improves solve time.
+        """
+        m = build_leyffer_two_cons()
+
+        # Define the uncertainty set
+        interval_cache = CustomExactBoundsUncertaintySet(
+            bounds=[(0.25, 2)], sleep_time=0.1, cache=True
+        )
+        interval_no_cache = CustomExactBoundsUncertaintySet(
+            bounds=[(0.25, 2)], sleep_time=0.1, cache=False
+        )
+
+        # Instantiate the PyROS solver
+        pyros_solver = SolverFactory("pyros")
+
+        # Define subsolvers utilized in the algorithm
+        local_subsolver = SolverFactory("ipopt")
+        global_subsolver = SolverFactory("ipopt")
+
+        # Call the PyROS solver
+        results_cache = pyros_solver.solve(
+            model=m,
+            first_stage_variables=[m.x1],
+            second_stage_variables=[m.x2],
+            uncertain_params=[m.u],
+            uncertainty_set=interval_cache,
+            local_solver=local_subsolver,
+            global_solver=global_subsolver,
+            options={
+                "objective_focus": ObjectiveType.worst_case,
+                "solve_master_globally": True,
+            },
+        )
+
+        results_no_cache = pyros_solver.solve(
+            model=m,
+            first_stage_variables=[m.x1],
+            second_stage_variables=[m.x2],
+            uncertain_params=[m.u],
+            uncertainty_set=interval_no_cache,
+            local_solver=local_subsolver,
+            global_solver=global_subsolver,
+            options={
+                "objective_focus": ObjectiveType.worst_case,
+                "solve_master_globally": True,
+            },
+        )
+
+        # caching should always result in less time,
+        # as not caching reruns the slow solver multiple times
+        self.assertGreater(results_no_cache.time, results_cache.time)
+
+    def test_pyros_cache_solutions(self):
+        """
+        Check that PyROS clears cache before/after and yields accurate results.
+        """
+        m = build_leyffer_two_cons()
+
+        # Define the uncertainty set
+        interval = CustomExactBoundsUncertaintySet(
+            bounds=[(25, 200)], sleep_time=0, cache=True
+        )
+        self.assertEqual(interval.parameter_bounds, [(25, 200)])
+
+        # change set attributes, leading to outdated parameter bounds
+        interval.bounds = [(0.25, 2)]
+        self.assertEqual(interval.parameter_bounds, [(25, 200)])
+
+        # Instantiate the PyROS solver
+        pyros_solver = SolverFactory("pyros")
+
+        # Define subsolvers utilized in the algorithm
+        local_subsolver = SolverFactory("ipopt")
+        global_subsolver = SolverFactory("ipopt")
+
+        # Solve with PyROS
+        exc_str = r"Uncertainty set cache has been modified."
+        with self.assertRaisesRegex(AssertionError, exc_str):
+            results = pyros_solver.solve(
+                model=m,
+                first_stage_variables=[m.x1],
+                second_stage_variables=[m.x2],
+                uncertain_params=[m.u],
+                uncertainty_set=interval,
+                local_solver=local_subsolver,
+                global_solver=global_subsolver,
+                options={
+                    "objective_focus": ObjectiveType.worst_case,
+                    "solve_master_globally": True,
+                },
+            )
+
+        # modify the cache
+        interval._cache[0, minimize] = 25
+        interval._cache[0, maximize] = 200
+
+        self.assertEqual(interval.parameter_bounds, [(25, 200)])
+
+        # Solve with PyROS
+        exc_str = r"Uncertainty set cache has been modified."
+        with self.assertRaisesRegex(AssertionError, exc_str):
+            results = pyros_solver.solve(
+                model=m,
+                first_stage_variables=[m.x1],
+                second_stage_variables=[m.x2],
+                uncertain_params=[m.u],
+                uncertainty_set=interval,
+                local_solver=local_subsolver,
+                global_solver=global_subsolver,
+                options={
+                    "objective_focus": ObjectiveType.worst_case,
+                    "solve_master_globally": True,
+                },
+            )
+
+
 if __name__ == "__main__":
     unittest.main()

pyomo/contrib/pyros/tests/test_uncertainty_sets.py

@@ -24,7 +24,7 @@
 )
 
 from pyomo.environ import SolverFactory
-from pyomo.core.base import ConcreteModel, Param, Var
+from pyomo.core.base import ConcreteModel, Param, Var, maximize, minimize, UnitInterval
 from pyomo.core.expr import RangedExpression
 from pyomo.core.expr.compare import assertExpressionsEqual
 
@@ -3736,6 +3736,37 @@ def parameter_bounds(self, val):
         self._parameter_bounds = val
 
 
+class CustomDomainUncertaintySet(CustomUncertaintySet):
+    """
+    Test simple custom uncertainty set with specified uncertain parameter domains.
+    """
+
+    def __init__(self, dim):
+        self._dim = dim
+        self._parameter_bounds = [(0, 1)] * self.dim
+
+    def set_as_constraint(self, uncertain_params=None, block=None):
+        blk, param_var_list, conlist, aux_vars = (
+            _setup_standard_uncertainty_set_constraint_block(
+                block=block,
+                uncertain_param_vars=uncertain_params,
+                dim=self.dim,
+                num_auxiliary_vars=None,
+            )
+        )
+        conlist.add(sum(param_var_list) <= 1)
+        for var in param_var_list:
+            conlist.add(0 <= var)
+            var.domain = UnitInterval
+
+        return UncertaintyQuantification(
+            block=blk,
+            uncertainty_cons=list(conlist.values()),
+            uncertain_param_vars=param_var_list,
+            auxiliary_vars=aux_vars,
+        )
+
+
 class TestCustomUncertaintySet(unittest.TestCase):
     """
     Test for a custom uncertainty set subclass.
@@ -3768,6 +3799,33 @@ def test_compute_exact_parameter_bounds(self):
             custom_set._compute_exact_parameter_bounds(baron), [(-1, 1)] * 2
         )
 
+    @unittest.skipUnless(baron_available, "BARON is not available")
+    def test_solve_exact_bounds_optimization(self):
+        """
+        Test parameter bounds computations are cached.
+        """
+        baron = SolverFactory("baron")
+        custom_set = CustomUncertaintySet(dim=2)
+
+        # check cache exists
+        self.assertTrue(hasattr(custom_set, "_cache"))
+
+        # check bounds calculation
+        self.assertEqual(
+            custom_set._solve_exact_bounds_optimization(
+                custom_set._create_bounding_model(), 0, minimize, baron
+            ),
+            -1.0,
+        )
+
+        # check cache access
+        self.assertIs(
+            custom_set._solve_exact_bounds_optimization(
+                custom_set._create_bounding_model(), 0, minimize, baron
+            ),
+            custom_set._cache[0, minimize],
+        )
+
     @unittest.skipUnless(baron_available, "BARON is not available")
     def test_solve_feasibility(self):
         """
@@ -3837,6 +3895,15 @@ def test_is_nonempty(self):
         with self.assertRaisesRegex(ValueError, exc_str):
             custom_set.is_nonempty(config=CONFIG)
 
+    def test_fbbt_values(self):
+        """
+        Test that `_fbbt_parameter_bounds` returns correct values.
+        """
+        CONFIG = pyros_config()
+        custom_set = CustomDomainUncertaintySet(dim=2)
+
+        self.assertEqual(custom_set._fbbt_parameter_bounds(config=CONFIG), [(0, 1)] * 2)
+
     @unittest.skipUnless(baron_available, "BARON is not available")
     def test_is_coordinate_fixed(self):
         """