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_mlpipeline(self):
primitives = [
'sklearn.ensemble.RandomForestClassifier'
]
init_params = {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}
}
pipeline = MLPipeline(primitives=primitives, init_params=init_params)
pipeline2 = MLPipeline(pipeline)
assert pipeline2.primitives == ['sklearn.ensemble.RandomForestClassifier']
assert pipeline2.init_params == {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}
def test_none(self):
primitives = [
'sklearn.ensemble.RandomForestClassifier'
]
init_params = {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}
}
pipeline = MLPipeline(primitives=primitives, init_params=init_params)
assert pipeline.primitives == ['sklearn.ensemble.RandomForestClassifier']
assert pipeline.init_params == {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}
def test_mlpipeline(self):
primitives = [
'sklearn.ensemble.RandomForestClassifier'
]
init_params = {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}
}
pipeline = MLPipeline(primitives=primitives, init_params=init_params)
pipeline2 = MLPipeline(pipeline)
assert pipeline2.primitives == ['sklearn.ensemble.RandomForestClassifier']
assert pipeline2.init_params == {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}
try:
primitive_path = os.path.join(MLBLOCKS_PRIMITIVES, primitive_filename)
with open(primitive_path, 'r') as f:
primitive = json.load(f)
primitive_name = primitive['name']
fixed_hyperparameters = primitive.get('hyperparameters', dict()).get('fixed', dict())
init_hyperparameters = dict()
for name, hyperparameter in fixed_hyperparameters.items():
if 'default' not in hyperparameter:
type_ = hyperparameter.get('type')
init_hyperparameters[name] = HYPERPARAMETER_DEFAULTS.get(type_)
block_name = primitive_name + '#1'
mlpipeline = MLPipeline(
primitives=[primitive_name],
init_params={block_name: init_hyperparameters}
)
# Validate methods
mlblock = mlpipeline.blocks[block_name]
if mlblock._class:
fit = primitive.get('fit')
if fit:
assert hasattr(mlblock.instance, fit['method'])
produce = primitive['produce']
assert hasattr(mlblock.instance, produce['method'])
except Exception:
raise ValueError("Invalid JSON primitive: {}".format(primitive_filename))
'n_estimators': 500
}
},
'input_names': {
'sklearn.ensemble.RandomForest#1': {
'X': 'X1'
}
},
'output_names': {
'sklearn.ensemble.RandomForest#1': {
'y': 'y1'
}
}
}
pipeline = MLPipeline(pipeline_dict)
assert pipeline.primitives == ['sklearn.ensemble.RandomForestClassifier']
assert pipeline.init_params == {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}
}
assert pipeline.input_names == {
'sklearn.ensemble.RandomForest#1': {
'X': 'X1'
}
}
assert pipeline.output_names == {
'sklearn.ensemble.RandomForest#1': {
'y': 'y1'
}
def test_list(self):
primitives = [
'sklearn.ensemble.RandomForestClassifier'
]
init_params = {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}
}
pipeline = MLPipeline(primitives, init_params=init_params)
assert pipeline.primitives == ['sklearn.ensemble.RandomForestClassifier']
assert pipeline.init_params == {
'sklearn.ensemble.RandomForest#1': {
'n_estimators': 500
}