|
| 1 | +import logging |
| 2 | +import numpy as np |
| 3 | +from numpy.random import default_rng |
| 4 | +from numpy import array |
| 5 | + |
| 6 | +# here we use train_model from spotPython.light.trainmodel |
| 7 | +# and not from spot.light.traintest: |
| 8 | +from spotPython.light.trainmodel import train_model |
| 9 | +from spotPython.hyperparameters.values import ( |
| 10 | + assign_values, |
| 11 | + generate_one_config_from_var_dict, |
| 12 | +) |
| 13 | + |
| 14 | +logger = logging.getLogger(__name__) |
| 15 | +py_handler = logging.FileHandler(f"{__name__}.log", mode="w") |
| 16 | +py_formatter = logging.Formatter("%(name)s %(asctime)s %(levelname)s %(message)s") |
| 17 | +py_handler.setFormatter(py_formatter) |
| 18 | +logger.addHandler(py_handler) |
| 19 | + |
| 20 | + |
| 21 | +class HyperLightning: |
| 22 | + """ |
| 23 | + Hyperparameter Tuning for Lightning. |
| 24 | + """ |
| 25 | + |
| 26 | + def __init__(self, seed: int = 126, log_level: int = 50) -> None: |
| 27 | + self.seed = seed |
| 28 | + self.rng = default_rng(seed=self.seed) |
| 29 | + self.fun_control = { |
| 30 | + "seed": None, |
| 31 | + "data": None, |
| 32 | + "step": 10_000, |
| 33 | + "horizon": None, |
| 34 | + "grace_period": None, |
| 35 | + "metric_river": None, |
| 36 | + "metric_sklearn": None, |
| 37 | + "weights": array([1, 0, 0]), |
| 38 | + "weight_coeff": 0.0, |
| 39 | + "log_level": log_level, |
| 40 | + "var_name": [], |
| 41 | + "var_type": [], |
| 42 | + } |
| 43 | + self.log_level = self.fun_control["log_level"] |
| 44 | + logger.setLevel(self.log_level) |
| 45 | + logger.info(f"Starting the logger at level {self.log_level} for module {__name__}:") |
| 46 | + |
| 47 | + def check_X_shape(self, X: np.ndarray) -> np.ndarray: |
| 48 | + """ |
| 49 | + Checks the shape of the input array X and raises an exception if it is not valid. |
| 50 | +
|
| 51 | + Args: |
| 52 | + X (np.ndarray): |
| 53 | + input array. |
| 54 | +
|
| 55 | + Returns: |
| 56 | + np.ndarray: |
| 57 | + input array with valid shape. |
| 58 | +
|
| 59 | + Raises: |
| 60 | + Exception: |
| 61 | + if the shape of the input array is not valid. |
| 62 | +
|
| 63 | + Examples: |
| 64 | + >>> hyper_light = HyperLight(seed=126, log_level=50) |
| 65 | + >>> X = np.array([[1, 2], [3, 4]]) |
| 66 | + >>> hyper_light.check_X_shape(X) |
| 67 | + array([[1, 2], |
| 68 | + [3, 4]]) |
| 69 | + """ |
| 70 | + try: |
| 71 | + X.shape[1] |
| 72 | + except ValueError: |
| 73 | + X = np.array([X]) |
| 74 | + if X.shape[1] != len(self.fun_control["var_name"]): |
| 75 | + raise Exception("Invalid shape of input array X.") |
| 76 | + return X |
| 77 | + |
| 78 | + def fun(self, X: np.ndarray, fun_control: dict = None) -> np.ndarray: |
| 79 | + """ |
| 80 | + Evaluates the function for the given input array X and control parameters. |
| 81 | +
|
| 82 | + Args: |
| 83 | + X (np.ndarray): |
| 84 | + input array. |
| 85 | + fun_control (dict): |
| 86 | + dictionary containing control parameters for the hyperparameter tuning. |
| 87 | +
|
| 88 | + Returns: |
| 89 | + (np.ndarray): |
| 90 | + array containing the evaluation results. |
| 91 | +
|
| 92 | + Examples: |
| 93 | + >>> hyper_light = HyperLight(seed=126, log_level=50) |
| 94 | + X = np.array([[1, 2], [3, 4]]) |
| 95 | + fun_control = {"weights": np.array([1, 0, 0])} |
| 96 | + hyper_light.fun(X, fun_control) |
| 97 | + array([nan, nan]) |
| 98 | + """ |
| 99 | + z_res = np.array([], dtype=float) |
| 100 | + if fun_control is not None: |
| 101 | + self.fun_control.update(fun_control) |
| 102 | + self.check_X_shape(X) |
| 103 | + var_dict = assign_values(X, self.fun_control["var_name"]) |
| 104 | + # type information and transformations are considered in generate_one_config_from_var_dict: |
| 105 | + for config in generate_one_config_from_var_dict(var_dict, self.fun_control): |
| 106 | + logger.debug(f"\nconfig: {config}") |
| 107 | + # extract parameters like epochs, batch_size, lr, etc. from config |
| 108 | + # config_id = generate_config_id(config) |
| 109 | + try: |
| 110 | + print("fun: Calling train_model") |
| 111 | + df_eval = train_model(config, self.fun_control) |
| 112 | + print("fun: train_model returned") |
| 113 | + except Exception as err: |
| 114 | + logger.error(f"Error in fun(). Call to train_model failed. {err=}, {type(err)=}") |
| 115 | + logger.error("Setting df_eval to np.nan") |
| 116 | + df_eval = np.nan |
| 117 | + z_val = self.fun_control["weights"] * df_eval |
| 118 | + z_res = np.append(z_res, z_val) |
| 119 | + return z_res |
0 commit comments