#!/usr/bin/env python3 # # Copyright (C) 2019 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 time from acts_contrib.test_utils.bt.A2dpBaseTest import A2dpBaseTest import acts_contrib.test_utils.bt.BleBaseTest as BleBT from acts_contrib.test_utils.power.IperfHelper import IperfHelper from acts_contrib.test_utils.bt.bt_test_utils import orchestrate_rfcomm_connection from concurrent.futures import ThreadPoolExecutor from acts_contrib.test_utils.wifi import wifi_test_utils as wutils from acts_contrib.test_utils.tel.tel_test_utils import WIFI_CONFIG_APBAND_2G from acts_contrib.test_utils.wifi import wifi_power_test_utils as wputils class BtMultiprofileBaseTest(A2dpBaseTest, BleBT.BleBaseTest): """Base class for BT mutiprofile related tests. Inherited from the A2DP Base class, Ble Base class """ # Iperf waiting time (margin) IPERF_MARGIN = 10 def mutiprofile_test(self, codec_config=None, mode=None, victim=None, aggressor=None, metric=None): if victim == 'A2DP' and aggressor == 'Ble_Scan' and metric == 'range': scan_callback = self.start_ble_scan(self.dut, mode) self.run_a2dp_to_max_range(codec_config) self.dut.droid.bleStopBleScan(scan_callback) self.log.info("BLE Scan stopped successfully") return True if victim == 'Ble_Scan' and aggressor == 'A2DP' and metric == 'scan_accuracy': scan_callback = self.start_ble_scan(self.dut, mode) recorded_file = self.play_and_record_audio( self.audio_params['duration']) self.dut.droid.bleStopBleScan(scan_callback) self.media.stop() self.log.info("BLE Scan & A2DP streaming stopped successfully") return True if victim == 'RFCOMM' and aggressor == 'Ble_Scan' and metric == 'throughput': self.remote_device = self.android_devices[2] scan_callback = self.start_ble_scan(self.dut, mode) if not orchestrate_rfcomm_connection(self.dut, self.remote_device): return False self.log.info("RFCOMM Connection established") self.measure_rfcomm_throughput(100) self.dut.droid.bleStopBleScan(scan_callback) self.log.info("BLE Scan stopped successfully") if victim == 'A2DP' and aggressor == 'Ble_Adv' and metric == 'range': advertise_callback = self.start_ble_adv(self.dut, mode, 2) self.run_a2dp_to_max_range(codec_config) self.dut.droid.bleStopBleAdvertising(advertise_callback) self.log.info("Advertisement stopped Successfully") return True if victim == 'A2DP' and aggressor == 'Ble_conn' and metric == 'range': self.start_ble_connection(self.dut, self.android_devices[2], mode) self.run_a2dp_to_max_range(codec_config) return True if victim == 'A2DP' and aggressor == 'wifi' and metric == 'range': self.setup_hotspot_and_connect_client() self.setup_iperf_and_run_throughput() self.run_a2dp_to_max_range(codec_config) self.process_iperf_results() return True if victim == 'Ble_Scan' and aggressor == 'wifi' and metric == 'scan_accuracy': scan_callback = self.start_ble_scan(self.dut, mode) self.setup_hotspot_and_connect_client() self.setup_iperf_and_run_throughput() time.sleep(self.audio_params['duration'] + self.IPERF_MARGIN + 2) self.log.info("BLE Scan & iPerf started successfully") self.process_iperf_results() self.dut.droid.bleStopBleScan(scan_callback) self.log.info("BLE Scan stopped successfully") return True if victim == 'Ble_Adv' and aggressor == 'wifi' and metric == 'periodic_adv': advertise_callback = self.start_ble_adv(self.dut, mode, 2) self.setup_hotspot_and_connect_client() self.setup_iperf_and_run_throughput() time.sleep(self.audio_params['duration'] + self.IPERF_MARGIN + 2) self.log.info("BLE Advertisement & iPerf started successfully") self.process_iperf_results() self.dut.droid.bleStopBleAdvertising(advertise_callback) self.log.info("Advertisement stopped Successfully") return True if victim == 'RFCOMM' and aggressor == 'wifi' and metric == 'throughput': self.remote_device = self.android_devices[2] if not orchestrate_rfcomm_connection(self.dut, self.remote_device): return False self.log.info("RFCOMM Connection established") self.setup_hotspot_and_connect_client() executor = ThreadPoolExecutor(2) throughput = executor.submit(self.measure_rfcomm_throughput, 100) executor.submit(self.setup_iperf_and_run_throughput, ) time.sleep(self.audio_params['duration'] + self.IPERF_MARGIN + 10) self.process_iperf_results() return True def measure_rfcomm_throughput(self, iteration): """Measures the throughput of a data transfer. Sends data over RFCOMM from the client device that is read by the server device. Calculates the throughput for the transfer. Args: iteration : An integer value that respesents number of RFCOMM data trasfer iteration Returns: The throughput of the transfer in bits per second. """ #An integer value designating the number of buffers to be sent. num_of_buffers = 1 #An integer value designating the size of each buffer, in bytes. buffer_size = 22000 throughput_list = [] for transfer in range(iteration): (self.dut.droid.bluetoothConnectionThroughputSend( num_of_buffers, buffer_size)) throughput = ( self.remote_device.droid.bluetoothConnectionThroughputRead( num_of_buffers, buffer_size)) throughput = throughput * 8 throughput_list.append(throughput) self.log.info( ("RFCOMM Throughput is :{} bits/sec".format(throughput))) throughput = statistics.mean(throughput_list) return throughput def setup_hotspot_and_connect_client(self): """ Setup hotspot on the remote device and client connects to hotspot """ self.network = { wutils.WifiEnums.SSID_KEY: 'Pixel_2G', wutils.WifiEnums.PWD_KEY: '1234567890' } # Setup tethering on dut wutils.start_wifi_tethering(self.android_devices[1], self.network[wutils.WifiEnums.SSID_KEY], self.network[wutils.WifiEnums.PWD_KEY], WIFI_CONFIG_APBAND_2G) # Connect client device to Hotspot wutils.wifi_connect(self.dut, self.network, check_connectivity=False) def setup_iperf_and_run_throughput(self): self.iperf_server_address = self.android_devices[ 1].droid.connectivityGetIPv4Addresses('wlan2')[0] # Create the iperf config iperf_config = { 'traffic_type': 'TCP', 'duration': self.audio_params['duration'] + self.IPERF_MARGIN, 'server_idx': 0, 'traffic_direction': 'UL', 'port': self.iperf_servers[0].port, 'start_meas_time': 4, } # Start iperf traffic (dut is the client) self.client_iperf_helper = IperfHelper(iperf_config) self.iperf_servers[0].start() wputils.run_iperf_client_nonblocking( self.dut, self.iperf_server_address, self.client_iperf_helper.iperf_args) def process_iperf_results(self): time.sleep(self.IPERF_MARGIN + 2) self.client_iperf_helper.process_iperf_results(self.dut, self.log, self.iperf_servers, self.test_name) self.iperf_servers[0].stop() return True