acts/framework/acts/test_utils/wifi/wifi_power_test_utils.py

#!/usr/bin/env python3.4
#
#   Copyright 2017 Google, Inc.
#
#   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 acts
import json
import logging
import os
import time
from acts import asserts
from acts import utils
from acts.controllers import monsoon
from acts.libs.proc import job
from acts.controllers.ap_lib import bridge_interface as bi
from acts.test_utils.wifi import wifi_test_utils as wutils
from bokeh.layouts import layout
from bokeh.models import CustomJS, ColumnDataSource
from bokeh.models import tools as bokeh_tools
from bokeh.models.widgets import DataTable, TableColumn
from bokeh.plotting import figure, output_file, save
from acts.controllers.ap_lib import hostapd_security
from acts.controllers.ap_lib import hostapd_ap_preset
from acts.test_utils.bt.bt_test_utils import enable_bluetooth
from acts.test_utils.bt.bt_test_utils import disable_bluetooth

# http://www.secdev.org/projects/scapy/
# On ubuntu, sudo pip3 install scapy-python3
import scapy.all as scapy

SETTINGS_PAGE = "am start -n com.android.settings/.Settings"
SCROLL_BOTTOM = "input swipe 0 2000 0 0"
UNLOCK_SCREEN = "input keyevent 82"
SCREENON_USB_DISABLE = "dumpsys battery unplug"
RESET_BATTERY_STATS = "dumpsys batterystats --reset"
AOD_OFF = "settings put secure doze_always_on 0"
MUSIC_IQ_OFF = "pm disable-user com.google.intelligence.sense"
# Command to disable gestures
LIFT = "settings put secure doze_pulse_on_pick_up 0"
DOUBLE_TAP = "settings put secure doze_pulse_on_double_tap 0"
JUMP_TO_CAMERA = "settings put secure camera_double_tap_power_gesture_disabled 1"
RAISE_TO_CAMERA = "settings put secure camera_lift_trigger_enabled 0"
FLIP_CAMERA = "settings put secure camera_double_twist_to_flip_enabled 0"
ASSIST_GESTURE = "settings put secure assist_gesture_enabled 0"
ASSIST_GESTURE_ALERT = "settings put secure assist_gesture_silence_alerts_enabled 0"
ASSIST_GESTURE_WAKE = "settings put secure assist_gesture_wake_enabled 0"
SYSTEM_NAVI = "settings put secure system_navigation_keys_enabled 0"
# End of command to disable gestures
AUTO_TIME_OFF = "settings put global auto_time 0"
AUTO_TIMEZONE_OFF = "settings put global auto_time_zone 0"
FORCE_YOUTUBE_STOP = "am force-stop com.google.android.youtube"
FORCE_DIALER_STOP = "am force-stop com.google.android.dialer"
IPERF_TIMEOUT = 180
THRESHOLD_TOLERANCE = 0.2
GET_FROM_PHONE = 'get_from_dut'
GET_FROM_AP = 'get_from_ap'
PHONE_BATTERY_VOLTAGE = 4.2
MONSOON_MAX_CURRENT = 8.0
MONSOON_RETRY_INTERVAL = 300
MEASUREMENT_RETRY_COUNT = 3
RECOVER_MONSOON_RETRY_COUNT = 3
MIN_PERCENT_SAMPLE = 95
ENABLED_MODULATED_DTIM = 'gEnableModulatedDTIM='
MAX_MODULATED_DTIM = 'gMaxLIModulatedDTIM='


def dut_rockbottom(ad):
    """Set the phone into Rock-bottom state.

    Args:
        ad: the target android device, AndroidDevice object

    """
    ad.log.info("Now set the device to Rockbottom State")
    utils.require_sl4a((ad, ))
    ad.droid.connectivityToggleAirplaneMode(False)
    time.sleep(5)
    ad.droid.connectivityToggleAirplaneMode(True)
    utils.set_ambient_display(ad, False)
    utils.set_auto_rotate(ad, False)
    utils.set_adaptive_brightness(ad, False)
    utils.sync_device_time(ad)
    utils.set_location_service(ad, False)
    utils.set_mobile_data_always_on(ad, False)
    utils.disable_doze_light(ad)
    utils.disable_doze(ad)
    wutils.reset_wifi(ad)
    wutils.wifi_toggle_state(ad, False)
    try:
        ad.droid.nfcDisable()
    except acts.controllers.sl4a_lib.rpc_client.Sl4aApiError:
        ad.log.info('NFC is not available')
    ad.droid.setScreenBrightness(0)
    ad.adb.shell(AOD_OFF)
    ad.droid.setScreenTimeout(2200)
    ad.droid.wakeUpNow()
    ad.adb.shell(LIFT)
    ad.adb.shell(DOUBLE_TAP)
    ad.adb.shell(JUMP_TO_CAMERA)
    ad.adb.shell(RAISE_TO_CAMERA)
    ad.adb.shell(FLIP_CAMERA)
    ad.adb.shell(ASSIST_GESTURE)
    ad.adb.shell(ASSIST_GESTURE_ALERT)
    ad.adb.shell(ASSIST_GESTURE_WAKE)
    ad.adb.shell(SCREENON_USB_DISABLE)
    ad.adb.shell(UNLOCK_SCREEN)
    ad.adb.shell(SETTINGS_PAGE)
    ad.adb.shell(SCROLL_BOTTOM)
    ad.adb.shell(MUSIC_IQ_OFF)
    ad.adb.shell(AUTO_TIME_OFF)
    ad.adb.shell(AUTO_TIMEZONE_OFF)
    ad.adb.shell(FORCE_YOUTUBE_STOP)
    ad.adb.shell(FORCE_DIALER_STOP)
    ad.droid.wifiSetCountryCode('US')
    ad.droid.wakeUpNow()
    ad.log.info('Device has been set to Rockbottom state')
    ad.log.info('Screen is ON')


def unpack_custom_file(file, test_class=None):
    """Unpack the pass_fail_thresholds from a common file.

    Args:
        file: the common file containing pass fail threshold.
    """
    with open(file, 'r') as f:
        params = json.load(f)
    if test_class:
        return params[test_class]
    else:
        return params


def pass_fail_check(test_class, test_result):
    """Check the test result and decide if it passed or failed.
    The threshold is provided in the config file

    Args:
        test_class: the specific test class where test is running
        test_result: the average current as the test result
    """
    test_name = test_class.current_test_name
    current_threshold = test_class.threshold[test_name]
    if test_result:
        asserts.assert_true(
            abs(test_result - current_threshold) / current_threshold <
            THRESHOLD_TOLERANCE,
            ("Measured average current in [%s]: %s, which is "
             "more than %d percent off than acceptable threshold %.2fmA") %
            (test_name, test_result, test_class.pass_fail_tolerance * 100,
             current_threshold))
        asserts.explicit_pass("Measurement finished for %s." % test_name)
    else:
        asserts.fail(
            "Something happened, measurement is not complete, test failed")


def monsoon_recover(mon):
    """Test loop to wait for monsoon recover from unexpected error.

    Wait for a certain time duration, then quit.0
    Args:
        mon: monsoon object
    """
    try:
        mon.reconnect_monsoon()
        time.sleep(2)
        mon.usb('on')
        logging.info("Monsoon recovered from unexpected error")
        time.sleep(2)
        return True
    except monsoon.MonsoonError:
        logging.info(mon.mon.ser.in_waiting)
        logging.warning("Unable to recover monsoon from unexpected error")
        return False


def monsoon_data_collect_save(ad, mon_info, test_name):
    """Current measurement and save the log file.

    Collect current data using Monsoon box and return the path of the
    log file. Take bug report if requested.

    Args:
        ad: the android device under test
        mon_info: dict with information of monsoon measurement, including
                  monsoon device object, measurement frequency, duration and
                  offset etc.
        test_name: current test name, used to contruct the result file name
        bug_report: indicator to take bug report or not, 0 or 1
    Returns:
        data_path: the absolute path to the log file of monsoon current
                   measurement
        avg_current: the average current of the test
    """

    log = logging.getLogger()
    tag = (test_name + '_' + ad.model + '_' + ad.build_info['build_id'])
    data_path = os.path.join(mon_info['data_path'], "%s.txt" % tag)
    total_expected_samples = mon_info['freq'] * (
        mon_info['duration'] + mon_info['offset'])
    min_required_samples = total_expected_samples * MIN_PERCENT_SAMPLE / 100
    # Retry counter for monsoon data aquisition
    retry_measure = 1
    # Indicator that need to re-collect data
    need_collect_data = 1
    result = None
    while retry_measure <= MEASUREMENT_RETRY_COUNT:
        try:
            # If need to retake data
            if need_collect_data == 1:
                #Resets the battery status right before the test started
                ad.adb.shell(RESET_BATTERY_STATS)
                log.info(
                    "Starting power measurement with monsoon box, try #{}".
                    format(retry_measure))
                #Start the power measurement using monsoon
                mon_info['dut'].monsoon_usb_auto()
                result = mon_info['dut'].measure_power(
                    mon_info['freq'],
                    mon_info['duration'],
                    tag=tag,
                    offset=mon_info['offset'])
                mon_info['dut'].reconnect_dut()
            # Reconnect to dut
            else:
                mon_info['dut'].reconnect_dut()
            # Reconnect and return measurement results if no error happens
            avg_current = result.average_current
            monsoon.MonsoonData.save_to_text_file([result], data_path)
            log.info(
                "Power measurement done within {} try".format(retry_measure))
            return data_path, avg_current
        # Catch monsoon errors during measurement
        except monsoon.MonsoonError:
            log.info(mon_info['dut'].mon.ser.in_waiting)
            # Break early if it's one count away from limit
            if retry_measure == MEASUREMENT_RETRY_COUNT:
                log.error('Test failed after maximum measurement retry')
                break

            log.warning('Monsoon error happened, now try to recover')
            # Retry loop to recover monsoon from error
            retry_monsoon = 1
            while retry_monsoon <= RECOVER_MONSOON_RETRY_COUNT:
                mon_status = monsoon_recover(mon_info['dut'])
                if mon_status:
                    break
                else:
                    retry_monsoon += 1
                    log.warning('Wait for {} second then try again'.format(
                        MONSOON_RETRY_INTERVAL))
                    time.sleep(MONSOON_RETRY_INTERVAL)

            # Break the loop to end test if failed to recover monsoon
            if not mon_status:
                log.error('Tried our best, still failed to recover monsoon')
                break
            else:
                # If there is no data or captured samples are less than min
                # required, re-take
                if not result:
                    log.warning('No data taken, need to remeasure')
                elif len(result._data_points) <= min_required_samples:
                    log.warning(
                        'More than {} percent of samples are missing due to monsoon error. Need to remeasure'.
                        format(100 - MIN_PERCENT_SAMPLE))
                else:
                    need_collect_data = 0
                    log.warning(
                        'Data collected is valid, try reconnect to DUT to finish test'
                    )
                retry_measure += 1

    if retry_measure > MEASUREMENT_RETRY_COUNT:
        log.error('Test failed after maximum measurement retry')


def monsoon_data_plot(mon_info, file_path, tag=""):
    """Plot the monsoon current data using bokeh interactive plotting tool.

    Plotting power measurement data with bokeh to generate interactive plots.
    You can do interactive data analysis on the plot after generating with the
    provided widgets, which make the debugging much easier. To realize that,
    bokeh callback java scripting is used. View a sample html output file:
    https://drive.google.com/open?id=0Bwp8Cq841VnpT2dGUUxLYWZvVjA

    Args:
        mon_info: dict with information of monsoon measurement, including
                  monsoon device object, measurement frequency, duration and
                  offset etc.
        file_path: the path to the monsoon log file with current data

    Returns:
        plot: the plotting object of bokeh, optional, will be needed if multiple
           plots will be combined to one html file.
        dt: the datatable object of bokeh, optional, will be needed if multiple
           datatables will be combined to one html file.
    """

    log = logging.getLogger()
    log.info("Plot the power measurement data")
    #Get results as monsoon data object from the input file
    results = monsoon.MonsoonData.from_text_file(file_path)
    #Decouple current and timestamp data from the monsoon object
    current_data = []
    timestamps = []
    voltage = results[0].voltage
    [current_data.extend(x.data_points) for x in results]
    [timestamps.extend(x.timestamps) for x in results]
    period = 1 / float(mon_info['freq'])
    time_relative = [x * period for x in range(len(current_data))]
    #Calculate the average current for the test
    current_data = [x * 1000 for x in current_data]
    avg_current = sum(current_data) / len(current_data)
    color = ['navy'] * len(current_data)

    #Preparing the data and source link for bokehn java callback
    source = ColumnDataSource(
        data=dict(x0=time_relative, y0=current_data, color=color))
    s2 = ColumnDataSource(
        data=dict(
            z0=[mon_info['duration']],
            y0=[round(avg_current, 2)],
            x0=[round(avg_current * voltage, 2)],
            z1=[round(avg_current * voltage * mon_info['duration'], 2)],
            z2=[round(avg_current * mon_info['duration'], 2)]))
    #Setting up data table for the output
    columns = [
        TableColumn(field='z0', title='Total Duration (s)'),
        TableColumn(field='y0', title='Average Current (mA)'),
        TableColumn(field='x0', title='Average Power (4.2v) (mW)'),
        TableColumn(field='z1', title='Average Energy (mW*s)'),
        TableColumn(field='z2', title='Normalized Average Energy (mA*s)')
    ]
    dt = DataTable(
        source=s2, columns=columns, width=1300, height=60, editable=True)

    plot_title = file_path[file_path.rfind('/') + 1:-4] + tag
    output_file("%s/%s.html" % (mon_info['data_path'], plot_title))
    TOOLS = ('box_zoom,box_select,pan,crosshair,redo,undo,reset,hover,save')
    # Create a new plot with the datatable above
    plot = figure(
        plot_width=1300,
        plot_height=700,
        title=plot_title,
        tools=TOOLS,
        output_backend="webgl")
    plot.add_tools(bokeh_tools.WheelZoomTool(dimensions="width"))
    plot.add_tools(bokeh_tools.WheelZoomTool(dimensions="height"))
    plot.line('x0', 'y0', source=source, line_width=2)
    plot.circle('x0', 'y0', source=source, size=0.5, fill_color='color')
    plot.xaxis.axis_label = 'Time (s)'
    plot.yaxis.axis_label = 'Current (mA)'
    plot.title.text_font_size = {'value': '15pt'}

    #Callback Java scripting
    source.callback = CustomJS(
        args=dict(mytable=dt),
        code="""
    var inds = cb_obj.get('selected')['1d'].indices;
    var d1 = cb_obj.get('data');
    var d2 = mytable.get('source').get('data');
    ym = 0
    ts = 0
    d2['x0'] = []
    d2['y0'] = []
    d2['z1'] = []
    d2['z2'] = []
    d2['z0'] = []
    min=max=d1['x0'][inds[0]]
    if (inds.length==0) {return;}
    for (i = 0; i < inds.length; i++) {
    ym += d1['y0'][inds[i]]
    d1['color'][inds[i]] = "red"
    if (d1['x0'][inds[i]] < min) {
      min = d1['x0'][inds[i]]}
    if (d1['x0'][inds[i]] > max) {
      max = d1['x0'][inds[i]]}
    }
    ym /= inds.length
    ts = max - min
    dx0 = Math.round(ym*4.2*100.0)/100.0
    dy0 = Math.round(ym*100.0)/100.0
    dz1 = Math.round(ym*4.2*ts*100.0)/100.0
    dz2 = Math.round(ym*ts*100.0)/100.0
    dz0 = Math.round(ts*1000.0)/1000.0
    d2['z0'].push(dz0)
    d2['x0'].push(dx0)
    d2['y0'].push(dy0)
    d2['z1'].push(dz1)
    d2['z2'].push(dz2)
    mytable.trigger('change');
    """)

    #Layout the plot and the datatable bar
    l = layout([[dt], [plot]])
    save(l)
    return [plot, dt]


def change_dtim(ad, gEnableModulatedDTIM, gMaxLIModulatedDTIM=10):
    """Function to change the DTIM setting in the phone.

    Args:
        ad: the target android device, AndroidDevice object
        gEnableModulatedDTIM: Modulated DTIM, int
        gMaxLIModulatedDTIM: Maximum modulated DTIM, int
    """
    # First trying to find the ini file with DTIM settings
    ini_file_phone = ad.adb.shell('ls /vendor/firmware/wlan/*/*.ini')
    ini_file_local = ini_file_phone.split('/')[-1]

    # Pull the file and change the DTIM to desired value
    ad.adb.pull('{} {}'.format(ini_file_phone, ini_file_local))

    with open(ini_file_local, 'r') as fin:
        for line in fin:
            if ENABLED_MODULATED_DTIM in line:
                gE_old = line.strip('\n')
                gEDTIM_old = line.strip(ENABLED_MODULATED_DTIM).strip('\n')
            if MAX_MODULATED_DTIM in line:
                gM_old = line.strip('\n')
                gMDTIM_old = line.strip(MAX_MODULATED_DTIM).strip('\n')
    fin.close()
    if int(gEDTIM_old) == gEnableModulatedDTIM and int(
            gMDTIM_old) == gMaxLIModulatedDTIM:
        ad.log.info('Current DTIM is already the desired value,'
                    'no need to reset it')
        return

    gE_new = ENABLED_MODULATED_DTIM + str(gEnableModulatedDTIM)
    gM_new = MAX_MODULATED_DTIM + str(gMaxLIModulatedDTIM)

    sed_gE = 'sed -i \'s/{}/{}/g\' {}'.format(gE_old, gE_new, ini_file_local)
    sed_gM = 'sed -i \'s/{}/{}/g\' {}'.format(gM_old, gM_new, ini_file_local)
    job.run(sed_gE)
    job.run(sed_gM)

    # Push the file to the phone
    push_file_to_phone(ad, ini_file_local, ini_file_phone)
    ad.log.info('DTIM changes checked in and rebooting...')
    ad.reboot()
    ad.log.info('DTIM updated and device back from reboot')


def push_file_to_phone(ad, file_local, file_phone):
    """Function to push local file to android phone.

    Args:
        ad: the target android device
        file_local: the locla file to push
        file_phone: the file/directory on the phone to be pushed
    """
    ad.adb.root()
    cmd_out = ad.adb.remount()
    if 'Permission denied' in cmd_out:
        ad.log.info('Need to disable verity first and reboot')
        ad.adb.disable_verity()
        time.sleep(1)
        ad.reboot()
        ad.log.info('Verity disabled and device back from reboot')
        ad.adb.root()
        ad.adb.remount()
    time.sleep(1)
    ad.adb.push('{} {}'.format(file_local, file_phone))


def ap_setup(ap, network, bandwidth=80):
    """Set up the whirlwind AP with provided network info.

    Args:
        ap: access_point object of the AP
        network: dict with information of the network, including ssid, password
                 bssid, channel etc.
        bandwidth: the operation bandwidth for the AP, default 80MHz
    Returns:
        brconfigs: the bridge interface configs
    """
    log = logging.getLogger()
    bss_settings = []
    ssid = network[wutils.WifiEnums.SSID_KEY]
    if "password" in network.keys():
        password = network["password"]
        security = hostapd_security.Security(
            security_mode="wpa", password=password)
    else:
        security = hostapd_security.Security(security_mode=None, password=None)
    channel = network["channel"]
    config = hostapd_ap_preset.create_ap_preset(
        channel=channel,
        ssid=ssid,
        security=security,
        bss_settings=bss_settings,
        vht_bandwidth=bandwidth,
        profile_name='whirlwind',
        iface_wlan_2g=ap.wlan_2g,
        iface_wlan_5g=ap.wlan_5g)
    config_bridge = ap.generate_bridge_configs(channel)
    brconfigs = bi.BridgeInterfaceConfigs(config_bridge[0], config_bridge[1],
                                          config_bridge[2])
    ap.bridge.startup(brconfigs)
    ap.start_ap(config)
    log.info("AP started on channel {} with SSID {}".format(channel, ssid))
    return brconfigs


def bokeh_plot(data_sets,
               legends,
               fig_property,
               shaded_region=None,
               output_file_path=None):
    """Plot bokeh figs.
        Args:
            data_sets: data sets including lists of x_data and lists of y_data
                       ex: [[[x_data1], [x_data2]], [[y_data1],[y_data2]]]
            legends: list of legend for each curve
            fig_property: dict containing the plot property, including title,
                      lables, linewidth, circle size, etc.
            shaded_region: optional dict containing data for plot shading
            output_file_path: optional path at which to save figure
        Returns:
            plot: bokeh plot figure object
    """
    TOOLS = ('box_zoom,box_select,pan,crosshair,redo,undo,reset,hover,save')
    plot = figure(
        plot_width=1300,
        plot_height=700,
        title=fig_property['title'],
        tools=TOOLS,
        output_backend="webgl")
    plot.add_tools(bokeh_tools.WheelZoomTool(dimensions="width"))
    plot.add_tools(bokeh_tools.WheelZoomTool(dimensions="height"))
    colors = [
        'red', 'green', 'blue', 'olive', 'orange', 'salmon', 'black', 'navy',
        'yellow', 'darkred', 'goldenrod'
    ]
    if shaded_region:
        band_x = shaded_region["x_vector"]
        band_x.extend(shaded_region["x_vector"][::-1])
        band_y = shaded_region["lower_limit"]
        band_y.extend(shaded_region["upper_limit"][::-1])
        plot.patch(
            band_x, band_y, color='#7570B3', line_alpha=0.1, fill_alpha=0.1)

    for x_data, y_data, legend in zip(data_sets[0], data_sets[1], legends):
        index_now = legends.index(legend)
        color = colors[index_now % len(colors)]
        plot.line(
            x_data, y_data, legend=str(legend), line_width=3, color=color)
        plot.circle(
            x_data, y_data, size=10, legend=str(legend), fill_color=color)

    #Plot properties
    plot.xaxis.axis_label = fig_property['x_label']
    plot.yaxis.axis_label = fig_property['y_label']
    plot.legend.location = "top_right"
    plot.legend.click_policy = "hide"
    plot.title.text_font_size = {'value': '15pt'}
    if output_file_path is not None:
        output_file(output_file_path)
        save(plot)
    return plot


def run_iperf_client_nonblocking(ad, server_host, extra_args=""):
    """Start iperf client on the device with nohup.

    Return status as true if iperf client start successfully.
    And data flow information as results.

    Args:
        ad: the android device under test
        server_host: Address of the iperf server.
        extra_args: A string representing extra arguments for iperf client,
            e.g. "-i 1 -t 30".

    """
    log = logging.getLogger()
    ad.adb.shell_nb("nohup iperf3 -c {} {} &".format(server_host, extra_args))
    log.info("IPerf client started")


def get_wifi_rssi(ad):
    """Get the RSSI of the device.

    Args:
        ad: the android device under test
    Returns:
        RSSI: the rssi level of the device
    """
    RSSI = ad.droid.wifiGetConnectionInfo()['rssi']
    return RSSI


def get_phone_ip(ad):
    """Get the WiFi IP address of the phone.

    Args:
        ad: the android device under test
    Returns:
        IP: IP address of the phone for WiFi, as a string
    """
    IP = ad.droid.connectivityGetIPv4Addresses('wlan0')[0]

    return IP


def get_phone_mac(ad):
    """Get the WiFi MAC address of the phone.

    Args:
        ad: the android device under test
    Returns:
        mac: MAC address of the phone for WiFi, as a string
    """
    mac = ad.droid.wifiGetConnectionInfo()["mac_address"]

    return mac


def get_phone_ipv6(ad):
    """Get the WiFi IPV6 address of the phone.

    Args:
        ad: the android device under test
    Returns:
        IPv6: IPv6 address of the phone for WiFi, as a string
    """
    IPv6 = ad.droid.connectivityGetLinkLocalIpv6Address('wlan0')[:-6]

    return IPv6


def get_if_addr6(intf, address_type):
    """Returns the Ipv6 address from a given local interface.

    Returns the desired IPv6 address from the interface 'intf' in human
    readable form. The address type is indicated by the IPv6 constants like
    IPV6_ADDR_LINKLOCAL, IPV6_ADDR_GLOBAL, etc. If no address is found,
    None is returned.

    Args:
        intf: desired interface name
        address_type: addrees typle like LINKLOCAL or GLOBAL

    Returns:
        Ipv6 address of the specified interface in human readable format
    """
    for if_list in scapy.in6_getifaddr():
        if if_list[2] == intf and if_list[1] == address_type:
            return if_list[0]

    return None


@utils.timeout(60)
def wait_for_dhcp(intf):
    """Wait the DHCP address assigned to desired interface.

    Getting DHCP address takes time and the wait time isn't constant. Utilizing
    utils.timeout to keep trying until success

    Args:
        intf: desired interface name
    Returns:
        ip: ip address of the desired interface name
    Raise:
        TimeoutError: After timeout, if no DHCP assigned, raise
    """
    log = logging.getLogger()
    reset_host_interface(intf)
    ip = '0.0.0.0'
    while ip == '0.0.0.0':
        ip = scapy.get_if_addr(intf)
    log.info('DHCP address assigned to {} as {}'.format(intf, ip))
    return ip


def reset_host_interface(intf):
    """Reset the host interface.

    Args:
        intf: the desired interface to reset
    """
    log = logging.getLogger()
    intf_down_cmd = 'ifconfig %s down' % intf
    intf_up_cmd = 'ifconfig %s up' % intf
    try:
        job.run(intf_down_cmd)
        time.sleep(10)
        job.run(intf_up_cmd)
        log.info('{} has been reset'.format(intf))
    except job.Error:
        raise Exception('No such interface')


def create_pkt_config(test_class):
    """Creates the config for generating multicast packets

    Args:
        test_class: object with all networking paramters

    Returns:
        Dictionary with the multicast packet config
    """
    addr_type = (scapy.IPV6_ADDR_LINKLOCAL
                 if test_class.ipv6_src_type == 'LINK_LOCAL' else
                 scapy.IPV6_ADDR_GLOBAL)

    mac_dst = test_class.mac_dst
    if GET_FROM_PHONE in test_class.mac_dst:
        mac_dst = get_phone_mac(test_class.dut)

    ipv4_dst = test_class.ipv4_dst
    if GET_FROM_PHONE in test_class.ipv4_dst:
        ipv4_dst = get_phone_ip(test_class.dut)

    ipv6_dst = test_class.ipv6_dst
    if GET_FROM_PHONE in test_class.ipv6_dst:
        ipv6_dst = get_phone_ipv6(test_class.dut)

    ipv4_gw = test_class.ipv4_gwt
    if GET_FROM_AP in test_class.ipv4_gwt:
        ipv4_gw = test_class.access_point.ssh_settings.hostname

    pkt_gen_config = {
        'interf': test_class.pkt_sender.interface,
        'subnet_mask': test_class.sub_mask,
        'src_mac': test_class.mac_src,
        'dst_mac': mac_dst,
        'src_ipv4': test_class.ipv4_src,
        'dst_ipv4': ipv4_dst,
        'src_ipv6': test_class.ipv6_src,
        'src_ipv6_type': addr_type,
        'dst_ipv6': ipv6_dst,
        'gw_ipv4': ipv4_gw
    }
    return pkt_gen_config


def create_monsoon_info(test_class):
    """Creates the config dictionary for monsoon

    Args:
        test_class: object with all the parameters

    Returns:
        Dictionary with the monsoon packet config
    """
    mon_info = {
        'dut': test_class.mon,
        'freq': test_class.mon_freq,
        'duration': test_class.mon_duration,
        'offset': test_class.mon_offset,
        'data_path': test_class.mon_data_path
    }
    return mon_info


def setup_phone_wireless(test_class,
                         bt_on,
                         wifi_on,
                         screen_status,
                         network=None,
                         regular_mode=False):
    """Sets the phone in rock-bottom and in the desired wireless mode

    Args:
        test_class: the specific test class where test is running
        bt_on: Enable/Disable BT
        wifi_on: Enable/Disable WiFi
        screen_status: screen ON or OFF
        network: a dict of information for the WiFi network to connect
        regular_mode: enable cellular data (i.e., disable airplane mode)
    """
    # Initialize the dut to rock-bottom state
    dut_rockbottom(test_class.dut)
    brconfigs = None
    time.sleep(1)

    if regular_mode:
        test_class.dut.droid.connectivityToggleAirplaneMode(False)
        utils.set_mobile_data_always_on(test_class.dut, True)
        time.sleep(2)

    # Turn ON/OFF BT
    if bt_on == 'ON':
        enable_bluetooth(test_class.dut.droid, test_class.dut.ed)
        test_class.dut.log.info('BT is ON')
    else:
        disable_bluetooth(test_class.dut.droid)
        test_class.dut.droid.bluetoothDisableBLE()
        test_class.dut.log.info('BT is OFF')
    time.sleep(2)

    # Turn ON/OFF Wifi
    if wifi_on == 'ON':
        wutils.wifi_toggle_state(test_class.dut, True)
        test_class.dut.log.info('WiFi is ON')
        if network:
            # Set attenuation and connect to AP
            for attn in range(test_class.num_atten):
                test_class.attenuators[attn].set_atten(
                    test_class.atten_level['zero_atten'][attn])
            test_class.log.info('Set attenuation level to all zero')
            brconfigs = ap_setup(test_class.access_point, network)
            wutils.wifi_connect(test_class.dut, network)
    else:
        wutils.wifi_toggle_state(test_class.dut, False)
        test_class.dut.log.info('WiFi is OFF')
    time.sleep(1)

    # Set the desired screen status
    if screen_status == 'OFF':
        test_class.dut.droid.goToSleepNow()
        test_class.dut.log.info('Screen is OFF')
    time.sleep(1)

    if brconfigs:
        return brconfigs
    else:
        return None