-
Notifications
You must be signed in to change notification settings - Fork 42
/
Metrics.py
149 lines (115 loc) · 4.56 KB
/
Metrics.py
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
from __future__ import division
from collections import OrderedDict, Iterable
import pandas as pd
import numpy as np
import pickle
class Metrics(object):
def __init__(self):
self.metrics = OrderedDict()
self.cache_dict = OrderedDict()
def register(self, name=None, value=None, formatter=None,
display_name=None, write_db = True, write_mail = True):
"""Register a new metric.
Params
------
name: str
Name of the metric. Name is used for computation and set as attribute.
display_name: str or None
Disoplay name of variable written in db and mail
value:
formatter:
Formatter to present value of metric. E.g. `'{:.2f}'.format`
write_db: boolean, default = True
Write value into db
write_mail: boolean, default = True
Write metric in result mail to user
"""
assert not name is None, 'No name specified'.format(name)
if not value:
value = 0
self.__setattr__( name, value)
if not display_name: display_name = name
self.metrics[name] = {
'name' : name,
'write_db' : write_db,
'formatter' : formatter,
'write_mail' : write_mail,
'display_name' : display_name
}
def cache(self, name=None, value=None, func=None):
assert not name is None, 'No name specified'.format(name)
self.__setattr__( name, value)
self.cache_dict[name] = {
'name' : name,
'func' : func
}
def __call__(self, name):
return self.metrics[name]
@property
def names(self):
"""Returns the name identifiers of all registered metrics."""
return [v['name'] for v in self.metrics.values()]
@property
def display_names(self):
"""Returns the display name identifiers of all registered metrics."""
return [v['display_name'] for v in self.metrics.values()]
@property
def formatters(self):
"""Returns the formatters for all metrics that have associated formatters."""
return dict([(v['display_name'], v['formatter']) for k, v in self.metrics.items() if not v['formatter'] is None])
#@property
def val_dict(self, display_name = False, object = "metrics"):
"""Returns dictionary of all registered values of object name or display_name as key.
Params
------
display_name: boolean, default = False
If True, display_name of keys in dict. (default names)
object: "cache" or "metrics", default = "metrics"
"""
if display_name: key_string = "display_name"
else: key_string = "name"
print("object dict: ", object)
val_dict = dict([(self.__getattribute__(object)[key][key_string], self.__getattribute__(key)) for key in self.__getattribute__(object).keys() ])
return val_dict
def val_db(self, display_name = True):
"""Returns dictionary of all registered values metrics to write in db."""
if display_name: key_string = "display_name"
else: key_string = "name"
val_dict = dict([(self.metrics[key][key_string], self.__getattribute__(key)) for key in self.metrics.keys() if self.metrics[key]["write_db"] ])
return val_dict
def val_mail(self, display_name = True):
"""Returns dictionary of all registered values metrics to write in mail."""
if display_name: key_string = "display_name"
else: key_string = "name"
val_dict = dict([(self.metrics[key][key_string], self.__getattribute__(key)) for key in self.metrics.keys() if self.metrics[key]["write_mail"] ])
return val_dict
def to_dataframe(self, display_name = False, type = None):
"""Returns pandas dataframe of all registered values metrics. """
if type=="mail":
self.df = pd.DataFrame(self.val_mail(display_name = display_name), index=[self.seqName])
else:
self.df = pd.DataFrame(self.val_dict(display_name = display_name), index=[self.seqName])
def update_values(self, value_dict = None):
"""Updates registered metrics with new values in value_dict. """
if value_dict:
for key, value in value_dict.items() :
if hasattr(self, key):
self.__setattr__(key, value)
def print_type(self, object = "metrics"):
"""Prints variable type of registered metrics or caches. """
print( "OBJECT " , object)
val_dict = self.val_dict(object = object)
for key, item in val_dict.items() :
print("%s: %s; Shape: %s" %(key, type(item), np.shape(item)))
def print_results(self):
"""Prints metrics. """
result_dict = self.val_dict()
for key, item in result_dict.items():
print(key)
print("%s: %s" %(key, self.metrics[key]["formatter"](item)))
def save_dict(self, path):
"""Save value dict to path as pickle file."""
with open(path, 'wb') as handle:
pickle.dump(self.__dict__, handle, protocol=pickle.HIGHEST_PROTOCOL)
def compute_metrics_per_sequence(self):
raise NotImplementedError