simplify code using PipelineNode

fedot/api/api_utils/predefined_model.py

@@ -6,7 +6,7 @@
 from fedot.core.log import Log
 from fedot.core.pipelines.node import PipelineNode
 from fedot.core.pipelines.pipeline import Pipeline
-from fedot.core.pipelines.verification import verify_pipeline, verifier_for_task
+from fedot.core.pipelines.verification import verify_pipeline
 
 
 class PredefinedModel:

fedot/core/composer/gp_composer/specific_operators.py

@@ -84,7 +84,7 @@ def boosting_mutation(pipeline: Pipeline, requirements, params, **kwargs) -> Pip
     node_boost = PipelineNode(new_model, nodes_from=[node_decompose])
 
     node_final = PipelineNode(choice(requirements.secondary),
-                               nodes_from=[existing_pipeline.root_node, node_boost])
+                              nodes_from=[existing_pipeline.root_node, node_boost])
     pipeline = Pipeline(node_final)
     return pipeline
 

fedot/core/composer/random_composer.py

@@ -39,9 +39,8 @@ def nodes_to_pipeline(nodes: List[Node]) -> Pipeline:
 class RandomGraphFactory:
     def __init__(self,
                  primary_candidates: Sequence[Any], secondary_candidates: Sequence[Any],
-                 primary_node_func: Callable = PipelineNode, secondary_node_func: Callable = PipelineNode):
-        self.__primary_node_func = primary_node_func
-        self.__secondary_node_func = secondary_node_func
+                 node_func: Callable = PipelineNode):
+        self.__node_func = node_func
         self.__primary_candidates = list(primary_candidates)
         self.__secondary_candidates = list(secondary_candidates)
 
@@ -54,21 +53,21 @@ def random_nodeset(self) -> Pipeline:
         # random primary nodes
         num_of_primary = randint(1, len(self.__primary_candidates))
         for _ in range(num_of_primary):
-            new_set.append(self.random_primary())
+            new_set.append(self.random_node())
 
         # random final node
         if len(new_set) > 1:
             parent_nodes = copy(new_set)
-            final_node = self.random_secondary(parent_nodes)
+            final_node = self.random_node(parent_nodes)
             new_set.append(final_node)
 
         return nodes_to_pipeline(new_set)
 
-    def random_primary(self):
-        return self.__primary_node_func(random.choice(self.__primary_candidates))
-
-    def random_secondary(self, parent_nodes):
-        return self.__secondary_node_func(random.choice(self.__secondary_candidates), parent_nodes)
+    def random_node(self, parent_nodes=None):
+        if parent_nodes:
+            return self.__node_func(random.choice(self.__secondary_candidates), parent_nodes)
+        else:
+            return self.__node_func(random.choice(self.__primary_candidates))
 
 
 class RandomSearchOptimizer(GraphOptimizer):

fedot/core/pipelines/adapters.py

@@ -32,11 +32,7 @@ def _transform_to_opt_node(node: Node) -> OptNode:
     def _transform_to_pipeline_node(node: OptNode) -> Node:
         # deepcopy to avoid accidental information sharing between opt graphs & pipelines
         content = deepcopy(node.content)
-        if not node.nodes_from:
-            return PipelineNode(operation_type=content['name'], content=content)
-        else:
-            # `nodes_from` are assigned on the step of overall graph mapping
-            return PipelineNode(operation_type=content['name'], content=content)
+        return PipelineNode(operation_type=content['name'], content=content)
 
     def _adapt(self, adaptee: Pipeline) -> OptGraph:
         adapted_nodes = map_dag_nodes(self._transform_to_opt_node, adaptee.nodes)

fedot/core/pipelines/node.py

@@ -439,3 +439,7 @@ def _combine_parents(parent_nodes: List[Node],
             target = prediction.target
 
     return parent_results, target
+
+
+PrimaryNode = PipelineNode
+SecondaryNode = PipelineNode

fedot/core/pipelines/pipeline_node_factory.py

@@ -55,5 +55,6 @@ def _return_node(candidates) -> Optional[OptNode]:
             return None
         return OptNode(content={'name': choice(candidates)})
 
-    def filter_specific_candidates(self, candidates: list):
+    @staticmethod
+    def filter_specific_candidates(candidates: list):
         return list(filter(lambda x: not check_for_specific_operations(x), candidates))

fedot/core/pipelines/template.py

@@ -334,18 +334,10 @@ def roll_pipeline_structure(self, operation_object: Union['OperationTemplate', '
 
         if operation_object.operation_type == atomized_model_type():
             atomized_model = operation_object.next_pipeline_template
-            if operation_object.nodes_from:
-                node = PipelineNode(operation_type=atomized_model)
-            else:
-                node = PipelineNode(operation_type=atomized_model)
+            node = PipelineNode(operation_type=atomized_model)
         else:
-            if operation_object.nodes_from:
-                node = PipelineNode(operation_object.operation_type)
-            else:
-                node = PipelineNode(operation_object.operation_type)
-
+            node = PipelineNode(operation_object.operation_type)
             node.parameters = operation_object.custom_params
-
             node.rating = operation_object.rating
 
         if hasattr(operation_object,

fedot/core/pipelines/verification_rules.py

@@ -171,8 +171,8 @@ def only_non_lagged_operations_are_primary(pipeline: Pipeline):
 
     # Check only primary nodes
     for node in pipeline.nodes:
-        if isinstance(node,
-                      PipelineNode) and node.is_primary and DataTypesEnum.ts not in node.operation.metadata.input_types:
+        if isinstance(node, PipelineNode) and node.is_primary and \
+                DataTypesEnum.ts not in node.operation.metadata.input_types:
             raise ValueError(
                 f'{ERROR_PREFIX} Pipeline for forecasting has not non_lagged preprocessing in primary nodes')
 

fedot/preprocessing/structure.py

@@ -12,7 +12,7 @@
 class PipelineStructureExplorer:
     """ Class for performing pipeline structure exploration.
     The class allows you to convert pipelines into a networkx graph and considers
-    all possible paths from PrimaryNode (or PrimaryNodes) to root node. If at least
+    all possible paths from primary Node (or primary Nodes) to root node. If at least
     one of the paths contains an invalid sequence of operations, the search performed
     by this class will detect it
     """
@@ -24,7 +24,7 @@ class PipelineStructureExplorer:
     def check_structure_by_tag(pipeline: 'Pipeline', tag_to_check: str, source_name: str = DEFAULT_SOURCE_NAME):
         """
         In the pipeline structure, a node with an operation with the appropriate tag is searched for.
-        In this case the operations must have priority in the pipeline - in the PrimaryNode or not far from it.
+        In this case the operations must have priority in the pipeline - in the primary Node or not far from it.
 
         Correct pipeline:
         operation with tag -> linear
@@ -72,7 +72,7 @@ def check_path(graph: nx.DiGraph, path: list, tag_to_check: str) -> Dict[str, An
         Checking the path for operations take right places in the pipeline.
 
         :param graph: graph for checking paths
-        :param path: path in the graph from PrimaryNode to root
+        :param path: path in the graph from primary Node to root
         :param tag_to_check: find appropriate operation by desired tag
         """
         operation_path, is_appropriate_operation, is_independent_operation = \

test/integration/quality/test_synthetic_tasks.py

@@ -18,7 +18,7 @@
 def get_regression_pipeline():
     first = PipelineNode(operation_type='scaling')
     final = PipelineNode(operation_type='ridge',
-                          nodes_from=[first])
+                         nodes_from=[first])
 
     pipeline = Pipeline(final)
     return pipeline