| # Copyright 2017 The ChromiumOS Authors |
| # 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.""" |
| |
| # Note: This is a py2/3 compatible file. |
| |
| from __future__ import print_function |
| |
| import collections |
| import json |
| import logging |
| import math |
| import os |
| |
| import numpy # pylint:disable=import-error |
| |
| STATS_PREFIX = "@@" |
| NAN_TAG = "*" |
| NAN_DESCRIPTION = "%s domains contain NaN samples" % NAN_TAG |
| |
| 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(title='Title Banner') |
| >>> stats.AddSample(TIME_KEY, 50.0) |
| >>> stats.AddSample(TIME_KEY, 25.0) |
| >>> stats.AddSample(TIME_KEY, 40.0) |
| >>> stats.AddSample(TIME_KEY, 10.0) |
| >>> stats.AddSample(TIME_KEY, 10.0) |
| >>> stats.AddSample('frobnicate', 11.5) |
| >>> stats.AddSample('frobnicate', 9.0) |
| >>> stats.AddSample('foobar', 11111.0) |
| >>> stats.AddSample('foobar', 22222.0) |
| >>> stats.CalculateStats() |
| >>> print(stats.SummaryToString()) |
| ` @@-------------------------------------------------------------- |
| ` @@ Title Banner |
| @@-------------------------------------------------------------- |
| @@ 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) |
| _smid: id supplied to differentiate data output to other StatsManager |
| instances that potentially save to the same directory |
| if smid all output files will be named |smid|_|fname| |
| _title: title to add as banner to formatted summary. If no title, |
| no banner gets added |
| _order: list of formatting order for domains. Domains not listed are |
| displayed in sorted order |
| _hide_domains: collection of domains to hide when formatting summary string |
| _accept_nan: flag to indicate if NaN samples are acceptable |
| _nan_domains: set to keep track of which domains contain NaN samples |
| _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. |
| """ |
| |
| # pylint: disable=W0102 |
| def __init__( |
| self, smid="", title="", order=[], hide_domains=[], accept_nan=True |
| ): |
| """Initialize infrastructure for data and their statistics.""" |
| self._title = title |
| self._data = collections.defaultdict(list) |
| self._unit = collections.defaultdict(str) |
| self._smid = smid |
| self._order = order |
| self._hide_domains = hide_domains |
| self._accept_nan = accept_nan |
| self._nan_domains = set() |
| self._summary = {} |
| self._logger = logging.getLogger(type(self).__name__) |
| |
| def AddSample(self, domain, sample): |
| """Add one sample for a domain. |
| |
| Args: |
| domain: the domain name for the sample. |
| sample: one time sample for domain, expect type float. |
| |
| Raises: |
| StatsManagerError: if trying to add NaN and |_accept_nan| is false |
| """ |
| try: |
| sample = float(sample) |
| except ValueError: |
| # if we don't accept nan this will be caught below |
| self._logger.debug( |
| "sample %s for domain %s is not a number. Making NaN", |
| sample, |
| domain, |
| ) |
| sample = float("NaN") |
| if not self._accept_nan and math.isnan(sample): |
| raise StatsManagerError( |
| "accept_nan is false. Cannot add NaN sample." |
| ) |
| self._data[domain].append(sample) |
| if math.isnan(sample): |
| self._nan_domains.add(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.warning( |
| "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.items(): |
| data_np = numpy.array(data) |
| self._summary[domain] = { |
| "mean": numpy.nanmean(data_np), |
| "min": numpy.nanmin(data_np), |
| "max": numpy.nanmax(data_np), |
| "stddev": numpy.nanstd(data_np), |
| "count": data_np.size, |
| } |
| |
| @property |
| def DomainsToDisplay(self): |
| """List of domains that the manager will output in summaries.""" |
| return set(self._summary.keys()) - set(self._hide_domains) |
| |
| @property |
| def NanInOutput(self): |
| """Return whether any of the domains to display have NaN values.""" |
| return bool(len(set(self._nan_domains) & self.DomainsToDisplay)) |
| |
| def _SummaryTable(self): |
| """Generate the matrix to output as a summary. |
| |
| Returns: |
| A 2d matrix of headers and their data for each domain |
| e.g. |
| [[NAME, COUNT, MEAN, STDDEV, MAX, MIN], |
| [pp5000_mw, 10, 50, 0, 50, 50]] |
| """ |
| headers = ("NAME", "COUNT", "MEAN", "STDDEV", "MAX", "MIN") |
| table = [headers] |
| # determine what domains to display & and the order |
| domains_to_display = self.DomainsToDisplay |
| display_order = [ |
| key for key in self._order if key in domains_to_display |
| ] |
| domains_to_display -= set(display_order) |
| display_order.extend(sorted(domains_to_display)) |
| for domain in display_order: |
| stats = self._summary[domain] |
| if not domain.endswith(self._unit[domain]): |
| domain = "%s_%s" % (domain, self._unit[domain]) |
| if domain in self._nan_domains: |
| domain = "%s%s" % (domain, NAN_TAG) |
| row = [domain] |
| row.append(str(stats["count"])) |
| for entry in headers[2:]: |
| row.append("%.2f" % stats[entry.lower()]) |
| table.append(row) |
| return table |
| |
| def SummaryToMarkdownString(self): |
| """Format the summary into a b/ compatible markdown table string. |
| |
| This requires this sort of output format |
| |
| | header1 | header2 | header3 | ... |
| | --------- | --------- | --------- | ... |
| | sample1h1 | sample1h2 | sample1h3 | ... |
| . |
| . |
| . |
| |
| Returns: |
| formatted summary string. |
| """ |
| # All we need to do before processing is insert a row of '-' between |
| # the headers, and the data |
| table = self._SummaryTable() |
| columns = len(table[0]) |
| # Using '-:' to allow the numbers to be right aligned |
| sep_row = ["-"] + ["-:"] * (columns - 1) |
| table.insert(1, sep_row) |
| text_rows = ["|".join(r) for r in table] |
| body = "\n".join(["|%s|" % r for r in text_rows]) |
| if self._title: |
| title_section = "**%s** \n\n" % self._title |
| body = title_section + body |
| # Make sure that the body is terminated with a newline. |
| return body + "\n" |
| |
| 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. |
| """ |
| table = self._SummaryTable() |
| 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_lines = [] |
| for row in table: |
| formatted_row = prefix + " " |
| for i in range(len(row)): |
| formatted_row += row[i].rjust(max_col_width[i] + 2) |
| formatted_lines.append(formatted_row) |
| if self.NanInOutput: |
| formatted_lines.append("%s %s" % (prefix, NAN_DESCRIPTION)) |
| |
| if self._title: |
| line_length = len(formatted_lines[0]) |
| dec_length = len(prefix) |
| # trim title to be at most as long as the longest line without the prefix |
| title = self._title[: (line_length - dec_length)] |
| # line is a seperator line consisting of ----- |
| line = "%s%s" % (prefix, "-" * (line_length - dec_length)) |
| # prepend the prefix to the centered title |
| padded_title = "%s%s" % ( |
| prefix, |
| title.center(line_length)[dec_length:], |
| ) |
| formatted_lines = ( |
| [line, padded_title, line] + formatted_lines + [line] |
| ) |
| formatted_output = "\n".join(formatted_lines) |
| return formatted_output |
| |
| def GetSummary(self): |
| """Getter for summary.""" |
| return self._summary |
| |
| def _MakeUniqueFName(self, fname): |
| """prepend |_smid| to fname & rotate fname to ensure uniqueness. |
| |
| Before saving a file through the StatsManager, make sure that the filename |
| is unique, first by prepending the smid if any and otherwise by appending |
| increasing integer suffixes until the filename is unique. |
| |
| If |smid| is defined /path/to/example/file.txt becomes |
| /path/to/example/{smid}_file.txt. |
| |
| The rotation works by changing /path/to/example/somename.txt to |
| /path/to/example/somename1.txt if the first one already exists on the |
| system. |
| |
| Note: this is not thread-safe. While it makes sense to use StatsManager |
| in a threaded data-collection, the data retrieval should happen in a |
| single threaded environment to ensure files don't get potentially clobbered. |
| |
| Args: |
| fname: filename to ensure uniqueness. |
| |
| Returns: |
| {smid_}fname{tag}.[b].ext |
| the smid portion gets prepended if |smid| is defined |
| the tag portion gets appended if necessary to ensure unique fname |
| """ |
| fdir = os.path.dirname(fname) |
| base, ext = os.path.splitext(os.path.basename(fname)) |
| if self._smid: |
| base = "%s_%s" % (self._smid, base) |
| unique_fname = os.path.join(fdir, "%s%s" % (base, ext)) |
| tag = 0 |
| while os.path.exists(unique_fname): |
| old_fname = unique_fname |
| unique_fname = os.path.join(fdir, "%s%d%s" % (base, tag, ext)) |
| self._logger.warning( |
| "Attempted to store stats information at %s, but " |
| "file already exists. Attempting to store at %s " |
| "now.", |
| old_fname, |
| unique_fname, |
| ) |
| tag += 1 |
| return unique_fname |
| |
| 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. |
| |
| Returns: |
| full path of summary save location |
| """ |
| summary_str = self.SummaryToString(prefix=prefix) + "\n" |
| return self._SaveSummary(summary_str, directory, fname) |
| |
| 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. |
| |
| Returns: |
| full path of summary save location |
| """ |
| data = {} |
| for domain in self._summary: |
| unit = LONG_UNIT.get(self._unit[domain], self._unit[domain]) |
| data_entry = {"mean": self._summary[domain]["mean"], "unit": unit} |
| data[domain] = data_entry |
| summary_str = json.dumps(data, indent=2) |
| return self._SaveSummary(summary_str, directory, fname) |
| |
| def SaveSummaryMD(self, directory, fname="summary.md"): |
| """Save summary into a MD file to paste into b/. |
| |
| Args: |
| directory: directory to save the MD summary in. |
| fname: filename to save summary under. |
| |
| Returns: |
| full path of summary save location |
| """ |
| summary_str = self.SummaryToMarkdownString() |
| return self._SaveSummary(summary_str, directory, fname) |
| |
| def _SaveSummary(self, output_str, directory, fname): |
| """Wrote |output_str| to |fname|. |
| |
| Args: |
| output_str: formatted output string |
| directory: directory to save the summary in. |
| fname: filename to save summary under. |
| |
| Returns: |
| full path of summary save location |
| """ |
| if not os.path.exists(directory): |
| os.makedirs(directory) |
| fname = self._MakeUniqueFName(os.path.join(directory, fname)) |
| with open(fname, "w") as f: |
| f.write(output_str) |
| return fname |
| |
| 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. |
| |
| Returns: |
| list of full path of each domain's raw data save location |
| """ |
| if not os.path.exists(directory): |
| os.makedirs(directory) |
| dirname = os.path.join(directory, dirname) |
| if not os.path.exists(dirname): |
| os.makedirs(dirname) |
| fnames = [] |
| for domain, data in self._data.items(): |
| if not domain.endswith(self._unit[domain]): |
| domain = "%s_%s" % (domain, self._unit[domain]) |
| fname = self._MakeUniqueFName( |
| os.path.join(dirname, "%s.txt" % domain) |
| ) |
| with open(fname, "w") as f: |
| f.write("\n".join("%.2f" % sample for sample in data) + "\n") |
| fnames.append(fname) |
| return fnames |
