CachedObjectiveAdapter.memory()   A
last analyzed

Complexity

Conditions 5

Size

Total Lines 18
Code Lines 14

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 14
dl 0
loc 18
rs 9.2333
c 0
b 0
f 0
cc 5
nop 3
1
# Author: Simon Blanke
2
# Email: [email protected]
3
# License: MIT License
4
5
6
import logging
7
import pandas as pd
8
9
from typing import Callable, Any, List
10
from multiprocessing.managers import DictProxy
11
12
from ._objective_adapter import ObjectiveAdapter
13
from ._result import Result
14
15
16
class CachedObjectiveAdapter(ObjectiveAdapter):
17
    def __init__(self, conv, objective):
18
        super().__init__(conv, objective)
19
20
        self.memory_dict = {}
21
        self.memory_dict_new = {}
22
23
    def memory(self, warm_start: pd.DataFrame, memory: Any = None):
24
        if isinstance(memory, DictProxy):
25
            self.memory_dict = memory
26
27
        if warm_start is None:
28
            return
29
30
        if not isinstance(warm_start, pd.DataFrame):
31
            logging.warning("Memory warm start must be of type pandas.DataFrame")
32
            logging.warning("Optimization will continue without memory warm start")
33
            return
34
35
        if warm_start.empty:
36
            logging.warning("Memory warm start has no values in current search space")
37
            logging.warning("Optimization will continue without memory warm start")
38
            return
39
40
        self.memory_dict.update(self._conv.dataframe2memory_dict(warm_start))
41
42
    def __call__(self, pos):
43
        pos_t = tuple(pos)
44
45
        if pos_t in self.memory_dict:
46
            params = self._conv.value2para(self._conv.position2value(pos))
47
48
            return self.memory_dict[pos_t], params
49
        else:
50
            result, params = self._call_objective(pos)
51
            self.memory_dict[pos_t] = result
52
            self.memory_dict_new[pos_t] = result
53
            return result, params
54