Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.op_list = []
if self.template == None: # default pipeline structure
step_in_type = np.ndarray
step_ret_type = Output_Array
for operator in self.operators:
arg_types = operator.parameter_types()[0][1:]
p_types = ([step_in_type] + arg_types, step_ret_type)
if operator.root:
# We need to add rooted primitives twice so that they can
# return both an Output_Array (and thus be the root of the tree),
# and return a np.ndarray so they can exist elsewhere in the tree.
self._pset.addPrimitive(operator, *p_types)
tree_p_types = ([step_in_type] + arg_types, step_in_type)
self._pset.addPrimitive(operator, *tree_p_types)
self._import_hash_and_add_terminals(operator, arg_types)
self._pset.addPrimitive(CombineDFs(), [step_in_type, step_in_type], step_in_type)
else:
gp_types = {}
for idx, step in enumerate(self._template_comp):
# input class in each step
if idx:
step_in_type = ret_types[-1]
else:
step_in_type = np.ndarray
if step != 'CombineDFs':
if idx < len(self._template_comp) - 1:
# create an empty for returning class for strongly-type GP
step_ret_type_name = 'Ret_{}'.format(idx)
step_ret_type = type(step_ret_type_name, (object,), {})
ret_types.append(step_ret_type)
else:
# input class in each step
if idx:
step_in_type = ret_types[-1]
else:
step_in_type = np.ndarray
if step != 'CombineDFs':
if idx < len(self._template_comp) - 1:
# create an empty for returning class for strongly-type GP
step_ret_type_name = 'Ret_{}'.format(idx)
step_ret_type = type(step_ret_type_name, (object,), {})
ret_types.append(step_ret_type)
else:
step_ret_type = Output_Array
check_template = True
if step == 'CombineDFs':
self._pset.addPrimitive(CombineDFs(), [step_in_type, step_in_type], step_in_type)
elif main_type.count(step): # if the step is a main type
ops = [op for op in self.operators if op.type() == step]
for operator in ops:
arg_types = operator.parameter_types()[0][1:]
p_types = ([step_in_type] + arg_types, step_ret_type)
self._pset.addPrimitive(operator, *p_types)
self._import_hash_and_add_terminals(operator, arg_types)
else: # is the step is a specific operator or a wrong input
try:
operator = next(op for op in self.operators if op.__name__ == step)
except:
raise ValueError(
'An error occured while attempting to read the specified '
'template. Please check a step named {}'.format(step)
)
arg_types = operator.parameter_types()[0][1:]