Rewrite for named tuples

.cspell.json

@@ -21,6 +21,7 @@
         "discretization",
         "discretize",
         "drivername",
+        "dropna",
         "dstpath",
         "dtype",
         "duckdb",

mlos_bench/mlos_bench/optimizers/mlos_core_optimizer.py

@@ -29,6 +29,7 @@
     OptimizerType,
     SpaceAdapterType,
 )
+from mlos_core.optimizers.observations import Observations
 
 _LOG = logging.getLogger(__name__)
 
@@ -128,7 +129,7 @@ def bulk_register(
 
         # TODO: Specify (in the config) which metrics to pass to the optimizer.
         # Issue: https://github.com/microsoft/MLOS/issues/745
-        self._opt.register(configs=df_configs, scores=df_scores)
+        self._opt.register(observations=Observations(config=df_configs, score=df_scores))
 
         if _LOG.isEnabledFor(logging.DEBUG):
             (score, _) = self.get_best_observation()
@@ -198,10 +199,10 @@ def suggest(self) -> TunableGroups:
         tunables = super().suggest()
         if self._start_with_defaults:
             _LOG.info("Use default values for the first trial")
-        df_config, _metadata = self._opt.suggest(defaults=self._start_with_defaults)
+        suggestion = self._opt.suggest(defaults=self._start_with_defaults)
         self._start_with_defaults = False
-        _LOG.info("Iteration %d :: Suggest:\n%s", self._iter, df_config)
-        return tunables.assign(configspace_data_to_tunable_values(df_config.loc[0].to_dict()))
+        _LOG.info("Iteration %d :: Suggest:\n%s", self._iter, suggestion.config)
+        return tunables.assign(configspace_data_to_tunable_values(suggestion.config.to_dict()))
 
     def register(
         self,
@@ -221,18 +222,20 @@ def register(
             # TODO: Specify (in the config) which metrics to pass to the optimizer.
             # Issue: https://github.com/microsoft/MLOS/issues/745
             self._opt.register(
-                configs=df_config,
-                scores=pd.DataFrame([registered_score], dtype=float),
+                observations=Observations(
+                    config=df_config,
+                    score=pd.DataFrame([registered_score], dtype=float),
+                )
             )
         return registered_score
 
     def get_best_observation(
         self,
     ) -> Union[Tuple[Dict[str, float], TunableGroups], Tuple[None, None]]:
-        (df_config, df_score, _df_context) = self._opt.get_best_observations()
-        if len(df_config) == 0:
+        best_observations = self._opt.get_best_observations()
+        if len(best_observations.config.index) == 0:
             return (None, None)
-        params = configspace_data_to_tunable_values(df_config.iloc[0].to_dict())
-        scores = self._adjust_signs_df(df_score).iloc[0].to_dict()
+        params = configspace_data_to_tunable_values(best_observations.config.iloc[0].to_dict())
+        scores = self._adjust_signs_df(best_observations.score).iloc[0].to_dict()
         _LOG.debug("Best observation: %s score: %s", params, scores)
         return (scores, self._tunables.copy().assign(params))

mlos_bench/mlos_bench/tests/optimizers/toy_optimization_loop_test.py

@@ -17,7 +17,8 @@
 from mlos_bench.optimizers.mock_optimizer import MockOptimizer
 from mlos_bench.tunables.tunable_groups import TunableGroups
 from mlos_core.optimizers.bayesian_optimizers.smac_optimizer import SmacOptimizer
-from mlos_core.util import config_to_dataframe
+from mlos_core.optimizers.observations import Suggestion
+from mlos_core.util import config_to_series
 
 # For debugging purposes output some warnings which are captured with failed tests.
 DEBUG = True
@@ -40,10 +41,13 @@ def _optimize(env: Environment, opt: Optimizer) -> Tuple[float, TunableGroups]:
             # pylint: disable=protected-access
             if isinstance(opt, MlosCoreOptimizer) and isinstance(opt._opt, SmacOptimizer):
                 config = tunable_values_to_configuration(tunables)
-                config_df = config_to_dataframe(config)
+                config_df = config_to_series(config)
                 logger("config: %s", str(config))
                 try:
-                    logger("prediction: %s", opt._opt.surrogate_predict(configs=config_df))
+                    logger(
+                        "prediction: %s",
+                        opt._opt.surrogate_predict(suggestion=Suggestion(config=config_df)),
+                    )
                 except RuntimeError:
                     pass
 

mlos_core/mlos_core/optimizers/bayesian_optimizers/bayesian_optimizer.py

@@ -5,57 +5,38 @@
 """Contains the wrapper classes for base Bayesian optimizers."""
 
 from abc import ABCMeta, abstractmethod
-from typing import Optional
 
 import numpy.typing as npt
-import pandas as pd
 
+from mlos_core.optimizers.observations import Suggestion
 from mlos_core.optimizers.optimizer import BaseOptimizer
 
 
 class BaseBayesianOptimizer(BaseOptimizer, metaclass=ABCMeta):
     """Abstract base class defining the interface for Bayesian optimization."""
 
     @abstractmethod
-    def surrogate_predict(
-        self,
-        *,
-        configs: pd.DataFrame,
-        context: Optional[pd.DataFrame] = None,
-    ) -> npt.NDArray:
+    def surrogate_predict(self, *, suggestion: Suggestion) -> npt.NDArray:
         """
         Obtain a prediction from this Bayesian optimizer's surrogate model for the given
         configuration(s).
 
         Parameters
         ----------
-        configs : pd.DataFrame
-            Dataframe of configs / parameters. The columns are parameter names and
-            the rows are the configs.
-
-        context : pd.DataFrame
-            Not Yet Implemented.
+        suggestion: Suggestion
+            The suggestion containing the configuration(s) to predict.
         """
         pass  # pylint: disable=unnecessary-pass # pragma: no cover
 
     @abstractmethod
-    def acquisition_function(
-        self,
-        *,
-        configs: pd.DataFrame,
-        context: Optional[pd.DataFrame] = None,
-    ) -> npt.NDArray:
+    def acquisition_function(self, *, suggestion: Suggestion) -> npt.NDArray:
         """
         Invokes the acquisition function from this Bayesian optimizer for the given
         configuration.
 
         Parameters
         ----------
-        configs : pd.DataFrame
-            Dataframe of configs / parameters. The columns are parameter names and
-            the rows are the configs.
-
-        context : pd.DataFrame
-            Not Yet Implemented.
+        suggestion: Suggestion
+            The suggestion containing the configuration(s) to evaluate.
         """
         pass  # pylint: disable=unnecessary-pass # pragma: no cover

mlos_core/mlos_core/optimizers/bayesian_optimizers/smac_optimizer.py

@@ -11,16 +11,19 @@
 from logging import warning
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
 from warnings import warn
 
 import ConfigSpace
+import numpy as np
 import numpy.typing as npt
 import pandas as pd
+from smac.utils.configspace import convert_configurations_to_array
 
 from mlos_core.optimizers.bayesian_optimizers.bayesian_optimizer import (
     BaseBayesianOptimizer,
 )
+from mlos_core.optimizers.observations import Observation, Observations, Suggestion
 from mlos_core.spaces.adapters.adapter import BaseSpaceAdapter
 from mlos_core.spaces.adapters.identity_adapter import IdentityAdapter
 
@@ -269,63 +272,49 @@ def _dummy_target_func(config: ConfigSpace.Configuration, seed: int = 0) -> None
         # release: https://github.com/automl/SMAC3/issues/946
         raise RuntimeError("This function should never be called.")
 
-    def _register(
-        self,
-        *,
-        configs: pd.DataFrame,
-        scores: pd.DataFrame,
-        context: Optional[pd.DataFrame] = None,
-        metadata: Optional[pd.DataFrame] = None,
-    ) -> None:
+    def _register(self, *, observation: Observation) -> None:
         """
         Registers the given configs and scores.
 
         Parameters
         ----------
-        configs : pd.DataFrame
-            Dataframe of configs / parameters. The columns are parameter names and
-            the rows are the configs.
-
-        scores : pd.DataFrame
-            Scores from running the configs. The index is the same as the index of
-            the configs.
-
-        context : pd.DataFrame
-            Not Yet Implemented.
-
-        metadata: pd.DataFrame
-            Not Yet Implemented.
+        observation: Observation
+            The observation to register.
         """
         from smac.runhistory import (  # pylint: disable=import-outside-toplevel
             StatusType,
             TrialInfo,
             TrialValue,
         )
 
-        if context is not None:
-            warn(f"Not Implemented: Ignoring context {list(context.columns)}", UserWarning)
-
-        # Register each trial (one-by-one)
-        for config, (_i, score) in zip(
-            self._to_configspace_configs(configs=configs), scores.iterrows()
-        ):
-            # Retrieve previously generated TrialInfo (returned by .ask()) or create
-            # new TrialInfo instance
-            info: TrialInfo = self.trial_info_map.get(
-                config,
-                TrialInfo(config=config, seed=self.base_optimizer.scenario.seed),
+        if observation.context is not None:
+            warn(
+                f"Not Implemented: Ignoring context {list(observation.context.index)}",
+                UserWarning,
             )
-            value = TrialValue(cost=list(score.astype(float)), time=0.0, status=StatusType.SUCCESS)
-            self.base_optimizer.tell(info, value, save=False)
+
+        # Retrieve previously generated TrialInfo (returned by .ask()) or create
+        # new TrialInfo instance
+        config = ConfigSpace.Configuration(
+            self.optimizer_parameter_space, values=observation.config.dropna().to_dict()
+        )
+        info: TrialInfo = self.trial_info_map.get(
+            config,
+            TrialInfo(config=config, seed=self.base_optimizer.scenario.seed),
+        )
+        value = TrialValue(
+            cost=list(observation.score.astype(float)), time=0.0, status=StatusType.SUCCESS
+        )
+        self.base_optimizer.tell(info, value, save=False)
 
         # Save optimizer once we register all configs
         self.base_optimizer.optimizer.save()
 
     def _suggest(
         self,
         *,
-        context: Optional[pd.DataFrame] = None,
-    ) -> Tuple[pd.DataFrame, Optional[pd.DataFrame]]:
+        context: Optional[pd.Series] = None,
+    ) -> Suggestion:
         """
         Suggests a new configuration.
 
@@ -336,49 +325,36 @@ def _suggest(
 
         Returns
         -------
-        configuration : pd.DataFrame
-            Pandas dataframe with a single row. Column names are the parameter names.
-
-        metadata : Optional[pd.DataFrame]
-            Not yet implemented.
+        suggestion: Suggestion
+            The suggestion to evaluate.
         """
         if TYPE_CHECKING:
             # pylint: disable=import-outside-toplevel,unused-import
             from smac.runhistory import TrialInfo
 
         if context is not None:
-            warn(f"Not Implemented: Ignoring context {list(context.columns)}", UserWarning)
+            warn(f"Not Implemented: Ignoring context {list(context.index)}", UserWarning)
 
         trial: TrialInfo = self.base_optimizer.ask()
         trial.config.is_valid_configuration()
         self.optimizer_parameter_space.check_configuration(trial.config)
         assert trial.config.config_space == self.optimizer_parameter_space
         self.trial_info_map[trial.config] = trial
-        config_df = pd.DataFrame(
-            [trial.config], columns=list(self.optimizer_parameter_space.keys())
-        )
-        return config_df, None
+        config_sr = pd.Series(dict(trial.config), dtype=object)
+        return Suggestion(config=config_sr, context=context, metadata=None)
 
-    def register_pending(
-        self,
-        *,
-        configs: pd.DataFrame,
-        context: Optional[pd.DataFrame] = None,
-        metadata: Optional[pd.DataFrame] = None,
-    ) -> None:
+    def register_pending(self, *, pending: Suggestion) -> None:
         raise NotImplementedError()
 
-    def surrogate_predict(
-        self,
-        *,
-        configs: pd.DataFrame,
-        context: Optional[pd.DataFrame] = None,
-    ) -> npt.NDArray:
+    def surrogate_predict(self, *, suggestion: Suggestion) -> npt.NDArray:
         # pylint: disable=import-outside-toplevel
         from smac.utils.configspace import convert_configurations_to_array
 
-        if context is not None:
-            warn(f"Not Implemented: Ignoring context {list(context.columns)}", UserWarning)
+        if suggestion.context is not None:
+            warn(
+                f"Not Implemented: Ignoring context {list(suggestion.context.index)}",
+                UserWarning,
+            )
         if self._space_adapter and not isinstance(self._space_adapter, IdentityAdapter):
             raise NotImplementedError("Space adapter not supported for surrogate_predict.")
 
@@ -392,55 +368,38 @@ def surrogate_predict(
         if self.base_optimizer._config_selector._model is None:
             raise RuntimeError("Surrogate model is not yet trained")
 
-        config_array: npt.NDArray = convert_configurations_to_array(
-            self._to_configspace_configs(configs=configs)
+        config_array = convert_configurations_to_array(
+            [
+                ConfigSpace.Configuration(
+                    self.optimizer_parameter_space, values=suggestion.config.to_dict()
+                )
+            ]
         )
         mean_predictions, _ = self.base_optimizer._config_selector._model.predict(config_array)
         return mean_predictions.reshape(
             -1,
         )
 
-    def acquisition_function(
-        self,
-        *,
-        configs: pd.DataFrame,
-        context: Optional[pd.DataFrame] = None,
-    ) -> npt.NDArray:
-        if context is not None:
-            warn(f"Not Implemented: Ignoring context {list(context.columns)}", UserWarning)
+    def acquisition_function(self, *, suggestion: Suggestion) -> npt.NDArray:
+        if suggestion.context is not None:
+            warn(
+                f"Not Implemented: Ignoring context {list(suggestion.context.index)}",
+                UserWarning,
+            )
         if self._space_adapter:
             raise NotImplementedError()
 
         # pylint: disable=protected-access
         if self.base_optimizer._config_selector._acquisition_function is None:
             raise RuntimeError("Acquisition function is not yet initialized")
 
-        cs_configs: list = self._to_configspace_configs(configs=configs)
-        return self.base_optimizer._config_selector._acquisition_function(cs_configs).reshape(
+        return self.base_optimizer._config_selector._acquisition_function(
+            suggestion.config.config_to_configspace(self.optimizer_parameter_space)
+        ).reshape(
             -1,
         )
 
     def cleanup(self) -> None:
         if hasattr(self, "_temp_output_directory") and self._temp_output_directory is not None:
             self._temp_output_directory.cleanup()
             self._temp_output_directory = None
-
-    def _to_configspace_configs(self, *, configs: pd.DataFrame) -> List[ConfigSpace.Configuration]:
-        """
-        Convert a dataframe of configs to a list of ConfigSpace configs.
-
-        Parameters
-        ----------
-        configs : pd.DataFrame
-            Dataframe of configs / parameters. The columns are parameter names and
-            the rows are the configs.
-
-        Returns
-        -------
-        configs : list
-            List of ConfigSpace configs.
-        """
-        return [
-            ConfigSpace.Configuration(self.optimizer_parameter_space, values=config.to_dict())
-            for (_, config) in configs.astype("O").iterrows()
-        ]