Redesign Adapter subsystem

examples/advanced/fedot_based_solutions/external_optimizer.py

@@ -35,12 +35,13 @@ def __init__(self,
     def optimise(self, objective: ObjectiveFunction):
 
         timer = OptimisationTimer(timeout=self.requirements.timeout)
-        dispatcher = SimpleDispatcher(self.graph_generation_params.adapter, timer)
+        dispatcher = SimpleDispatcher(timer)
         evaluator = dispatcher.dispatch(objective)
 
         num_iter = 0
-        initial_graph = self.graph_generation_params.adapter.adapt(choice(self.initial_graphs))
-        best = Individual(initial_graph)
+
+        initial_individuals = [Individual(graph) for graph in self.initial_graphs]
+        best = choice(initial_individuals)
         evaluator([best])
 
         with timer as t:
@@ -52,7 +53,7 @@ def optimise(self, objective: ObjectiveFunction):
                     best = new
                 num_iter += 1
 
-        return [self.graph_generation_params.adapter.restore(best.graph)]
+        return self.graph_generation_params.adapter.restore(best)
 
 
 def run_with_random_search_composer():

examples/advanced/fedot_based_solutions/graph_model_optimization.py

@@ -5,17 +5,15 @@
 import numpy as np
 import pandas as pd
 
+from fedot.core.adapter import DirectAdapter, register_native
 from fedot.core.dag.verification_rules import has_no_cycle, has_no_self_cycled_nodes
 from fedot.core.log import default_log
-from fedot.core.optimisers.adapters import DirectAdapter
-from fedot.core.optimisers.gp_comp.gp_optimizer import (
-    EvoGraphOptimizer,
-    GeneticSchemeTypesEnum
-)
+from fedot.core.optimisers.gp_comp.gp_optimizer import EvoGraphOptimizer
 from fedot.core.optimisers.gp_comp.gp_params import GPGraphOptimizerParameters
-from fedot.core.optimisers.gp_comp.pipeline_composer_requirements import PipelineComposerRequirements
 from fedot.core.optimisers.gp_comp.operators.crossover import CrossoverTypesEnum
+from fedot.core.optimisers.gp_comp.operators.inheritance import GeneticSchemeTypesEnum
 from fedot.core.optimisers.gp_comp.operators.regularization import RegularizationTypesEnum
+from fedot.core.optimisers.gp_comp.pipeline_composer_requirements import PipelineComposerRequirements
 from fedot.core.optimisers.graph import OptGraph, OptNode
 from fedot.core.optimisers.objective import Objective, ObjectiveEvaluate
 from fedot.core.optimisers.optimizer import GraphGenerationParams
@@ -52,6 +50,7 @@ def _has_no_duplicates(graph):
     return True
 
 
+@register_native
 def custom_mutation(graph: OptGraph, **kwargs):
     num_mut = 10
     try:

fedot/api/api_utils/api_composer.py

@@ -15,7 +15,7 @@
 from fedot.core.composer.gp_composer.specific_operators import boosting_mutation, parameter_change_mutation
 from fedot.core.constants import DEFAULT_TUNING_ITERATIONS_NUMBER
 from fedot.core.data.data import InputData
-from fedot.core.log import LoggerAdapter
+from fedot.core.log import default_log
 from fedot.core.optimisers.adapters import PipelineAdapter
 from fedot.core.optimisers.gp_comp.evaluation import determine_n_jobs
 from fedot.core.optimisers.gp_comp.gp_params import GPGraphOptimizerParameters
@@ -39,11 +39,16 @@
 class ApiComposer:
 
     def __init__(self, problem: str):
+        self.log = default_log(self)
         self.metrics = ApiMetrics(problem)
         self.pipelines_cache: Optional[OperationsCache] = None
         self.preprocessing_cache: Optional[PreprocessingCache] = None
         self.preset_name = None
         self.timer = None
+        # status flag indicating that composer step was applied
+        self.was_optimised = False
+        # status flag indicating that tuner step was applied
+        self.was_tuned = False
 
     def obtain_metric(self, task: Task, metric: Union[str, Callable]) -> Sequence[MetricType]:
         """Chooses metric to use for quality assessment of pipeline during composition"""
@@ -169,7 +174,6 @@ def _init_graph_generation_params(task: Task, preset: str, available_operations:
     def compose_fedot_model(self, api_params: dict, composer_params: dict, tuning_params: dict) \
             -> Tuple[Pipeline, Sequence[Pipeline], OptHistory]:
         """ Function for composing FEDOT pipeline model """
-        log: LoggerAdapter = api_params['logger']
         task: Task = api_params['task']
         train_data = api_params['train_data']
         timeout = api_params['timeout']
@@ -189,7 +193,7 @@ def compose_fedot_model(self, api_params: dict, composer_params: dict, tuning_pa
                 assumption_handler.fit_assumption_and_check_correctness(initial_assumption[0],
                                                                         pipelines_cache=self.pipelines_cache,
                                                                         preprocessing_cache=self.preprocessing_cache)
-        log.message(
+        self.log.message(
             f'Initial pipeline was fitted in {round(self.timer.assumption_fit_spend_time.total_seconds())} sec.')
 
         n_jobs = determine_n_jobs(api_params['n_jobs'])
@@ -205,28 +209,26 @@ def compose_fedot_model(self, api_params: dict, composer_params: dict, tuning_pa
                                                preset=preset,
                                                available_operations=composer_params.get('available_operations'),
                                                requirements=composer_requirements)
-        log.message(f"AutoML configured."
-                    f" Parameters tuning: {with_tuning}"
-                    f" Time limit: {timeout} min"
-                    f" Set of candidate models: {available_operations}")
+        self.log.message(f"AutoML configured."
+                         f" Parameters tuning: {with_tuning}"
+                         f" Time limit: {timeout} min"
+                         f" Set of candidate models: {available_operations}")
 
         best_pipeline, best_pipeline_candidates, gp_composer = self.compose_pipeline(task, train_data,
                                                                                      fitted_assumption,
                                                                                      metric_functions,
                                                                                      composer_requirements,
                                                                                      composer_params,
-                                                                                     graph_generation_params,
-                                                                                     log)
+                                                                                     graph_generation_params)
         if with_tuning:
-            self.tune_final_pipeline(task, train_data,
-                                     metric_functions[0],
-                                     composer_requirements,
-                                     best_pipeline,
-                                     log)
+            best_pipeline = self.tune_final_pipeline(task, train_data,
+                                                     metric_functions[0],
+                                                     composer_requirements,
+                                                     best_pipeline)
         # enforce memory cleaning
         gc.collect()
 
-        log.message('Model generation finished')
+        self.log.message('Model generation finished')
         return best_pipeline, best_pipeline_candidates, gp_composer.history
 
     def compose_pipeline(self, task: Task,
@@ -236,7 +238,7 @@ def compose_pipeline(self, task: Task,
                          composer_requirements: PipelineComposerRequirements,
                          composer_params: dict,
                          graph_generation_params: GraphGenerationParams,
-                         log: LoggerAdapter) -> Tuple[Pipeline, List[Pipeline], GPComposer]:
+                         ) -> Tuple[Pipeline, List[Pipeline], GPComposer]:
 
         multi_objective = len(metric_functions) > 1
         optimizer_params = ApiComposer._init_optimizer_parameters(composer_params,
@@ -259,18 +261,20 @@ def compose_pipeline(self, task: Task,
         if self.timer.have_time_for_composing(composer_params['pop_size'], n_jobs):
             # Launch pipeline structure composition
             with self.timer.launch_composing():
-                log.message('Pipeline composition started.')
+                self.log.message('Pipeline composition started.')
+                self.was_optimised = False
                 best_pipelines = gp_composer.compose_pipeline(data=train_data)
                 best_pipeline_candidates = gp_composer.best_models
+                self.was_optimised = True
         else:
             # Use initial pipeline as final solution
-            log.message(f'Timeout is too small for composing and is skipped '
-                        f'because fit_time is {self.timer.assumption_fit_spend_time.total_seconds()} sec.')
+            self.log.message(f'Timeout is too small for composing and is skipped '
+                             f'because fit_time is {self.timer.assumption_fit_spend_time.total_seconds()} sec.')
             best_pipelines = fitted_assumption
             best_pipeline_candidates = [fitted_assumption]
 
         for pipeline in best_pipeline_candidates:
-            pipeline.log = log
+            pipeline.log = self.log
         best_pipeline = best_pipelines[0] if isinstance(best_pipelines, Sequence) else best_pipelines
         return best_pipeline, best_pipeline_candidates, gp_composer
 
@@ -279,7 +283,7 @@ def tune_final_pipeline(self, task: Task,
                             metric_function: Optional[MetricType],
                             composer_requirements: PipelineComposerRequirements,
                             pipeline_gp_composed: Pipeline,
-                            log: LoggerAdapter) -> Pipeline:
+                            ) -> Pipeline:
         """ Launch tuning procedure for obtained pipeline by composer """
         timeout_for_tuning = abs(self.timer.determine_resources_for_tuning()) / 60
         tuner = TunerBuilder(task) \
@@ -294,14 +298,16 @@ def tune_final_pipeline(self, task: Task,
         if self.timer.have_time_for_tuning():
             # Tune all nodes in the pipeline
             with self.timer.launch_tuning():
-                log.message(f'Hyperparameters tuning started with {round(timeout_for_tuning)} sec. timeout')
+                self.log.message(f'Hyperparameters tuning started with {round(timeout_for_tuning)} sec. timeout')
+                self.was_tuned = False
                 tuned_pipeline = tuner.tune(pipeline_gp_composed)
-                log.message('Hyperparameters tuning finished')
+                self.was_tuned = True
+                self.log.message('Hyperparameters tuning finished')
         else:
-            log.message(f'Time for pipeline composing was {str(self.timer.composing_spend_time)}.\n'
-                        f'The remaining {max(0, timeout_for_tuning)} seconds are not enough '
-                        f'to tune the hyperparameters.')
-            log.message('Composed pipeline returned without tuning.')
+            self.log.message(f'Time for pipeline composing was {str(self.timer.composing_spend_time)}.\n'
+                             f'The remaining {max(0, timeout_for_tuning)} seconds are not enough '
+                             f'to tune the hyperparameters.')
+            self.log.message('Composed pipeline returned without tuning.')
             tuned_pipeline = pipeline_gp_composed
 
         return tuned_pipeline
@@ -312,8 +318,7 @@ def _divide_parameters(common_dict: dict) -> List[dict]:
 
     :param common_dict: dictionary with parameters for all AutoML modules
     """
-    api_params_dict = dict(train_data=None, task=Task, logger=LoggerAdapter, timeout=5, n_jobs=1,
-                           show_progress=True)
+    api_params_dict = dict(train_data=None, task=Task, timeout=5, n_jobs=1, show_progress=True, logger=None)
 
     composer_params_dict = dict(max_depth=None, max_arity=None, pop_size=None, num_of_generations=None,
                                 keep_n_best=None, available_operations=None, metric=None,

fedot/core/adapter/__init__.py

@@ -0,0 +1,2 @@
+from .adapter import BaseOptimizationAdapter, DirectAdapter
+from .adapt_registry import AdaptRegistry, register_native

fedot/core/adapter/adapt_registry.py

@@ -0,0 +1,140 @@
+from copy import copy
+from functools import partial
+from typing import Callable
+
+from fedot.core.utilities.singleton_meta import SingletonMeta
+
+
+class AdaptRegistry(metaclass=SingletonMeta):
+    """Registry of callables that require adaptation of argument/return values.
+    AdaptRegistry together with :class:``BaseOptimizationAdapter`` enables
+    automatic transformation between internal and domain graph representations.
+
+    **Short description of the use-case.**
+
+    Operators & verification rules that operate on internal representation
+    of graphs must be marked as native with ``register_native`` decorator.
+
+    Usually this is the case when users of the framework provide custom
+    operators for internal optimization graphs. When custom operators
+    operate on domain graphs, nothing is required.
+
+    **Extended description.**
+
+    Optimiser operates with generic graph representation.
+    Because of this any domain function requires adaptation
+    of its graph arguments. Adapter can automatically adapt
+    arguments to generic form in such cases.
+
+    Important notions:
+    - 'Domain' functions operate with domain-specific graphs.
+    - 'Native' functions operate with generic graphs used by optimiser.
+    - 'External' functions are functions defined by users of optimiser.
+    (most notably, custom mutations and custom verifier rules).
+    - 'Internal' functions are those defined by graph optimiser.
+    (most notably, the default set of mutations and verifier rules).
+    All internal functions are native.
+
+    Adaptation registry usage and behavior:
+    - Domain functions are adapted by default.
+    - Native functions don't require adaptation of their arguments.
+    - External functions are considered 'domain' functions by default.
+    Hence, they're their arguments are adapted, unless users of optimiser
+    exclude them from the process of automatic adaptation. It can be done
+    by registering them as 'native'.
+
+    AdaptRegistry can be safely used with multiprocessing
+    insofar as all relevant functions are registered as native
+    in the main process before child processes are started.
+    """
+
+    _native_flag_attr_name_ = '_fedot_is_optimizer_native'
+
+    def __init__(self):
+        self._registered_native_callables = []
+
+    def register_native(self, fun: Callable) -> Callable:
+        """Registers callable object as an internal function
+        that can work with internal graph representation.
+        Hence, it doesn't require adaptation when called by the optimiser.
+
+        Implementation details.
+        Works by setting a special attribute on the object.
+        This attribute then is checked by ``is_native`` used by adapters.
+
+        Args:
+            fun: function or callable to be registered as native
+
+        Returns:
+            Callable: same function with special private attribute set
+        """
+        original_function = AdaptRegistry._get_underlying_func(fun)
+        setattr(original_function, AdaptRegistry._native_flag_attr_name_, True)
+        self._registered_native_callables.append(original_function)
+        return fun
+
+    def unregister_native(self, fun: Callable) -> Callable:
+        """Unregisters callable object. See ``register_native``.
+
+        Args:
+            fun: function or callable to be unregistered as native
+
+        Returns:
+            Callable: same function with special private attribute unset
+        """
+        original_function = AdaptRegistry._get_underlying_func(fun)
+        if hasattr(original_function, AdaptRegistry._native_flag_attr_name_):
+            delattr(original_function, AdaptRegistry._native_flag_attr_name_)
+            self._registered_native_callables.remove(original_function)
+        return fun
+
+    @staticmethod
+    def is_native(fun: Callable) -> bool:
+        """Tests callable object for a presence of specific attribute
+        that tells that this function must not be restored with Adapter.
+
+        Args:
+            fun: tested Callable (function, method, functools.partial, or any callable object)
+
+        Returns:
+            bool: True if the callable was registered as native, False otherwise.
+        """
+        original_function = AdaptRegistry._get_underlying_func(fun)
+        is_native = getattr(original_function, AdaptRegistry._native_flag_attr_name_, False)
+        return is_native
+
+    def clear_registered_callables(self):
+        # copy is to avoid removing elements from list while iterating
+        for f in copy(self._registered_native_callables):
+            self.unregister_native(f)
+
+    @staticmethod
+    def _get_underlying_func(obj: Callable) -> Callable:
+        """Recursively unpacks 'partial' and 'method' objects to get underlying function.
+
+        Args:
+            obj: callable to try unpacking
+
+        Returns:
+            Callable: unpacked function that underlies the callable, or the unchanged object itself
+        """
+        while True:
+            if isinstance(obj, partial):  # if it is a 'partial'
+                obj = obj.func
+            elif hasattr(obj, '__func__'):  # if it is a 'method'
+                obj = obj.__func__
+            else:
+                return obj  # return the unpacked underlying function or the original object
+
+
+def register_native(fun: Callable) -> Callable:
+    """Out-of-class version of the ``register_native``
+    function that's intended to be used as a decorator.
+
+    Args:
+        fun: function or callable to be registered as native
+
+    Returns:
+        Callable: same function with special private attribute set
+    """
+    return AdaptRegistry().register_native(fun)