hyperactive.extensions.memory.memory.apply_tobytes() - Code Metrics - Inspection of "improve performance of _load_data_into_memory" - SimonBlanke/Hyperactive - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( a0fddf...e61612 )

by Simon

created 2020-01-16 09:07 UTC

apply_tobytes() A

↳ Parent: hyperactive.extensions.memory.memory

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	2
rs	10
c	0
b	0
f	0

# Author: Simon Blanke
# Email: [email protected]
# License: MIT License

import os
import glob
import json
import dill
import time
import datetime
import hashlib
import inspect

import numpy as np
import pandas as pd

def apply_tobytes(df):
    return df.values.tobytes()


class Memory:
    def __init__(self, _space_, _main_args_, _cand_):
        self._space_ = _space_
        self._main_args_ = _main_args_

        self.pos_best = None
        self.score_best = -np.inf

        self.memory_type = _main_args_.memory
        self.memory_dict = {}

        self.meta_data_found = False


class ShortTermMemory(Memory):
    def __init__(self, _space_, _main_args_, _cand_):
        super().__init__(_space_, _main_args_, _cand_)


class LongTermMemory(Memory):
    def __init__(self, _space_, _main_args_, _cand_):
        super().__init__(_space_, _main_args_, _cand_)

        self.nth_process = _cand_.nth_process

        self.score_col_name = "mean_test_score"

        self.feature_hash = self._get_hash(_main_args_.X)
        self.label_hash = self._get_hash(_main_args_.y)

        current_path = os.path.realpath(__file__)
        meta_learn_path, _ = current_path.rsplit("/", 1)

        self.datetime = "run_data/" + datetime.datetime.now().strftime("%d.%m.%Y - %H:%M:%S")
        func_str = self._get_func_str(_cand_.func_)
        self.func_path_ = self._get_hash(func_str.encode("utf-8")) + "/"

        self.meta_path = meta_learn_path + "/meta_data/"
        self.func_path = self.meta_path + self.func_path_
        self.date_path = self.meta_path + self.func_path_ + self.datetime + "/"

        if not os.path.exists(self.date_path):
            os.makedirs(self.date_path, exist_ok=True)

    def load_memory(self, _cand_, _verb_):
        c_time = time.time()
        para, score = self._read_func_metadata(_cand_.func_, _verb_)
        if para is None or score is None:
            return
        print("Time _read_func_metadata:", round(time.time()- c_time, 2))

        _verb_.load_samples(para)

        c_time = time.time()
        _cand_.eval_time = list(para["eval_time"])

        self._load_data_into_memory(para, score)
        print("Time _load_data_into_memory:", round(time.time()- c_time, 2))

    def save_memory(self, _main_args_, _opt_args_, _cand_):
        path = self._get_file_path(_cand_.func_)
        meta_data = self._collect(_cand_)

        meta_data["run"] = self.datetime

        self._save_toCSV(meta_data, path)

        obj_func_path = self.func_path + "objective_function.py"
        if not os.path.exists(obj_func_path):
            file = open(obj_func_path, "w")
            file.write(self._get_func_str(_cand_.func_))
            file.close()

        search_config_path = self.date_path + "search_config.py"
        search_config_temp = dict(self._main_args_.search_config)

        for key in search_config_temp.keys():
            if isinstance(key, str):
                continue
            search_config_temp[key.__name__] = search_config_temp[key]
            del search_config_temp[key]

        search_config_str = "search_config = " + str(search_config_temp)

        if not os.path.exists(search_config_path):
            file = open(search_config_path, "w")
            file.write(search_config_str)
            file.close()

        """
        os.chdir(self.date_path)
        os.system("black search_config.py")
        os.getcwd()
        """

        run_data = {
            "random_state": self._main_args_.random_state,
            "max_time": self._main_args_.random_state,
            "n_iter": self._main_args_.n_iter,
            "optimizer": self._main_args_.optimizer,
            "n_jobs": self._main_args_.n_jobs,
            "eval_time": np.array(_cand_.eval_time).sum(),
            "total_time": _cand_.total_time,
        }

        with open(self.date_path + "run_data.json", "w") as f:
            json.dump(run_data, f, indent=4)

        """
        print("_opt_args_.kwargs_opt", _opt_args_.kwargs_opt)

        opt_para = pd.DataFrame.from_dict(_opt_args_.kwargs_opt, dtype=object)
        print("opt_para", opt_para)
        opt_para.to_csv(self.date_path + "opt_para", index=False)
        """

    def _save_toCSV(self, meta_data_new, path):
        if os.path.exists(path):
            meta_data_old = pd.read_csv(path)
            meta_data = meta_data_old.append(meta_data_new)

            columns = list(meta_data.columns)
            noScore = ["mean_test_score", "cv_default_score", "eval_time", "run"]
            columns_noScore = [c for c in columns if c not in noScore]

            meta_data = meta_data.drop_duplicates(subset=columns_noScore)
        else:
            meta_data = meta_data_new

        meta_data.to_csv(path, index=False)

    def _read_func_metadata(self, model_func, _verb_):
        paths = self._get_func_data_names()

        meta_data_list = []
        for path in paths:
            meta_data = pd.read_csv(path)
            meta_data_list.append(meta_data)
            self.meta_data_found = True

        if len(meta_data_list) > 0:
            meta_data = pd.concat(meta_data_list, ignore_index=True)

            column_names = meta_data.columns
            score_name = [name for name in column_names if self.score_col_name in name]

            para = meta_data.drop(score_name, axis=1)
            score = meta_data[score_name]

            _verb_.load_meta_data()
            return para, score

        else:
            _verb_.no_meta_data(model_func)
            return None, None

    def _get_opt_meta_data(self):
        results_dict = {}
        para_list = []
        score_list = []

        for key in self.memory_dict.keys():
            pos = np.fromstring(key, dtype=int)
            para = self._space_.pos2para(pos)
            score = self.memory_dict[key]

            for key in para.keys():
                if (
                    not isinstance(para[key], int)
                    and not isinstance(para[key], float)
                    and not isinstance(para[key], str)
                ):

                    para_dill = dill.dumps(para[key])
                    para_hash = self._get_hash(para_dill)

                    with open(
                        self.func_path + str(para_hash) + ".pkl", "wb"
                    ) as pickle_file:
                        dill.dump(para_dill, pickle_file)

                    para[key] = para_hash

            if score != 0:
                para_list.append(para)
                score_list.append(score)

        results_dict["params"] = para_list
        results_dict["mean_test_score"] = score_list

        return results_dict

    def _load_data_into_memory(self, paras, scores):

        paras = paras.replace(self._hash2obj())
        paras = self.para2pos(paras)

        df_temp = pd.DataFrame()
        df_temp["pos_str"] = paras.apply(apply_tobytes, axis=1)
        df_temp["score"] = scores

        self.memory_dict = df_temp.set_index('pos_str').to_dict()['score']

        scores = np.array(scores)
        paras = np.array(paras)

        idx = np.argmax(scores)
        self.score_best = scores[idx]
        self.pos_best = paras[idx]

    def para2pos(self, paras):
        paras = paras[self._space_.para_names]
        for pos_key in self._space_.search_space:
            paras[pos_key] = paras[pos_key].apply(
                self._space_.search_space[pos_key].index
            )

        return paras

    def _collect(self, _cand_):
        results_dict = self._get_opt_meta_data()

        para_pd = pd.DataFrame(results_dict["params"])
        metric_pd = pd.DataFrame(
            results_dict["mean_test_score"], columns=["mean_test_score"]
        )

        eval_time = pd.DataFrame(_cand_.eval_time, columns=["eval_time"])
        md_model = pd.concat(
            [para_pd, metric_pd, eval_time], axis=1, ignore_index=False
        )

        return md_model

    def _get_hash(self, object):
        return hashlib.sha1(object).hexdigest()

    def _get_func_str(self, func):
        return inspect.getsource(func)

    def _get_subdirs(self):
        subdirs = glob.glob(self.func_path + "*/")

        return subdirs

    def _get_func_data_names1(self):
        subdirs = self._get_subdirs()

        path_list = []
        for subdir in subdirs:
            paths = glob.glob(subdir + "*.csv")
            path_list = path_list + paths

        return path_list

    def _get_func_data_names(self):
        paths = glob.glob(
            self.func_path + (self.feature_hash + "_" + self.label_hash + "_.csv")
        )

        return paths

    def _read_dill(self, value):
        paths = self._get_pkl_hash(value)
        for path in paths:
            with open(path, "rb") as fp:
                value = dill.load(fp)
                value = dill.loads(value)
                break

        return value

    def _hash2obj(self):
        hash2obj_dict = {}
        para_hash_list = self._get_para_hash_list()

        for para_hash in para_hash_list:
            obj = self._read_dill(para_hash)
            hash2obj_dict[para_hash] = obj

        return hash2obj_dict

    def _get_para_hash_list(self):
        para_hash_list = []
        for key in self._space_.search_space.keys():
            values = self._space_.search_space[key]

            for value in values:
                if (
                    not isinstance(value, int)
                    and not isinstance(value, float)
                    and not isinstance(value, str)
                ):

                    para_dill = dill.dumps(value)
                    para_hash = self._get_hash(para_dill)
                    para_hash_list.append(para_hash)

        return para_hash_list

    def _get_pkl_hash(self, hash):
        paths = glob.glob(self.func_path + hash + "*.pkl")

        return paths

    def _get_file_path(self, model_func):
        if not os.path.exists(self.date_path):
            os.makedirs(self.date_path)

        return self.func_path + (self.feature_hash + "_" + self.label_hash + "_.csv")


1			# Author: Simon Blanke
2			# Email: [email protected]
3			# License: MIT License
4
5			import os
6			import glob
7			import json
8			import dill
9			import time
10			import datetime
11			import hashlib
12			import inspect
13
14			import numpy as np
15			import pandas as pd
16
17			def apply_tobytes(df):
18			return df.values.tobytes()
19
20
21			class Memory:
22			def __init__(self, _space_, _main_args_, _cand_):
23			self._space_ = _space_
24			self._main_args_ = _main_args_
25
26			self.pos_best = None
27			self.score_best = -np.inf
28
29			self.memory_type = _main_args_.memory
30			self.memory_dict = {}
31
32			self.meta_data_found = False
33
34
35			class ShortTermMemory(Memory):
36			def __init__(self, _space_, _main_args_, _cand_):
37			super().__init__(_space_, _main_args_, _cand_)
38
39
40			class LongTermMemory(Memory):
41			def __init__(self, _space_, _main_args_, _cand_):
42			super().__init__(_space_, _main_args_, _cand_)
43
44			self.nth_process = _cand_.nth_process
45
46			self.score_col_name = "mean_test_score"
47
48			self.feature_hash = self._get_hash(_main_args_.X)
49			self.label_hash = self._get_hash(_main_args_.y)
50
51			current_path = os.path.realpath(__file__)
52			meta_learn_path, _ = current_path.rsplit("/", 1)
53
54			self.datetime = "run_data/" + datetime.datetime.now().strftime("%d.%m.%Y - %H:%M:%S")
55			func_str = self._get_func_str(_cand_.func_)
56			self.func_path_ = self._get_hash(func_str.encode("utf-8")) + "/"
57
58			self.meta_path = meta_learn_path + "/meta_data/"
59			self.func_path = self.meta_path + self.func_path_
60			self.date_path = self.meta_path + self.func_path_ + self.datetime + "/"
61
62			if not os.path.exists(self.date_path):
63			os.makedirs(self.date_path, exist_ok=True)
64
65			def load_memory(self, _cand_, _verb_):
66			c_time = time.time()
67			para, score = self._read_func_metadata(_cand_.func_, _verb_)
68			if para is None or score is None:
69			return
70			print("Time _read_func_metadata:", round(time.time()- c_time, 2))
71
72			_verb_.load_samples(para)
73
74			c_time = time.time()
75			_cand_.eval_time = list(para["eval_time"])
76
77			self._load_data_into_memory(para, score)
78			print("Time _load_data_into_memory:", round(time.time()- c_time, 2))
79
80			def save_memory(self, _main_args_, _opt_args_, _cand_):
81			path = self._get_file_path(_cand_.func_)
82			meta_data = self._collect(_cand_)
83
84			meta_data["run"] = self.datetime
85
86			self._save_toCSV(meta_data, path)
87
88			obj_func_path = self.func_path + "objective_function.py"
89			if not os.path.exists(obj_func_path):
90			file = open(obj_func_path, "w")
91			file.write(self._get_func_str(_cand_.func_))
92			file.close()
93
94			search_config_path = self.date_path + "search_config.py"
95			search_config_temp = dict(self._main_args_.search_config)
96
97			for key in search_config_temp.keys():
98			if isinstance(key, str):
99			continue
100			search_config_temp[key.__name__] = search_config_temp[key]
101			del search_config_temp[key]
102
103			search_config_str = "search_config = " + str(search_config_temp)
104
105			if not os.path.exists(search_config_path):
106			file = open(search_config_path, "w")
107			file.write(search_config_str)
108			file.close()
109
110			"""
111			os.chdir(self.date_path)
112			os.system("black search_config.py")
113			os.getcwd()
114			"""
115
116			run_data = {
117			"random_state": self._main_args_.random_state,
118			"max_time": self._main_args_.random_state,
119			"n_iter": self._main_args_.n_iter,
120			"optimizer": self._main_args_.optimizer,
121			"n_jobs": self._main_args_.n_jobs,
122			"eval_time": np.array(_cand_.eval_time).sum(),
123			"total_time": _cand_.total_time,
124			}
125
126			with open(self.date_path + "run_data.json", "w") as f:
127			json.dump(run_data, f, indent=4)
128
129			"""
130			print("_opt_args_.kwargs_opt", _opt_args_.kwargs_opt)
131
132			opt_para = pd.DataFrame.from_dict(_opt_args_.kwargs_opt, dtype=object)
133			print("opt_para", opt_para)
134			opt_para.to_csv(self.date_path + "opt_para", index=False)
135			"""
136
137			def _save_toCSV(self, meta_data_new, path):
138			if os.path.exists(path):
139			meta_data_old = pd.read_csv(path)
140			meta_data = meta_data_old.append(meta_data_new)
141
142			columns = list(meta_data.columns)
143			noScore = ["mean_test_score", "cv_default_score", "eval_time", "run"]
144			columns_noScore = [c for c in columns if c not in noScore]
145
146			meta_data = meta_data.drop_duplicates(subset=columns_noScore)
147			else:
148			meta_data = meta_data_new
149
150			meta_data.to_csv(path, index=False)
151
152			def _read_func_metadata(self, model_func, _verb_):
153			paths = self._get_func_data_names()
154
155			meta_data_list = []
156			for path in paths:
157			meta_data = pd.read_csv(path)
158			meta_data_list.append(meta_data)
159			self.meta_data_found = True
160
161			if len(meta_data_list) > 0:
162			meta_data = pd.concat(meta_data_list, ignore_index=True)
163
164			column_names = meta_data.columns
165			score_name = [name for name in column_names if self.score_col_name in name]
166
167			para = meta_data.drop(score_name, axis=1)
168			score = meta_data[score_name]
169
170			_verb_.load_meta_data()
171			return para, score
172
173			else:
174			_verb_.no_meta_data(model_func)
175			return None, None
176
177			def _get_opt_meta_data(self):
178			results_dict = {}
179			para_list = []
180			score_list = []
181
182			for key in self.memory_dict.keys():
183			pos = np.fromstring(key, dtype=int)
184			para = self._space_.pos2para(pos)
185			score = self.memory_dict[key]
186
187			for key in para.keys():
188			if (
189			not isinstance(para[key], int)
190			and not isinstance(para[key], float)
191			and not isinstance(para[key], str)
192			):
193
194			para_dill = dill.dumps(para[key])
195			para_hash = self._get_hash(para_dill)
196
197			with open(
198			self.func_path + str(para_hash) + ".pkl", "wb"
199			) as pickle_file:
200			dill.dump(para_dill, pickle_file)
201
202			para[key] = para_hash
203
204			if score != 0:
205			para_list.append(para)
206			score_list.append(score)
207
208			results_dict["params"] = para_list
209			results_dict["mean_test_score"] = score_list
210
211			return results_dict
212
213			def _load_data_into_memory(self, paras, scores):
214
215			paras = paras.replace(self._hash2obj())
216			paras = self.para2pos(paras)
217
218			df_temp = pd.DataFrame()
219			df_temp["pos_str"] = paras.apply(apply_tobytes, axis=1)
220			df_temp["score"] = scores
221
222			self.memory_dict = df_temp.set_index('pos_str').to_dict()['score']
223
224			scores = np.array(scores)
225			paras = np.array(paras)
226
227			idx = np.argmax(scores)
228			self.score_best = scores[idx]
229			self.pos_best = paras[idx]
230
231			def para2pos(self, paras):
232			paras = paras[self._space_.para_names]
233			for pos_key in self._space_.search_space:
234			paras[pos_key] = paras[pos_key].apply(
235			self._space_.search_space[pos_key].index
236			)
237
238			return paras
239
240			def _collect(self, _cand_):
241			results_dict = self._get_opt_meta_data()
242
243			para_pd = pd.DataFrame(results_dict["params"])
244			metric_pd = pd.DataFrame(
245			results_dict["mean_test_score"], columns=["mean_test_score"]
246			)
247
248			eval_time = pd.DataFrame(_cand_.eval_time, columns=["eval_time"])
249			md_model = pd.concat(
250			[para_pd, metric_pd, eval_time], axis=1, ignore_index=False
251			)
252
253			return md_model
254
255			def _get_hash(self, object):
256			return hashlib.sha1(object).hexdigest()
257
258			def _get_func_str(self, func):
259			return inspect.getsource(func)
260
261			def _get_subdirs(self):
262			subdirs = glob.glob(self.func_path + "*/")
263
264			return subdirs
265
266			def _get_func_data_names1(self):
267			subdirs = self._get_subdirs()
268
269			path_list = []
270			for subdir in subdirs:
271			paths = glob.glob(subdir + "*.csv")
272			path_list = path_list + paths
273
274			return path_list
275
276			def _get_func_data_names(self):
277			paths = glob.glob(
278			self.func_path + (self.feature_hash + "_" + self.label_hash + "_.csv")
279			)
280
281			return paths
282
283			def _read_dill(self, value):
284			paths = self._get_pkl_hash(value)
285			for path in paths:
286			with open(path, "rb") as fp:
287			value = dill.load(fp)
288			value = dill.loads(value)
289			break
290
291			return value
292
293			def _hash2obj(self):
294			hash2obj_dict = {}
295			para_hash_list = self._get_para_hash_list()
296
297			for para_hash in para_hash_list:
298			obj = self._read_dill(para_hash)
299			hash2obj_dict[para_hash] = obj
300
301			return hash2obj_dict
302
303			def _get_para_hash_list(self):
304			para_hash_list = []
305			for key in self._space_.search_space.keys():
306			values = self._space_.search_space[key]
307
308			for value in values:
309			if (
310			not isinstance(value, int)
311			and not isinstance(value, float)
312			and not isinstance(value, str)
313			):
314
315			para_dill = dill.dumps(value)
316			para_hash = self._get_hash(para_dill)
317			para_hash_list.append(para_hash)
318
319			return para_hash_list
320
321			def _get_pkl_hash(self, hash):
322			paths = glob.glob(self.func_path + hash + "*.pkl")
323
324			return paths
325
326			def _get_file_path(self, model_func):
327			if not os.path.exists(self.date_path):
328			os.makedirs(self.date_path)
329
330			return self.func_path + (self.feature_hash + "_" + self.label_hash + "_.csv")
331

SimonBlanke / Hyperactive

Push — master ( a0fddf...e61612 )

apply_tobytes() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like