import os
import json
import numpy as np
from .json import json_safe
from smt_optim.utils.multi_obj import hypervolume, spacing, get_pf_from_dataset
[docs]
class ConsoleLogger:
def __init__(self, config):
self.config = config
self.headers = [
"iter",
"budget",
"fmin",
"rscv",
"fidelity",
"gp_time",
"acq_time",
]
self.widths: list | None = None
self.header_fmt: str | None = None
self.row_fmt: str | None = None
self.update_header_format()
self.formats = {
"iter": ".0f",
"budget": ".3f",
"fmin": ".5e",
"HV": ".5e", # hypervolume (for multi-obj only)
"spacing": ".5e", # spacing (for multi-obj only)
"rscv": ".3e",
"fidelity": ".0f",
"gp_time": ".3f",
"acq_time": ".3f",
}
self.iter = 0
self.repeat_header = 10
self.multi_obj_ref = None # reference objective values (for multi-obj only)
[docs]
def on_iter_end(self, state) -> None:
if self.iter % self.repeat_header == 0:
if state.problem.num_obj > 1 and "fmin" in self.headers:
self.headers[2] = "HV"
self.headers[3] = "spacing"
# self.headers.pop(3)
self.update_header_format()
dataset = state.dataset.export_as_dict()
obj = dataset["obj"]
dataset["rscv"]
# self.multi_obj_ref = np.array([
# obj[:, 0][rscv <= 1e-4].max(),
# obj[:, 1][rscv <= 1e-4].max(),
# ])
self.multi_obj_ref = np.empty(state.problem.num_obj)
for obj_idx in range(state.problem.num_obj):
self.multi_obj_ref[obj_idx] = obj[:, obj_idx].max()
self.print_header()
sample = state.get_best_sample(ctol=1e-4)
iter_log = getattr(state, "iter_log", {}) or {}
data = {
"iter": state.iter,
"budget": state.budget,
# "rscv": sample.metadata["rscv"],
"fidelity": iter_log.get("fidelity", np.nan),
"gp_time": iter_log.get("gp_training_time", np.nan),
"acq_time": iter_log.get("acq_opt_time", np.nan),
}
if state.problem.num_obj == 1:
data["fmin"] = sample.obj[0]
data["rscv"] = sample.metadata["rscv"]
elif state.problem.num_obj > 1:
# dataset = state.dataset.export_as_dict()
# obj = dataset["obj"]
# rscv = dataset["rscv"]
# obj = obj[rscv <= 1e-4, :]
# pf = get_pareto_front(obj)
pf = get_pf_from_dataset(state.dataset, ctol=1e-4)
# hv_indicator = HV(ref_point=self.multi_obj_ref)
if pf.shape[0] > 0:
hv = hypervolume(pf, self.multi_obj_ref)
sp = spacing(pf)
else:
hv = np.nan
sp = np.nan
data["HV"] = hv
data["spacing"] = sp
else:
data["HV"] = np.nan
data["spacing"] = np.nan
row = [format_value(data[h], self.formats[h]) for h in self.headers]
print(self.row_fmt.format(*row))
self.iter += 1
[docs]
class JsonLogger:
def __init__(self, config):
self.dir = config.results_dir
[docs]
def on_iter_end(self, state) -> None:
path = os.path.join(self.dir, "stats.jsonl")
os.makedirs(self.dir, exist_ok=True)
with open(path, "a") as file:
safe_iter_log = json_safe(state.iter_log)
file.write(json.dumps(safe_iter_log) + "\n")
# json.dump(safe_iter_log, file, indent=4)
# class DoeLogger:
# def __init__(self, config):
# self.config = config
# self.num_saved = 0
#
#
# def log_sample(self, state, sample) -> None:
# """
# Log sample data once sampled
#
# :param state:
# :param sample:
# :return:
# """
#
# if self.config.results_dir is None:
# return None
#
# try:
# row = dict()
#
# row["iter"] = sample.metadata["iter"]
# row["budget"] = np.nan # self.compute_used_budget() # self.budget
#
# # save variables
# for i in range(len(sample.x)):
# row[f"x{i}"] = sample.x[i]
#
# # save objectives
# for i in range(len(sample.obj)):
# row[f"f{i}"] = sample.obj[i]
#
# # save constraints
# for i in range(len(sample.cstr)):
# row[f"c{i}"] = sample.cstr[i]
#
# row["time"] = np.sum(sample.eval_time)
#
# path = os.path.join(self.config.results_dir, "doe.csv")
# file_exists = os.path.isfile(path)
#
# # possibly does not work on Windows -> to be tested
# with open(path, 'a') as file:
# writer = csv.DictWriter(file, fieldnames=row.keys())
#
# if not file_exists:
# writer.writeheader()
#
# writer.writerow(row)
#
# self.num_saved += 1
#
# except Exception as e:
# print(f"Error while saving the DoE: {e}")
#
# pass
#
# def on_iter_end(self, state) -> None:
# """
# DOE data should be logged right after sampling the blackbox function (to avoid loss of data)
#
# :param state:
# :return:
# """
# num_samples = len(state.dataset.samples)
#
# for idx in range(self.num_saved, num_samples):
#
# sample = state.dataset.samples[idx]
#
# try:
# row = dict()
#
# row["iter"] = sample.metadata["iter"]
# row["budget"] = np.nan # self.compute_used_budget() # self.budget
#
# # save variables
# for i in range(len(sample.x)):
# row[f"x{i}"] = sample.x[i]
#
# # save objectives
# for i in range(len(sample.obj)):
# row[f"f{i}"] = sample.obj[i]
#
# # save constraints
# for i in range(len(sample.cstr)):
# row[f"c{i}"] = sample.cstr[i]
#
# row["time"] = np.sum(sample.eval_time)
#
# path = os.path.join(self.config.results_dir, "DOE")
# os.makedirs(path, exist_ok=True)
#
# path = os.path.join(self.config.results_dir, "DOE", f"doe_fidelity_{sample.fidelity}.csv")
# file_exists = os.path.isfile(path)
#
# # possibly does not work on Windows -> to be tested
# with open(path, 'a') as file:
# writer = csv.DictWriter(file, fieldnames=row.keys())
#
# if not file_exists:
# writer.writeheader()
#
# writer.writerow(row)
#
# self.num_saved += 1
#
# except Exception as e:
# print(f"Error while saving the DoE: {e}")
