HyperactiveWrapper.__init__()   A
last analyzed

Complexity

Conditions 2

Size

Total Lines 23
Code Lines 18

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 18
dl 0
loc 23
rs 9.5
c 0
b 0
f 0
cc 2
nop 7
1
# Author: Simon Blanke
2
# Email: [email protected]
3
# License: MIT License
4
5
import os
6
import dill
7
import datetime
8
import numpy as np
9
import pandas as pd
10
import hashlib
11
import inspect
12
13
from .paths import HypermemoryPaths
14
from .io_helpers import (
15
    save_object,
16
    load_object,
17
    save_json,
18
    load_json,
19
    save_dataframe,
20
    load_dataframes,
21
)
22
23
from .verbosity import VerbosityLVL0, VerbosityLVL1
24
from .dataset_features import dataset_features
25
from .memory_conv import memory_dict2dataframe, convert_dataframe, dataframe2memory_dict
26
27
28
def get_datetime():
29
    return datetime.datetime.now().strftime("%d.%m.%Y - %H:%M:%S:%f")
30
31
32
class HyperactiveWrapper:
33
    def __init__(self, main_path, X, y, model, search_space, verbosity=1):
34
        self.paths = HypermemoryPaths(main_path)
35
        self.paths.add_directory(name="X", prefix="X_ID:", id_type="array", object_=X)
36
        self.paths.add_directory(name="y", prefix="y_ID:", id_type="array", object_=y)
37
        self.paths.add_directory(
38
            name="model", prefix="model_ID:", id_type="function", object_=model
39
        )
40
        self.paths.add_directory(
41
            name="search_space",
42
            prefix="search_space_ID:",
43
            id_type="dictionary",
44
            object_=search_space,
45
        )
46
47
        self.X = X
48
        self.y = y
49
        self.model = model
50
        self.search_space = search_space
51
52
        if verbosity == 0:
53
            self.verb = VerbosityLVL0()
54
        else:
55
            self.verb = VerbosityLVL1()
56
57
    def _drop_duplicates(self, dataframe):
58
        columns_drop = list(self.search_space.keys())
59
        return dataframe.drop_duplicates(subset=columns_drop, keep="last")
60
61
    def _load_dataframes(self):
62
        subdirs = self.paths.subdirs("model")
63
64
        dataframes_all = []
65
        for subdir in subdirs:
66
            search_space = load_object(path=subdir, name="search_space")
67
            dataframes = load_dataframes(subdir)
68
69
            for dataframe in dataframes:
70
                dataframe = convert_dataframe(
71
                    dataframe, search_space, self.search_space
72
                )
73
                dataframes_all.append(dataframe)
74
75
        return dataframes_all
76
77
    def load(self):
78
        self.verb.load_search_data(self.model)
79
80
        dataframes_all = self._load_dataframes()
81
        if len(dataframes_all) == 0:
82
            return {}
83
84
        dataframe = pd.concat(dataframes_all, axis=0)
85
        dataframe = self._drop_duplicates(dataframe)
86
87
        memory_dict = dataframe2memory_dict(dataframe, self.search_space)
88
        self.verb.load_search_data_success(self.model, dataframe)
89
90
        return memory_dict
91
92
    def save(self, memory_dict):
93
        dataframe = memory_dict2dataframe(memory_dict, self.search_space)
94
95
        if len(dataframe) == 0:
96
            self.verb.save_search_data_canceled(self.model)
97
98
        else:
99
            self.verb.save_search_data(self.model)
100
101
            X_info = dataset_features(self.X)
102
            y_info = dataset_features(self.y)
103
104
            io_X_path = self.paths.path_dict["X"]
105
            io_y_path = self.paths.path_dict["y"]
106
            io_model_path = self.paths.path_dict["model"]
107
            io_search_space_path = self.paths.path_dict["search_space"]
108
109
            save_json(io_X_path, "X_meta_data", X_info)
110
            save_json(io_y_path, "y_meta_data", y_info)
111
            save_object(io_model_path, "model", self.model)
112
            save_object(io_search_space_path, "search_space", self.search_space)
113
            save_dataframe(
114
                io_search_space_path, "search_data_" + str(get_datetime()), dataframe
115
            )
116
117
            self.verb.save_search_data_success(self.model, dataframe)
118
119