Replacing Monkey-Patching

fedot/api/api_utils/api_params_repository.py

@@ -75,7 +75,9 @@ def get_params_for_gp_algorithm_params(self, params: dict) -> dict:
         if params.get('genetic_scheme') == 'steady_state':
             gp_algorithm_params['genetic_scheme_type'] = GeneticSchemeTypesEnum.steady_state
 
-        gp_algorithm_params['mutation_types'] = ApiParamsRepository._get_default_mutations(self.task_type, params)
+        # gp_algorithm_params['mutation_types'] = ApiParamsRepository._get_default_mutations(self.task_type, params)
+        gp_algorithm_params['mutation_types'] = context.api_params_repository__get_default_mutations(self.task_type,
+                                                                                                     params)
         gp_algorithm_params['seed'] = params['seed']
         return gp_algorithm_params
 

fedot/api/main.py

@@ -52,6 +52,8 @@
 from fedot.utilities.memory import MemoryAnalytics
 from fedot.utilities.project_import_export import export_project_to_zip, import_project_from_zip
 
+from fedot.core.context.context import resolve_context
+
 NOT_FITTED_ERR_MSG = 'Model not fitted yet'
 
 
@@ -102,20 +104,23 @@ def __init__(self,
                  logging_level: int = logging.ERROR,
                  safe_mode: bool = False,
                  n_jobs: int = -1,
+                 context: Optional[str] = None,
                  **composer_tuner_params
                  ):
 
         set_random_seed(seed)
         self.log = self._init_logger(logging_level)
 
+        self.context = resolve_context(context)
+
         # Attributes for dealing with metrics, data sources and hyperparameters
         self.params = ApiParams(composer_tuner_params, problem, task_params, n_jobs, timeout, seed)
 
         default_metrics = MetricByTask.get_default_quality_metrics(self.params.task.task_type)
         passed_metrics = self.params.get('metric')
         self.metrics = ensure_wrapped_in_sequence(passed_metrics) if passed_metrics else default_metrics
 
-        self.api_composer = ApiComposer(self.params, self.metrics)
+        self.api_composer = ApiComposer(self.params, self.metrics, self.context)
 
         # Initialize data processors for data preprocessing and preliminary data analysis
         self.data_processor = ApiDataProcessor(task=self.params.task,

fedot/core/composer/composer_builder.py

@@ -13,6 +13,7 @@
 from fedot.core.caching.operations_cache import OperationsCache
 from fedot.core.caching.preprocessing_cache import PreprocessingCache
 from fedot.core.caching.predictions_cache import PredictionsCache
+from fedot.core.context import ExecutionContext
 from fedot.core.composer.composer import Composer
 from fedot.core.composer.gp_composer.gp_composer import GPComposer
 from fedot.core.optimisers.objective.metrics_objective import MetricsObjective
@@ -58,6 +59,12 @@ def __init__(self, task: Task):
         self.preprocessing_cache: Optional[PreprocessingCache] = None
         self.predictions_cache: Optional[PredictionsCache] = None
 
+        self.context: Optional[ExecutionContext] = None
+
+    def with_context(self, context: ExecutionContext):
+        self.context = context or ExecutionContext()
+        return self
+
     def with_composer(self, composer_cls: Optional[Type[Composer]]):
         if composer_cls is not None:
             self.composer_cls = composer_cls
@@ -111,7 +118,8 @@ def with_cache(self,
     @staticmethod
     def _get_default_composer_params(task: Task) -> PipelineComposerRequirements:
         # Get all available operations for task
-        operations = get_operations_for_task(task=task, mode='all')
+        # operations = get_operations_for_task(task=task, mode='all')
+        operations = self.context.operation_registry.get_operation_for_task(task=task, mode='all')
         return PipelineComposerRequirements(primary=operations, secondary=operations)
 
     def _get_default_graph_generation_params(self) -> GraphGenerationParams:
@@ -161,6 +169,7 @@ def build(self) -> Composer:
                                      self.composer_requirements,
                                      self.operations_cache,
                                      self.preprocessing_cache,
-                                     self.predictions_cache)
+                                     self.predictions_cache,
+                                     self.context)
 
         return composer

fedot/core/composer/gp_composer/gp_composer.py

@@ -11,6 +11,7 @@
 from fedot.core.caching.operations_cache import OperationsCache
 from fedot.core.caching.predictions_cache import PredictionsCache
 from fedot.core.caching.preprocessing_cache import PreprocessingCache
+from fedot.core.context import ExecutionContext
 from fedot.core.composer.composer import Composer
 from fedot.core.data.data import InputData
 from fedot.core.data.multi_modal import MultiModalData
@@ -24,6 +25,7 @@
 )
 from fedot.core.utils import default_fedot_data_dir
 
+from functools import partial
 
 class GPComposer(Composer):
     """
@@ -40,7 +42,8 @@ def __init__(self, optimizer: GraphOptimizer,
                  composer_requirements: PipelineComposerRequirements,
                  operations_cache: Optional[OperationsCache] = None,
                  preprocessing_cache: Optional[PreprocessingCache] = None,
-                 predictions_cache: Optional[PredictionsCache] = None):
+                 predictions_cache: Optional[PredictionsCache] = None,
+                 context: Optional[ExectuionContext] = None):
         super().__init__(optimizer, composer_requirements)
         self.composer_requirements = composer_requirements
         self.operations_cache: Optional[OperationsCache] = operations_cache
@@ -49,11 +52,14 @@ def __init__(self, optimizer: GraphOptimizer,
 
         self.best_models: Collection[Pipeline] = ()
 
+        self.context = context or ExecutionContext()
+
     def compose_pipeline(self, data: Union[InputData, MultiModalData]) -> Union[Pipeline, Sequence[Pipeline]]:
         # Define data source
         data_splitter = DataSourceSplitter(self.composer_requirements.cv_folds,
                                            shuffle=True)
-        data_producer = data_splitter.build(data)
+
+        data_producer = self.context.data_source_splitter_build(data_splitter, data)
 
         parallelization_mode = self.composer_requirements.parallelization_mode
         if parallelization_mode == 'populational':
@@ -72,7 +78,9 @@ def compose_pipeline(self, data: Union[InputData, MultiModalData]) -> Union[Pipe
                                                         predictions_cache=self.predictions_cache,
                                                         validation_blocks=data_splitter.validation_blocks,
                                                         eval_n_jobs=n_jobs_for_evaluation)
-        objective_function = objective_evaluator.evaluate
+
+        # objective_function = objective_evaluator.evaluate
+        objective_function = partial(self.context.evaluator_evaluate, objective_evaluator)
 
         # Define callback for computing intermediate metrics if needed
         if self.composer_requirements.collect_intermediate_metric:

fedot/core/context/context.py

@@ -0,0 +1,169 @@
+from fedot.core.data.merge.data_merger import ImageDataMerger, TSDataMerger, DataMerger
+from fedot.core.operations.evaluation.operation_implementations.data_operations.topological.fast_topological_extractor import (
+    TopologicalFeaturesImplementation, )
+from fedot.core.operations.evaluation.operation_implementations.data_operations.ts_transformations import (
+    LaggedImplementation,
+    TsSmoothingImplementation,
+)
+from fedot.core.operations.operation import Operation
+from fedot.core.optimisers.objective import PipelineObjectiveEvaluate
+from fedot.core.optimisers.objective.data_source_splitter import DataSourceSplitter
+from fedot.core.pipelines.pipeline import Pipeline
+from fedot.core.pipelines.tuning.search_space import PipelineSearchSpace
+from fedot.core.pipelines.verification import class_rules, ts_rules, common_rules
+from fedot.core.repository.operation_types_repository import OperationTypesRepository
+from fedot.core.data.data_split import _split_any, _split_time_series
+from fedot.api.api_utils.api_params_repository import ApiParamsRepository
+from fedot.api.api_utils.api_composer import ApiComposer
+from golem.core.tuning.optuna_tuner import OptunaTuner
+from golem.core.optimisers.genetic.operators.reproduction import ReproductionController
+
+import fedot.core.data.data_split as fedot_data_split
+import golem.core.tuning.optuna_tuner as OptunaImpl
+
+def resolve_context(context_name: str = "core", backend: str = "default") -> ExecutionContext:
+    if context_name == "core":
+        return ExecutionContext(backend=backend)
+
+    from fedot.extensions.registry import get_registered_extension, get_registered_extensions
+
+    ext = get_registered_extension(context_name)
+    if ext is not None:
+        factory = ext.value.manifest.protocols.get("context_factory")
+        if factory:
+            return factory(backend=backend)
+
+    raise ValueError(f"Unknown context: {context_name}")
+
+class ExecutionContext:
+    def __init__(self, backend: str = "default") -> None:
+        """Initializes ExecutionContext with default configuration."""
+        self.backend = backend
+        self._init_defaults()
+        self._apply_protocols()
+
+    def _init_defaults(self):
+        """Sets default implementations for all pipeline components."""
+        self.evaluator_evaluate = PipelineObjectiveEvaluate.evaluate
+        self.search_space_get_parameters_dict = PipelineSearchSpace.get_parameters_dict
+        self.api_params_repository__get_default_mutations = ApiParamsRepository._get_default_mutations
+        self.merger_find_main_output = DataMerger.find_main_output
+        self.merger_get = DataMerger.get
+        self.merger_merge_predicts = DataMerger.merge_predicts
+        self.image_merger_preprocess_predicts = ImageDataMerger.preprocess_predicts
+        self.image_merger_merge_predicts = ImageDataMerger.merge_predicts
+        self.ts_merger_merge_predicts = TSDataMerger.merge_predicts
+        self.ts_merger_merge_targets = TSDataMerger.merge_targets
+        self.ts_merger_postprocess_predicts = TSDataMerger.postprocess_predicts
+        self.ts_merger_preprocess_predicts = TSDataMerger.preprocess_predicts
+        self.data_source_splitter_build = DataSourceSplitter.build
+        self.data_split__split_any = fedot_data_split._split_any
+        self.data_split__split_time_series = fedot_data_split._split_time_series
+        self.operation__predict = Operation._predict
+        self.operation_predict = Operation.predict
+        self.operation_predict_for_fit = Operation.predict_for_fit
+        self.lagged__update_column_types = LaggedImplementation._update_column_types
+        self.lagged_transform = LaggedImplementation.transform
+        self.lagged_transform_for_fit = LaggedImplementation.transform_for_fit
+        self.lagged__check_and_correct_window_size = LaggedImplementation._check_and_correct_window_size
+        self.topo_features_fit = TopologicalFeaturesImplementation.fit
+        self.topo_features_transform = TopologicalFeaturesImplementation.transform
+        self.ts_smoothing_transform = TsSmoothingImplementation.transform
+        self.optuna_optuna_tuner = OptunaImpl.OptunaTuner
+        self.api_composer_tune_final_pipeline = ApiComposer.tune_final_pipeline
+        self.reproduction_reproduce = ReproductionController.reproduce
+        self.reproduction_reproduce_uncontrolled = ReproductionController.reproduce_uncontrolled
+        self.class_rules = class_rules.copy()
+        self.ts_rules = ts_rules.copy()
+        self.common_rules = common_rules.copy()
+
+    def _apply_protocols(self):
+        # Splitters
+        splitters = resolve_protocol_instance("splitters", backend=self.backend)
+        if splitters:
+            self.data_split__split_any = splitters.split_any
+            self.data_split__split_time_series = splitters.split_time_series
+
+        # Mergers
+        mergers = resolve_protocol_instance("mergers", backend=self.backend)
+        if mergers:
+            self.merger_find_main_output = mergers.find_main_output
+            self.merger_get = mergers.get
+            self.merger_merge_predicts = mergers.merge_predicts
+            if hasattr(mergers, 'preprocess_predicts'):
+                self.image_merger_preprocess_predicts = mergers.preprocess_predicts
+                self.ts_merger_preprocess_predicts = mergers.preprocess_predicts
+            if hasattr(mergers, 'postprocess_predicts'):
+                self.ts_merger_postprocess_predicts = mergers.postprocess_predicts
+            if hasattr(mergers, 'merge_targets'):
+                self.ts_merger_merge_targets = mergers.merge_targets
+            # Image merge обычно совпадает с основным
+            self.image_merger_merge_predicts = mergers.merge_predicts
+
+        # DataSourceSplitter
+        splitter_builder = resolve_protocol_instance("data_source_splitter", backend=self.backend)
+        if splitter_builder:
+            self.data_source_splitter_build = splitter_builder.build
+
+        # Tuner class
+        tuner_class = resolve_protocol_instance("tuner_class", backend=self.backend)
+        if tuner_class:
+            self.optuna_optuna_tuner = tuner_class
+
+        # Reproduction
+        reproduction = resolve_protocol_instance("reproduction", backend=self.backend)
+        if reproduction:
+            self.reproduction_reproduce = reproduction.reproduce
+            if hasattr(reproduction, 'reproduce_uncontrolled'):
+                self.reproduction_reproduce_uncontrolled = reproduction.reproduce_uncontrolled
+
+        # Evaluator
+        evaluator = resolve_protocol_instance("evaluator", backend=self.backend)
+        if evaluator:
+            self.evaluator_evaluate = evaluator.evaluate
+
+        # Search space
+        search_space = resolve_protocol_instance("search_space", backend=self.backend)
+        if search_space:
+            self.search_space_get_parameters_dict = search_space.get_parameters_dict
+
+        # Mutations
+        mutations = resolve_protocol_instance("default_mutations", backend=self.backend)
+        if mutations:
+            self.api_params_repository__get_default_mutations = mutations
+
+        # Operation predict
+        op_predict = resolve_protocol_instance("operation_predict", backend=self.backend)
+        if op_predict:
+            self.operation_predict = op_predict.predict
+            self.operation_predict_for_fit = op_predict.predict_for_fit
+            if hasattr(op_predict, '_predict'):
+                self.operation__predict = op_predict._predict
+
+        # Lagged transformer
+        lagged = resolve_protocol_instance("lagged_transformer", backend=self.backend)
+        if lagged:
+            self.lagged__update_column_types = lagged._update_column_types
+            self.lagged_transform = lagged.transform
+            self.lagged_transform_for_fit = lagged.transform_for_fit
+            self.lagged__check_and_correct_window_size = lagged._check_and_correct_window_size
+
+        # Topological features
+        topo = resolve_protocol_instance("topological_features", backend=self.backend)
+        if topo:
+            self.topo_features_fit = topo.fit
+            self.topo_features_transform = topo.transform
+
+        # TS Smoothing
+        smoothing = resolve_protocol_instance("ts_smoothing", backend=self.backend)
+        if smoothing:
+            self.ts_smoothing_transform = smoothing.transform
+
+        # ApiComposer tune
+        tune = resolve_protocol_instance("api_composer_tune", backend=self.backend)
+        if tune:
+            self.api_composer_tune_final_pipeline = tune
+
+        @cached_property
+        def set_operation_registry(self) -> OperationTypesRepository:
+            return OperationTypesRepository()

fedot/core/context/factories.py

@@ -0,0 +1,60 @@
+from fedot.core.context.industrial_backend import (IndustrialSplitter, IndustrialDataMerger, IndustrialImageMerger,
+                                                   IndustrialTSMerger, IndustrialTextMerger,
+                                                   IndustrialDataSourceSplitterBuilder, IndustrialTunerClass,
+                                                   IndustrialReproduction, IndustrialEvaluator, IndustrialSearchSpace,
+                                                   IndustrialDefaultMutations,IndustrialOperationPredict,
+                                                   IndustrialLaggedTransformer, IndustrialTopologicalFeatures,
+                                                   IndustrialTsSmoothing, IndustrialApiComposerTune)
+
+
+
+def industrial_context_factory(backend: str = "default"):
+    return IndustrialContext(backend=backend)
+
+def splitters_factory():
+    return IndustrialSplitter()
+
+def data_merger_factory():
+    return IndustrialDataMerger()
+
+def image_merger_factory():
+    return IndustrialImageMerger()
+
+def ts_merger_factory():
+    return IndustrialTSMerger()
+
+def text_merger_factory():
+    return IndustrialTextMerger()
+
+def data_source_splitter_factory():
+    return IndustrialDataSourceSplitterBuilder()
+
+def tuner_class_factory(backend: str = "default"):
+    return IndustrialTunerClass(backend)
+
+def reproduction_factory():
+    return IndustrialReproduction()
+
+def evaluator_factory():
+    return IndustrialEvaluator()
+
+def search_space_factory():
+    return IndustrialSearchSpace()
+
+def mutations_factory():
+    return IndustrialDefaultMutations()
+
+def operation_predict_factory():
+    return IndustrialOperationPredict()
+
+def lagged_transformer_factory():
+    return IndustrialLaggedTransformer()
+
+def topo_features_factory():
+    return IndustrialTopologicalFeatures()
+
+def ts_smoothing_factory():
+    return IndustrialTsSmoothing()
+
+def api_composer_tune_factory():
+    return IndustrialApiComposerTune()