VOCS APOSMM direct implementation

.claude/skills/generate-scripts/references/results_metadata.md

@@ -39,4 +39,3 @@ If the minimum objective value is exactly 0.0, check whether those rows have
 `sim_ended == True`. Unevaluated rows often have fields initialized to zero.
 This is common for the last few rows when the simulation budget is exhausted — they were
 allocated by the generator but never evaluated.
-

.pre-commit-config.yaml

@@ -37,4 +37,4 @@ repos:
   rev: v1.19.1
   hooks:
     - id: mypy
-      exclude: ^docs/conf\.py$|libensemble/utils/(launcher|loc_stack|runners|pydantic|output_directory)\.py$|libensemble/tests/(regression_tests|functionality_tests|unit_tests|scaling_tests)/.*
+      pass_filenames: false

libensemble/gen_classes/sampling.py

@@ -156,9 +156,7 @@ class UniformSampleWithVariableResources(LibensembleGenerator):
        path was tested with the default alloc.
     """
 
-    def __init__(
-        self, vocs: VOCS, max_resource_sets: int, random_seed: int = 1, *args, **kwargs
-    ):
+    def __init__(self, vocs: VOCS, max_resource_sets: int, random_seed: int = 1, *args, **kwargs):
         super().__init__(vocs, *args, **kwargs)
         self.rng = np.random.default_rng(random_seed)
         self.max_rsets = max_resource_sets

libensemble/tests/regression_tests/test_aposmm_nlopt.py

@@ -92,12 +92,13 @@ def six_hump_camel_func(x):
     H, _, _ = workflow.run()
 
     if workflow.is_manager:
-        print("[Manager]:", H[np.where(H["local_min"])]["x"])
+        x_min = np.column_stack([H[H["local_min"]]["x0"], H[H["local_min"]]["x1"]])
+        print("[Manager]:", x_min)
         print("[Manager]: Time taken =", time() - start_time, flush=True)
 
         tol = 1e-5
         for m in minima:
             # The minima are known on this test problem.
             # We use their values to test APOSMM has identified all minima
-            print(np.min(np.sum((H[H["local_min"]]["x"] - m) ** 2, 1)), flush=True)
-            assert np.min(np.sum((H[H["local_min"]]["x"] - m) ** 2, 1)) < tol
+            print(np.min(np.sum((x_min - m) ** 2, 1)), flush=True)
+            assert np.min(np.sum((x_min - m) ** 2, 1)) < tol

libensemble/tests/regression_tests/test_aposmm_scipy.py

@@ -89,15 +89,16 @@ def six_hump_camel_func(x):
     H, _, _ = workflow.run()
 
     if workflow.is_manager:
-        print("[Manager]:", H[np.where(H["local_min"])]["x"])
+        x_min = np.column_stack([H[H["local_min"]]["x0"], H[H["local_min"]]["x1"]])
+        print("[Manager]:", x_min)
         print("[Manager]: Time taken =", time() - start_time, flush=True)
 
         tol = 1e-3
         min_found = 0
         for m in minima:
             # The minima are known on this test problem.
             # We use their values to test APOSMM has identified all minima
-            print(np.min(np.sum((H[H["local_min"]]["x"] - m) ** 2, 1)), flush=True)
-            if np.min(np.sum((H[H["local_min"]]["x"] - m) ** 2, 1)) < tol:
+            print(np.min(np.sum((x_min - m) ** 2, 1)), flush=True)
+            if np.min(np.sum((x_min - m) ** 2, 1)) < tol:
                 min_found += 1
         assert min_found >= 2, f"Found {min_found} minima"

libensemble/tests/unit_tests/test_persistent_aposmm.py

@@ -470,7 +470,7 @@ def test_asktell_ingest_first():
 
 @pytest.mark.extra
 def test_asktell_consecutive_during_sample():
-    """Test consecutive suggest and ingest during sample"""
+    """Test consecutive ingest during sampling"""
 
     from gest_api.vocs import VOCS
 
@@ -503,12 +503,11 @@ def test_asktell_consecutive_during_sample():
         dist_to_bound_multiple=0.01,
     )
 
-    # Test consecutive suggest
     first = my_APOSMM.suggest(1)
     first[0]["energy"] = six_hump_camel_func(np.array([first[0]["core"], first[0]["edge"]]))
     my_APOSMM.ingest(first)
-    second = my_APOSMM.suggest(1)
-    second += my_APOSMM.suggest(4)
+    second = my_APOSMM.suggest(5)
+
     for point in second:
         point["energy"] = six_hump_camel_func(np.array([point["core"], point["edge"]]))
     # Test consecutive ingest
@@ -522,8 +521,8 @@ def test_asktell_consecutive_during_sample():
 
     while total_evals < eval_max:
 
-        sample, detected_minima = my_APOSMM.suggest(3), my_APOSMM.suggest_updates()
-        sample += my_APOSMM.suggest(3)
+        sample = my_APOSMM.suggest(6)
+        detected_minima = my_APOSMM.suggest_updates()
         if len(detected_minima):
             for m in detected_minima:
                 potential_minima.append(m)

libensemble/tools/test_support.py

@@ -244,9 +244,9 @@ def check_gpu_setting(task, assert_setting=True, print_setting=False, resources=
     if assert_setting:
         if isinstance(expected, dict):
             for key, value in expected.items():
-                assert key in gpu_setting, (
-                    f"Worker {task.workerID}: Expected env key '{key}' not found in GPU setting: {gpu_setting}"
-                )
+                assert (
+                    key in gpu_setting
+                ), f"Worker {task.workerID}: Expected env key '{key}' not found in GPU setting: {gpu_setting}"
                 assert gpu_setting[key] == value, (
                     f"Worker {task.workerID}: GPU setting key '{key}' has value '{gpu_setting[key]}', "
                     f"expected '{value}'"

libensemble/utils/runners.py

@@ -200,31 +200,33 @@ def _result(self, calc_in: npt.NDArray, persis_info: dict, libE_info: dict) -> (
 
 
 class LibensembleGenRunner(StandardGenRunner):
-    def _get_initial_suggest(self, libE_info) -> npt.NDArray:
-        """Get initial batch from a LibensembleGenerator.
+    def _get_mapping_for_outputs(self) -> dict:
+        """Return mappings whose internal fields are declared generator outputs."""
+        output_names = {field[0] for field in self.specs.get("out", [])}
+        return {
+            name: fields for name, fields in getattr(self.gen, "variables_mapping", {}).items() if name in output_names
+        }
 
-        LibensembleGenerator.suggest_numpy emits VOCS-field-named structured arrays
-        (e.g. x0/x1, energy). The manager-side history expects mapped fields (x, f)
-        unless the user explicitly requested otherwise.
-        """
+    def _get_initial_suggest(self, libE_info) -> npt.NDArray:
+        """Get initial batch from a LibensembleGenerator."""
         initial_batch = self.specs.get("initial_batch_size") or self.specs.get("batch_size") or libE_info["batch_size"]
         H_out = self.gen.suggest_numpy(initial_batch)
-        return map_numpy_array(H_out, mapping=getattr(self.gen, "variables_mapping", {}))
+        return map_numpy_array(H_out, mapping=self._get_mapping_for_outputs())
 
     def _get_points_updates(self, batch_size: int) -> (npt.NDArray, list):
         numpy_out = self.gen.suggest_numpy(batch_size)
-        numpy_out = map_numpy_array(numpy_out, mapping=getattr(self.gen, "variables_mapping", {}))
+        numpy_out = map_numpy_array(numpy_out, mapping=self._get_mapping_for_outputs())
         if callable(getattr(self.gen, "suggest_updates", None)):
             updates = self.gen.suggest_updates()
         else:
             updates = None
         return numpy_out, updates
 
     def _convert_ingest(self, x: npt.NDArray) -> list:
-        self.gen.ingest_numpy(unmap_numpy_array(x, mapping=getattr(self.gen, "variables_mapping", {})))
+        self.gen.ingest_numpy(unmap_numpy_array(x, mapping=self._get_mapping_for_outputs()))
 
     def _convert_initial_ingest(self, x: npt.NDArray) -> list:
-        self.gen.ingest_numpy(unmap_numpy_array(x, mapping=getattr(self.gen, "variables_mapping", {})))
+        self.gen.ingest_numpy(unmap_numpy_array(x, mapping=self._get_mapping_for_outputs()))
 
 
 class LibensembleGenThreadRunner(StandardGenRunner):