Replace event based logging

bayes_opt/__init__.py

@@ -5,10 +5,10 @@
 import importlib.metadata
 
 from bayes_opt import acquisition
-from bayes_opt.bayesian_optimization import BayesianOptimization, Events
+from bayes_opt.bayesian_optimization import BayesianOptimization
 from bayes_opt.constraint import ConstraintModel
 from bayes_opt.domain_reduction import SequentialDomainReductionTransformer
-from bayes_opt.logger import JSONLogger, ScreenLogger
+from bayes_opt.logger import ScreenLogger
 from bayes_opt.target_space import TargetSpace
 
 __version__ = importlib.metadata.version("bayesian-optimization")
@@ -19,8 +19,6 @@
     "BayesianOptimization",
     "TargetSpace",
     "ConstraintModel",
-    "Events",
     "ScreenLogger",
-    "JSONLogger",
     "SequentialDomainReductionTransformer",
 ]

bayes_opt/bayesian_optimization.py

@@ -21,14 +21,13 @@
 from bayes_opt import acquisition
 from bayes_opt.constraint import ConstraintModel
 from bayes_opt.domain_reduction import DomainTransformer
-from bayes_opt.event import DEFAULT_EVENTS, Events
-from bayes_opt.logger import _get_default_logger
+from bayes_opt.logger import ScreenLogger
 from bayes_opt.parameter import wrap_kernel
 from bayes_opt.target_space import TargetSpace
 from bayes_opt.util import ensure_rng
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Iterable, Mapping
+    from collections.abc import Callable, Mapping
 
     from numpy.random import RandomState
     from numpy.typing import NDArray
@@ -41,35 +40,7 @@
     Float = np.floating[Any]
 
 
-class Observable:
-    """Inspired by https://www.protechtraining.com/blog/post/879#simple-observer."""
-
-    def __init__(self, events: Iterable[Any]) -> None:
-        # maps event names to subscribers
-        # str -> dict
-        self._events = {event: dict() for event in events}
-
-    def get_subscribers(self, event: Any) -> Any:
-        """Return the subscribers of an event."""
-        return self._events[event]
-
-    def subscribe(self, event: Any, subscriber: Any, callback: Callable[..., Any] | None = None) -> None:
-        """Add subscriber to an event."""
-        if callback is None:
-            callback = subscriber.update
-        self.get_subscribers(event)[subscriber] = callback
-
-    def unsubscribe(self, event: Any, subscriber: Any) -> None:
-        """Remove a subscriber for a particular event."""
-        del self.get_subscribers(event)[subscriber]
-
-    def dispatch(self, event: Any) -> None:
-        """Trigger callbacks for subscribers of an event."""
-        for callback in self.get_subscribers(event).values():
-            callback(event, self)
-
-
-class BayesianOptimization(Observable):
+class BayesianOptimization:
     """Handle optimization of a target function over a specific target space.
 
     This class takes the function to optimize as well as the parameters bounds
@@ -105,6 +76,27 @@ class BayesianOptimization(Observable):
         This behavior may be desired in high noise situations where repeatedly probing
         the same point will give different answers. In other situations, the acquisition
         may occasionally generate a duplicate point.
+
+    Attributes
+    ----------
+    space : TargetSpace
+        The target space object containing the function to optimize.
+
+    acquisition_function : AcquisitionFunction
+        The acquisition function used to propose new points.
+
+    constraint : ConstraintModel or None
+        The constraint model, if any.
+
+    max : dict or None
+        Maximum target value observed and corresponding parameters.
+
+    res : list
+        All target values observed and corresponding parameters and timestamps.
+
+    logger : ScreenLogger
+        The logger used for displaying optimization progress.
+        You can customize the logger's properties (e.g., colors, verbosity, formatting).
     """
 
     def __init__(
@@ -173,7 +165,13 @@ def __init__(
             self._bounds_transformer.initialize(self._space)
 
         self._sorting_warning_already_shown = False  # TODO: remove in future version
-        super().__init__(events=DEFAULT_EVENTS)
+
+        # Initialize logger
+        self.logger = ScreenLogger(
+            verbose=self._verbose,
+            is_constrained=self.is_constrained,
+            params_config=self._space._params_config,
+        )
 
     @property
     def space(self) -> TargetSpace:
@@ -236,7 +234,7 @@ def register(
             warn(msg, stacklevel=1)
             self._sorting_warning_already_shown = True
         self._space.register(params, target, constraint_value)
-        self.dispatch(Events.OPTIMIZATION_STEP)
+        self.logger.log_optimization_step(self._space.keys, self._space.res()[-1], self.max)
 
     def probe(self, params: ParamsType, lazy: bool = True) -> None:
         """Evaluate the function at the given points.
@@ -268,7 +266,7 @@ def probe(self, params: ParamsType, lazy: bool = True) -> None:
             self._queue.append(params)
         else:
             self._space.probe(params)
-            self.dispatch(Events.OPTIMIZATION_STEP)
+            self.logger.log_optimization_step(self._space.keys, self._space.res()[-1], self.max)
 
     def suggest(self) -> dict[str, float | NDArray[Float]]:
         """Suggest a promising point to probe next."""
@@ -295,13 +293,6 @@ def _prime_queue(self, init_points: int) -> None:
             sample = self._space.random_sample(random_state=self._random_state)
             self._queue.append(self._space.array_to_params(sample))
 
-    def _prime_subscriptions(self) -> None:
-        if not any([len(subs) for subs in self._events.values()]):
-            _logger = _get_default_logger(self._verbose, self.is_constrained)
-            self.subscribe(Events.OPTIMIZATION_START, _logger)
-            self.subscribe(Events.OPTIMIZATION_STEP, _logger)
-            self.subscribe(Events.OPTIMIZATION_END, _logger)
-
     def maximize(self, init_points: int = 5, n_iter: int = 25) -> None:
         r"""
         Maximize the given function over the target space.
@@ -324,8 +315,10 @@ def maximize(self, init_points: int = 5, n_iter: int = 25) -> None:
             optimization routine, make sure to fit it manually, e.g. by calling
             ``optimizer._gp.fit(optimizer.space.params, optimizer.space.target)``.
         """
-        self._prime_subscriptions()
-        self.dispatch(Events.OPTIMIZATION_START)
+        # Log optimization start
+        self.logger.log_optimization_start(self._space.keys)
+
+        # Prime the queue with random points
         self._prime_queue(init_points)
 
         iteration = 0
@@ -342,7 +335,8 @@ def maximize(self, init_points: int = 5, n_iter: int = 25) -> None:
                 # the init_points points (only for the true iterations)
                 self.set_bounds(self._bounds_transformer.transform(self._space))
 
-        self.dispatch(Events.OPTIMIZATION_END)
+        # Log optimization end
+        self.logger.log_optimization_end()
 
     def set_bounds(self, new_bounds: BoundsMapping) -> None:
         """Modify the bounds of the search space.