extract cache and tuner setup rules from api composer

Author: v1docq

fedot/api/api_utils/api_composer.py

@@ -7,6 +7,7 @@
 from golem.core.optimisers.opt_history_objects.opt_history import OptHistory
 from golem.core.tuning.simultaneous import SimultaneousTuner
 
+from fedot.api.api_utils.api_composer_rules import build_cache_init_plan, build_tuner_plan
 from fedot.api.api_utils.api_run_planner import build_composer_execution_plan
 from fedot.api.api_utils.assumptions.assumptions_handler import AssumptionsHandler
 from fedot.api.api_utils.params import ApiParams
@@ -41,23 +42,23 @@ def __init__(self, api_params: ApiParams, metrics: Union[MetricIDType, Sequence[
         self.init_cache()
 
     def init_cache(self):
-        use_operations_cache = self.params.get('use_operations_cache')
-        use_preprocessing_cache = self.params.get('use_preprocessing_cache')
-        use_predictions_cache = self.params.get('use_predictions_cache')
-        use_input_preprocessing = self.params.get('use_input_preprocessing')
-        cache_dir = self.params.get('cache_dir')
-        use_stats = self.params.get('use_stats')
-        if use_operations_cache:
-            self.operations_cache = OperationsCache(cache_dir=cache_dir, use_stats=use_stats)
-            #  in case of previously generated singleton cache
+        cache_plan = build_cache_init_plan(
+            use_operations_cache=self.params.get('use_operations_cache'),
+            use_preprocessing_cache=self.params.get('use_preprocessing_cache'),
+            use_predictions_cache=self.params.get('use_predictions_cache'),
+            use_input_preprocessing=self.params.get('use_input_preprocessing'),
+            cache_dir=self.params.get('cache_dir'),
+            use_stats=self.params.get('use_stats'),
+        )
+
+        if cache_plan.use_operations_cache:
+            self.operations_cache = OperationsCache(cache_dir=cache_plan.cache_dir, use_stats=cache_plan.use_stats)
             self.operations_cache.reset()
-        if use_input_preprocessing and use_preprocessing_cache:
-            self.preprocessing_cache = PreprocessingCache(cache_dir=cache_dir, use_stats=use_stats)
-            #  in case of previously generated singleton cache
+        if cache_plan.use_preprocessing_cache:
+            self.preprocessing_cache = PreprocessingCache(cache_dir=cache_plan.cache_dir, use_stats=cache_plan.use_stats)
             self.preprocessing_cache.reset()
-        if use_predictions_cache:
-            self.predictions_cache = PredictionsCache(cache_dir=cache_dir, use_stats=use_stats)
-            #  in case of previously generated singleton cache
+        if cache_plan.use_predictions_cache:
+            self.predictions_cache = PredictionsCache(cache_dir=cache_plan.cache_dir, use_stats=cache_plan.use_stats)
             self.predictions_cache.reset()
 
     def obtain_model(self, train_data: InputData) -> Tuple[Pipeline, Sequence[Pipeline], OptHistory]:
@@ -106,7 +107,6 @@ def obtain_model(self, train_data: InputData) -> Tuple[Pipeline, Sequence[Pipeli
         if gp_composer.history:
             adapter = self.params.graph_generation_params.adapter
             gp_composer.history.tuning_result = adapter.adapt(best_pipeline)
-        # enforce memory cleaning
         gc.collect()
 
         self.log.message('Model generation finished')
@@ -166,15 +166,13 @@ def compose_pipeline(self, train_data: InputData, initial_assumption: Sequence[P
         )
 
         if execution_plan.should_compose:
-            # Launch pipeline structure composition
             with self.timer.launch_composing():
                 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
             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
@@ -192,18 +190,23 @@ def tune_final_pipeline(self, train_data: InputData,
         """ Launch tuning procedure for obtained pipeline by composer """
         timeout_for_tuning = execution_plan.tuning_timeout_minutes if execution_plan else abs(
             self.timer.determine_resources_for_tuning()) / 60
+        tuner_plan = build_tuner_plan(
+            metrics=self.metrics,
+            timeout_minutes=timeout_for_tuning,
+            iterations=DEFAULT_TUNING_ITERATIONS_NUMBER,
+        )
         tuner = (TunerBuilder(self.params.task)
                  .with_tuner(SimultaneousTuner)
-                 .with_metric(self.metrics[0])
-                 .with_iterations(DEFAULT_TUNING_ITERATIONS_NUMBER)
-                 .with_timeout(datetime.timedelta(minutes=timeout_for_tuning))
+                 .with_metric(tuner_plan.metric)
+                 .with_iterations(tuner_plan.iterations)
+                 .with_timeout(datetime.timedelta(minutes=tuner_plan.timeout_minutes))
                  .with_eval_time_constraint(self.params.composer_requirements.max_graph_fit_time)
                  .with_requirements(self.params.composer_requirements)
                  .build(train_data))
 
         with self.timer.launch_tuning():
             self.was_tuned = False
-            self.log.message(f'Hyperparameters tuning started with {round(timeout_for_tuning)} min. timeout')
+            self.log.message(f'Hyperparameters tuning started with {round(tuner_plan.timeout_minutes)} min. timeout')
             tuned_pipeline = tuner.tune(pipeline_gp_composed)
             self.log.message('Hyperparameters tuning finished')
         self.was_tuned = tuner.was_tuned

fedot/api/api_utils/api_composer_rules.py

@@ -0,0 +1,41 @@
+from dataclasses import dataclass
+from typing import Any, Sequence
+
+
+@dataclass(frozen=True)
+class CacheInitPlan:
+    use_operations_cache: bool
+    use_preprocessing_cache: bool
+    use_predictions_cache: bool
+    cache_dir: str | None
+    use_stats: bool
+
+
+@dataclass(frozen=True)
+class TunerPlan:
+    metric: Any
+    iterations: int
+    timeout_minutes: float
+
+
+def build_cache_init_plan(use_operations_cache: bool,
+                          use_preprocessing_cache: bool,
+                          use_predictions_cache: bool,
+                          use_input_preprocessing: bool,
+                          cache_dir,
+                          use_stats: bool) -> CacheInitPlan:
+    return CacheInitPlan(
+        use_operations_cache=bool(use_operations_cache),
+        use_preprocessing_cache=bool(use_input_preprocessing and use_preprocessing_cache),
+        use_predictions_cache=bool(use_predictions_cache),
+        cache_dir=cache_dir,
+        use_stats=bool(use_stats),
+    )
+
+
+def build_tuner_plan(metrics: Sequence[Any], timeout_minutes: float, iterations: int) -> TunerPlan:
+    return TunerPlan(
+        metric=metrics[0],
+        iterations=iterations,
+        timeout_minutes=max(0.0, timeout_minutes),
+    )

tests/api/api_utils/test_api_composer.py

@@ -0,0 +1,40 @@
+import fedot.api.api_utils.api_composer as composer_module
+from fedot.api.api_utils.api_composer import ApiComposer
+
+
+class _FakeCache:
+    def __init__(self, cache_dir=None, use_stats=False):
+        self.cache_dir = cache_dir
+        self.use_stats = use_stats
+        self.was_reset = False
+
+    def reset(self):
+        self.was_reset = True
+
+
+class _FakeParams(dict):
+    timeout = 1
+    n_jobs = -1
+
+
+def test_api_composer_init_cache_uses_typed_cache_plan(monkeypatch):
+    monkeypatch.setattr(composer_module, 'OperationsCache', _FakeCache)
+    monkeypatch.setattr(composer_module, 'PreprocessingCache', _FakeCache)
+    monkeypatch.setattr(composer_module, 'PredictionsCache', _FakeCache)
+
+    params = _FakeParams(
+        use_operations_cache=True,
+        use_preprocessing_cache=True,
+        use_predictions_cache=True,
+        use_input_preprocessing=False,
+        cache_dir='cache_dir',
+        use_stats=True,
+    )
+
+    composer = ApiComposer(params, metrics=['f1'])
+
+    assert isinstance(composer.operations_cache, _FakeCache)
+    assert composer.operations_cache.was_reset is True
+    assert composer.preprocessing_cache is None
+    assert isinstance(composer.predictions_cache, _FakeCache)
+    assert composer.predictions_cache.was_reset is True

tests/api/api_utils/test_api_composer_rules.py

@@ -0,0 +1,26 @@
+from fedot.api.api_utils.api_composer_rules import build_cache_init_plan, build_tuner_plan
+
+
+def test_build_cache_init_plan_respects_input_preprocessing_boundary():
+    plan = build_cache_init_plan(
+        use_operations_cache=True,
+        use_preprocessing_cache=True,
+        use_predictions_cache=True,
+        use_input_preprocessing=False,
+        cache_dir='cache',
+        use_stats=True,
+    )
+
+    assert plan.use_operations_cache is True
+    assert plan.use_preprocessing_cache is False
+    assert plan.use_predictions_cache is True
+    assert plan.cache_dir == 'cache'
+    assert plan.use_stats is True
+
+
+def test_build_tuner_plan_is_deterministic_and_clamps_timeout():
+    plan = build_tuner_plan(metrics=['f1', 'roc_auc'], timeout_minutes=-3, iterations=42)
+
+    assert plan.metric == 'f1'
+    assert plan.iterations == 42
+    assert plan.timeout_minutes == 0.0