• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3.4
2#
3#   Copyright 2018 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the 'License');
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an 'AS IS' BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16import acts
17import json
18import logging
19import math
20import os
21import time
22import acts.controllers.iperf_server as ipf
23from acts import asserts
24from acts import base_test
25from acts import utils
26from acts.controllers import monsoon
27from acts.metrics.loggers.blackbox import BlackboxMetricLogger
28from acts.test_utils.wifi import wifi_test_utils as wutils
29from acts.test_utils.wifi import wifi_power_test_utils as wputils
30
31SETTINGS_PAGE = 'am start -n com.android.settings/.Settings'
32SCROLL_BOTTOM = 'input swipe 0 2000 0 0'
33UNLOCK_SCREEN = 'input keyevent 82'
34SET_BATTERY_LEVEL = 'dumpsys battery set level 100'
35SCREENON_USB_DISABLE = 'dumpsys battery unplug'
36RESET_BATTERY_STATS = 'dumpsys batterystats --reset'
37AOD_OFF = 'settings put secure doze_always_on 0'
38MUSIC_IQ_OFF = 'pm disable-user com.google.intelligence.sense'
39# Command to disable gestures
40LIFT = 'settings put secure doze_pulse_on_pick_up 0'
41DOUBLE_TAP = 'settings put secure doze_pulse_on_double_tap 0'
42JUMP_TO_CAMERA = 'settings put secure camera_double_tap_power_gesture_disabled 1'
43RAISE_TO_CAMERA = 'settings put secure camera_lift_trigger_enabled 0'
44FLIP_CAMERA = 'settings put secure camera_double_twist_to_flip_enabled 0'
45ASSIST_GESTURE = 'settings put secure assist_gesture_enabled 0'
46ASSIST_GESTURE_ALERT = 'settings put secure assist_gesture_silence_alerts_enabled 0'
47ASSIST_GESTURE_WAKE = 'settings put secure assist_gesture_wake_enabled 0'
48SYSTEM_NAVI = 'settings put secure system_navigation_keys_enabled 0'
49# End of command to disable gestures
50AUTO_TIME_OFF = 'settings put global auto_time 0'
51AUTO_TIMEZONE_OFF = 'settings put global auto_time_zone 0'
52FORCE_YOUTUBE_STOP = 'am force-stop com.google.android.youtube'
53FORCE_DIALER_STOP = 'am force-stop com.google.android.dialer'
54IPERF_TIMEOUT = 180
55THRESHOLD_TOLERANCE = 0.2
56GET_FROM_PHONE = 'get_from_dut'
57GET_FROM_AP = 'get_from_ap'
58PHONE_BATTERY_VOLTAGE = 4.2
59MONSOON_MAX_CURRENT = 8.0
60MONSOON_RETRY_INTERVAL = 300
61MEASUREMENT_RETRY_COUNT = 3
62RECOVER_MONSOON_RETRY_COUNT = 3
63MIN_PERCENT_SAMPLE = 95
64ENABLED_MODULATED_DTIM = 'gEnableModulatedDTIM='
65MAX_MODULATED_DTIM = 'gMaxLIModulatedDTIM='
66TEMP_FILE = '/sdcard/Download/tmp.log'
67IPERF_DURATION = 'iperf_duration'
68INITIAL_ATTEN = [0, 0, 90, 90]
69
70
71class ObjNew():
72    """Create a random obj with unknown attributes and value.
73
74    """
75
76    def __init__(self, **kwargs):
77        self.__dict__.update(kwargs)
78
79    def __contains__(self, item):
80        """Function to check if one attribute is contained in the object.
81
82        Args:
83            item: the item to check
84        Return:
85            True/False
86        """
87        return hasattr(self, item)
88
89
90class PowerBaseTest(base_test.BaseTestClass):
91    """Base class for all wireless power related tests.
92
93    """
94
95    def __init__(self, controllers):
96
97        base_test.BaseTestClass.__init__(self, controllers)
98        BlackboxMetricLogger.for_test_case(
99            metric_name='avg_power', result_attr='power_consumption')
100        self.start_meas_time = 0
101
102    def setup_class(self):
103
104        self.log = logging.getLogger()
105        self.tests = self._get_all_test_names()
106
107        # Setup the must have controllers, phone and monsoon
108        self.dut = self.android_devices[0]
109        self.mon_data_path = os.path.join(self.log_path, 'Monsoon')
110        self.mon = self.monsoons[0]
111        self.mon.set_max_current(8.0)
112        self.mon.set_voltage(PHONE_BATTERY_VOLTAGE)
113        self.mon.attach_device(self.dut)
114
115        # Unpack the test/device specific parameters
116        TEST_PARAMS = self.TAG + '_params'
117        req_params = [TEST_PARAMS, 'custom_files']
118        self.unpack_userparams(req_params)
119        # Unpack the custom files based on the test configs
120        for file in self.custom_files:
121            if 'pass_fail_threshold_' + self.dut.model in file:
122                self.threshold_file = file
123            elif 'attenuator_setting' in file:
124                self.attenuation_file = file
125            elif 'network_config' in file:
126                self.network_file = file
127
128        # Unpack test specific configs
129        self.unpack_testparams(getattr(self, TEST_PARAMS))
130        if hasattr(self, 'attenuators'):
131            self.num_atten = self.attenuators[0].instrument.num_atten
132            self.atten_level = self.unpack_custom_file(self.attenuation_file)
133            self.set_attenuation(INITIAL_ATTEN)
134        self.threshold = self.unpack_custom_file(self.threshold_file)
135        self.mon_info = self.create_monsoon_info()
136
137        # Onetime task for each test class
138        # Temporary fix for b/77873679
139        self.adb_disable_verity()
140        self.dut.adb.shell('mv /vendor/bin/chre /vendor/bin/chre_renamed')
141        self.dut.adb.shell('pkill chre')
142
143    def setup_test(self):
144        """Set up test specific parameters or configs.
145
146        """
147        # Reset the power consumption to 0 before each tests
148        self.power_consumption = 0
149        # Set the device into rockbottom state
150        self.dut_rockbottom()
151        # Wait for extra time if needed for the first test
152        if hasattr(self, 'extra_wait'):
153            self.more_wait_first_test()
154
155    def teardown_test(self):
156        """Tear down necessary objects after test case is finished.
157
158        """
159        self.log.info('Tearing down the test case')
160        self.mon.usb('on')
161
162    def teardown_class(self):
163        """Clean up the test class after tests finish running
164
165        """
166        self.log.info('Tearing down the test class')
167        self.mon.usb('on')
168
169    def unpack_testparams(self, bulk_params):
170        """Unpack all the test specific parameters.
171
172        Args:
173            bulk_params: dict with all test specific params in the config file
174        """
175        for key in bulk_params.keys():
176            setattr(self, key, bulk_params[key])
177
178    def unpack_custom_file(self, file, test_specific=True):
179        """Unpack the pass_fail_thresholds from a common file.
180
181        Args:
182            file: the common file containing pass fail threshold.
183        """
184        with open(file, 'r') as f:
185            params = json.load(f)
186        if test_specific:
187            try:
188                return params[self.TAG]
189            except KeyError:
190                pass
191        else:
192            return params
193
194    def decode_test_configs(self, attrs, indices):
195        """Decode the test config/params from test name.
196
197        Remove redundant function calls when tests are similar.
198        Args:
199            attrs: a list of the attrs of the test config obj
200            indices: a list of the location indices of keyword in the test name.
201        """
202        # Decode test parameters for the current test
203        test_params = self.current_test_name.split('_')
204        values = [test_params[x] for x in indices]
205        config_dict = dict(zip(attrs, values))
206        self.test_configs = ObjNew(**config_dict)
207
208    def more_wait_first_test(self):
209        # For the first test, increase the offset for longer wait time
210        if self.current_test_name == self.tests[0]:
211            self.mon_info.offset = self.mon_offset + self.extra_wait
212        else:
213            self.mon_info.offset = self.mon_offset
214
215    def set_attenuation(self, atten_list):
216        """Function to set the attenuator to desired attenuations.
217
218        Args:
219            atten_list: list containing the attenuation for each attenuator.
220        """
221        if len(atten_list) != self.num_atten:
222            raise Exception('List given does not have the correct length')
223        for i in range(self.num_atten):
224            self.attenuators[i].set_atten(atten_list[i])
225
226    def dut_rockbottom(self):
227        """Set the phone into Rock-bottom state.
228
229        """
230        self.dut.log.info('Now set the device to Rockbottom State')
231        utils.require_sl4a((self.dut, ))
232        self.dut.droid.connectivityToggleAirplaneMode(False)
233        time.sleep(2)
234        self.dut.droid.connectivityToggleAirplaneMode(True)
235        time.sleep(2)
236        utils.set_ambient_display(self.dut, False)
237        utils.set_auto_rotate(self.dut, False)
238        utils.set_adaptive_brightness(self.dut, False)
239        utils.sync_device_time(self.dut)
240        utils.set_location_service(self.dut, False)
241        utils.set_mobile_data_always_on(self.dut, False)
242        utils.disable_doze_light(self.dut)
243        utils.disable_doze(self.dut)
244        wutils.reset_wifi(self.dut)
245        wutils.wifi_toggle_state(self.dut, False)
246        try:
247            self.dut.droid.nfcDisable()
248        except acts.controllers.sl4a_lib.rpc_client.Sl4aApiError:
249            self.dut.log.info('NFC is not available')
250        self.dut.droid.setScreenBrightness(0)
251        self.dut.adb.shell(AOD_OFF)
252        self.dut.droid.setScreenTimeout(2200)
253        self.dut.droid.wakeUpNow()
254        self.dut.adb.shell(LIFT)
255        self.dut.adb.shell(DOUBLE_TAP)
256        self.dut.adb.shell(JUMP_TO_CAMERA)
257        self.dut.adb.shell(RAISE_TO_CAMERA)
258        self.dut.adb.shell(FLIP_CAMERA)
259        self.dut.adb.shell(ASSIST_GESTURE)
260        self.dut.adb.shell(ASSIST_GESTURE_ALERT)
261        self.dut.adb.shell(ASSIST_GESTURE_WAKE)
262        self.dut.adb.shell(SET_BATTERY_LEVEL)
263        self.dut.adb.shell(SCREENON_USB_DISABLE)
264        self.dut.adb.shell(UNLOCK_SCREEN)
265        self.dut.adb.shell(SETTINGS_PAGE)
266        self.dut.adb.shell(SCROLL_BOTTOM)
267        self.dut.adb.shell(MUSIC_IQ_OFF)
268        self.dut.adb.shell(AUTO_TIME_OFF)
269        self.dut.adb.shell(AUTO_TIMEZONE_OFF)
270        self.dut.adb.shell(FORCE_YOUTUBE_STOP)
271        self.dut.adb.shell(FORCE_DIALER_STOP)
272        self.dut.droid.wifiSetCountryCode('US')
273        self.dut.droid.wakeUpNow()
274        self.dut.log.info('Device has been set to Rockbottom state')
275        self.dut.log.info('Screen is ON')
276
277    def measure_power_and_validate(self):
278        """The actual test flow and result processing and validate.
279
280        """
281        self.collect_power_data()
282        self.pass_fail_check()
283
284    def collect_power_data(self):
285        """Measure power, plot and take log if needed.
286
287        """
288        tag = ''
289        # Collecting current measurement data and plot
290        begin_time = utils.get_current_epoch_time()
291        self.file_path, self.test_result = self.monsoon_data_collect_save()
292        self.power_consumption = self.test_result * PHONE_BATTERY_VOLTAGE
293        wputils.monsoon_data_plot(self.mon_info, self.file_path, tag=tag)
294        # Take Bugreport
295        if self.bug_report:
296            self.dut.take_bug_report(self.test_name, begin_time)
297
298    def pass_fail_check(self):
299        """Check the test result and decide if it passed or failed.
300
301        The threshold is provided in the config file. In this class, result is
302        current in mA.
303        """
304
305        if not self.threshold or self.test_name not in self.threshold:
306            self.log.error("No threshold is provided for the test '{}' in "
307                           "the configuration file.".format(self.test_name))
308            return
309
310        current_threshold = self.threshold[self.test_name]
311        if self.test_result:
312            asserts.assert_true(
313                abs(self.test_result - current_threshold) / current_threshold <
314                THRESHOLD_TOLERANCE,
315                ('Measured average current in [{}]: {}, which is '
316                 'more than {} percent off than acceptable threshold {:.2f}mA'
317                 ).format(self.test_name, self.test_result,
318                          self.pass_fail_tolerance * 100, current_threshold))
319            asserts.explicit_pass('Measurement finished for {}.'.format(
320                self.test_name))
321        else:
322            asserts.fail(
323                'Something happened, measurement is not complete, test failed')
324
325    def create_monsoon_info(self):
326        """Creates the config dictionary for monsoon
327
328        Returns:
329            mon_info: Dictionary with the monsoon packet config
330        """
331        if hasattr(self, IPERF_DURATION):
332            self.mon_duration = self.iperf_duration - 10
333        mon_info = ObjNew(
334            dut=self.mon,
335            freq=self.mon_freq,
336            duration=self.mon_duration,
337            offset=self.mon_offset,
338            data_path=self.mon_data_path)
339        return mon_info
340
341    def monsoon_recover(self):
342        """Test loop to wait for monsoon recover from unexpected error.
343
344        Wait for a certain time duration, then quit.0
345        Args:
346            mon: monsoon object
347        Returns:
348            True/False
349        """
350        try:
351            self.mon.reconnect_monsoon()
352            time.sleep(2)
353            self.mon.usb('on')
354            logging.info('Monsoon recovered from unexpected error')
355            time.sleep(2)
356            return True
357        except monsoon.MonsoonError:
358            logging.info(self.mon.mon.ser.in_waiting)
359            logging.warning('Unable to recover monsoon from unexpected error')
360            return False
361
362    def monsoon_data_collect_save(self):
363        """Current measurement and save the log file.
364
365        Collect current data using Monsoon box and return the path of the
366        log file. Take bug report if requested.
367
368        Returns:
369            data_path: the absolute path to the log file of monsoon current
370                       measurement
371            avg_current: the average current of the test
372        """
373
374        tag = '{}_{}_{}'.format(self.test_name, self.dut.model,
375                                self.dut.build_info['build_id'])
376        data_path = os.path.join(self.mon_info.data_path, '{}.txt'.format(tag))
377        total_expected_samples = self.mon_info.freq * (
378            self.mon_info.duration + self.mon_info.offset)
379        min_required_samples = total_expected_samples * MIN_PERCENT_SAMPLE / 100
380        # Retry counter for monsoon data aquisition
381        retry_measure = 1
382        # Indicator that need to re-collect data
383        need_collect_data = 1
384        result = None
385        while retry_measure <= MEASUREMENT_RETRY_COUNT:
386            try:
387                # If need to retake data
388                if need_collect_data == 1:
389                    #Resets the battery status right before the test started
390                    self.dut.adb.shell(RESET_BATTERY_STATS)
391                    self.log.info(
392                        'Starting power measurement with monsoon box, try #{}'.
393                        format(retry_measure))
394                    #Start the power measurement using monsoon
395                    self.mon_info.dut.monsoon_usb_auto()
396                    result = self.mon_info.dut.measure_power(
397                        self.mon_info.freq,
398                        self.mon_info.duration,
399                        tag=tag,
400                        offset=self.mon_info.offset)
401                    self.mon_info.dut.reconnect_dut()
402                # Reconnect to dut
403                else:
404                    self.mon_info.dut.reconnect_dut()
405                # Reconnect and return measurement results if no error happens
406                avg_current = result.average_current
407                monsoon.MonsoonData.save_to_text_file([result], data_path)
408                self.log.info('Power measurement done within {} try'.format(
409                    retry_measure))
410                return data_path, avg_current
411            # Catch monsoon errors during measurement
412            except monsoon.MonsoonError:
413                self.log.info(self.mon_info.dut.mon.ser.in_waiting)
414                # Break early if it's one count away from limit
415                if retry_measure == MEASUREMENT_RETRY_COUNT:
416                    self.log.error(
417                        'Test failed after maximum measurement retry')
418                    break
419
420                self.log.warning('Monsoon error happened, now try to recover')
421                # Retry loop to recover monsoon from error
422                retry_monsoon = 1
423                while retry_monsoon <= RECOVER_MONSOON_RETRY_COUNT:
424                    mon_status = self.monsoon_recover()
425                    if mon_status:
426                        break
427                    else:
428                        retry_monsoon += 1
429                        self.log.warning(
430                            'Wait for {} second then try again'.format(
431                                MONSOON_RETRY_INTERVAL))
432                        time.sleep(MONSOON_RETRY_INTERVAL)
433
434                # Break the loop to end test if failed to recover monsoon
435                if not mon_status:
436                    self.log.error(
437                        'Tried our best, still failed to recover monsoon')
438                    break
439                else:
440                    # If there is no data, or captured samples are less than min
441                    # required, re-take
442                    if not result:
443                        self.log.warning('No data taken, need to remeasure')
444                    elif len(result._data_points) <= min_required_samples:
445                        self.log.warning(
446                            'More than {} percent of samples are missing due to monsoon error. Need to remeasure'.
447                            format(100 - MIN_PERCENT_SAMPLE))
448                    else:
449                        need_collect_data = 0
450                        self.log.warning(
451                            'Data collected is valid, try reconnect to DUT to finish test'
452                        )
453                    retry_measure += 1
454
455        if retry_measure > MEASUREMENT_RETRY_COUNT:
456            self.log.error('Test failed after maximum measurement retry')
457
458    def setup_ap_connection(self, network, bandwidth=80, connect=True,
459                            ap=None):
460        """Setup AP and connect DUT to it.
461
462        Args:
463            network: the network config for the AP to be setup
464            bandwidth: bandwidth of the WiFi network to be setup
465            connect: indicator of if connect dut to the network after setup
466            ap: access point object, default is None to find the main AP
467        Returns:
468            self.brconfigs: dict for bridge interface configs
469        """
470        wutils.wifi_toggle_state(self.dut, True)
471        if not ap:
472            if hasattr(self, 'access_points'):
473                self.brconfigs = wputils.ap_setup(
474                    self.access_point, network, bandwidth=bandwidth)
475        else:
476            self.brconfigs = wputils.ap_setup(ap, network, bandwidth=bandwidth)
477        if connect:
478            wutils.wifi_connect(self.dut, network, num_of_tries=3)
479
480        if ap or (not ap and hasattr(self, 'access_points')):
481            return self.brconfigs
482
483    def process_iperf_results(self):
484        """Get the iperf results and process.
485
486        Returns:
487             throughput: the average throughput during tests.
488        """
489        # Get IPERF results and add this to the plot title
490        RESULTS_DESTINATION = os.path.join(self.iperf_server.log_path,
491                                           'iperf_client_output_{}.log'.format(
492                                               self.current_test_name))
493        PULL_FILE = '{} {}'.format(TEMP_FILE, RESULTS_DESTINATION)
494        self.dut.adb.pull(PULL_FILE)
495        # Calculate the average throughput
496        if self.use_client_output:
497            iperf_file = RESULTS_DESTINATION
498        else:
499            iperf_file = self.iperf_server.log_files[-1]
500        try:
501            iperf_result = ipf.IPerfResult(iperf_file)
502
503            # Compute the throughput in Mbit/s
504            throughput = (math.fsum(
505                iperf_result.instantaneous_rates[self.start_meas_time:-1]
506            ) / len(iperf_result.instantaneous_rates[self.start_meas_time:-1])
507                          ) * 8 * (1.024**2)
508
509            self.log.info('The average throughput is {}'.format(throughput))
510        except ValueError:
511            self.log.warning('Cannot get iperf result. Setting to 0')
512            throughput = 0
513        return throughput
514
515    # TODO(@qijiang)Merge with tel_test_utils.py
516    def adb_disable_verity(self):
517        """Disable verity on the device.
518
519        """
520        if self.dut.adb.getprop("ro.boot.veritymode") == "enforcing":
521            self.dut.adb.disable_verity()
522            self.dut.reboot()
523            self.dut.adb.root()
524            self.dut.adb.remount()
525            self.dut.adb.shell(SET_BATTERY_LEVEL)
526