| #!/usr/bin/python |
| |
| # Copyright (C) 2014 The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import logging |
| import os.path |
| import select |
| import sys |
| import time |
| import collections |
| import socket |
| import gflags as flags # http://code.google.com/p/python-gflags/ |
| import pkgutil |
| import threading |
| import Queue |
| import traceback |
| import math |
| import bisect |
| from bisect import bisect_left |
| |
| """ |
| scipy, numpy and matplotlib are python packages that can be installed |
| from: http://www.scipy.org/ |
| |
| """ |
| import scipy |
| import matplotlib.pyplot as plt |
| |
| # let this script know about the power monitor implementations |
| sys.path = [os.path.basename(__file__)] + sys.path |
| available_monitors = [ |
| name |
| for _, name, _ in pkgutil.iter_modules( |
| [os.path.join(os.path.dirname(__file__), "power_monitors")]) |
| if not name.startswith("_")] |
| |
| APK = os.path.join(os.path.dirname(__file__), "..", "CtsVerifier.apk") |
| |
| FLAGS = flags.FLAGS |
| |
| # DELAY_SCREEN_OFF is the number of seconds to wait for baseline state |
| DELAY_SCREEN_OFF = 20.0 |
| |
| # whether to log data collected to a file for each sensor run: |
| LOG_DATA_TO_FILE = True |
| |
| logging.getLogger().setLevel(logging.ERROR) |
| |
| |
| def do_import(name): |
| """import a module by name dynamically""" |
| mod = __import__(name) |
| components = name.split(".") |
| for comp in components[1:]: |
| mod = getattr(mod, comp) |
| return mod |
| |
| class PowerTestException(Exception): |
| """ |
| Definition of specialized Exception class for CTS power tests |
| """ |
| def __init__(self, message): |
| self._error_message = message |
| def __str__(self): |
| return self._error_message |
| |
| class PowerTest: |
| """Class to run a suite of power tests. This has methods for obtaining |
| measurements from the power monitor (through the driver) and then |
| processing it to determine baseline and AP suspend state and |
| measure ampere draw of various sensors. |
| Ctrl+C causes a keyboard interrupt exception which terminates the test.""" |
| |
| # Thresholds for max allowed power usage per sensor tested |
| # TODO: Accel, Mag and Gyro have no maximum power specified in the CDD; |
| # the following numbers are bogus and will be replaced soon by what |
| # the device reports (from Sensor.getPower()) |
| MAX_ACCEL_AMPS = 0.08 # Amps |
| MAX_MAG_AMPS = 0.08 # Amps |
| MAX_GYRO_AMPS = 0.08 # Amps |
| MAX_SIGMO_AMPS = 0.08 # Amps |
| |
| # TODO: The following numbers for step counter, etc must be replaced by |
| # the numbers specified in CDD for low-power sensors. The expected current |
| # draw must be computed from the specified power and the voltage used to |
| # power the device (specified from a config file). |
| MAX_STEP_COUNTER_AMPS = 0.08 # Amps |
| MAX_STEP_DETECTOR_AMPS = 0.08 # Amps |
| # The variable EXPECTED_AMPS_VARIATION_HALF_RANGE denotes the expected |
| # variation of the ampere measurements |
| # around the mean value at baseline state. i.e. we expect most of the |
| # ampere measurements at baseline state to vary around the mean by |
| # between +/- of the number below |
| EXPECTED_AMPS_VARIATION_HALF_RANGE = 0.0005 |
| # The variable THRESHOLD_BASELINE_SAMPLES_FRACTION denotes the minimum fraction of samples that must |
| # be in the range of variation defined by EXPECTED_AMPS_VARIATION_HALF_RANGE |
| # around the mean baseline for us to decide that the phone has settled into |
| # its baseline state |
| THRESHOLD_BASELINE_SAMPLES_FRACTION = 0.86 |
| # The variable MAX_PERCENTILE_AP_SCREEN_OFF_AMPS denotes the maximum ampere |
| # draw that the device can consume when it has gone to suspend state with |
| # one or more sensors registered and batching samples (screen and AP are |
| # off in this case) |
| MAX_PERCENTILE_AP_SCREEN_OFF_AMPS = 0.030 # Amps |
| # The variable PERCENTILE_MAX_AP_SCREEN_OFF denotes the fraction of ampere |
| # measurements that must be below the specified maximum amperes |
| # MAX_PERCENTILE_AP_SCREEN_OFF_AMPS for us to decide that the phone has |
| # reached suspend state. |
| PERCENTILE_MAX_AP_SCREEN_OFF = 0.95 |
| DOMAIN_NAME = "/android/cts/powertest" |
| # SAMPLE_COUNT_NOMINAL denotes the typical number of measurements of amperes |
| # to collect from the power monitor |
| SAMPLE_COUNT_NOMINAL = 1000 |
| # RATE_NOMINAL denotes the nominal frequency at which ampere measurements |
| # are taken from the monsoon power monitor |
| RATE_NOMINAL = 100 |
| ENABLE_PLOTTING = False |
| |
| REQUEST_EXTERNAL_STORAGE = "EXTERNAL STORAGE?" |
| REQUEST_EXIT = "EXIT" |
| REQUEST_RAISE = "RAISE %s %s" |
| REQUEST_USER_RESPONSE = "USER RESPONSE %s" |
| REQUEST_SET_TEST_RESULT = "SET TEST RESULT %s %s %s" |
| REQUEST_SENSOR_SWITCH = "SENSOR %s %s" |
| REQUEST_SENSOR_AVAILABILITY = "SENSOR? %s" |
| REQUEST_SCREEN_OFF = "SCREEN OFF" |
| REQUEST_SHOW_MESSAGE = "MESSAGE %s" |
| |
| NEGATIVE_AMPERE_ERROR_MESSAGE = ( |
| "Negative ampere draw measured, possibly due to power " |
| "supply from USB cable. Check the setup of device and power " |
| "monitor to make sure that the device is not connected " |
| "to machine via USB directly. The device should be " |
| "connected to the USB slot in the power monitor. It is okay " |
| "to change the wiring when the test is in progress.") |
| |
| |
| def __init__(self, max_baseline_amps): |
| """ |
| Args: |
| max_baseline_amps: The maximum value of baseline amperes |
| that we expect the device to consume at baseline state. |
| This can be different between models of phones. |
| """ |
| power_monitors = do_import("power_monitors.%s" % FLAGS.power_monitor) |
| testid = time.strftime("%d_%m_%Y__%H__%M_%S") |
| self._power_monitor = power_monitors.Power_Monitor(log_file_id = testid) |
| self._tcp_connect_port = 0 # any available port |
| print ("Establishing connection to device...") |
| self.setUsbEnabled(True) |
| status = self._power_monitor.GetStatus() |
| self._native_hz = status["sampleRate"] * 1000 |
| # the following describes power test being run (i.e on what sensor |
| # and what type of test. This is used for logging. |
| self._current_test = "None" |
| self._external_storage = self.executeOnDevice(PowerTest.REQUEST_EXTERNAL_STORAGE) |
| self._max_baseline_amps = max_baseline_amps |
| |
| def __del__(self): |
| self.finalize() |
| |
| def finalize(self): |
| """To be called upon termination of host connection to device""" |
| if self._tcp_connect_port > 0: |
| # tell device side to exit connection loop, and remove the forwarding |
| # connection |
| self.executeOnDevice(PowerTest.REQUEST_EXIT, reportErrors = False) |
| self.executeLocal("adb forward --remove tcp:%d" % self._tcp_connect_port) |
| self._tcp_connect_port = 0 |
| if self._power_monitor: |
| self._power_monitor.Close() |
| self._power_monitor = None |
| |
| def _send(self, msg, report_errors = True): |
| """Connect to the device, send the given command, and then disconnect""" |
| if self._tcp_connect_port == 0: |
| # on first attempt to send a command, connect to device via any open port number, |
| # forwarding that port to a local socket on the device via adb |
| logging.debug("Seeking port for communication...") |
| # discover an open port |
| dummysocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| dummysocket.bind(("localhost", 0)) |
| (_, self._tcp_connect_port) = dummysocket.getsockname() |
| dummysocket.close() |
| assert(self._tcp_connect_port > 0) |
| |
| status = self.executeLocal("adb forward tcp:%d localabstract:%s" % |
| (self._tcp_connect_port, PowerTest.DOMAIN_NAME)) |
| # If the status !=0, then the host machine is unable to |
| # forward requests to client over adb. Ending the test and logging error message |
| # to the console on the host. |
| self.endTestIfLostConnection( |
| status != 0, |
| "Unable to forward requests to client over adb") |
| logging.info("Forwarding requests over local port %d", |
| self._tcp_connect_port) |
| |
| link = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| |
| try: |
| logging.debug("Connecting to device...") |
| link.connect(("localhost", self._tcp_connect_port)) |
| logging.debug("Connected.") |
| except socket.error as serr: |
| print "Socket connection error: ", serr |
| print "Finalizing and exiting the test" |
| self.endTestIfLostConnection( |
| report_errors, |
| "Unable to communicate with device: connection refused") |
| except: |
| print "Non socket-related exception at this block in _send(); re-raising now." |
| raise |
| logging.debug("Sending '%s'", msg) |
| link.sendall(msg) |
| logging.debug("Getting response...") |
| response = link.recv(4096) |
| logging.debug("Got response '%s'", response) |
| link.close() |
| return response |
| |
| def queryDevice(self, query): |
| """Post a yes/no query to the device, return True upon successful query, False otherwise""" |
| logging.info("Querying device with '%s'", query) |
| return self._send(query) == "OK" |
| |
| # TODO: abstract device communication (and string commands) into its own class |
| def executeOnDevice(self, cmd, reportErrors = True): |
| """Execute a (string) command on the remote device""" |
| return self._send(cmd, reportErrors) |
| |
| def executeLocal(self, cmd, check_status = True): |
| """execute a shell command locally (on the host)""" |
| from subprocess import call |
| status = call(cmd.split(" ")) |
| if status != 0 and check_status: |
| logging.error("Failed to execute \"%s\"", cmd) |
| else: |
| logging.debug("Executed \"%s\"", cmd) |
| return status |
| |
| def reportErrorRaiseExceptionIf(self, condition, msg): |
| """Report an error condition to the device if condition is True. |
| Will raise an exception on the device if condition is True. |
| Args: |
| condition: If true, this reports error |
| msg: Message related to exception |
| Raises: |
| A PowerTestException encapsulating the message provided in msg |
| """ |
| if condition: |
| try: |
| logging.error("Exiting on error: %s" % msg) |
| self.executeOnDevice(PowerTest.REQUEST_RAISE % (self._current_test, msg), |
| reportErrors = True) |
| except: |
| logging.error("Unable to communicate with device to report " |
| "error: %s" % msg) |
| self.finalize() |
| sys.exit(msg) |
| raise PowerTestException(msg) |
| |
| def endTestIfLostConnection(self, lost_connection, error_message): |
| """ |
| This function ends the test if lost_connection was true, |
| which indicates that the connection to the device was lost. |
| Args: |
| lost_connection: boolean variable, if True it indicates that |
| connection to device was lost and the test must be terminated. |
| error_message: String to print to the host console before exiting the test |
| (if lost_connection is True) |
| Returns: |
| None. |
| """ |
| if lost_connection: |
| logging.error(error_message) |
| self.finalize() |
| sys.exit(error_message) |
| |
| def setUsbEnabled(self, enabled, verbose = True): |
| if enabled: |
| val = 1 |
| else: |
| val = 0 |
| self._power_monitor.SetUsbPassthrough(val) |
| tries = 0 |
| |
| # Sometimes command won't go through first time, particularly if immediately after a data |
| # collection, so allow for retries |
| # TODO: Move this retry mechanism to the power monitor driver. |
| status = self._power_monitor.GetStatus() |
| while status is None and tries < 5: |
| tries += 1 |
| time.sleep(2.0) |
| logging.error("Retrying get status call...") |
| self._power_monitor.StopDataCollection() |
| self._power_monitor.SetUsbPassthrough(val) |
| status = self._power_monitor.GetStatus() |
| |
| if enabled: |
| if verbose: |
| print("...USB enabled, waiting for device") |
| self.executeLocal("adb wait-for-device") |
| if verbose: |
| print("...device online") |
| else: |
| if verbose: |
| logging.info("...USB disabled") |
| # re-establish port forwarding |
| if enabled and self._tcp_connect_port > 0: |
| status = self.executeLocal("adb forward tcp:%d localabstract:%s" % |
| (self._tcp_connect_port, PowerTest.DOMAIN_NAME)) |
| self.reportErrorRaiseExceptionIf(status != 0, msg = "Unable to forward requests to client over adb") |
| |
| def computeBaselineState(self, measurements): |
| """ |
| Args: |
| measurements: List of floats containing ampere draw measurements |
| taken from the monsoon power monitor. |
| Must be atleast 100 measurements long |
| Returns: |
| A tuple (isBaseline, mean_current) where isBaseline is a |
| boolean that is True only if the baseline state for the phone is |
| detected. mean_current is an estimate of the average baseline |
| current for the device, which is valid only if baseline state is |
| detected (if not, it is set to -1). |
| """ |
| |
| # Looks at the measurements to see if it is in baseline state |
| if len(measurements) < 100: |
| print( |
| "Need at least 100 measurements to determine if baseline state has" |
| " been reached") |
| return (False, -1) |
| |
| # Assumption: At baseline state, the power profile is Gaussian distributed |
| # with low-variance around the mean current draw. |
| # Ideally we should find the mode from a histogram bin to find an estimated mean. |
| # Assuming here that the median is very close to this value; later we check that the |
| # variance of the samples is low enough to validate baseline. |
| sorted_measurements = sorted(measurements) |
| number_measurements = len(measurements) |
| if not number_measurements % 2: |
| median_measurement = (sorted_measurements[(number_measurements - 1) / 2] + |
| sorted_measurements[(number_measurements + 1) / 2]) / 2 |
| else: |
| median_measurement = sorted_measurements[number_measurements / 2] |
| |
| # Assume that at baseline state, a large fraction of power measurements |
| # are within +/- EXPECTED_AMPS_VARIATION_HALF_RANGE milliAmperes of |
| # the average baseline current. Find all such measurements in the |
| # sorted measurement vector. |
| left_index = ( |
| bisect_left( |
| sorted_measurements, |
| median_measurement - |
| PowerTest.EXPECTED_AMPS_VARIATION_HALF_RANGE)) |
| right_index = ( |
| bisect_left( |
| sorted_measurements, |
| median_measurement + |
| PowerTest.EXPECTED_AMPS_VARIATION_HALF_RANGE)) |
| |
| average_baseline_amps = scipy.mean( |
| sorted_measurements[left_index: (right_index - 1)]) |
| |
| detected_baseline = True |
| # We enforce that a fraction of more than 'THRESHOLD_BASELINE_SAMPLES_FRACTION' |
| # of samples must be within +/- EXPECTED_AMPS_VARIATION_HALF_RANGE |
| # milliAmperes of the mean baseline current, which we have estimated as |
| # the median. |
| if ((right_index - left_index) < PowerTest.THRESHOLD_BASELINE_SAMPLES_FRACTION * len( |
| measurements)): |
| detected_baseline = False |
| |
| # We check for the maximum limit of the expected baseline |
| if median_measurement > self._max_baseline_amps: |
| detected_baseline = False |
| if average_baseline_amps < 0: |
| print PowerTest.NEGATIVE_AMPERE_ERROR_MESSAGE |
| detected_baseline = False |
| |
| print("%s baseline state" % ("Could detect" if detected_baseline else "Could NOT detect")) |
| print( |
| "median amps = %f, avg amps = %f, fraction of good samples = %f" % |
| (median_measurement, average_baseline_amps, |
| float(right_index - left_index) / len(measurements))) |
| if PowerTest.ENABLE_PLOTTING: |
| plt.plot(measurements) |
| plt.show() |
| print("To continue test, please close the plot window manually.") |
| return (detected_baseline, average_baseline_amps) |
| |
| def isApInSuspendState(self, measurements_amps, nominal_max_amps, test_percentile): |
| """ |
| This function detects AP suspend and display off state of phone |
| after a sensor has been registered. |
| |
| Because the power profile can be very different between sensors and |
| even across builds, it is difficult to specify a tight threshold for |
| mean current draw or mandate that the power measurements must have low |
| variance. We use a criteria that allows for a certain fraction of |
| peaks in power spectrum and checks that test_percentile fraction of |
| measurements must be below the specified value nominal_max_amps |
| Args: |
| measurements_amps: amperes draw measurements from power monitor |
| test_percentile: the fraction of measurements we require to be below |
| a specified amps value |
| nominal_max_amps: the specified value of the max current draw |
| Returns: |
| returns a boolean which is True if and only if the AP suspend and |
| display off state is detected |
| """ |
| count_good = len([m for m in measurements_amps if m < nominal_max_amps]) |
| count_negative = len([m for m in measurements_amps if m < 0]) |
| if count_negative > 0: |
| print PowerTest.NEGATIVE_AMPERE_ERROR_MESSAGE |
| return False; |
| return count_good > test_percentile * len(measurements_amps) |
| |
| def getBaselineState(self): |
| """This function first disables all sensors, then collects measurements |
| through the power monitor and continuously evaluates if baseline state |
| is reached. Once baseline state is detected, it returns a tuple with |
| status information. If baseline is not detected in a preset maximum |
| number of trials, it returns as well. |
| |
| Returns: |
| Returns a tuple (isBaseline, mean_current) where isBaseline is a |
| boolean that is True only if the baseline state for the phone is |
| detected. mean_current is an estimate of the average baseline current |
| for the device, which is valid only if baseline state is detected |
| (if not, it is set to -1) |
| """ |
| self.setPowerOn("ALL", False) |
| self.setUsbEnabled(False) |
| print("Waiting %d seconds for baseline state" % DELAY_SCREEN_OFF) |
| time.sleep(DELAY_SCREEN_OFF) |
| |
| MEASUREMENT_DURATION_SECONDS_BASELINE_DETECTION = 5 # seconds |
| NUMBER_MEASUREMENTS_BASELINE_DETECTION = ( |
| PowerTest.RATE_NOMINAL * |
| MEASUREMENT_DURATION_SECONDS_BASELINE_DETECTION) |
| NUMBER_MEASUREMENTS_BASELINE_VERIFICATION = ( |
| NUMBER_MEASUREMENTS_BASELINE_DETECTION * 5) |
| MAX_TRIALS = 50 |
| |
| collected_baseline_measurements = False |
| |
| for tries in xrange(MAX_TRIALS): |
| print("Trial number %d of %d..." % (tries, MAX_TRIALS)) |
| measurements = self.collectMeasurements( |
| NUMBER_MEASUREMENTS_BASELINE_DETECTION, PowerTest.RATE_NOMINAL, |
| verbose = False) |
| if self.computeBaselineState(measurements)[0] is True: |
| collected_baseline_measurements = True |
| break |
| |
| if collected_baseline_measurements: |
| print("Verifying baseline state over a longer interval " |
| "in order to double check baseline state") |
| measurements = self.collectMeasurements( |
| NUMBER_MEASUREMENTS_BASELINE_VERIFICATION, PowerTest.RATE_NOMINAL, |
| verbose = False) |
| self.reportErrorRaiseExceptionIf( |
| not measurements, "No background measurements could be taken") |
| retval = self.computeBaselineState(measurements) |
| if retval[0]: |
| print("Verified baseline.") |
| if measurements and LOG_DATA_TO_FILE: |
| with open("/tmp/cts-power-tests-background-data.log", "w") as f: |
| for m in measurements: |
| f.write("%.4f\n" % m) |
| return retval |
| else: |
| return (False, -1) |
| |
| def waitForApSuspendMode(self): |
| """This function repeatedly collects measurements until AP suspend and display off |
| mode is detected. After a maximum number of trials, if this state is not reached, it |
| raises an error. |
| Returns: |
| boolean which is True if device was detected to be in suspend state |
| Raises: |
| Power monitor-related exception |
| """ |
| print("waitForApSuspendMode(): Sleeping for %d seconds" % DELAY_SCREEN_OFF) |
| time.sleep(DELAY_SCREEN_OFF) |
| |
| NUMBER_MEASUREMENTS = 200 |
| # Maximum trials for which to collect measurements to get to Ap suspend |
| # state |
| MAX_TRIALS = 50 |
| |
| got_to_suspend_state = False |
| for count in xrange(MAX_TRIALS): |
| print ("waitForApSuspendMode(): Trial %d of %d" % (count, MAX_TRIALS)) |
| measurements = self.collectMeasurements(NUMBER_MEASUREMENTS, |
| PowerTest.RATE_NOMINAL, |
| verbose = False) |
| if self.isApInSuspendState( |
| measurements, PowerTest.MAX_PERCENTILE_AP_SCREEN_OFF_AMPS, |
| PowerTest.PERCENTILE_MAX_AP_SCREEN_OFF): |
| got_to_suspend_state = True |
| break |
| self.reportErrorRaiseExceptionIf( |
| got_to_suspend_state is False, |
| msg = "Unable to determine application processor suspend mode status.") |
| print("Got to AP suspend state") |
| return got_to_suspend_state |
| |
| def collectMeasurements(self, measurementCount, rate, verbose = True): |
| """Args: |
| measurementCount: Number of measurements to collect from the power |
| monitor |
| rate: The integer frequency in Hertz at which to collect measurements from |
| the power monitor |
| Returns: |
| A list containing measurements from the power monitor; that has the |
| requested count of the number of measurements at the specified rate |
| """ |
| assert (measurementCount > 0) |
| decimate_by = self._native_hz / rate or 1 |
| |
| self._power_monitor.StartDataCollection() |
| sub_measurements = [] |
| measurements = [] |
| tries = 0 |
| if verbose: print("") |
| try: |
| while len(measurements) < measurementCount and tries < 5: |
| if tries: |
| self._power_monitor.StopDataCollection() |
| self._power_monitor.StartDataCollection() |
| time.sleep(1.0) |
| tries += 1 |
| additional = self._power_monitor.CollectData() |
| if additional is not None: |
| tries = 0 |
| sub_measurements.extend(additional) |
| while len(sub_measurements) >= decimate_by: |
| sub_avg = sum(sub_measurements[0:decimate_by]) / decimate_by |
| measurements.append(sub_avg) |
| sub_measurements = sub_measurements[decimate_by:] |
| if verbose: |
| # "\33[1A\33[2K" is a special Linux console control |
| # sequence for moving to the previous line, and |
| # erasing it; and reprinting new text on that |
| # erased line. |
| sys.stdout.write("\33[1A\33[2K") |
| print ("MEASURED[%d]: %f" % (len(measurements), measurements[-1])) |
| finally: |
| self._power_monitor.StopDataCollection() |
| |
| self.reportErrorRaiseExceptionIf(measurementCount > len(measurements), |
| "Unable to collect all requested measurements") |
| return measurements |
| |
| def requestUserAcknowledgment(self, msg): |
| """Post message to user on screen and wait for acknowledgment""" |
| response = self.executeOnDevice(PowerTest.REQUEST_USER_RESPONSE % msg) |
| self.reportErrorRaiseExceptionIf( |
| response != "OK", "Unable to request user acknowledgment") |
| |
| def setTestResult(self, test_name, test_result, test_message): |
| """ |
| Reports the result of a test to the device |
| Args: |
| test_name: name of the test |
| test_result: Boolean result of the test (True means Pass) |
| test_message: Relevant message |
| """ |
| print ("Test %s : %s" % (test_name, test_result)) |
| |
| response = ( |
| self.executeOnDevice( |
| PowerTest.REQUEST_SET_TEST_RESULT % |
| (test_name, test_result, test_message))) |
| self.reportErrorRaiseExceptionIf( |
| response != "OK", "Unable to send test status to Verifier") |
| |
| def setPowerOn(self, sensor, powered_on): |
| response = self.executeOnDevice(PowerTest.REQUEST_SENSOR_SWITCH % |
| (("ON" if powered_on else "OFF"), sensor)) |
| self.reportErrorRaiseExceptionIf( |
| response == "ERR", "Unable to set sensor %s state" % sensor) |
| logging.info("Set %s %s", sensor, ("ON" if powered_on else "OFF")) |
| return response |
| |
| def runSensorPowerTest( |
| self, sensor, max_amperes_allowed, baseline_amps, user_request = None): |
| """ |
| Runs power test for a specific sensor; i.e. measures the amperes draw |
| of the phone using monsoon, with the specified sensor mregistered |
| and the phone in suspend state; and verifies that the incremental |
| consumed amperes is within expected bounds. |
| Args: |
| sensor: The specified sensor for which to run the power test |
| max_amperes_allowed: Maximum ampere draw of the device with the |
| sensor registered and device in suspend state |
| baseline_amps: The power draw of the device when it is in baseline |
| state (no sensors registered, display off, AP asleep) |
| """ |
| self._current_test = ("%s_Power_Test_While_%s" % ( |
| sensor, ("Under_Motion" if user_request is not None else "Still"))) |
| try: |
| print ("\n\n---------------------------------") |
| if user_request is not None: |
| print ("Running power test on %s under motion." % sensor) |
| else: |
| print ("Running power test on %s while device is still." % sensor) |
| print ("---------------------------------") |
| response = self.executeOnDevice( |
| PowerTest.REQUEST_SENSOR_AVAILABILITY % sensor) |
| if response == "UNAVAILABLE": |
| self.setTestResult( |
| self._current_test, test_result = "SKIPPED", |
| test_message = "Sensor %s not available on this platform" % sensor) |
| self.setPowerOn("ALL", False) |
| if response == "UNAVAILABLE": |
| self.setTestResult( |
| self._current_test, test_result = "SKIPPED", |
| test_message = "Sensor %s not available on this device" % sensor) |
| return |
| self.reportErrorRaiseExceptionIf(response != "OK", "Unable to set all sensor off") |
| self.executeOnDevice(PowerTest.REQUEST_SCREEN_OFF) |
| self.setUsbEnabled(False) |
| self.setUsbEnabled(True) |
| self.setPowerOn(sensor, True) |
| if user_request is not None: |
| print("===========================================\n" + |
| "==> Please follow the instructions presented on the device\n" + |
| "===========================================") |
| self.requestUserAcknowledgment(user_request) |
| self.executeOnDevice(PowerTest.REQUEST_SCREEN_OFF) |
| self.setUsbEnabled(False) |
| self.reportErrorRaiseExceptionIf( |
| response != "OK", "Unable to set sensor %s ON" % sensor) |
| |
| self.waitForApSuspendMode() |
| print ("Collecting sensor %s measurements" % sensor) |
| measurements = self.collectMeasurements(PowerTest.SAMPLE_COUNT_NOMINAL, |
| PowerTest.RATE_NOMINAL) |
| |
| if measurements and LOG_DATA_TO_FILE: |
| with open("/tmp/cts-power-tests-%s-%s-sensor-data.log" % (sensor, |
| ("Under_Motion" if user_request is not None else "Still")), "w") as f: |
| for m in measurements: |
| f.write("%.4f\n" % m) |
| self.setUsbEnabled(True, verbose = False) |
| print("Saving raw data files to device...") |
| self.executeLocal("adb shell mkdir -p %s" % self._external_storage, False) |
| self.executeLocal("adb push %s %s/." % (f.name, self._external_storage)) |
| self.setUsbEnabled(False, verbose = False) |
| self.reportErrorRaiseExceptionIf( |
| not measurements, "No measurements could be taken for %s" % sensor) |
| avg = sum(measurements) / len(measurements) |
| squared = [(m - avg) * (m - avg) for m in measurements] |
| |
| stddev = math.sqrt(sum(squared) / len(squared)) |
| current_diff = avg - baseline_amps |
| self.setUsbEnabled(True) |
| max_power = max(measurements) - avg |
| if current_diff <= max_amperes_allowed: |
| # TODO: fail the test of background > current |
| message = ( |
| "Draw is within limits. Sensor delta:%f mAmp Baseline:%f " |
| "mAmp Sensor: %f mAmp Stddev : %f mAmp Peak: %f mAmp") % ( |
| current_diff * 1000.0, baseline_amps * 1000.0, avg * 1000.0, |
| stddev * 1000.0, max_power * 1000.0) |
| else: |
| message = ( |
| "Draw is too high. Current:%f Background:%f Measured: %f " |
| "Stddev: %f Peak: %f") % ( |
| current_diff * 1000.0, baseline_amps * 1000.0, avg * 1000.0, |
| stddev * 1000.0, max_power * 1000.0) |
| self.setTestResult( |
| self._current_test, |
| ("PASS" if (current_diff <= max_amperes_allowed) else "FAIL"), |
| message) |
| print("Result: " + message) |
| except: |
| traceback.print_exc() |
| self.setTestResult(self._current_test, test_result = "FAIL", |
| test_message = "Exception occurred during run of test.") |
| raise |
| |
| @staticmethod |
| def runTests(max_baseline_amps): |
| testrunner = None |
| try: |
| GENERIC_MOTION_REQUEST = ("\n===> Please press Next and when the " |
| "screen is off, keep the device under motion with only tiny, " |
| "slow movements until the screen turns on again.\nPlease " |
| "refrain from interacting with the screen or pressing any side " |
| "buttons while measurements are taken.") |
| USER_STEPS_REQUEST = ("\n===> Please press Next and when the " |
| "screen is off, then move the device to simulate step motion " |
| "until the screen turns on again.\nPlease refrain from " |
| "interacting with the screen or pressing any side buttons " |
| "while measurements are taken.") |
| testrunner = PowerTest(max_baseline_amps) |
| testrunner.executeOnDevice( |
| PowerTest.REQUEST_SHOW_MESSAGE % "Connected. Running tests...") |
| is_baseline_success, baseline_amps = testrunner.getBaselineState() |
| |
| if is_baseline_success: |
| testrunner.setUsbEnabled(True) |
| # TODO: Enable testing a single sensor |
| testrunner.runSensorPowerTest( |
| "SIGNIFICANT_MOTION", PowerTest.MAX_SIGMO_AMPS, baseline_amps, |
| user_request = GENERIC_MOTION_REQUEST) |
| testrunner.runSensorPowerTest( |
| "STEP_DETECTOR", PowerTest.MAX_STEP_DETECTOR_AMPS, baseline_amps, |
| user_request = USER_STEPS_REQUEST) |
| testrunner.runSensorPowerTest( |
| "STEP_COUNTER", PowerTest.MAX_STEP_COUNTER_AMPS, baseline_amps, |
| user_request = USER_STEPS_REQUEST) |
| testrunner.runSensorPowerTest( |
| "ACCELEROMETER", PowerTest.MAX_ACCEL_AMPS, baseline_amps, |
| user_request = GENERIC_MOTION_REQUEST) |
| testrunner.runSensorPowerTest( |
| "MAGNETIC_FIELD", PowerTest.MAX_MAG_AMPS, baseline_amps, |
| user_request = GENERIC_MOTION_REQUEST) |
| testrunner.runSensorPowerTest( |
| "GYROSCOPE", PowerTest.MAX_GYRO_AMPS, baseline_amps, |
| user_request = GENERIC_MOTION_REQUEST) |
| testrunner.runSensorPowerTest( |
| "ACCELEROMETER", PowerTest.MAX_ACCEL_AMPS, baseline_amps, |
| user_request = None) |
| testrunner.runSensorPowerTest( |
| "MAGNETIC_FIELD", PowerTest.MAX_MAG_AMPS, baseline_amps, |
| user_request = None) |
| testrunner.runSensorPowerTest( |
| "GYROSCOPE", PowerTest.MAX_GYRO_AMPS, baseline_amps, |
| user_request = None) |
| testrunner.runSensorPowerTest( |
| "SIGNIFICANT_MOTION", PowerTest.MAX_SIGMO_AMPS, baseline_amps, |
| user_request = None) |
| testrunner.runSensorPowerTest( |
| "STEP_DETECTOR", PowerTest.MAX_STEP_DETECTOR_AMPS, baseline_amps, |
| user_request = None) |
| testrunner.runSensorPowerTest( |
| "STEP_COUNTER", PowerTest.MAX_STEP_COUNTER_AMPS, baseline_amps, |
| user_request = None) |
| else: |
| print("Could not get to baseline state. This is either because " |
| "in several trials, the monitor could not measure a set " |
| "of power measurements that had the specified low " |
| "variance or the mean measurements were below the " |
| "expected value. None of the sensor power measurement " |
| " tests were performed due to not being able to detect " |
| "baseline state. Please re-run the power tests.") |
| except KeyboardInterrupt: |
| print "Keyboard interrupt from user." |
| raise |
| except: |
| import traceback |
| traceback.print_exc() |
| finally: |
| logging.info("TESTS COMPLETE") |
| if testrunner: |
| try: |
| testrunner.finalize() |
| except socket.error: |
| sys.exit( |
| "===================================================\n" |
| "Unable to connect to device under test. Make sure \n" |
| "the device is connected via the usb pass-through, \n" |
| "the CtsVerifier app is running the SensorPowerTest on \n" |
| "the device, and USB pass-through is enabled.\n" |
| "===================================================") |
| |
| def main(argv): |
| """ Simple command-line interface for a power test application.""" |
| useful_flags = ["voltage", "status", "usbpassthrough", |
| "samples", "current", "log", "power_monitor"] |
| if not [f for f in useful_flags if FLAGS.get(f, None) is not None]: |
| print __doc__.strip() |
| print FLAGS.MainModuleHelp() |
| return |
| |
| if FLAGS.avg and FLAGS.avg < 0: |
| logging.error("--avg must be greater than 0") |
| return |
| |
| if FLAGS.voltage is not None: |
| if FLAGS.voltage > 5.5: |
| print("!!WARNING: Voltage higher than typical values!!!") |
| try: |
| response = raw_input( |
| "Voltage of %.3f requested. Confirm this is correct (Y/N)" % |
| FLAGS.voltage) |
| if response.upper() != "Y": |
| sys.exit("Aborting") |
| except: |
| sys.exit("Aborting.") |
| |
| if not FLAGS.power_monitor: |
| sys.exit( |
| "You must specify a '--power_monitor' option to specify which power " |
| "monitor type " + |
| "you are using.\nOne of:\n \n ".join(available_monitors)) |
| power_monitors = do_import("power_monitors.%s" % FLAGS.power_monitor) |
| try: |
| mon = power_monitors.Power_Monitor(device = FLAGS.device) |
| except: |
| import traceback |
| |
| traceback.print_exc() |
| sys.exit("No power monitors found") |
| |
| if FLAGS.voltage is not None: |
| |
| if FLAGS.ramp is not None: |
| mon.RampVoltage(mon.start_voltage, FLAGS.voltage) |
| else: |
| mon.SetVoltage(FLAGS.voltage) |
| |
| if FLAGS.current is not None: |
| mon.SetMaxCurrent(FLAGS.current) |
| |
| if FLAGS.status: |
| items = sorted(mon.GetStatus().items()) |
| print "\n".join(["%s: %s" % item for item in items]) |
| |
| if FLAGS.usbpassthrough: |
| if FLAGS.usbpassthrough == "off": |
| mon.SetUsbPassthrough(0) |
| elif FLAGS.usbpassthrough == "on": |
| mon.SetUsbPassthrough(1) |
| elif FLAGS.usbpassthrough == "auto": |
| mon.SetUsbPassthrough(2) |
| else: |
| mon.Close() |
| sys.exit("bad pass-through flag: %s" % FLAGS.usbpassthrough) |
| |
| if FLAGS.samples: |
| # Make sure state is normal |
| mon.StopDataCollection() |
| status = mon.GetStatus() |
| native_hz = status["sampleRate"] * 1000 |
| |
| # Collect and average samples as specified |
| mon.StartDataCollection() |
| |
| # In case FLAGS.hz doesn't divide native_hz exactly, use this invariant: |
| # 'offset' = (consumed samples) * FLAGS.hz - (emitted samples) * native_hz |
| # This is the error accumulator in a variation of Bresenham's algorithm. |
| emitted = offset = 0 |
| collected = [] |
| history_deque = collections.deque() # past n samples for rolling average |
| |
| # TODO: Complicated lines of code below. Refactoring needed |
| try: |
| last_flush = time.time() |
| while emitted < FLAGS.samples or FLAGS.samples == -1: |
| # The number of raw samples to consume before emitting the next output |
| need = (native_hz - offset + FLAGS.hz - 1) / FLAGS.hz |
| if need > len(collected): # still need more input samples |
| samples = mon.CollectData() |
| if not samples: break |
| collected.extend(samples) |
| else: |
| # Have enough data, generate output samples. |
| # Adjust for consuming 'need' input samples. |
| offset += need * FLAGS.hz |
| while offset >= native_hz: # maybe multiple, if FLAGS.hz > native_hz |
| this_sample = sum(collected[:need]) / need |
| |
| if FLAGS.timestamp: print int(time.time()), |
| |
| if FLAGS.avg: |
| history_deque.appendleft(this_sample) |
| if len(history_deque) > FLAGS.avg: history_deque.pop() |
| print "%f %f" % (this_sample, |
| sum(history_deque) / len(history_deque)) |
| else: |
| print "%f" % this_sample |
| sys.stdout.flush() |
| |
| offset -= native_hz |
| emitted += 1 # adjust for emitting 1 output sample |
| collected = collected[need:] |
| now = time.time() |
| if now - last_flush >= 0.99: # flush every second |
| sys.stdout.flush() |
| last_flush = now |
| except KeyboardInterrupt: |
| print("interrupted") |
| return 1 |
| finally: |
| mon.Close() |
| return 0 |
| |
| if FLAGS.run: |
| if not FLAGS.power_monitor: |
| sys.exit( |
| "When running power tests, you must specify which type of power " |
| "monitor to use" + |
| " with '--power_monitor <type of power monitor>'") |
| try: |
| PowerTest.runTests(FLAGS.max_baseline_amps) |
| except KeyboardInterrupt: |
| print "Keyboard interrupt from user" |
| |
| if __name__ == "__main__": |
| flags.DEFINE_boolean("status", None, "Print power meter status") |
| flags.DEFINE_integer("avg", None, |
| "Also report average over last n data points") |
| flags.DEFINE_float("voltage", None, "Set output voltage (0 for off)") |
| flags.DEFINE_float("current", None, "Set max output current") |
| flags.DEFINE_string("usbpassthrough", None, "USB control (on, off, auto)") |
| flags.DEFINE_integer("samples", None, "Collect and print this many samples") |
| flags.DEFINE_integer("hz", 5000, "Print this many samples/sec") |
| flags.DEFINE_string("device", None, |
| "Path to the device in /dev/... (ex:/dev/ttyACM1)") |
| flags.DEFINE_boolean("timestamp", None, |
| "Also print integer (seconds) timestamp on each line") |
| flags.DEFINE_boolean("ramp", True, "Gradually increase voltage") |
| flags.DEFINE_boolean("log", False, "Log progress to a file or not") |
| flags.DEFINE_boolean("run", False, "Run the test suite for power") |
| flags.DEFINE_string("power_monitor", None, "Type of power monitor to use") |
| flags.DEFINE_float("max_baseline_amps", 0.005, |
| "Set maximum baseline current for device being tested") |
| sys.exit(main(FLAGS(sys.argv))) |
