1#!/usr/bin/env python3 2# 3# Copyright (C) 2019 The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); you may not 6# use this file except in compliance with the License. You may obtain a copy of 7# 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, WITHOUT 13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 14# License for the specific language governing permissions and limitations under 15# the License. 16 17import time 18from acts_contrib.test_utils.bt.A2dpBaseTest import A2dpBaseTest 19import acts_contrib.test_utils.bt.BleBaseTest as BleBT 20from acts_contrib.test_utils.power.IperfHelper import IperfHelper 21from acts_contrib.test_utils.bt.bt_test_utils import orchestrate_rfcomm_connection 22from concurrent.futures import ThreadPoolExecutor 23from acts_contrib.test_utils.wifi import wifi_test_utils as wutils 24from acts_contrib.test_utils.tel.tel_test_utils import WIFI_CONFIG_APBAND_2G 25from acts_contrib.test_utils.wifi import wifi_power_test_utils as wputils 26 27 28class BtMultiprofileBaseTest(A2dpBaseTest, BleBT.BleBaseTest): 29 """Base class for BT mutiprofile related tests. 30 31 Inherited from the A2DP Base class, Ble Base class 32 """ 33 # Iperf waiting time (margin) 34 IPERF_MARGIN = 10 35 36 def mutiprofile_test(self, 37 codec_config=None, 38 mode=None, 39 victim=None, 40 aggressor=None, 41 metric=None): 42 43 if victim == 'A2DP' and aggressor == 'Ble_Scan' and metric == 'range': 44 scan_callback = self.start_ble_scan(self.dut, mode) 45 self.run_a2dp_to_max_range(codec_config) 46 self.dut.droid.bleStopBleScan(scan_callback) 47 self.log.info("BLE Scan stopped successfully") 48 return True 49 50 if victim == 'Ble_Scan' and aggressor == 'A2DP' and metric == 'scan_accuracy': 51 scan_callback = self.start_ble_scan(self.dut, mode) 52 recorded_file = self.play_and_record_audio( 53 self.audio_params['duration']) 54 self.dut.droid.bleStopBleScan(scan_callback) 55 self.media.stop() 56 self.log.info("BLE Scan & A2DP streaming stopped successfully") 57 return True 58 59 if victim == 'RFCOMM' and aggressor == 'Ble_Scan' and metric == 'throughput': 60 self.remote_device = self.android_devices[2] 61 scan_callback = self.start_ble_scan(self.dut, mode) 62 if not orchestrate_rfcomm_connection(self.dut, self.remote_device): 63 return False 64 self.log.info("RFCOMM Connection established") 65 self.measure_rfcomm_throughput(100) 66 self.dut.droid.bleStopBleScan(scan_callback) 67 self.log.info("BLE Scan stopped successfully") 68 69 if victim == 'A2DP' and aggressor == 'Ble_Adv' and metric == 'range': 70 advertise_callback = self.start_ble_adv(self.dut, mode, 2) 71 self.run_a2dp_to_max_range(codec_config) 72 self.dut.droid.bleStopBleAdvertising(advertise_callback) 73 self.log.info("Advertisement stopped Successfully") 74 return True 75 76 if victim == 'A2DP' and aggressor == 'Ble_conn' and metric == 'range': 77 self.start_ble_connection(self.dut, self.android_devices[2], mode) 78 self.run_a2dp_to_max_range(codec_config) 79 return True 80 81 if victim == 'A2DP' and aggressor == 'wifi' and metric == 'range': 82 self.setup_hotspot_and_connect_client() 83 self.setup_iperf_and_run_throughput() 84 self.run_a2dp_to_max_range(codec_config) 85 self.process_iperf_results() 86 return True 87 88 if victim == 'Ble_Scan' and aggressor == 'wifi' and metric == 'scan_accuracy': 89 scan_callback = self.start_ble_scan(self.dut, mode) 90 self.setup_hotspot_and_connect_client() 91 self.setup_iperf_and_run_throughput() 92 time.sleep(self.audio_params['duration'] + self.IPERF_MARGIN + 2) 93 self.log.info("BLE Scan & iPerf started successfully") 94 self.process_iperf_results() 95 self.dut.droid.bleStopBleScan(scan_callback) 96 self.log.info("BLE Scan stopped successfully") 97 return True 98 99 if victim == 'Ble_Adv' and aggressor == 'wifi' and metric == 'periodic_adv': 100 advertise_callback = self.start_ble_adv(self.dut, mode, 2) 101 self.setup_hotspot_and_connect_client() 102 self.setup_iperf_and_run_throughput() 103 time.sleep(self.audio_params['duration'] + self.IPERF_MARGIN + 2) 104 self.log.info("BLE Advertisement & iPerf started successfully") 105 self.process_iperf_results() 106 self.dut.droid.bleStopBleAdvertising(advertise_callback) 107 self.log.info("Advertisement stopped Successfully") 108 return True 109 110 if victim == 'RFCOMM' and aggressor == 'wifi' and metric == 'throughput': 111 self.remote_device = self.android_devices[2] 112 if not orchestrate_rfcomm_connection(self.dut, self.remote_device): 113 return False 114 self.log.info("RFCOMM Connection established") 115 self.setup_hotspot_and_connect_client() 116 executor = ThreadPoolExecutor(2) 117 throughput = executor.submit(self.measure_rfcomm_throughput, 100) 118 executor.submit(self.setup_iperf_and_run_throughput, ) 119 time.sleep(self.audio_params['duration'] + self.IPERF_MARGIN + 10) 120 self.process_iperf_results() 121 return True 122 123 def measure_rfcomm_throughput(self, iteration): 124 """Measures the throughput of a data transfer. 125 126 Sends data over RFCOMM from the client device that is read by the server device. 127 Calculates the throughput for the transfer. 128 129 Args: 130 iteration : An integer value that respesents number of RFCOMM data trasfer iteration 131 132 Returns: 133 The throughput of the transfer in bits per second. 134 """ 135 #An integer value designating the number of buffers to be sent. 136 num_of_buffers = 1 137 #An integer value designating the size of each buffer, in bytes. 138 buffer_size = 22000 139 throughput_list = [] 140 for transfer in range(iteration): 141 (self.dut.droid.bluetoothConnectionThroughputSend( 142 num_of_buffers, buffer_size)) 143 144 throughput = ( 145 self.remote_device.droid.bluetoothConnectionThroughputRead( 146 num_of_buffers, buffer_size)) 147 throughput = throughput * 8 148 throughput_list.append(throughput) 149 self.log.info( 150 ("RFCOMM Throughput is :{} bits/sec".format(throughput))) 151 throughput = statistics.mean(throughput_list) 152 return throughput 153 154 def setup_hotspot_and_connect_client(self): 155 """ 156 Setup hotspot on the remote device and client connects to hotspot 157 158 """ 159 self.network = { 160 wutils.WifiEnums.SSID_KEY: 'Pixel_2G', 161 wutils.WifiEnums.PWD_KEY: '1234567890' 162 } 163 # Setup tethering on dut 164 wutils.start_wifi_tethering(self.android_devices[1], 165 self.network[wutils.WifiEnums.SSID_KEY], 166 self.network[wutils.WifiEnums.PWD_KEY], 167 WIFI_CONFIG_APBAND_2G) 168 169 # Connect client device to Hotspot 170 wutils.wifi_connect(self.dut, self.network, check_connectivity=False) 171 172 def setup_iperf_and_run_throughput(self): 173 self.iperf_server_address = self.android_devices[ 174 1].droid.connectivityGetIPv4Addresses('wlan2')[0] 175 # Create the iperf config 176 iperf_config = { 177 'traffic_type': 'TCP', 178 'duration': self.audio_params['duration'] + self.IPERF_MARGIN, 179 'server_idx': 0, 180 'traffic_direction': 'UL', 181 'port': self.iperf_servers[0].port, 182 'start_meas_time': 4, 183 } 184 # Start iperf traffic (dut is the client) 185 self.client_iperf_helper = IperfHelper(iperf_config) 186 self.iperf_servers[0].start() 187 wputils.run_iperf_client_nonblocking( 188 self.dut, self.iperf_server_address, 189 self.client_iperf_helper.iperf_args) 190 191 def process_iperf_results(self): 192 time.sleep(self.IPERF_MARGIN + 2) 193 self.client_iperf_helper.process_iperf_results(self.dut, self.log, 194 self.iperf_servers, 195 self.test_name) 196 self.iperf_servers[0].stop() 197 return True 198