summaryrefslogtreecommitdiff
path: root/extra/usb_power/stats_manager.py
blob: 02f984f09737fb33724660a52618e230629831f0 (plain)
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
# 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 numpy
import os

STATS_PREFIX = '@@'
KEY_PREFIX = '__'
# This prefix is used for keys that should not be shown in the summary tab, such
# as timeline keys.
NOSHOW_PREFIX = '!!'

class StatsManager(object):
  """Calculates statistics for several lists of data(float)."""

  def __init__(self):
    """Initialize infrastructure for data and their statistics."""
    self._data = collections.defaultdict(list)
    self._summary = {}

  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
    print('Warning: value %s for domain %s is not a number, thus ignored.' %
          (value, domain))

  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.

    Args:
      prefix: start every row in summary string with prefix, for easier reading.
    """
    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]
      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 PrintSummary(self, prefix=STATS_PREFIX):
    """Print the formatted summary.

    Args:
      prefix: start every row in summary string with prefix, for easier reading.
    """
    summary_str = self._SummaryToString(prefix=prefix)
    print(summary_str)

  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 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():
      fname = domain + '.txt'
      fname = os.path.join(dirname, fname)
      with open(fname, 'w') as f:
        f.write('\n'.join('%.2f' % value for value in data) + '\n')