blob: e7dd78c9925d8847058874ccdbc45ad006d66882 [file] [log] [blame] [edit]
# Copyright 2015 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 error, logging, os, serial, shutil, threading, time
_power_play_data_file = '/tmp/power_play_data'
class PowerPlay(object):
"""Class to record serial over USB data from Power Play (go/powerplay).
It detects if powerplay is connected to the DUT over USB and opens the
serial port to start receiving powerplay data. It also opens a text file to
save this data after some formatting.
"""
version = 1
def __init__(self, test_obj, record_interval=0):
"""Initialize PowerPlay.
@param test_obj: test object.
@param record_interval: Power play data recording interval in seconds.
"""
self.test = test_obj
self.ser = None
self.recording_interval = record_interval
self.momentary_curr_list = list()
self.record_thread = None
def extract_current(self, pp_data):
"""Extract momentary current value from each line of powerplay data.
@param pp_data: Single line of powerplay data with eight comma separated
values.
@return list containing momentary current values.
"""
if pp_data[0].isdigit():
self.momentary_curr_list.append(float(pp_data[pp_data.index(',')+1:]
[:pp_data[pp_data.index(',')+1:].index(',')]))
return self.momentary_curr_list
def start_recording_power_play_data(self):
"""Starts a new thread to record power play data."""
self.record_thread = threading.Thread(target=self.start_record_thread)
self.record_thread.daemon = True
self.record_thread.start()
def start_record_thread(self):
"""Start recording power play data.
Get a list of connected USB devices and try to establish a serial
connection. Once the connection is established, open a text file and
start reading serial data and write it to the text file after some
formatting.
"""
devices = [x for x in os.listdir('/dev/') if x.startswith('ttyUSB')]
for device in devices:
device_link = '/dev/' + device
try:
if self.ser == None:
logging.info('Trying ... %s', device_link)
self.ser = serial.Serial(device_link, 115200)
logging.info('Successfully connected to %s', device_link)
break
except serial.SerialException, e:
raise error.TestError('Failed to connect to %s becuase of %s' %
(device_link, str(e)))
self.text_file = open(_power_play_data_file, 'w')
if self.ser != None:
title_row = ('time,powerplay_timestamp,momentary_current (A),' +
'momentary_charge (AH),average_current (A),' +
'total_standby_time,total_wake_time,num_wakes,is_awake?\n')
self.text_file.write(title_row)
start_time = time.time()
while self.ser.readline():
current_timestamp = (('{:>10.3f}'.
format(time.time() - start_time)).replace(' ', ''))
pp_data = (self.ser.readline().replace('\00', '').
replace(' ', ',').replace('\r', ''))
if (not pp_data.startswith('#') and (len(pp_data) > 30) and
not self.text_file.closed):
self.text_file.write(current_timestamp + ',' + pp_data)
self.momentary_curr_list = self.extract_current(pp_data)
time.sleep(self.recording_interval)
self.ser.flushInput()
else:
self.text_file.write('No data from powerplay. Check connection.')
def stop_recording_power_play_data(self):
"""Stop recording power play data.
Close the text file and copy it to the test log results directory. Also
report current data to the performance dashboard.
"""
if not self.text_file.closed:
self.text_file.close()
shutil.copy(_power_play_data_file, self.test.resultsdir)
self.test.output_perf_value(description='momentary_current_draw',
value=self.momentary_curr_list,
units='Amps', higher_is_better=False)