src/tools/telemetry/telemetry/page/csv_page_measurement_results.py

   1 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4 import csv
   5
   6 from telemetry.page import page_measurement_results
   7
   8 class CsvPageMeasurementResults(
   9     page_measurement_results.PageMeasurementResults):
  10   def __init__(self, output_stream, output_after_every_page=None):
  11     super(CsvPageMeasurementResults, self).__init__()
  12     self._output_stream = output_stream
  13     self._results_writer = csv.writer(self._output_stream)
  14     self._did_output_header = False
  15     self._header_names_written_to_writer = None
  16     self._output_after_every_page = output_after_every_page
  17
  18   def DidMeasurePage(self):
  19     assert self.values_for_current_page, 'Failed to call WillMeasurePage'
  20
  21     if (not self.values_for_current_page.values or
  22         not self._output_after_every_page):
  23       # Do not output if no results were added on this page or if output flag
  24       # is not set.
  25       super(CsvPageMeasurementResults, self).DidMeasurePage()
  26       return
  27
  28     if not self._did_output_header:
  29       self._OutputHeader()
  30     else:
  31       self._ValidateOutputNamesForCurrentPage()
  32
  33     self._OutputValuesForPage(self.values_for_current_page)
  34
  35     super(CsvPageMeasurementResults, self).DidMeasurePage()
  36
  37   def PrintSummary(self):
  38     if not self._output_after_every_page:
  39       self._OutputHeader()
  40       for page_values in self.page_results:
  41         self._OutputValuesForPage(page_values)
  42
  43     super(CsvPageMeasurementResults, self).PrintSummary()
  44
  45   def _ValidateOutputNamesForCurrentPage(self):
  46     assert self._did_output_header
  47     current_page_measurement_names = \
  48         set(self.values_for_current_page.measurement_names)
  49     header_names_written_to_writer = \
  50         set(self._header_names_written_to_writer)
  51     if header_names_written_to_writer == current_page_measurement_names:
  52       return
  53     assert False, """To use CsvPageMeasurementResults, you must add the same
  54 result names for every page. In this case, first page output:
  55 %s
  56
  57 Thus, all subsequent pages must output this as well. Instead, the current page
  58 output:
  59 %s
  60
  61 Change your test to produce the same thing each time, or modify
  62 PageMeasurement.results_are_the_same_on_every_page to return False.
  63 """ % (repr(header_names_written_to_writer),
  64        repr(current_page_measurement_names))
  65
  66   def _OutputHeader(self):
  67     assert not self._did_output_header
  68     all_measurement_names = list(
  69       self.all_measurements_that_have_been_seen.keys())
  70     all_measurement_names.sort()
  71     self._did_output_header = True
  72     self._header_names_written_to_writer = list(all_measurement_names)
  73
  74     row = ['page_name']
  75     for measurement_name in all_measurement_names:
  76       measurement_data = \
  77           self.all_measurements_that_have_been_seen[measurement_name]
  78       row.append('%s (%s)' % (measurement_name, measurement_data['units']))
  79     self._results_writer.writerow(row)
  80     self._output_stream.flush()
  81
  82   def _OutputValuesForPage(self, page_values):
  83     row = [page_values.page.display_name]
  84     for measurement_name in self._header_names_written_to_writer:
  85       value = page_values.FindValueByMeasurementName(measurement_name)
  86       if value:
  87         row.append('%s' % value.output_value)
  88       else:
  89         row.append('-')
  90     self._results_writer.writerow(row)
  91     self._output_stream.flush()