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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
|
# Lint as: python2, python3
# Copyright 2018 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.
import logging
import re
import time
from autotest_lib.client.bin import test
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros import arc_common
from autotest_lib.client.common_lib.cros import retry
from autotest_lib.client.common_lib.cros.network import interface
from autotest_lib.client.cros import service_stopper
from autotest_lib.client.cros.camera import camera_utils
from autotest_lib.client.cros.power import force_discharge_utils
from autotest_lib.client.cros.power import power_dashboard
from autotest_lib.client.cros.power import power_status
from autotest_lib.client.cros.power import power_telemetry_utils
from autotest_lib.client.cros.power import power_utils
from autotest_lib.client.cros.video import histogram_verifier
class power_Test(test.test):
"""Optional base class power related tests."""
version = 1
keypress_histogram = 'Event.Latency.EndToEnd.KeyPress'
histogram_re = 'Histogram: %s recorded (\d+) samples, mean = (\d+\.\d+)'
hist_percentile_re = '^(\d+).+\{(\d+)\.\d+\%\}'
def initialize(self,
seconds_period=20.,
pdash_note='',
force_discharge='false',
check_network=False,
run_arc=True):
"""Perform necessary initialization prior to power test run.
@param seconds_period: float of probing interval in seconds.
@param pdash_note: note of the current run to send to power dashboard.
@param force_discharge: string of whether to tell ec to discharge
battery even when the charger is plugged in. 'false' means no
forcing discharge; 'true' means forcing discharge and raising an
error when it fails; 'optional' means forcing discharge when
possible but not raising an error when it fails, which is more
friendly to devices without a battery.
@param check_network: check that Ethernet interface is not running.
@param run_arc: bool, whether to run with ARC (if available)
@var backlight: power_utils.Backlight object.
@var keyvals: dictionary of result keyvals.
@var status: power_status.SysStat object.
@var _checkpoint_logger: power_status.CheckpointLogger to track
checkpoint data.
@var _psr: power_utils.DisplayPanelSelfRefresh object to monitor PSR.
@var _services: service_stopper.ServiceStopper object.
@var _start_time: float of time in seconds since Epoch test started.
@var _stats: power_status.StatoMatic object.
@var _meas_logs: list of power_status.MeasurementLoggers
"""
super(power_Test, self).initialize()
self.keyvals = dict()
self.status = power_status.get_status()
self._checkpoint_logger = power_status.CheckpointLogger()
self._seconds_period = seconds_period
self._force_discharge_success = force_discharge_utils.process(
force_discharge, self.status)
self.backlight = power_utils.Backlight(
force_battery=self._force_discharge_success)
self.backlight.set_default()
ifaces = [iface for iface in interface.get_interfaces()
if (not iface.is_wifi_device() and
iface.name.startswith('eth'))]
logging.debug('Ethernet interfaces include: %s',
str([iface.name for iface in ifaces]))
for iface in ifaces:
if check_network and iface.is_lower_up:
raise error.TestError('Ethernet interface is active. '
'Please remove Ethernet cable.')
self._psr = power_utils.DisplayPanelSelfRefresh()
self._services = service_stopper.ServiceStopper(
service_stopper.ServiceStopper.POWER_DRAW_SERVICES)
self._services.stop_services()
self._stats = power_status.StatoMatic()
self._meas_logs = power_status.create_measurement_loggers(
seconds_period, self._checkpoint_logger)
logging.debug('measurement loggers (%d):', len(self._meas_logs))
for log in self._meas_logs:
logging.debug('%s: %s', type(log).__name__, ", ".join(log.domains))
self._pdash_note = pdash_note
self._failure_messages = []
self._arc_mode = arc_common.ARC_MODE_DISABLED
if run_arc and utils.is_arc_available():
self._arc_mode = arc_common.ARC_MODE_ENABLED
self.keyvals['arc_mode'] = self._arc_mode
def get_extra_browser_args_for_camera_test(self):
"""Return Chrome args for camera power test."""
ret = [
# No pop up to ask permission to record video.
'--use-fake-ui-for-media-stream',
# Allow 2 windows side by side.
'--force-tablet-mode=clamshell',
# Prefer using constant frame rate for camera streaming.
'--enable-features=PreferConstantFrameRate',
# Bypass HID detection for Chromebox / Chromebase.
'--disable-hid-detection-on-oobe',
# Disable test account info sync, eg. Wi-Fi credentials,
# so that each test run does not remember info from last test
# run.
'--disable-sync'
]
# Use fake camera for DUT without camera, e.g. chromebox.
if not camera_utils.has_builtin_or_vivid_camera():
ret.append('--use-fake-device-for-media-stream')
self.keyvals['use_fake_camera'] = 1
else:
self.keyvals['use_fake_camera'] = 0
return ret
def warmup(self, warmup_time=30):
"""Warm up.
Wait between initialization and run_once for new settings to stabilize.
@param warmup_time: integer of seconds to warmup.
"""
time.sleep(warmup_time)
def start_measurements(self):
"""Start measurements."""
for log in self._meas_logs:
log.start()
self._start_time = time.time()
if self.status.battery:
self._start_energy = self.status.battery.energy
self._keyvallogger = power_dashboard.KeyvalLogger(self._start_time)
power_telemetry_utils.start_measurement()
def loop_sleep(self, loop, sleep_secs):
"""Jitter free sleep.
@param loop: integer of loop (1st is zero).
@param sleep_secs: integer of desired sleep seconds.
"""
next_time = self._start_time + (loop + 1) * sleep_secs
time.sleep(next_time - time.time())
def checkpoint_measurements(self, name, start_time=None):
"""Checkpoint measurements.
@param name: string name of measurement being checkpointed.
@param start_time: float of time in seconds since Epoch that
measurements being checkpointed began.
"""
if not start_time:
start_time = self._start_time
self.status.refresh()
self._checkpoint_logger.checkpoint(name, start_time)
self._psr.refresh()
@retry.retry(Exception, timeout_min=1, delay_sec=2)
def collect_keypress_latency(self, cr):
"""Collect keypress latency information from Histograms.
@param cr: object, the Chrome instance
"""
keypress_histogram_end = histogram_verifier.get_histogram(
cr, self.keypress_histogram)
matches = re.search((self.histogram_re % self.keypress_histogram),
keypress_histogram_end)
if matches:
count = int(matches.group(1))
mean_latency = float(matches.group(2))
logging.info('latency count %d mean %f', count, mean_latency)
self.keyvals['keypress_cnt'] = count
self.keyvals['keypress_latency_us_avg'] = mean_latency
self.output_perf_value(description='keypress_cnt', value=count,
higher_is_better=True)
self.output_perf_value(description='keypress_latency_us_avg',
value=mean_latency,
higher_is_better=False)
self._keyvallogger.add_item('keypress_cnt', count, 'point',
'keypress')
self._keyvallogger.add_item('keypress_latency_us_avg',
mean_latency, 'point', 'keypress')
# Capture the first bucket >= 90th percentile
for s in keypress_histogram_end.splitlines():
matches = re.search((self.hist_percentile_re), s)
if matches:
lat = int(matches.group(1))
perc = int(matches.group(2))
if perc >= 90:
self.keyvals['keypress_latency_us_high'] = lat
self.keyvals['keypress_high_percentile'] = perc
self.output_perf_value(
description='keypress_latency_us_high', value=lat,
higher_is_better=False)
self.output_perf_value(
description='keypress_high_percentile', value=perc,
higher_is_better=False)
self._keyvallogger.add_item('keypress_latency_us_high',
lat, 'point', 'keypress')
self._keyvallogger.add_item('keypress_high_percentile',
perc, 'point', 'keypress')
break
def publish_keyvals(self):
"""Publish power result keyvals."""
keyvals = self._stats.publish()
keyvals['level_backlight_max'] = self.backlight.get_max_level()
keyvals['level_backlight_current'] = self.backlight.get_level()
# record battery stats if battery exists
keyvals['b_on_ac'] = int(not self._force_discharge_success
and self.status.on_ac())
keyvals['force_discharge'] = int(self._force_discharge_success)
for key in [
'b_on_ac', 'force_discharge', 'percent_usb_suspended_time'
]:
self._keyvallogger.add_item(key, keyvals[key], 'point', 'perf')
if self.status.battery:
keyvals['ah_charge_full'] = self.status.battery.charge_full
keyvals['ah_charge_full_design'] = \
self.status.battery.charge_full_design
keyvals['ah_charge_now'] = self.status.battery.charge_now
keyvals['a_current_now'] = self.status.battery.current_now
keyvals['wh_energy_full'] = self.status.battery.energy_full
keyvals['wh_energy_full_design'] = \
self.status.battery.energy_full_design
keyvals['wh_energy_now'] = self.status.battery.energy
keyvals['wh_energy_start'] = self._start_energy
energy_used = self._start_energy - self.status.battery.energy
runtime_minutes = (time.time() - self._start_time) / 60.
keyvals['wh_energy_used'] = energy_used
keyvals['minutes_tested'] = runtime_minutes
self._keyvallogger.add_item('minutes_tested',
keyvals['minutes_tested'], 'point',
'perf')
low_batt = power_utils.get_low_battery_shutdown_percent()
keyvals['percent_sys_low_battery'] = low_batt
if energy_used > 0 and runtime_minutes > 1:
keyvals['w_energy_rate'] = energy_used * 60. / runtime_minutes
self._keyvallogger.add_item('w_energy_rate',
keyvals['w_energy_rate'], 'point',
'perf')
energy_avail = self.status.battery.energy_full_design * \
((100. - low_batt) / 100.)
keyvals['minutes_battery_life'] = energy_avail / energy_used * \
runtime_minutes
self._keyvallogger.add_item('minutes_battery_life',
keyvals['minutes_battery_life'],
'point', 'perf')
keyvals['hours_battery_life'] = \
keyvals['minutes_battery_life'] / 60.
self._keyvallogger.add_item('hours_battery_life',
keyvals['hours_battery_life'],
'point', 'perf')
keyvals['v_voltage_min_design'] = \
self.status.battery.voltage_min_design
keyvals['v_voltage_now'] = self.status.battery.voltage_now
for log in self._meas_logs:
keyvals.update(log.calc())
keyvals.update(self._psr.get_keyvals())
self.keyvals.update(keyvals)
core_keyvals = power_utils.get_core_keyvals(self.keyvals)
self.write_perf_keyval(core_keyvals)
def publish_dashboard(self):
"""Report results to chromeperf & power dashboard."""
self.publish_keyvals()
# publish power values
for key, values in self.keyvals.items():
if key.endswith('pwr_avg'):
self.output_perf_value(description=key, value=values, units='W',
higher_is_better=False, graph='power')
# publish temperature values
for key, values in self.keyvals.items():
if key.endswith('temp_avg'):
self.output_perf_value(description=key, value=values, units='C',
higher_is_better=False, graph='temperature')
# publish fps values
for key, values in self.keyvals.items():
if key.endswith('fps_avg'):
self.output_perf_value(description=key, value=values,
units='fps', higher_is_better=True, graph='fps')
# include KeyvalLogger in dashboard
self._meas_logs.append(self._keyvallogger)
# publish to power dashboard
dashboard_factory = power_dashboard.get_dashboard_factory()
for log in self._meas_logs:
dashboard = dashboard_factory.createDashboard(log,
self.tagged_testname, self.resultsdir, note=self._pdash_note)
dashboard.upload()
def _save_results(self):
"""Save results of each logger in resultsdir."""
for log in self._meas_logs:
log.save_results(self.resultsdir)
self._checkpoint_logger.save_checkpoint_data(self.resultsdir)
def postprocess_iteration(self):
"""Write keyval and send data to dashboard."""
power_telemetry_utils.end_measurement()
self.status.refresh()
for log in self._meas_logs:
log.done = True
super(power_Test, self).postprocess_iteration()
self.publish_dashboard()
self._save_results()
power_dashboard.generate_parallax_report(self.outputdir)
if self._failure_messages:
raise error.TestFail('Test has failed with messages: %s' %
self._failure_messages)
def cleanup(self):
"""Reverse setting change in initialization."""
force_discharge_utils.restore(self._force_discharge_success)
if self.backlight:
self.backlight.restore()
self._services.restore_services()
super(power_Test, self).cleanup()
|