Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_apply_on_pipeline_with_meta_step_and_positional_argument_should_call_method_on_each_steps():
pipeline = Pipeline([OutputTransformerWrapper(MultiplyByN(1)), MultiplyByN(1)])
pipeline.apply('set_hyperparams', hyperparams=HyperparameterSamples({'multiply_by': 2}))
assert pipeline.get_hyperparams()['multiply_by'] == 2
assert pipeline['OutputTransformerWrapper'].wrapped.get_hyperparams()['multiply_by'] == 2
assert pipeline['MultiplyByN'].get_hyperparams()['multiply_by'] == 2
lambda step: step.set_hyperparams(HyperparameterSamples({'multiply_by': 2}))
)
def create_callback_step(tape_step_name, hyperparams):
step = (
tape_step_name,
TransformCallbackStepWithMockHasher(
callback_function=TapeCallbackFunction().callback,
more_arguments=[tape_step_name],
hyperparams=HyperparameterSamples(hyperparams)
)
)
return step
# When
pipeline_save = create_pipeline(
tmpdir=tmpdir,
pickle_checkpoint_step=Identity(),
tape=TapeCallbackFunction(),
hyperparameters=HyperparameterSamples({"a__learning_rate": 1}),
different=True,
save_pipeline=False
)
pipeline_save.fit_transform(data_inputs, expected_outputs)
pipeline_load = create_pipeline(
tmpdir=tmpdir,
pickle_checkpoint_step=pickle_checkpoint_step,
tape=tape,
hyperparameters=HyperparameterSamples({"a__learning_rate": 1})
)
pipeline_load, actual_data_inputs = pipeline_load.fit_transform(data_inputs, expected_outputs)
# Then
actual_tape = tape.get_name_tape()
assert np.array_equal(actual_data_inputs, data_inputs)
assert actual_tape == ["1", "2", "3"]
def test_apply_on_pipeline_with_meta_step_should_call_method_on_each_steps():
pipeline = Pipeline([OutputTransformerWrapper(MultiplyByN(1)), MultiplyByN(1)])
pipeline.apply('set_hyperparams', HyperparameterSamples({'multiply_by': 2}))
assert pipeline.get_hyperparams()['multiply_by'] == 2
assert pipeline['OutputTransformerWrapper'].wrapped.get_hyperparams()['multiply_by'] == 2
assert pipeline['MultiplyByN'].get_hyperparams()['multiply_by'] == 2
from testing.test_pipeline import SomeStep
SOME_STEP_HP_KEY = 'somestep_hyperparam'
RAND_INT_SOME_STEP = RandInt(-10, 0)
RAND_INT_STEP_CLONER = RandInt(0, 10)
META_STEP_HP = 'metastep_hyperparam'
SOME_STEP_HP = "SomeStep__somestep_hyperparam"
META_STEP_HP_VALUE = 1
SOME_STEP_HP_VALUE = 2
HYPE_SPACE = HyperparameterSpace({
"a__test": Boolean()
})
HYPE_SAMPLE = HyperparameterSamples({
"a__test": True
})
class SomeMetaStepMixin(NonTransformableMixin, NonFittableMixin, MetaStepMixin, BaseStep):
pass
class SomeStepInverseTransform(SomeStep):
def fit_transform(self, data_inputs, expected_outputs=None):
return self, 'fit_transform'
def inverse_transform(self, processed_outputs):
return 'inverse_transform'
def test_hyperparams_repository_should_create_new_trial(tmpdir):
hyperparams_json_repository = HyperparamsJSONRepository(tmpdir)
hyperparams = HyperparameterSamples(HYPERPARAMS)
hyperparams_json_repository.create_new_trial(hyperparams)
trial_hash = hyperparams_json_repository._get_trial_hash(hyperparams.to_flat_as_dict_primitive())
file_name = 'NEW_' + trial_hash + '.json'
path = os.path.join(tmpdir, file_name)
with open(path) as f:
trial_json = json.load(f)
assert trial_json['hyperparams'] == hyperparams.to_flat_as_dict_primitive()
assert trial_json['score'] is None
Will create an equivalent flat HyperparameterSpace, as a dict.
:return: an HyperparameterSpace like self, flattened.
"""
return nested_dict_to_flat(self, dict_ctor=OrderedDict)
def to_nested_dict_as_ordered_dict_primitive(self) -> OrderedDict:
"""
Will create an equivalent nested dict HyperparameterSpace, as a dict.
:return: a nested primitive dict type of self.
"""
return flat_to_nested_dict(self, dict_ctor=OrderedDict)
class HyperparameterSpace(HyperparameterSamples):
"""Wraps an hyperparameter nested dict or flat dict, and offer a few more functions to process
all contained HyperparameterDistribution.
This can be set on a Pipeline with the method ``set_hyperparams_space``.
Calling ``.rvs()`` on an ``HyperparameterSpace`` results in ``HyperparameterSamples``."""
def rvs(self) -> 'HyperparameterSamples':
"""
Sample the space of random variables.
:return: a random HyperparameterSamples, sampled from a point of the present HyperparameterSpace.
"""
new_items = []
for k, v in self.items():
if isinstance(v, HyperparameterDistribution) or isinstance(v, HyperparameterSpace):
def __init__(self, multiply_by=1):
NonFittableMixin.__init__(self)
BaseStep.__init__(
self,
hyperparams=HyperparameterSamples({
'multiply_by': multiply_by
})
def get_params(self, deep=True):
neuraxle_params = HyperparameterSamples(self.p.get_hyperparams()).to_flat_as_dict_primitive()
return neuraxle_params