extract operation parameter normalization and change tracking rules

Author: v1docq

fedot/core/operations/operation_parameter_rules.py

@@ -0,0 +1,25 @@
+from typing import Dict, Iterable, Tuple
+
+
+
+def merge_operation_default_params(default_parameters: Dict, passed_parameters: Dict) -> Dict:
+    return {
+        **dict(default_parameters or {}),
+        **dict(passed_parameters or {}),
+    }
+
+
+
+def collect_changed_keys(current_parameters: Dict, updated_parameters: Dict, existing_changed_keys: Iterable[str]) -> Tuple[str, ...]:
+    changed_keys = list(existing_changed_keys)
+    for key, value in updated_parameters.items():
+        if key not in changed_keys and current_parameters.get(key) != value:
+            changed_keys.append(key)
+    return tuple(changed_keys)
+
+
+
+def resolve_setdefault_value(current_parameters: Dict, key, value):
+    if key in current_parameters:
+        return current_parameters[key], False
+    return value, True

fedot/core/operations/operation_parameters.py

@@ -1,6 +1,11 @@
 from copy import deepcopy
 from typing import Iterable
 
+from fedot.core.operations.operation_parameter_rules import (
+    collect_changed_keys,
+    merge_operation_default_params,
+    resolve_setdefault_value,
+)
 from fedot.core.repository.default_params_repository import DefaultOperationParamsRepository
 
 
@@ -28,23 +33,21 @@ def __bool__(self):
     @staticmethod
     def from_operation_type(operation_type: str, **parameters):
         default_parameters = get_default_params(operation_type)
-        parameters = {**default_parameters, **parameters}
+        parameters = merge_operation_default_params(default_parameters, parameters)
         return OperationParameters(**parameters)
 
     def update(self, **params):
-        for key, value in params.items():
-            if key not in self._changed_keys:
-                if self._parameters.get(key) != value:
-                    self._changed_keys.append(key)
-            self._parameters.update({key: value})
+        self._changed_keys = list(collect_changed_keys(self._parameters, params, self._changed_keys))
+        self._parameters.update(params)
 
     def get(self, key, default_value=None):
         return self._parameters.get(key, default_value)
 
     def setdefault(self, key, value):
-        if key not in self._parameters.keys():
-            self.update(**{key: value})
-        return self.get(key)
+        resolved_value, should_update = resolve_setdefault_value(self._parameters, key, value)
+        if should_update:
+            self.update(**{key: resolved_value})
+        return resolved_value
 
     def to_dict(self) -> dict:
         return deepcopy(self._parameters)

tests/core/operations/test_operation_parameter_rules.py

@@ -0,0 +1,25 @@
+from fedot.core.operations.operation_parameter_rules import (
+    collect_changed_keys,
+    merge_operation_default_params,
+    resolve_setdefault_value,
+)
+
+
+
+def test_operation_parameter_rules_merge_defaults_and_track_changes():
+    merged = merge_operation_default_params({'a': 1, 'b': 2}, {'b': 3, 'c': 4})
+    changed_keys = collect_changed_keys({'a': 1, 'b': 2}, {'a': 1, 'b': 3, 'd': 4}, ())
+
+    assert merged == {'a': 1, 'b': 3, 'c': 4}
+    assert changed_keys == ('b', 'd')
+
+
+
+def test_operation_parameter_rules_resolve_setdefault_value_explicitly():
+    existing_value, should_update_existing = resolve_setdefault_value({'a': 1}, 'a', 2)
+    missing_value, should_update_missing = resolve_setdefault_value({'a': 1}, 'b', 3)
+
+    assert existing_value == 1
+    assert should_update_existing is False
+    assert missing_value == 3
+    assert should_update_missing is True

tests/core/operations/test_operation_parameters.py

@@ -0,0 +1,43 @@
+from fedot.core.operations.operation_parameters import OperationParameters, get_default_params
+
+
+
+def test_params_keeper_update():
+    params = {'a': 1, 'b': 2, 'c': 3}
+    keeper = OperationParameters(**params)
+    new_params = {'a': 1, 'b': 3, 'd': 4}
+    keeper.update(**new_params)
+    expected_params = {'a': 1, 'b': 3, 'c': 3, 'd': 4}
+    actual_params = keeper.to_dict()
+    changed_params = keeper.changed_parameters.keys()
+    assert actual_params == expected_params
+    assert 'a' not in changed_params
+    assert 'b' in changed_params
+    assert 'd' in changed_params
+
+
+
+def test_params_keeper_get():
+    params = {'a': 1, 'b': 2, 'c': 3}
+    keeper = OperationParameters(**params)
+    a = keeper.get('a')
+    b = keeper.get('b', -1)
+    d = keeper.get('d', 5)
+    assert a == 1
+    assert b == 2
+    assert d == 5
+
+
+
+def test_params_keeper_setdefault_and_defaults_from_repository():
+    keeper = OperationParameters(alpha=1.0)
+    existing_value = keeper.setdefault('alpha', 2.0)
+    missing_value = keeper.setdefault('beta', 3.0)
+    default_params = get_default_params('ridge')
+    merged_keeper = OperationParameters.from_operation_type('ridge', alpha=0.75)
+
+    assert existing_value == 1.0
+    assert missing_value == 3.0
+    assert keeper.get('beta') == 3.0
+    assert merged_keeper.get('alpha') == 0.75
+    assert set(default_params).issubset(set(merged_keeper.to_dict()))