Add tests for JSON encoder and optimize_experiments error handling in DoE module

Author: smondal13

pyomo/contrib/doe/tests/test_doe_build.py

@@ -22,7 +22,7 @@
 
 from pyomo.common.fileutils import this_file_dir
 import pyomo.common.unittest as unittest
-from pyomo.contrib.doe.doe import ObjectiveLib
+from pyomo.contrib.doe.doe import ObjectiveLib, _DoEResultsJSONEncoder
 
 if not (numpy_available and scipy_available):
     raise unittest.SkipTest("Pyomo.DoE needs scipy and numpy to run tests")
@@ -881,6 +881,19 @@ def test_maximize_objective_set_contents(self):
         self.assertNotIn(ObjectiveLib.condition_number, maximize)
         self.assertNotIn(ObjectiveLib.zero, maximize)
 
+    def test_doe_results_json_encoder_handles_numpy_and_enum(self):
+        payload = {
+            "scalar": np.int64(7),
+            "array": np.array([1.0, 2.0]),
+            "objective": ObjectiveLib.trace,
+        }
+        encoded = json.dumps(payload, cls=_DoEResultsJSONEncoder)
+        decoded = json.loads(encoded)
+
+        self.assertEqual(decoded["scalar"], 7)
+        self.assertEqual(decoded["array"], [1.0, 2.0])
+        self.assertEqual(decoded["objective"], str(ObjectiveLib.trace))
+
     def test_symmetrize_lower_tri_helper(self):
         m = np.array([[1.0, 0.0, 0.0], [2.0, 3.0, 0.0], [4.0, 5.0, 6.0]])
         got = DesignOfExperiments._symmetrize_lower_tri(m)

pyomo/contrib/doe/tests/test_doe_errors.py

@@ -6,6 +6,7 @@
 # Solutions of Sandia, LLC, the U.S. Government retains certain rights in this
 # software.  This software is distributed under the 3-clause BSD License.
 # ____________________________________________________________________________________
+import json
 import warnings
 from pyomo.common.dependencies import (
     numpy as np,
@@ -24,7 +25,10 @@
 
 if scipy_available:
     from pyomo.contrib.doe import DesignOfExperiments
-    from pyomo.contrib.doe.doe import InitializationMethod
+    from pyomo.contrib.doe.doe import (
+        InitializationMethod,
+        _DoEResultsJSONEncoder,
+    )
     from pyomo.contrib.doe.tests.experiment_class_example_flags import (
         BadExperiment,
         RooneyBieglerExperimentFlag,
@@ -109,6 +113,16 @@ def test_experiment_none_error(self):
 
             doe_obj = DesignOfExperiments(**DoE_args)
 
+    def test_experiment_list_empty_error(self):
+        with self.assertRaisesRegex(
+            ValueError, "The 'experiment_list' cannot be empty"
+        ):
+            DesignOfExperiments(experiment_list=[], objective_option="pseudo_trace")
+
+    def test_doe_results_json_encoder_unsupported_object_raises(self):
+        with self.assertRaises(TypeError):
+            json.dumps({"x": object()}, cls=_DoEResultsJSONEncoder)
+
     def test_reactor_check_no_get_labeled_model(self):
         fd_method = "central"
         obj_used = "pseudo_trace"
@@ -844,6 +858,102 @@ def test_optimize_experiments_invalid_init_n_samples_float(self):
                 init_n_samples=2.5,
             )
 
+    def test_optimize_experiments_lhs_requires_template_mode(self):
+        doe_obj = DesignOfExperiments(
+            experiment_list=[
+                RooneyBieglerMultiExperiment(hour=2.0),
+                RooneyBieglerMultiExperiment(hour=3.0),
+            ],
+            objective_option="pseudo_trace",
+        )
+        with self.assertRaisesRegex(
+            ValueError,
+            r"``initialization_method='lhs'`` is currently supported only in template mode",
+        ):
+            doe_obj.optimize_experiments(initialization_method="lhs")
+
+    def test_optimize_experiments_lhs_requires_scipy(self):
+        doe_obj = DesignOfExperiments(
+            experiment_list=[RooneyBieglerMultiExperiment(hour=2.0)],
+            objective_option="pseudo_trace",
+        )
+        with patch("pyomo.contrib.doe.doe.scipy_available", False):
+            with self.assertRaisesRegex(
+                ImportError, r"LHS initialization requires scipy"
+            ):
+                doe_obj.optimize_experiments(initialization_method="lhs")
+
+    def test_optimize_experiments_init_parallel_requires_bool(self):
+        doe_obj = DesignOfExperiments(
+            experiment_list=[RooneyBieglerMultiExperiment(hour=2.0)],
+            objective_option="pseudo_trace",
+        )
+        with self.assertRaisesRegex(
+            ValueError, r"``init_parallel`` must be a bool, got 1."
+        ):
+            doe_obj.optimize_experiments(initialization_method="lhs", init_parallel=1)
+
+    def test_optimize_experiments_init_combo_parallel_requires_bool(self):
+        doe_obj = DesignOfExperiments(
+            experiment_list=[RooneyBieglerMultiExperiment(hour=2.0)],
+            objective_option="pseudo_trace",
+        )
+        with self.assertRaisesRegex(
+            ValueError, r"``init_combo_parallel`` must be a bool"
+        ):
+            doe_obj.optimize_experiments(
+                initialization_method="lhs", init_combo_parallel="yes"
+            )
+
+    def test_optimize_experiments_init_n_workers_must_be_positive_integer(self):
+        doe_obj = DesignOfExperiments(
+            experiment_list=[RooneyBieglerMultiExperiment(hour=2.0)],
+            objective_option="pseudo_trace",
+        )
+        with self.assertRaisesRegex(
+            ValueError, r"``init_n_workers`` must be None or a positive integer"
+        ):
+            doe_obj.optimize_experiments(initialization_method="lhs", init_n_workers=0)
+
+    def test_optimize_experiments_init_combo_chunk_size_must_be_positive_integer(self):
+        doe_obj = DesignOfExperiments(
+            experiment_list=[RooneyBieglerMultiExperiment(hour=2.0)],
+            objective_option="pseudo_trace",
+        )
+        with self.assertRaisesRegex(
+            ValueError,
+            r"``init_combo_chunk_size`` must be a positive integer",
+        ):
+            doe_obj.optimize_experiments(
+                initialization_method="lhs", init_combo_chunk_size=0
+            )
+
+    def test_optimize_experiments_init_combo_parallel_threshold_positive_integer(self):
+        doe_obj = DesignOfExperiments(
+            experiment_list=[RooneyBieglerMultiExperiment(hour=2.0)],
+            objective_option="pseudo_trace",
+        )
+        with self.assertRaisesRegex(
+            ValueError,
+            r"``init_combo_parallel_threshold`` must be a positive integer",
+        ):
+            doe_obj.optimize_experiments(
+                initialization_method="lhs", init_combo_parallel_threshold=0
+            )
+
+    def test_optimize_experiments_init_max_wall_clock_time_must_be_positive(self):
+        doe_obj = DesignOfExperiments(
+            experiment_list=[RooneyBieglerMultiExperiment(hour=2.0)],
+            objective_option="pseudo_trace",
+        )
+        with self.assertRaisesRegex(
+            ValueError,
+            r"``init_max_wall_clock_time`` must be None or a positive number",
+        ):
+            doe_obj.optimize_experiments(
+                initialization_method="lhs", init_max_wall_clock_time=0
+            )
+
     def test_optimize_experiments_n_exp_with_multi_list(self):
         doe_obj = DesignOfExperiments(
             experiment_list=[

pyomo/contrib/doe/tests/test_doe_solve.py

@@ -1462,6 +1462,92 @@ def test_optimize_experiments_trace_expected_values(self):
         self.assertStructuredAlmostEqual(got_hours, expected_hours, abstol=1e-3)
         self.assertAlmostEqual(scenario["log10 A-opt"], -2.2347, places=3)
 
+    def test_optimize_experiments_termination_message_bytes(self):
+        doe = self._make_template_doe("pseudo_trace")
+        original_solve = doe.solver.solve
+
+        def _solve_with_bytes_message(*args, **kwargs):
+            res = original_solve(*args, **kwargs)
+            res.solver.message = b"byte-message"
+            return res
+
+        with patch.object(doe.solver, "solve", side_effect=_solve_with_bytes_message):
+            doe.optimize_experiments(n_exp=1)
+
+        self.assertEqual(doe.results["Termination Message"], "byte-message")
+
+    def test_optimize_experiments_termination_message_fallback_to_str(self):
+        doe = self._make_template_doe("pseudo_trace")
+        original_solve = doe.solver.solve
+
+        class _CustomMessage:
+            def __str__(self):
+                return "custom-message"
+
+        def _solve_with_custom_message(*args, **kwargs):
+            res = original_solve(*args, **kwargs)
+            res.solver.message = _CustomMessage()
+            return res
+
+        with patch.object(doe.solver, "solve", side_effect=_solve_with_custom_message):
+            doe.optimize_experiments(n_exp=1)
+
+        self.assertEqual(doe.results["Termination Message"], "custom-message")
+
+    def test_optimize_experiments_safe_metric_failure_sets_nan(self):
+        doe = self._make_template_doe("pseudo_trace")
+        with patch("pyomo.contrib.doe.doe.np.linalg.inv", side_effect=RuntimeError("boom")):
+            with self.assertLogs("pyomo.contrib.doe.doe", level="WARNING") as log_cm:
+                doe.optimize_experiments(n_exp=1)
+
+        scenario = doe.results["Scenarios"][0]
+        self.assertTrue(np.isnan(scenario["log10 A-opt"]))
+        self.assertTrue(
+            any("failed to compute log10 A-opt" in msg for msg in log_cm.output)
+        )
+
+    def test_optimize_experiments_non_cholesky_determinant_initialization(self):
+        exp = RooneyBieglerMultiExperiment(hour=2.0, y=10.0)
+        solver = SolverFactory("ipopt")
+        solver.options["linear_solver"] = "ma57"
+        solver.options["halt_on_ampl_error"] = "yes"
+        solver.options["max_iter"] = 3000
+        doe = DesignOfExperiments(
+            experiment_list=[exp],
+            objective_option="determinant",
+            step=1e-2,
+            solver=solver,
+            _Cholesky_option=False,
+            _only_compute_fim_lower=False,
+        )
+        original_solve = doe.solver.solve
+        solve_count = {"n": 0}
+
+        class _MockSolverInfo:
+            status = "ok"
+            termination_condition = "optimal"
+            message = "mock-solve"
+
+        class _MockResults:
+            solver = _MockSolverInfo()
+
+        def _solve_first_real_then_mock(*args, **kwargs):
+            solve_count["n"] += 1
+            if solve_count["n"] == 1:
+                return original_solve(*args, **kwargs)
+            return _MockResults()
+
+        with patch.object(doe.solver, "solve", side_effect=_solve_first_real_then_mock):
+            doe.optimize_experiments(n_exp=1)
+
+        scenario_block = doe.model.param_scenario_blocks[0]
+        self.assertTrue(hasattr(scenario_block.obj_cons, "determinant"))
+        total_fim = np.array(doe.results["Scenarios"][0]["Total FIM"])
+        expected_det = np.linalg.det(total_fim)
+        self.assertAlmostEqual(
+            pyo.value(scenario_block.obj_cons.determinant), expected_det, places=6
+        )
+
 
 if __name__ == "__main__":
     unittest.main()