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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
|
# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Calculates statistics for lists of data and pretty print them."""
from __future__ import print_function
import collections
import json
import logging
import os
import numpy
STATS_PREFIX = '@@'
# used to aid sorting of dict keys
KEY_PREFIX = '__'
# This prefix is used for keys that should not be shown in the summary tab, such
# as timeline keys.
NOSHOW_PREFIX = '!!'
LONG_UNIT = {
'': 'N/A',
'mW': 'milliwatt',
'uW': 'microwatt',
'mV': 'millivolt',
'uA': 'microamp',
'uV': 'microvolt'
}
class StatsManagerError(Exception):
"""Errors in StatsManager class."""
pass
class StatsManager(object):
"""Calculates statistics for several lists of data(float).
Example usage:
>>> stats = StatsManager()
>>> stats.AddValue(TIME_KEY, 50.0)
>>> stats.AddValue(TIME_KEY, 25.0)
>>> stats.AddValue(TIME_KEY, 40.0)
>>> stats.AddValue(TIME_KEY, 10.0)
>>> stats.AddValue(TIME_KEY, 10.0)
>>> stats.AddValue('frobnicate', 11.5)
>>> stats.AddValue('frobnicate', 9.0)
>>> stats.AddValue('foobar', 11111.0)
>>> stats.AddValue('foobar', 22222.0)
>>> stats.CalculateStats()
>>> print(stats.SummaryToString())
@@ NAME COUNT MEAN STDDEV MAX MIN
@@ sample_msecs 4 31.25 15.16 50.00 10.00
@@ foobar 2 16666.50 5555.50 22222.00 11111.00
@@ frobnicate 2 10.25 1.25 11.50 9.00
Attributes:
_data: dict of list of readings for each domain(key)
_unit: dict of unit for each domain(key)
_summary: dict of stats per domain (key): min, max, count, mean, stddev
_logger = StatsManager logger
Note:
_summary is empty until CalculateStats() is called, and is updated when
CalculateStats() is called.
"""
def __init__(self):
"""Initialize infrastructure for data and their statistics."""
self._data = collections.defaultdict(list)
self._unit = collections.defaultdict(str)
self._summary = {}
self._logger = logging.getLogger('StatsManager')
def AddValue(self, domain, value):
"""Add one value for a domain.
Args:
domain: the domain name for the value.
value: one time reading for domain, expect type float.
"""
if isinstance(value, int):
value = float(value)
if isinstance(value, float):
self._data[domain].append(value)
return
self._logger.warn('value %s for domain %s is not a number, thus ignored.',
value, domain)
def SetUnit(self, domain, unit):
"""Set the unit for a domain.
There can be only one unit for each domain. Setting unit twice will
overwrite the original unit.
Args:
domain: the domain name.
unit: unit of the domain.
"""
if domain in self._unit:
self._logger.warn('overwriting the unit of %s, old unit is %s, new unit '
'is %s.', domain, self._unit[domain], unit)
self._unit[domain] = unit
def CalculateStats(self):
"""Calculate stats for all domain-data pairs.
First erases all previous stats, then calculate stats for all data.
"""
self._summary = {}
for domain, data in self._data.iteritems():
data_np = numpy.array(data)
self._summary[domain] = {
'mean': data_np.mean(),
'min': data_np.min(),
'max': data_np.max(),
'stddev': data_np.std(),
'count': data_np.size,
}
def SummaryToString(self, prefix=STATS_PREFIX):
"""Format summary into a string, ready for pretty print.
See class description for format example.
Args:
prefix: start every row in summary string with prefix, for easier reading.
Returns:
formatted summary string.
"""
headers = ('NAME', 'COUNT', 'MEAN', 'STDDEV', 'MAX', 'MIN')
table = [headers]
for domain in sorted(self._summary.keys()):
if domain.startswith(NOSHOW_PREFIX):
continue
stats = self._summary[domain]
if not domain.endswith(self._unit[domain]):
domain = '%s_%s' % (domain, self._unit[domain])
row = [domain.lstrip(KEY_PREFIX)]
row.append(str(stats['count']))
for entry in headers[2:]:
row.append('%.2f' % stats[entry.lower()])
table.append(row)
max_col_width = []
for col_idx in range(len(table[0])):
col_item_widths = [len(row[col_idx]) for row in table]
max_col_width.append(max(col_item_widths))
formatted_table = []
for row in table:
formatted_row = prefix + ' '
for i in range(len(row)):
formatted_row += row[i].rjust(max_col_width[i] + 2)
formatted_table.append(formatted_row)
return '\n'.join(formatted_table)
def GetSummary(self):
"""Getter for summary."""
return self._summary
def SaveSummary(self, directory, fname='summary.txt', prefix=STATS_PREFIX):
"""Save summary to file.
Args:
directory: directory to save the summary in.
fname: filename to save summary under.
prefix: start every row in summary string with prefix, for easier reading.
"""
summary_str = self.SummaryToString(prefix=prefix) + '\n'
if not os.path.exists(directory):
os.makedirs(directory)
fname = os.path.join(directory, fname)
with open(fname, 'w') as f:
f.write(summary_str)
def SaveSummaryJSON(self, directory, fname='summary.json'):
"""Save summary (only MEAN) into a JSON file.
Args:
directory: directory to save the JSON summary in.
fname: filename to save summary under.
"""
data = {}
for domain in self._summary:
if domain.startswith(NOSHOW_PREFIX):
continue
unit = LONG_UNIT.get(self._unit[domain], self._unit[domain])
data_entry = {'mean': self._summary[domain]['mean'], 'unit': unit}
data[domain] = data_entry
if not os.path.exists(directory):
os.makedirs(directory)
fname = os.path.join(directory, fname)
with open(fname, 'w') as f:
json.dump(data, f)
def GetRawData(self):
"""Getter for all raw_data."""
return self._data
def SaveRawData(self, directory, dirname='raw_data'):
"""Save raw data to file.
Args:
directory: directory to create the raw data folder in.
dirname: folder in which raw data live.
"""
if not os.path.exists(directory):
os.makedirs(directory)
dirname = os.path.join(directory, dirname)
if not os.path.exists(dirname):
os.makedirs(dirname)
for domain, data in self._data.iteritems():
if not domain.endswith(self._unit[domain]):
domain = '%s_%s' % (domain, self._unit[domain])
fname = '%s.txt' % domain
fname = os.path.join(dirname, fname)
with open(fname, 'w') as f:
f.write('\n'.join('%.2f' % value for value in data) + '\n')
|
