merged redundant BOEngine.__init__s

This commit is contained in:
2026-08-14 19:28:54 +09:00
parent 1ddfc9d246
commit 785760d23d
3 changed files with 60 additions and 121 deletions
+57 -4
View File
@@ -1,15 +1,68 @@
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
import os
import numpy as np
class BOEngine(ABC): class BOEngine(ABC):
def __init__(self, config): def __init__(self, config):
self.config = config self.config = config
self.state = None self.params = [key for key, spec in config["bo"]["params"].items() if spec is not None]
self.param_types = {key: config["bo"]["params"][key].get("type", "float") for key in self.params}
self.integer_indices = [i for i, param in enumerate(self.params) if self.param_types[param] == 'int']
self.bounds = np.empty((2, len(self.params)))
for i, param in enumerate(self.params):
center = config["ptycho"]["params"][param]
radius = config["bo"]["params"][param]["radius"]
self.bounds[0, i] = center - radius
self.bounds[1, i] = center + radius
self.train_x = np.empty((0, len(self.params))) # shape: (BOiter, BOparam)
self.train_y = np.empty((0,)) # shape: (BOiter,)
train_x_path = config["bo"].get("train_x")
train_y_path = config["bo"].get("train_y")
if train_x_path is not None and train_y_path is not None:
if os.path.exists(train_x_path) and os.path.exists(train_y_path):
train_x = np.load(train_x_path)
train_y = np.load(train_y_path)
assert train_x.ndim == 2, "loaded train_x must be 2D"
assert train_x.shape[1] == len(self.params), "loaded train_x shape(1) does not match number of variable parameters"
assert train_y.ndim == 1, "loaded train_y must be 1D"
assert train_y.shape[0] == train_x.shape[0], "loaded train_x and train_y shape(0) have unequal iterations"
self.train_x = train_x
self.train_y = train_y
@abstractmethod @abstractmethod
def ask(self, n: int = 1) -> list[dict]: def ask(self, n: int = 1) -> list[dict]:
pass pass
@abstractmethod
def tell(self, job_config, y_value): def tell(self, job_config, y_value) -> None:
pass config = self.config
x_value = []
for param in self.params:
x_value.append(job_config['ptycho']['params'][param])
self.train_x = np.vstack([
self.train_x,
np.array(x_value).reshape(1, -1)
])
self.train_y = np.concatenate([
self.train_y,
np.array([y_value])
])
train_x_path = config['bo'].get('train_x')
train_y_path = config['bo'].get('train_y')
if train_x_path is not None and train_y_path is not None:
np.save(train_x_path, self.train_x)
np.save(train_y_path, self.train_y)
else:
result_dir = config['io']['result_dir']
np.save(os.path.join(result_dir, 'train_x.npy'), self.train_x)
np.save(os.path.join(result_dir, 'train_y.npy'), self.train_y)
+3 -61
View File
@@ -1,4 +1,3 @@
import os
import copy import copy
import numpy as np import numpy as np
from bo.base import BOEngine from bo.base import BOEngine
@@ -9,43 +8,15 @@ class RandomBOEngine(BOEngine):
def __init__(self, config): def __init__(self, config):
super().__init__(config) super().__init__(config)
self.params = [key for key, spec in config["bo"]["params"].items() if spec is not None]
self.param_types = {key: config["bo"]["params"][key].get("type", "float") for key in self.params}
self.integer_indices = [i for i, param in enumerate(self.params) if self.param_types[param] == 'int']
self.bounds = np.empty((2, len(self.params)))
for i, param in enumerate(self.params):
center = config["ptycho"]["params"][param]
radius = config["bo"]["params"][param]["radius"]
self.bounds[0, i] = center - radius
self.bounds[1, i] = center + radius
self.train_x = np.empty((0, len(self.params))) # shape: (BOiter, BOparam)
self.train_y = np.empty((0,)) # shape: (BOiter,)
train_x_path = config["bo"].get("train_x")
train_y_path = config["bo"].get("train_y")
if train_x_path is not None and train_y_path is not None:
if os.path.exists(train_x_path) and os.path.exists(train_y_path):
train_x = np.load(train_x_path)
train_y = np.load(train_y_path)
assert train_x.ndim == 2, "loaded train_x must be 2D"
assert train_x.shape[1] == len(self.params), "loaded train_x shape(1) does not match number of variable parameters"
assert train_y.ndim == 1, "loaded train_y must be 1D"
assert train_y.shape[0] == train_x.shape[0], "loaded train_x and train_y shape(0) have unequal iterations"
self.train_x = train_x
self.train_y = train_y
def ask(self, n = 1): def ask(self, n = 1):
config = self.config
next_configs = [] next_configs = []
for _ in range(n): for _ in range(n):
next_config = copy.deepcopy(config) next_config = copy.deepcopy(self.config)
for param in self.params: for param in self.params:
radius = config['bo']['params'][param]['radius'] radius = self.config['bo']['params'][param]['radius']
center = config['ptycho']['params'][param] center = self.config['ptycho']['params'][param]
modulation = radius * (np.random.rand() - 0.5) * 2 modulation = radius * (np.random.rand() - 0.5) * 2
next_value = center + modulation next_value = center + modulation
if self.param_types[param] == 'int': if self.param_types[param] == 'int':
@@ -54,32 +25,3 @@ class RandomBOEngine(BOEngine):
next_configs.append(next_config) next_configs.append(next_config)
return next_configs return next_configs
def tell(self, job_config, y_value):
config = self.config
x_value = []
for param in self.params:
x_value.append(job_config['ptycho']['params'][param])
self.train_x = np.vstack([
self.train_x,
np.array(x_value).reshape(1, -1)
])
self.train_y = np.concatenate([
self.train_y,
np.array([y_value])
])
train_x_path = config['bo'].get('train_x')
train_y_path = config['bo'].get('train_y')
if train_x_path is not None and train_y_path is not None:
np.save(train_x_path, self.train_x)
np.save(train_y_path, self.train_y)
else:
result_dir = config['io']['result_dir']
np.save(os.path.join(result_dir, 'train_x.npy'), self.train_x)
np.save(os.path.join(result_dir, 'train_y.npy'), self.train_y)
-56
View File
@@ -21,34 +21,6 @@ from bo.base import BOEngine
class SingleObjectiveBOEngine(BOEngine): class SingleObjectiveBOEngine(BOEngine):
def __init__(self, config): def __init__(self, config):
super().__init__(config) super().__init__(config)
self.params = [key for key, spec in config["bo"]["params"].items() if spec is not None]
self.param_types = {key: config["bo"]["params"][key].get("type", "float") for key in self.params}
self.integer_indices = [i for i, param in enumerate(self.params) if self.param_types[param] == 'int']
self.bounds = np.empty((2, len(self.params)))
for i, param in enumerate(self.params):
center = config["ptycho"]["params"][param]
radius = config["bo"]["params"][param]["radius"]
self.bounds[0, i] = center - radius
self.bounds[1, i] = center + radius
self.train_x = np.empty((0, len(self.params))) # shape: (BOiter, BOparam)
self.train_y = np.empty((0,)) # shape: (BOiter,)
train_x_path = config["bo"].get("train_x")
train_y_path = config["bo"].get("train_y")
if train_x_path is not None and train_y_path is not None:
if os.path.exists(train_x_path) and os.path.exists(train_y_path):
train_x = np.load(train_x_path)
train_y = np.load(train_y_path)
assert train_x.ndim == 2, "loaded train_x must be 2D"
assert train_x.shape[1] == len(self.params), "loaded train_x shape(1) does not match number of variable parameters"
assert train_y.ndim == 1, "loaded train_y must be 1D"
assert train_y.shape[0] == train_x.shape[0], "loaded train_x and train_y shape(0) have unequal iterations"
self.train_x = train_x
self.train_y = train_y
self.acquisition = config['bo']['acquisition'] self.acquisition = config['bo']['acquisition']
@@ -146,34 +118,6 @@ class SingleObjectiveBOEngine(BOEngine):
return X_out return X_out
def tell(self, job_config, y_value):
config = self.config
x_value = []
for param in self.params:
x_value.append(job_config['ptycho']['params'][param])
self.train_x = np.vstack([
self.train_x,
np.array(x_value).reshape(1, -1)
])
self.train_y = np.concatenate([
self.train_y,
np.array([y_value])
])
train_x_path = config['bo'].get('tain_x')
train_y_path = config['bo'].get('train_y')
if train_x_path is not None and train_y_path is not None:
np.save(train_x_path, self.train_x)
np.save(train_y_path, self.train_y)
else:
result_dir = config['io']['result_dir']
np.save(os.path.join(result_dir, 'train_x.npy'), self.train_x)
np.save(os.path.join(result_dir, 'train_y.npy'), self.train_y)