FEDOT

1
import os
2
import shutil
3
from pathlib import Path
4
from typing import Optional, Tuple, Union
5

6
import golem.core.paths
7
from golem.core.log import LoggerAdapter, default_log
8
from golem.core.optimisers.opt_history_objects.opt_history import OptHistory
9

10
from fedot.core.data.data import InputData
11
from fedot.core.pipelines.pipeline import Pipeline
12

13
DEFAULT_PATH = Path(golem.core.paths.default_data_dir())
14
DEFAULT_PROJECTS_PATH = DEFAULT_PATH.joinpath('projects')
15

16

17
def export_project_to_zip(zip_name: Union[str, Path], pipeline: Pipeline, train_data: InputData, test_data: InputData,
18
                          opt_history: Optional[OptHistory] = None, log_file_name: str = 'log.txt'):
19
    """
20
    Convert pipeline to JSON, data to csv, compress them to zip
21
    archive and save to 'DEFAULT_PROJECTS_PATH/projects' with logs.
22

23
    :param zip_name: absolute or relative path the zip file with exported project
24
    :param pipeline: pipeline object to export
25
    :param train_data: train InputData object to export
26
    :param test_data: test InputData object to export
27
    :param opt_history: history of model optimisation to export (if available)
28
    :param log_file_name: name of the file with log to export
29
    """
30

31
    log = default_log(prefix='fedot.utilities.project_import_export')
32
    absolute_folder_path, absolute_zip_path, folder_name, zip_name = _prepare_paths(zip_name)
33
    _check_for_existing_project(absolute_folder_path, absolute_zip_path)
34

35
    # Converts python objects to files for compression
36
    pipeline_path = os.path.join(absolute_folder_path, 'pipeline', 'pipeline.json')
37
    pipeline.save(pipeline_path, is_datetime_in_path=False, create_subdir=False)
38
    train_data.to_csv(absolute_folder_path.joinpath('train_data.csv'))
39
    test_data.to_csv(absolute_folder_path.joinpath('test_data.csv'))
40
    if opt_history is not None:
41
        opt_history.save(Path(absolute_folder_path, 'opt_history.json'))
42

43
    _copy_log_file(log_file_name, absolute_folder_path)
44

45
    shutil.make_archive(base_name=absolute_zip_path.with_suffix(''), format='zip', root_dir=absolute_folder_path)
46
    folder_to_delete = DEFAULT_PROJECTS_PATH.joinpath(
47
        absolute_folder_path.relative_to(DEFAULT_PROJECTS_PATH).parts[0])
48
    shutil.rmtree(folder_to_delete)
49

50
    log.info(f'The exported project was saved on the path: {absolute_zip_path}')
51

52

53
def import_project_from_zip(zip_path: str) -> Tuple[Pipeline, InputData, InputData, OptHistory]:
54
    """
55
    Unzipping zip file. Zip file should contain:
56
    - pipeline.json: json performance,
57
    - train_data.csv: csv with first line which contains task_type and data_type of train InputData object,
58
    - test_data.csv: csv with first line which contains task_type and data_type of test InputData object.
59

60
    Created Pipeline and InputData objects. Ready to work with it.
61

62
    :param zip_path: path to zip archive
63
    :return imported classes
64
    """
65
    log = default_log(prefix='fedot.utilities.project_import_export')
66

67
    folder_path, absolute_zip_path, _, zip_name = _prepare_paths(zip_path)
68

69
    zip_path = _check_zip_path(absolute_zip_path, log)
70

71
    if folder_path.exists():
72
        # ensure temporary folder is clear
73
        shutil.rmtree(folder_path)
74
    shutil.unpack_archive(zip_path, folder_path)
75

76
    message = f'The project "{zip_name}" was unpacked to the "{folder_path}".'
77
    log.info(message)
78
    pipeline, train_data, test_data, opt_history = [None] * 4
79
    for root, dirs, files in os.walk(folder_path):
80
        for file in files:
81
            if file == 'pipeline.json':
82
                pipeline = Pipeline.from_serialized(os.path.join(root, file))
83
            elif file == 'train_data.csv':
84
                train_data = InputData.from_csv(os.path.join(root, file))
85
            elif file == 'test_data.csv':
86
                test_data = InputData.from_csv(os.path.join(root, file))
87
            elif file == 'opt_history.json':
88
                opt_history = OptHistory.load(os.path.join(root, file))
89

90
    shutil.rmtree(folder_path)
91
    return pipeline, train_data, test_data, opt_history
92

93

94
def _check_zip_path(zip_path: Path, log: LoggerAdapter) -> Path:
95
    """Check 'zip_path' for correctness."""
96

97
    zip_path = zip_path.with_suffix('.zip')
98

99
    if not zip_path.exists():
100
        message = f'File with the path "{zip_path}" could not be found.'
101
        log.error(message)
102
        raise FileNotFoundError(message)
103
    return zip_path
104

105

106
def _copy_log_file(log_file_name: Optional[str], absolute_folder_path: Path):
107
    """Copy log file to folder which will be compressed."""
108

109
    if log_file_name is None:
110
        return
111

112
    log_file_name = Path(log_file_name)
113

114
    if not log_file_name.is_absolute():
115
        log_file_name = DEFAULT_PATH.joinpath(log_file_name).resolve()
116

117
    if log_file_name.exists():
118
        shutil.copy2(log_file_name,
119
                     absolute_folder_path.joinpath(log_file_name.stem).with_suffix(log_file_name.suffix))
120

121

122
def _prepare_paths(zip_path: Union[str, Path]) -> Tuple[Path, Path, Path, Path]:
123
    """Prepared paths and names: absolute folder path, absolute zip path, folder name, zip name."""
124

125
    zip_path = Path(zip_path)
126
    if Path(zip_path).suffix:
127
        folder_name = Path(zip_path.with_suffix('').name)
128
    else:
129
        folder_name = Path(zip_path.name)
130
        zip_path = zip_path.with_suffix('.zip')
131

132
    absolute_folder_path = DEFAULT_PROJECTS_PATH.joinpath(folder_name)
133
    if not zip_path.is_absolute():
134
        absolute_zip_path = Path.cwd().joinpath(zip_path)
135
    else:
136
        absolute_zip_path = zip_path
137
        zip_path = Path(zip_path.name)
138
    return absolute_folder_path, absolute_zip_path, folder_name, zip_path
139

140

141
def _check_for_existing_project(absolute_folder_path: Path, zip_path: Path):
142
    """Check for existing folder and zipfile of project. Create it, if it is no exists."""
143
    if zip_path.exists():
144
        message = f'Zipfile with the name "{zip_path}" exists.'
145
        raise FileExistsError(message)
146

147
    if absolute_folder_path.exists():
148
        message = f'Project with the name "{absolute_folder_path}" exists.'
149
        raise FileExistsError(message)
150
    else:
151
        absolute_folder_path.mkdir(parents=True)
152