Enable context for SMAC Optimizer

.vscode/settings.json

@@ -17,7 +17,6 @@
         // See Also:
         // - https://github.com/microsoft/vscode/issues/2809#issuecomment-1544387883
         // - mlos_bench/config/schemas/README.md
-
         {
             "fileMatch": [
                 "mlos_bench/mlos_bench/tests/config/schemas/environments/test-cases/**/*.jsonc",
@@ -131,13 +130,17 @@
     "[python]": {
         "editor.defaultFormatter": "ms-python.autopep8",
         "editor.formatOnSave": true,
-        "editor.formatOnSaveMode": "modifications"
+        "editor.formatOnSaveMode": "modifications",
+        "editor.codeActionsOnSave": {
+            "source.organizeImports": true,
+            "source.fixAll": "always"
+        },
     },
     // See Also .vscode/launch.json for environment variable args to pytest during debug sessions.
     // For the rest, see setup.cfg
     "python.testing.pytestArgs": [
-        "--log-level=DEBUG",
         "."
     ],
-    "python.testing.unittestEnabled": false
+    "python.testing.unittestEnabled": false,
+    "restructuredtext.pythonRecommendation.disabled": true
 }

conda-envs/condaenv.ib_tklms.requirements.txt

@@ -0,0 +1,15 @@
+autopep8>=1.7.0
+bump2version
+check-jsonschema
+licenseheaders
+mypy
+pandas-stubs
+types-beautifulsoup4
+types-colorama
+types-jsonschema
+types-pygments
+types-pytest-lazy-fixture
+types-requests
+types-setuptools
+--editable ../mlos_core[full-tests]
+--editable ../mlos_bench[full-tests]

mlos_bench/mlos_bench/optimizers/base_optimizer.py

@@ -9,25 +9,23 @@
 
 import logging
 from abc import ABCMeta, abstractmethod
-from distutils.util import strtobool    # pylint: disable=deprecated-module
-
+from distutils.util import strtobool  # pylint: disable=deprecated-module
 from types import TracebackType
 from typing import Dict, Optional, Sequence, Tuple, Type, Union
-from typing_extensions import Literal
 
 from ConfigSpace import ConfigurationSpace
-
+from typing_extensions import Literal
 from mlos_bench.config.schemas import ConfigSchema
-from mlos_bench.services.base_service import Service
 from mlos_bench.environments.status import Status
+from mlos_bench.optimizers.convert_configspace import tunable_groups_to_configspace
+from mlos_bench.services.base_service import Service
 from mlos_bench.tunables.tunable import TunableValue
 from mlos_bench.tunables.tunable_groups import TunableGroups
-from mlos_bench.optimizers.convert_configspace import tunable_groups_to_configspace
 
 _LOG = logging.getLogger(__name__)
 
 
-class Optimizer(metaclass=ABCMeta):     # pylint: disable=too-many-instance-attributes
+class Optimizer(metaclass=ABCMeta):  # pylint: disable=too-many-instance-attributes
     """
     An abstract interface between the benchmarking framework and mlos_core optimizers.
     """
@@ -40,11 +38,13 @@ class Optimizer(metaclass=ABCMeta):     # pylint: disable=too-many-instance-attr
         "start_with_defaults",
     }
 
-    def __init__(self,
-                 tunables: TunableGroups,
-                 config: dict,
-                 global_config: Optional[dict] = None,
-                 service: Optional[Service] = None):
+    def __init__(
+        self,
+        tunables: TunableGroups,
+        config: dict,
+        global_config: Optional[dict] = None,
+        service: Optional[Service] = None,
+    ):
         """
         Create a new optimizer for the given configuration space defined by the tunables.
 
@@ -68,25 +68,30 @@ def __init__(self,
         self._seed = int(config.get("seed", 42))
         self._in_context = False
 
-        experiment_id = self._global_config.get('experiment_id')
+        experiment_id = self._global_config.get("experiment_id")
         self.experiment_id = str(experiment_id).strip() if experiment_id else None
 
         self._iter = 0
         # If False, use the optimizer to suggest the initial configuration;
         # if True (default), use the already initialized values for the first iteration.
         self._start_with_defaults: bool = bool(
-            strtobool(str(self._config.pop('start_with_defaults', True))))
-        self._max_iter = int(self._config.pop('max_suggestions', 100))
+            strtobool(str(self._config.pop("start_with_defaults", True)))
+        )
+        self._max_iter = int(self._config.pop("max_suggestions", 100))
 
-        opt_targets: Dict[str, str] = self._config.pop('optimization_targets', {'score': 'min'})
+        opt_targets: Dict[str, str] = self._config.pop(
+            "optimization_targets", {"score": "min"}
+        )
         self._opt_targets: Dict[str, Literal[1, -1]] = {}
-        for (opt_target, opt_dir) in opt_targets.items():
+        for opt_target, opt_dir in opt_targets.items():
             if opt_dir == "min":
                 self._opt_targets[opt_target] = 1
             elif opt_dir == "max":
                 self._opt_targets[opt_target] = -1
             else:
-                raise ValueError(f"Invalid optimization direction: {opt_dir} for {opt_target}")
+                raise ValueError(
+                    f"Invalid optimization direction: {opt_dir} for {opt_target}"
+                )
 
     def _validate_json_config(self, config: dict) -> None:
         """
@@ -108,7 +113,7 @@ def __repr__(self) -> str:
         )
         return f"{self.name}({opt_targets},config={self._config})"
 
-    def __enter__(self) -> 'Optimizer':
+    def __enter__(self) -> "Optimizer":
         """
         Enter the optimizer's context.
         """
@@ -117,9 +122,12 @@ def __enter__(self) -> 'Optimizer':
         self._in_context = True
         return self
 
-    def __exit__(self, ex_type: Optional[Type[BaseException]],
-                 ex_val: Optional[BaseException],
-                 ex_tb: Optional[TracebackType]) -> Literal[False]:
+    def __exit__(
+        self,
+        ex_type: Optional[Type[BaseException]],
+        ex_val: Optional[BaseException],
+        ex_tb: Optional[TracebackType],
+    ) -> Literal[False]:
         """
         Exit the context of the optimizer.
         """
@@ -191,7 +199,9 @@ def config_space(self) -> ConfigurationSpace:
             The ConfigSpace representation of the tunable parameters.
         """
         if self._config_space is None:
-            self._config_space = tunable_groups_to_configspace(self._tunables, self._seed)
+            self._config_space = tunable_groups_to_configspace(
+                self._tunables, self._seed
+            )
             _LOG.debug("ConfigSpace: %s", self._config_space)
         return self._config_space
 
@@ -204,7 +214,7 @@ def name(self) -> str:
         return self.__class__.__name__
 
     @property
-    def targets(self) -> Dict[str, Literal['min', 'max']]:
+    def targets(self) -> Dict[str, Literal["min", "max"]]:
         """
         A dictionary of {target: direction} of optimization targets.
         """
@@ -221,10 +231,12 @@ def supports_preload(self) -> bool:
         return True
 
     @abstractmethod
-    def bulk_register(self,
-                      configs: Sequence[dict],
-                      scores: Sequence[Optional[Dict[str, TunableValue]]],
-                      status: Optional[Sequence[Status]] = None) -> bool:
+    def bulk_register(
+        self,
+        configs: Sequence[dict],
+        scores: Sequence[Optional[Dict[str, TunableValue]]],
+        status: Optional[Sequence[Status]] = None,
+    ) -> bool:
         """
         Pre-load the optimizer with the bulk data from previous experiments.
 
@@ -242,8 +254,12 @@ def bulk_register(self,
         is_not_empty : bool
             True if there is data to register, false otherwise.
         """
-        _LOG.info("Update the optimizer with: %d configs, %d scores, %d status values",
-                  len(configs or []), len(scores or []), len(status or []))
+        _LOG.info(
+            "Update the optimizer with: %d configs, %d scores, %d status values",
+            len(configs or []),
+            len(scores or []),
+            len(status or []),
+        )
         if len(configs or []) != len(scores or []):
             raise ValueError("Numbers of configs and scores do not match.")
         if status is not None and len(configs or []) != len(status or []):
@@ -272,8 +288,12 @@ def suggest(self) -> TunableGroups:
         return self._tunables.copy()
 
     @abstractmethod
-    def register(self, tunables: TunableGroups, status: Status,
-                 score: Optional[Dict[str, TunableValue]] = None) -> Optional[Dict[str, float]]:
+    def register(
+        self,
+        tunables: TunableGroups,
+        status: Status,
+        score: Optional[Dict[str, TunableValue]] = None,
+    ) -> Optional[Dict[str, float]]:
         """
         Register the observation for the given configuration.
 
@@ -294,15 +314,22 @@ def register(self, tunables: TunableGroups, status: Status,
             Benchmark scores extracted (and possibly transformed)
             from the dataframe that's being MINIMIZED.
         """
-        _LOG.info("Iteration %d :: Register: %s = %s score: %s",
-                  self._iter, tunables, status, score)
+        _LOG.info(
+            "Iteration %d :: Register: %s = %s score: %s",
+            self._iter,
+            tunables,
+            status,
+            score,
+        )
         if status.is_succeeded() == (score is None):  # XOR
             raise ValueError("Status and score must be consistent.")
         return self._get_scores(status, score)
 
-    def _get_scores(self, status: Status,
-                    scores: Optional[Union[Dict[str, TunableValue], Dict[str, float]]]
-                    ) -> Optional[Dict[str, float]]:
+    def _get_scores(
+        self,
+        status: Status,
+        scores: Optional[Union[Dict[str, TunableValue], Dict[str, float]]],
+    ) -> Optional[Dict[str, float]]:
         """
         Extract a scalar benchmark score from the dataframe.
         Change the sign if we are maximizing.
@@ -329,7 +356,7 @@ def _get_scores(self, status: Status,
 
         assert scores is not None
         target_metrics: Dict[str, float] = {}
-        for (opt_target, opt_dir) in self._opt_targets.items():
+        for opt_target, opt_dir in self._opt_targets.items():
             val = scores[opt_target]
             assert val is not None
             target_metrics[opt_target] = float(val) * opt_dir
@@ -344,7 +371,9 @@ def not_converged(self) -> bool:
         return self._iter < self._max_iter
 
     @abstractmethod
-    def get_best_observation(self) -> Union[Tuple[Dict[str, float], TunableGroups], Tuple[None, None]]:
+    def get_best_observation(
+        self,
+    ) -> Union[Tuple[Dict[str, float], TunableGroups], Tuple[None, None]]:
         """
         Get the best observation so far.