# SPDX-License-Identifier: MPL-2.0
# Copyright (C) 2025- The University of Tokyo
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import numpy as np
import copy
from .. import utility
MAX_SEARCH = int(30000)
[docs]
class History:
def __init__(self, dim: int):
self.dim = dim
self.num_runs = int(0)
self.total_num_search = int(0)
self.fx = np.zeros(MAX_SEARCH, dtype=float)
self.action_X = np.zeros((MAX_SEARCH, self.dim), dtype=float)
self.terminal_num_run = np.zeros(MAX_SEARCH, dtype=int)
# index of the best valid observation so far; -1 until the first
# valid (finite) observation
self.best_index = np.full(MAX_SEARCH, -1, dtype=int)
self.time_total_ = np.zeros(MAX_SEARCH, dtype=float)
self.time_update_predictor_ = np.zeros(MAX_SEARCH, dtype=float)
self.time_get_action_ = np.zeros(MAX_SEARCH, dtype=float)
self.time_run_simulator_ = np.zeros(MAX_SEARCH, dtype=float)
@property
def time_total(self):
return copy.copy(self.time_total_[0 : self.num_runs])
@property
def time_update_predictor(self):
return copy.copy(self.time_update_predictor_[0 : self.num_runs])
@property
def time_get_action(self):
return copy.copy(self.time_get_action_[0 : self.num_runs])
@property
def time_run_simulator(self):
return copy.copy(self.time_run_simulator_[0 : self.num_runs])
@property
def valid_mask(self):
"""
Mask of valid observations (True) vs failed ones (False).
An observation is failed when its objective value is not finite
(NaN or +-Inf). Failed observations are kept in the history but
excluded from the training data and the best-value tracking.
Returns
-------
numpy.ndarray of bool, shape (total_num_search,)
"""
return utility.finite_mask(self.fx[0 : self.total_num_search])
[docs]
def export_valid(self):
"""
Export the valid (successfully evaluated) observations.
Returns
-------
X: numpy.ndarray
Inputs of the valid observations (N_valid x dim).
fx: numpy.ndarray
Objective values of the valid observations.
"""
N = self.total_num_search
mask = self.valid_mask
return self.action_X[0:N, :][mask], self.fx[0:N][mask]
[docs]
def write(
self,
t,
action_X,
time_total=None,
time_update_predictor=None,
time_get_action=None,
time_run_simulator=None,
):
"""
Overwrite fx and action_X by t and action_X.
Parameters
----------
t: numpy.ndarray
N dimensional array. The negative energy of each search candidate (value of the objective function to be optimized).
action_X: numpy.ndarray
N x d dimensional array. The input of each search candidate.
time_total: numpy.ndarray
N dimenstional array. The total elapsed time in each step.
If None (default), filled by 0.0.
time_update_predictor: numpy.ndarray
N dimenstional array. The elapsed time for updating predictor (e.g., learning hyperparemters) in each step.
If None (default), filled by 0.0.
time_get_action: numpy.ndarray
N dimenstional array. The elapsed time for getting next action in each step.
If None (default), filled by 0.0.
time_run_simulator: numpy.ndarray
N dimenstional array. The elapsed time for running the simulator in each step.
If None (default), filled by 0.0.
Returns
-------
"""
N = utility.length_vector(t)
st = self.total_num_search
en = st + N
self.terminal_num_run[self.num_runs] = en
self.fx[st:en] = t
self.action_X[st:en, :] = action_X
self._update_best_index(st, en)
self.num_runs += 1
self.total_num_search += N
if time_total is None:
time_total = np.zeros(N, dtype=float)
self.time_total_[st:en] = time_total
if time_update_predictor is None:
time_update_predictor = np.zeros(N, dtype=float)
self.time_update_predictor_[st:en] = time_update_predictor
if time_get_action is None:
time_get_action = np.zeros(N, dtype=float)
self.time_get_action_[st:en] = time_get_action
if time_run_simulator is None:
time_run_simulator = np.zeros(N, dtype=float)
self.time_run_simulator_[st:en] = time_run_simulator
def _update_best_index(self, st, en):
"""
Update best_index[st:en] from fx.
Failed (non-finite) observations are skipped, and best_index stays -1
until the first valid one.
"""
for n in range(st, en):
prev = self.best_index[n - 1] if n > 0 else -1
if np.isfinite(self.fx[n]) and (prev < 0 or self.fx[n] > self.fx[prev]):
self.best_index[n] = n
else:
self.best_index[n] = prev
[docs]
def export_sequence_best_fx(self):
"""
Export best fx and X at each sequence (each call of write function).
Returns
-------
best_fx: numpy.ndarray (num_runs)
The best fx at each sequence.
best_X: numpy.ndarray (num_runs, dim)
The best X at each sequence.
"""
# NaN until the first valid observation
best_fx = np.full(self.num_runs, np.nan, dtype=float)
best_X = np.full((self.num_runs, self.dim), np.nan, dtype=float)
for r in range(self.num_runs):
n = self.terminal_num_run[r] - 1
b = self.best_index[n]
if b >= 0:
best_fx[r] = self.fx[b]
best_X[r, :] = self.action_X[b, :]
return best_fx, best_X
[docs]
def export_all_sequence_best_fx(self):
"""
Export all fx and actions at each sequence.
(The total number of data is total_num_research.)
Returns
-------
best_fx: numpy.ndarray
best_actions: numpy.ndarray
"""
# NaN until the first valid observation
best_fx = np.full(self.total_num_search, np.nan, dtype=float)
best_X = np.full((self.total_num_search, self.dim), np.nan, dtype=float)
for n in range(self.total_num_search):
b = self.best_index[n]
if b >= 0:
best_fx[n] = self.fx[b]
best_X[n, :] = self.action_X[b, :]
return best_fx, best_X
[docs]
def save(self, filename):
"""
Save the information of the history.
Parameters
----------
filename: str
The name of the file which stores the information of the history
Returns
-------
"""
N = self.total_num_search
M = self.num_runs
np.savez_compressed(
filename,
num_runs=M,
total_num_search=N,
fx=self.fx[0:N],
action_X=self.action_X[0:N, :],
best_index=self.best_index[0:N],
terminal_num_run=self.terminal_num_run[0:M],
time_total=self.time_total_[0:N],
time_update_predictor=self.time_update_predictor_[0:N],
time_get_action=self.time_get_action_[0:N],
time_run_simulator=self.time_run_simulator_[0:N],
)
[docs]
def load(self, filename):
"""
Load the information of the history.
Parameters
----------
filename: str
The name of the file which stores the information of the history
Returns
-------
"""
data = np.load(filename)
M = int(data["num_runs"])
N = int(data["total_num_search"])
self.num_runs = M
self.total_num_search = N
self.fx[0:N] = data["fx"]
self.action_X[0:N, :] = data["action_X"]
self.terminal_num_run[0:M] = data["terminal_num_run"]
# best_index is recomputed from fx: the saved one may refer to a
# failed observation if the file was written by an older version
self._update_best_index(0, N)
self.time_total_[0:N] = data["time_total"]
self.time_update_predictor_[0:N] = data["time_update_predictor"]
self.time_get_action_[0:N] = data["time_get_action"]
self.time_run_simulator_[0:N] = data["time_run_simulator"]
[docs]
def show_search_results(self, N):
n = self.total_num_search
index = self.best_index[n - 1]
if index >= 0:
best_msg = f"current best f(x) = {self.fx[index]:.6f} (best action={self.action_X[index, :]})"
else:
best_msg = "current best f(x) = (no valid observation yet)"
if N == 1:
print(
f"{n:04d}-th step: f(x) = {self.fx[n - 1]:.6f} (action={self.action_X[n - 1, :]})"
)
print(" " + best_msg + " \n")
else:
print(best_msg)
print("list of simulation results")
st = self.total_num_search - N
en = self.total_num_search
for n in range(st, en):
print(f"f(x)={self.fx[n]:.6f} (action = {self.action_X[n, :]})")
print("\n")