add logger

Author: nychiang

src/Drivers/hiopbbpy/BODriverEX.py

@@ -102,6 +102,7 @@ def cons_jac_vec(x):
 
       options = {
         'acquisition_type': acq_type,
+        'log_level': 'none',
         'bo_maxiter': 10,
         'opt_solver': 'IPOPT',  #"SLSQP" "IPOPT" "trust-constr"
         'batch_size': 3,

src/hiopbbpy/opt/boalgorithm.py

@@ -9,12 +9,11 @@
 from numpy.random import uniform
 from scipy.optimize import minimize
 from scipy.stats import qmc
-import warnings
 from ..surrogate_modeling.gp import GaussianProcess
 from .acquisition import LCBacquisition, EIacquisition
 from ..problems.problem import Problem
 from .optproblem import IpoptProb
-from ..utils.util import Evaluator
+from ..utils.util import Evaluator, Logger
 
 # A base class defining a general framework for Bayesian Optimization
 class BOAlgorithmBase:
@@ -35,6 +34,7 @@ def __init__(self):
     self.x_opt = None             # Best observed point
     self.y_opt = None             # Best observed value
     self.idx_opt = None           # Index of the best observed value in the history
+    self.logger = Logger()        # logger
 
   # Sets the acquisition function type and batch size
   def setAcquisitionType(self, acquisition_type, batch_size=1):
@@ -100,6 +100,8 @@ def __init__(self, prob:Problem, gpsurrogate:GaussianProcess, xtrain, ytrain,
     self.opt_evaluator = options.get('opt_evaluator', self.opt_evaluator)
     assert isinstance(self.opt_evaluator, Evaluator)
 
+    self.logger.setlevel(options.get('log_level', "INFO"))
+
     if options and 'opt_solver' in options:
       opt_solver = options['opt_solver']
       assert opt_solver in ["SLSQP", "trust-constr", "IPOPT"], f"Invalid opt_solver: {opt_solver}"
@@ -219,17 +221,17 @@ def optimize(self):
         self.x_hist.append(x_train[-j].flatten())
         self.y_hist.append(y_train[-j].flatten())
       if self.batch_size == 1:
-        print(f"Sample point X:")
+        self.logger.info(f"Sample point X:")
       else:
-        print(f"Sample points X:")
+        self.logger.info(f"Sample points X:")
       for j in range(self.batch_size):
-        print(f"{x_train[-j-1]}")
+        self.logger.info(f"{x_train[-j-1]}")
       if self.batch_size == 1:
-        print(f"\nObservation Y:")
+        self.logger.info(f"Observation Y:")
       else:
-        print(f"\nObservations Y:")
+        self.logger.info(f"Observations Y:")
       for j in range(self.batch_size):
-        print(f"{y_new[-j-1]}")
+        self.logger.info(f"{y_new[-j-1]}")
 
     # Save the optimal results and all the training data
     self.idx_opt = np.argmin(self.y_hist)
@@ -238,9 +240,7 @@ def optimize(self):
     self.setTrainingData(x_train, y_train)
 
     print(f"\n\nOptimal at BO iteration: {self.idx_opt+1} ")
-    print(f"Optimal point: {self.x_opt.flatten()}, Optimal value: {self.y_opt}")
-    print()
-
+    print(f"Optimal point: {self.x_opt.flatten()}, Optimal value: {self.y_opt}\n\n")
 
 class minimizer_wrapper:
   def __init__(self, fun, method, bounds, constraints, solver_options):
@@ -260,15 +260,15 @@ def minimizer_callback(self, x0s):
           y = minimize(self.fun['obj'], x0, method=self.method, bounds=self.bounds, constraints=self.constraints, options=self.solver_options)
         success = y.success
         if not success:
-          print(y.message)
+          self.logger.warning(y.message)
         xopt = y.x
         yopt = y.fun
       elif self.method == "trust-constr":
         nonlinear_constraint = NonlinearConstraint(self.constraints['cons'], self.constraints['cl'], self.constraints['cu'], jac=self.constraints['jac'])
         y = minimize(self.fun['obj'], x0, method=self.method, bounds=self.bounds, constraints=[nonlinear_constraint], options=self.solver_options)
         success = y.success
         if not success:
-          print(y.message)
+          self.logger.warning(y.message)
         xopt = y.x
         yopt = y.fun
       else:
@@ -281,7 +281,7 @@ def minimizer_callback(self, x0s):
           # ipopt returns 0 as success
           success = True
         else:
-          warnings.warn(f"Ipopt failed to solve the problem. Status msg: {msg}")
+          self.logger.warning(f"Ipopt failed to solve the problem. Status msg: {msg}")
           success = False
 
         yopt = info['obj_val']

src/hiopbbpy/utils/util.py

@@ -26,7 +26,7 @@
 """
 import numpy as np
 from .evaluation_manager import EvaluationManager, is_running_with_mpi
-
+import logging
 
 
 def check_required_keys(user_dict, required_keys):
@@ -95,3 +95,41 @@ def run(self, fun, Xin):
       Y = [Fi[0] for Fi in Fout]
     return Y
 
+class Logger:
+  """
+  A simple wrapper for Python's logging module that sets up a reusable logger.
+  Logs to the console using a consistent format.
+  """
+
+  def __init__(self, name='hiopbbpy', level='ERROR'):
+    # Create a logger instance with a given name        
+    self._logger = logging.getLogger(name)
+
+    # Create a console output handler
+    ch = logging.StreamHandler()
+
+    # Define the output format: timestamp, logger name, level, and message
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+
+    # Add the handle
+    ch.setFormatter(formatter)
+    self._logger.addHandler(ch)
+
+    # Set log level
+    self.setlevel(level)
+
+  def setlevel(self, level_str):
+    """Set the log level using a string from 'NONE', 'DEBUG', 'INFO', 'WARNING', 'ERROR' and 'CRITICAL' """
+    if isinstance(level_str, str) and level_str.strip().lower() == 'none':
+        level = logging.CRITICAL + 1  # Disable all logging
+    else:
+        # Default to INFO if string is unrecognized
+        level = getattr(logging, str(level_str).upper(), logging.INFO)
+  
+    self._logger.setLevel(level)
+    for handler in self._logger.handlers:
+      handler.setLevel(level)
+
+  def __getattr__(self, attr):
+    # Forward all unknown attributes to the underlying logger
+    return getattr(self._logger, attr)