Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:param low_noise_std_deviation: Standard deviation of Gaussian observation noise on low fidelity observations.
Defaults to zero.
:return: Tuple of user function object and parameter space
"""
parameter_space = ParameterSpace([
ContinuousParameter('borehole_radius', 0.05, 0.15),
ContinuousParameter('radius_of_influence', 100, 50000),
ContinuousParameter('upper_aquifer_transmissivity', 63070, 115600),
ContinuousParameter('upper_aquifer_head', 990, 1110),
ContinuousParameter('lower_aquifer_transmissivity', 63.1, 116),
ContinuousParameter('lower_aquifer_head', 700, 820),
ContinuousParameter('borehole_length', 1120, 1680),
ContinuousParameter('hydraulic_conductivity', 9855, 12045),
InformationSourceParameter(2)])
user_function = MultiSourceFunctionWrapper([
lambda x: _borehole_low(x, low_noise_std_deviation),
lambda x: _borehole_high(x, high_noise_std_deviation)])
return user_function, parameter_space
x1 = x[:, 0]
x2 = x[:, 1]
result = (10.0 * np.sqrt(_branin(x - 2.0)[:, 0]) + 2.0 * (x1 - 0.5) - 3.0 * (
3.0 * x2 - 1.0) - 1.0) / 100.
return result[:, None]
def branin_low_fidelity(x):
x2 = x[:, 1]
result = (branin_medium_fidelity(1.2 * (x + 2.0))[:, 0] * 100. - 3.0 * x2 + 1.0) / 100.
return result[:, None]
parameter_space = ParameterSpace([ContinuousParameter('x1', -5, 10), ContinuousParameter('x2', 0, 15),
InformationSourceParameter(3)])
branin_high = lambda x: _branin(x)/100
return MultiSourceFunctionWrapper([branin_low_fidelity, branin_medium_fidelity, branin_high]), parameter_space
function_input = np.array([1, 0])
msfw = MultiSourceFunctionWrapper(functions)
msfw.evaluate(function_input)
# invalid function output
with pytest.raises(ValueError):
functions = [lambda x: np.array([2])]
function_input = np.array([[1, 0]])
msfw = MultiSourceFunctionWrapper(functions)
msfw.evaluate(function_input)
# invalid function output type
with pytest.raises(ValueError):
functions = [lambda x: [2]]
function_input = np.array([[1, 0]])
msfw = MultiSourceFunctionWrapper(functions)
msfw.evaluate(function_input)
def test_multi_source_function_wrapper_evaluation_with_multiple_extra_arguments():
functions = [lambda x: (2 * x, np.array([[1]] * x.shape[0]), np.array([[1]] * x.shape[0])),
lambda x: (4 * x, np.array([[2]] * x.shape[0]), np.array([[1]] * x.shape[0]))]
function_input = np.array([[1, 0], [2, 1], [3, 0], [4, 0], [5, 1]])
source_index = -1
msfw = MultiSourceFunctionWrapper(functions, source_index, extra_output_names=['cost', 'constraint'])
output = msfw.evaluate(function_input)
assert len(output) == function_input.shape[0]
for i, record in enumerate(output):
assert_array_equal(output[i].X, function_input[i])
this_function = functions[function_input[i, source_index]]
this_function_input = np.delete(function_input[i], source_index)
assert_array_equal(output[i].Y, this_function(this_function_input)[0])
assert_array_equal(output[i].cost, this_function(this_function_input)[1][0])
assert_array_equal(output[i].constraint, this_function(this_function_input)[2][0])
def test_multi_source_function_wrapper_too_many_outputs_outputs_fails():
functions = [lambda x: (2 * x, np.array([[1]] * x.shape[0]), np.array([[1]] * x.shape[0])),
lambda x: (4 * x, np.array([[2]] * x.shape[0]), np.array([[1]] * x.shape[0]))]
function_input = np.array([[1, 0], [2, 1], [3, 0], [4, 0], [5, 1]])
source_index = -1
msfw = MultiSourceFunctionWrapper(functions, source_index)
with pytest.raises(ValueError):
msfw.evaluate(function_input)
def test_multi_source_function_wrapper_too_few_outputs_outputs_fails():
functions = [lambda x: 2 * x,
lambda x: 4 * x]
function_input = np.array([[1, 0], [2, 1], [3, 0], [4, 0], [5, 1]])
source_index = -1
msfw = MultiSourceFunctionWrapper(functions, source_index, extra_output_names=['cost'])
with pytest.raises(ValueError):
msfw.evaluate(function_input)
def test_multi_source_function_wrapper_evaluation_single_output():
functions = [lambda x: 2 * x, lambda x: 4 * x]
function_input = np.array([[1, 0], [2, 1], [3, 0], [4, 0], [5, 1]])
source_index = -1
msfw = MultiSourceFunctionWrapper(functions, source_index)
output = msfw.evaluate(function_input)
assert len(output) == function_input.shape[0]
for i, record in enumerate(output):
assert_array_equal(output[i].X, function_input[i])
this_function = functions[function_input[i, source_index]]
this_function_input = np.delete(function_input[i], source_index)
assert_array_equal(output[i].Y, this_function(this_function_input))
def test_multi_source_function_wrapper_evaluation_with_cost():
functions = [lambda x: (2 * x, np.array([[1]] * x.shape[0])),
lambda x: (4 * x, np.array([[2]] * x.shape[0]))]
function_input = np.array([[1, 0], [2, 1], [3, 0], [4, 0], [5, 1]])
source_index = -1
msfw = MultiSourceFunctionWrapper(functions, source_index, extra_output_names=['cost'])
output = msfw.evaluate(function_input)
assert len(output) == function_input.shape[0]
for i, record in enumerate(output):
assert_array_equal(output[i].X, function_input[i])
this_function = functions[function_input[i, source_index]]
this_function_input = np.delete(function_input[i], source_index)
assert_array_equal(output[i].Y, this_function(this_function_input)[0])
assert_array_equal(output[i].cost, this_function(this_function_input)[1][0])