syndrome/paramprocessor.py at master · diegomcarvalho/syndrome · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
import ray
import lmfit as lm

import os
import json
import numpy as np

import dpath.util as du


class NpEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, np.integer):
            return int(obj)
        elif isinstance(obj, np.floating):
            return float(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        else:
            return super(NpEncoder, self).default(obj)


@ray.remote(num_cpus=0.1)
class ParamProcessor(object):

    def __init__(self, store, var_names):
        self.param_dict = dict()
        self.store = store
        self.var_names = var_names
        return

    def dump(self, filename):
        with open(filename, 'w') as f:
            for tag in self.param_dict.keys():
                keys = list(self.param_dict[tag].keys())
                f.write('tag, ct')
                for vname in self.var_names:
                    f.write(f', {vname}')
                f.write('\n')
                for ct in keys:
                    f.write(f'{tag}, {ct}')
                    for vname in self.var_names:
                        val = self.param_dict[tag][ct][vname].value
                        f.write(f', {val}')
                    f.write('\n')
        return

    def get_param(self, key, tag):
        if tag in self.param_dict.keys():
            if key in self.param_dict[tag].keys():
                return self.param_dict[tag][key]
        params = lm.Parameters()

        try:
            with open(f'{self.store}/{tag}/{key}.json', 'r') as f:
                if tag not in self.param_dict.keys():
                    self.param_dict[tag] = dict()

                self.param_dict[tag][key] = params.load(f)
        except:
            return params

        return self.param_dict[tag][key]

    def set_param(self, key, tag, param):
        # Create a new dict if there isn't this tag
        if tag not in self.param_dict.keys():
            self.param_dict[tag] = dict()

        # Store the new param
        self.param_dict[tag][key] = param

        # Make a temporary copy onto the filesystem
        tmp_storage_directory = f'{self.store}/tmp/{tag}'
        if not os.path.exists(tmp_storage_directory):
            try:
                os.makedirs(tmp_storage_directory)
            except Exception as e:
                print(f'ParamProcessor: cannot create {tmp_storage_directory}')
                raise e

        with open(f'{self.store}/tmp/{tag}/{key}.json', 'w') as f:
            f.write(self.param_dict[tag][key].dumps(cls=NpEncoder))

        return


@ray.remote(num_cpus=0.1)
class MemoryProcessor(object):

    def __init__(self, var_names):
        self.memory = dict()
        du.new(self.memory, '/memory/var_names', sorted(var_names))
        return

    def set_val(self, ct, tag, variable, value):
        if variable not in du.get(self.memory, '/memory/var_names'):
            print(f'ValProcessor: {variable} is not in my memory')

        du.new(self.memory, f'/data/{ct}/{tag}/{variable}', value)
        return

    def get_val(self, ct, tag, variable):
        if variable not in du.get(self.memory, '/memory/var_names'):
            print(f'ValProcessor: {variable} is not in my memory')

        return du.get(self.memory, f'/data/{ct}/{tag}/{variable}')

    def set_service(self, service, value):
        du.new(self.memory, f'/service/{service}', value)
        return

    def get_service(self, service):
        print(service)
        print(self.memory)
        return du.get(self.memory, f'/service/{service}')


    def dump(self, filename):
        with open(filename, 'w') as f:
            f.write(json.dumps(self.memory, indent=4, sort_keys=True))
        return

    def to_csv(self, tag, filename):
        with open(filename, 'w') as f:
            f.write('ct, tag')
            for vname in du.get(self.memory, '/memory/var_names'):
                f.write(f', {vname}')
            f.write('\n')
            data = du.search(self.memory, f'/data/*/{tag}/*')['data']
            for ct in data.keys():
                f.write(f'{ct}, {tag}')
                for vname in du.get(self.memory, '/memory/var_names'):
                    vvalue = data[ct][tag].get(vname, 'n.a')
                    f.write(f', {vvalue}')
                f.write('\n')
        return

    def add_log(self, ct, tag, val):
        try:
            du.get(self.memory, f'/log/{ct}/{tag}').append(val)
        except KeyError as e:
            du.new(self.memory, f'/log/{ct}/{tag}', [val])
        return

    def dump_log(self, ct, filename):
        with open(filename, 'w') as f:
            f.write(f'tag, line\n')
            for tag, line in du.get(self.memory, f'/log/{ct}').items():
                f.write(f'{tag}, {line}\n')
        return