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_hyperparams_space_round_robin(to_nested_dict_func_name, to_flat_func_name):
orig_space = copy.deepcopy(HYPE_SPACE)
print(orig_space.keys())
nestened = HyperparameterSpace(getattr(
orig_space,
to_nested_dict_func_name
)())
print(nestened)
flattened = HyperparameterSpace(getattr(
nestened,
to_flat_func_name
)())
print(flattened.keys())
assert flattened.to_flat_as_dict_primitive() == orig_space.to_flat_as_dict_primitive()
from neuraxle.base import BaseStep, TruncableSteps, NonFittableMixin, MetaStepMixin
from neuraxle.hyperparams.distributions import LogUniform, Quantized, RandInt, Boolean
from neuraxle.hyperparams.space import HyperparameterSpace, HyperparameterSamples
HYPERPARAMETERS_SPACE = HyperparameterSpace({
'learning_rate': LogUniform(0.0001, 0.1),
'l2_weight_reg': LogUniform(0.0001, 0.1),
'momentum': LogUniform(0.01, 1.0),
'hidden_size': Quantized(LogUniform(16, 512)),
'num_layers': RandInt(1, 4),
'num_lstm_layers': RandInt(1, 2),
'use_xavier_init': Boolean(),
'use_max_pool_else_avg_pool': Boolean(),
'dropout_drop_proba': LogUniform(0.3, 0.7)
})
HYPERPARAMETERS = HyperparameterSamples({
'learning_rate': 0.1,
'l2_weight_reg': 0.001,
'hidden_size': 32,
'num_layers': 3,
def test_meta_step_mixin_should_get_hyperparams_space():
p = SomeMetaStepMixin(SomeStep())
p.set_hyperparams_space(HyperparameterSpace({
META_STEP_HP: RAND_INT_META_STEP,
SOME_STEP_HP: RAND_INT_SOME_STEP
}))
hyperparams_space = p.get_hyperparams_space()
assert hyperparams_space[META_STEP_HP] == RAND_INT_META_STEP
assert hyperparams_space[SOME_STEP_HP] == RAND_INT_SOME_STEP
:rtype: BaseStep
.. seealso::
* :class:`HyperparameterSpace`
"""
self.is_invalidated = True
hyperparams_space: HyperparameterSpace = HyperparameterSpace(hyperparams_space).to_nested_dict()
remainders = dict()
for name, hparams in hyperparams_space.items():
if name in self.steps.keys():
self.steps[name].set_hyperparams_space(hparams)
else:
remainders[name] = hparams
self.hyperparams_space = HyperparameterSpace(remainders)
return self
def main():
p = Pipeline([
('step1', MultiplyByN()),
('step2', MultiplyByN()),
Pipeline([
Identity(),
Identity(),
SKLearnWrapper(PCA(n_components=4))
])
])
p.set_hyperparams_space(HyperparameterSpace({
'step1__multiply_by': RandInt(42, 50),
'step2__multiply_by': RandInt(-10, 0),
'Pipeline__SKLearnWrapper_PCA__n_components': RandInt(2, 3)
}))
samples = p.get_hyperparams_space().rvs()
p.set_hyperparams(samples)
samples = p.get_hyperparams()
assert 42 <= samples['step1__multiply_by'] <= 50
assert -10 <= samples['step2__multiply_by'] <= 0
assert samples['Pipeline__SKLearnWrapper_PCA__n_components'] in [2, 3]
assert p['Pipeline']['SKLearnWrapper_PCA'].get_wrapped_sklearn_predictor().n_components in [2, 3]
p.set_hyperparams_space(HyperparameterSpace({
'learning_rate': RandInt(0,5),
'some_step__learning_rate': RandInt(0, 10) # will set SomeStep() 'learning_rate' hyperparam space to RandInt(0, 10)
}))
:param hyperparams_space: hyperparameters space
:type hyperparams_space: Union[HyperparameterSpace, OrderedDict, dict]
:return: self
:rtype: BaseStep
.. seealso::
* :class:`HyperparameterSpace`
"""
self.is_invalidated = True
hyperparams_space: HyperparameterSpace = HyperparameterSpace(hyperparams_space).to_nested_dict()
remainders = dict()
for name, hparams in hyperparams_space.items():
if name in self.steps.keys():
self.steps[name].set_hyperparams_space(hparams)
else:
remainders[name] = hparams
self.hyperparams_space = HyperparameterSpace(remainders)
return self
def unnarrow(self) -> 'HyperparameterSpace':
"""
Return the original space before narrowing of the distribution. If the distribution was never narrowed,
the values in the dict will be copies.
:return: the original HyperparameterSpace before narrowing.
"""
new_items = []
for k, v in self.items():
if isinstance(v, HyperparameterDistribution) or isinstance(v, HyperparameterSpace):
v = v.unnarrow()
new_items.append((k, v))
return HyperparameterSpace(new_items)
):
if hyperparams is None:
hyperparams = dict()
if hyperparams_space is None:
hyperparams_space = dict()
if name is None:
name = self.__class__.__name__
if savers is None:
savers = []
if hashers is None:
hashers = [HashlibMd5Hasher()]
self.hyperparams: HyperparameterSamples = HyperparameterSamples(hyperparams)
self.hyperparams = self.hyperparams.to_flat()
self.hyperparams_space: HyperparameterSpace = HyperparameterSpace(hyperparams_space)
self.hyperparams_space = self.hyperparams_space.to_flat()
self.name: str = name
self.savers: List[BaseSaver] = savers # TODO: doc. First is the most stripped.
self.hashers: List[BaseHasher] = hashers
self.pending_mutate: ('BaseStep', str, str) = (None, None, None)
self.is_initialized = False
self.is_invalidated = True
self.is_train: bool = True
def main(tmpdir, sleep_time: float = 0, n_iter: int = 10):
DATA_INPUTS = np.array(range(100))
EXPECTED_OUTPUTS = np.array(range(100, 200))
HYPERPARAMETER_SPACE = HyperparameterSpace({
'multiplication_1__multiply_by': RandInt(1, 2),
'multiplication_2__multiply_by': RandInt(1, 2),
'multiplication_3__multiply_by': RandInt(1, 2),
})
print('Classic Pipeline:')
pipeline = Pipeline([
('multiplication_1', MultiplyByN()),
('sleep_1', ForEachDataInput(Sleep(sleep_time))),
('multiplication_2', MultiplyByN()),
('sleep_2', ForEachDataInput(Sleep(sleep_time))),
('multiplication_3', MultiplyByN()),
]).set_hyperparams_space(HYPERPARAMETER_SPACE)
time_a = time.time()
:type hyperparams_space: HyperparameterSpace
:return: self
"""
self.is_invalidated = True
hyperparams_space: HyperparameterSpace = HyperparameterSpace(hyperparams_space.to_nested_dict())
remainders = dict()
for name, hparams in hyperparams_space.items():
if name == self.wrapped.name:
self.wrapped.set_hyperparams_space(hparams)
else:
remainders[name] = hparams
self.hyperparams_space = HyperparameterSpace(remainders)
return self