extract pipeline node parameter normalization rules

Author: v1docq

fedot/core/pipelines/node.py

@@ -15,6 +15,11 @@
 from fedot.core.operations.factory import OperationFactory
 from fedot.core.operations.operation import Operation
 from fedot.core.operations.operation_parameters import OperationParameters
+from fedot.core.pipelines.pipeline_node_rules import (
+    merge_node_parameters,
+    normalize_node_parameters,
+    should_update_node_parameters,
+)
 from fedot.core.repository.operation_types_repository import OperationTypesRepository
 from fedot.core.utils import DEFAULT_PARAMS_STUB, NESTED_PARAMS_LABEL
 
@@ -124,7 +129,7 @@ def update_params(self):
         """Updates :attr:`custom_params` with changed parameters"""
         new_params = self.fitted_operation.get_params()
         changed_parameters = new_params.changed_parameters
-        updated_parameters = {**self.parameters, **changed_parameters}
+        updated_parameters = merge_node_parameters(self.parameters, changed_parameters)
         self.parameters = updated_parameters
 
     @property
@@ -218,9 +223,7 @@ def fit(self,
                                                                descriptive_id=self.descriptive_id)
 
         # Update parameters after operation fitting (they can be corrected)
-        not_atomized_operation = 'atomized' not in self.operation.operation_type
-
-        if not_atomized_operation and 'correct_params' in self.operation.metadata.tags:
+        if should_update_node_parameters(self.operation.operation_type, self.operation.metadata.tags):
             self.update_params()
         return operation_predict
 
@@ -357,15 +360,9 @@ def parameters(self, params: dict):
         Args:
             params: new parameters to be placed instead of existing
         """
-        if params is not None:
-            # The check for "default_params" is needed for backward compatibility.
-            if params == DEFAULT_PARAMS_STUB:
-                params = {}
-            # take nested params if they appeared (mostly used for tuning)
-            if NESTED_PARAMS_LABEL in params:
-                params = params[NESTED_PARAMS_LABEL]
-            self._parameters = OperationParameters.from_operation_type(self.operation.operation_type, **params)
-            self.content['params'] = self._parameters.to_dict()
+        normalized_params = normalize_node_parameters(params, DEFAULT_PARAMS_STUB, NESTED_PARAMS_LABEL)
+        self._parameters = OperationParameters.from_operation_type(self.operation.operation_type, **normalized_params)
+        self.content['params'] = self._parameters.to_dict()
 
     def __str__(self) -> str:
         """Returns ``str`` representation of the node

fedot/core/pipelines/pipeline_node_rules.py

@@ -0,0 +1,27 @@
+from typing import Dict, Iterable, Optional
+
+
+
+def normalize_node_parameters(params: Optional[dict], default_params_stub, nested_params_label: str) -> Dict:
+    if params is None:
+        return {}
+    if params == default_params_stub:
+        return {}
+    if nested_params_label in params:
+        return dict(params[nested_params_label])
+    return dict(params)
+
+
+
+def merge_node_parameters(current_parameters: Optional[dict], changed_parameters: Optional[dict]) -> Dict:
+    return {
+        **dict(current_parameters or {}),
+        **dict(changed_parameters or {}),
+    }
+
+
+
+def should_update_node_parameters(operation_type: str, operation_tags: Optional[Iterable[str]]) -> bool:
+    if 'atomized' in operation_type:
+        return False
+    return 'correct_params' in set(operation_tags or ())

tests/core/pipelines/test_node.py

@@ -0,0 +1,37 @@
+from fedot.core.operations.operation_parameters import OperationParameters
+from fedot.core.pipelines.node import PipelineNode
+from fedot.core.utils import DEFAULT_PARAMS_STUB, NESTED_PARAMS_LABEL
+
+
+class _FittedOperationWithParams:
+    def __init__(self, params):
+        self._params = params
+
+    def get_params(self):
+        return self._params
+
+
+
+def test_pipeline_node_parameters_setter_normalizes_default_and_nested_params():
+    default_node = PipelineNode(operation_type='ridge')
+    nested_node = PipelineNode(operation_type='ridge')
+
+    default_node.parameters = DEFAULT_PARAMS_STUB
+    nested_node.parameters = {NESTED_PARAMS_LABEL: {'alpha': 1.0}}
+
+    assert default_node.parameters == {}
+    assert nested_node.parameters['alpha'] == 1.0
+
+
+
+def test_pipeline_node_update_params_uses_typed_merge_rule():
+    node = PipelineNode(operation_type='ridge')
+    node.parameters = {'alpha': 1.0}
+    fitted_params = OperationParameters(alpha=1.0)
+    fitted_params.update(beta=2.0)
+    node.fitted_operation = _FittedOperationWithParams(fitted_params)
+
+    node.update_params()
+
+    assert node.parameters['alpha'] == 1.0
+    assert node.parameters['beta'] == 2.0

tests/core/pipelines/test_pipeline_node_rules.py

@@ -0,0 +1,27 @@
+from fedot.core.pipelines.pipeline_node_rules import (
+    merge_node_parameters,
+    normalize_node_parameters,
+    should_update_node_parameters,
+)
+from fedot.core.utils import DEFAULT_PARAMS_STUB, NESTED_PARAMS_LABEL
+
+
+
+def test_normalize_node_parameters_handles_default_stub_and_nested_params():
+    assert normalize_node_parameters(DEFAULT_PARAMS_STUB, DEFAULT_PARAMS_STUB, NESTED_PARAMS_LABEL) == {}
+    assert normalize_node_parameters(
+        {NESTED_PARAMS_LABEL: {'alpha': 1.0}},
+        DEFAULT_PARAMS_STUB,
+        NESTED_PARAMS_LABEL,
+    ) == {'alpha': 1.0}
+    assert normalize_node_parameters({'beta': 2.0}, DEFAULT_PARAMS_STUB, NESTED_PARAMS_LABEL) == {'beta': 2.0}
+
+
+
+def test_merge_node_parameters_and_update_rule_are_explicit():
+    merged = merge_node_parameters({'alpha': 1.0}, {'beta': 2.0})
+
+    assert merged == {'alpha': 1.0, 'beta': 2.0}
+    assert should_update_node_parameters('ridge', ['correct_params']) is True
+    assert should_update_node_parameters('atomized_operation', ['correct_params']) is False
+    assert should_update_node_parameters('ridge', ['linear']) is False