FEDOT

1
import os
2
from datetime import timedelta
3
from typing import Optional, Callable
4
from uuid import uuid4
5

6
from fedot.core.pipelines.pipeline import Pipeline
7
from fedot.core.utils import default_fedot_data_dir
8
from fedot.remote.infrastructure.clients.client import Client
9
from fedot.remote.run_pipeline import fit_pipeline
10

11

12
class TestClient(Client):
13
    def __init__(self, connect_params: dict, exec_params: dict, output_path: Optional[str] = None):
14
        self.connect_params = connect_params
15
        self.exec_params = exec_params
16
        self.output_path = output_path if output_path else \
17
            os.path.join(default_fedot_data_dir(), 'remote_fit_results')
18
        super().__init__(connect_params, exec_params, output_path)
19

20
    def create_task(self, config) -> str:
21
        fit_pipeline(config)
22
        return str(uuid4())
23

24
    def wait_until_ready(self) -> timedelta:
25
        return timedelta()
26

27
    def download_result(self, execution_id: int, result_cls=Pipeline) -> Pipeline:
28
        results_path_out = os.path.join(self.output_path)
29
        results_folder = os.listdir(results_path_out)[0]
30
        pipeline = result_cls.from_serialized(os.path.join(results_path_out, results_folder, 'fitted_pipeline.json'))
31
        return pipeline
32