acts_tests/acts_contrib/test_utils/bt/BtInterferenceBaseTest.py - platform/tools/test/connectivity - Gitiles

 #!/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.
 """Stream music through connected device from phone across different
 attenuations."""

 import json
 import math
 import time
 import acts.controllers.iperf_client as ipc
 import acts.controllers.iperf_server as ipf
 import acts_contrib.test_utils.bt.bt_test_utils as btutils
 from acts import asserts
 from acts_contrib.test_utils.bt.A2dpBaseTest import A2dpBaseTest
 from acts_contrib.test_utils.bt.loggers import bluetooth_metric_logger as log
 from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wpeutils
 from acts_contrib.test_utils.wifi import wifi_power_test_utils as wputils
 from acts_contrib.test_utils.wifi import wifi_test_utils as wutils
 from acts_contrib.test_utils.power.PowerBaseTest import ObjNew

 MAX_ATTENUATION = 95
 TEMP_FILE = '/sdcard/Download/tmp.log'
 IPERF_CLIENT_ERROR = 'the client has unexpectedly closed the connection'


 def setup_ap_connection(dut, network, ap, bandwidth=20):
     """Setup AP and connect DUT to it.

     Args:
         dut: the android device to connect and run traffic
         network: the network config for the AP to be setup
         ap: access point object
         bandwidth: bandwidth of the WiFi network to be setup
     Returns:
         self.brconfigs: dict for bridge interface configs
     """
     wutils.wifi_toggle_state(dut, True)
     brconfigs = wputils.ap_setup(ap, network, bandwidth=bandwidth)
     wutils.wifi_connect(dut, network, num_of_tries=3)
     return brconfigs


 def start_iperf_client(traffic_pair_obj, duration):
     """Setup iperf traffic for TCP downlink.
     Args:
         traffic_pair_obj: obj to contain info on traffic pair
         duration: duration of iperf traffic to run
     """
     # Construct the iperf command based on the test params
     iperf_cmd = 'iperf3 -c {} -i 1 -t {} -p {} -J -R > {}'.format(
         traffic_pair_obj.server_address, duration,
         traffic_pair_obj.iperf_server.port, TEMP_FILE)
     # Start IPERF client
     traffic_pair_obj.dut.adb.shell_nb(iperf_cmd)


 def unpack_custom_file(file):
     """Unpack the json file to .

     Args:
         file: custom json file.
     """
     with open(file, 'r') as f:
         params = json.load(f)
     return params


 def get_iperf_results(iperf_server_obj):
     """Get the iperf results and process.

     Args:
         iperf_server_obj: the IperfServer object
     Returns:
          throughput: the average throughput during tests.
     """
     # Get IPERF results and add this to the plot title
     iperf_file = iperf_server_obj.log_files[-1]
     try:
         iperf_result = ipf.IPerfResult(iperf_file)
         # Compute the throughput in Mbit/s
         if iperf_result.error == IPERF_CLIENT_ERROR:
             rates = []
             for item in iperf_result.result['intervals']:
                 rates.append(item['sum']['bits_per_second'] / 8 / 1024 / 1024)
             throughput = ((math.fsum(rates) / len(rates))) * 8 * (1.024**2)
         else:
             throughput = (math.fsum(iperf_result.instantaneous_rates) / len(
                 iperf_result.instantaneous_rates)) * 8 * (1.024**2)
     except (ValueError, TypeError):
         throughput = 0
     return throughput


 class BtInterferenceBaseTest(A2dpBaseTest):
     def __init__(self, configs):
         super().__init__(configs)
         self.bt_logger = log.BluetoothMetricLogger.for_test_case()
         self.start_time = time.time()
         req_params = [
             'attenuation_vector', 'wifi_networks', 'codecs', 'custom_files',
             'audio_params'
         ]
         self.unpack_userparams(req_params)
         for file in self.custom_files:
             if 'static_interference' in file:
                 self.static_wifi_interference = unpack_custom_file(file)
             elif 'dynamic_interference' in file:
                 self.dynamic_wifi_interference = unpack_custom_file(file)

     def setup_class(self):
         super().setup_class()
         # Build object to store all necessary information for each pair of wifi
         # interference setup: phone, ap, network, channel, iperf server port/ip
         # object and bridge interface configs
         if len(self.android_devices) < 5 or len(self.attenuators) < 4:
             self.log.error('Need a 4 channel attenuator and 5 android phones'
                            'please update the config file')
         self.wifi_int_pairs = []
         for i in range(len(self.attenuators) - 1):
             tmp_dict = {
                 'dut': self.android_devices[i + 1],
                 'ap': self.access_points[i],
                 'network': self.wifi_networks[i],
                 'channel': self.wifi_networks[i]['channel'],
                 'iperf_server': self.iperf_servers[i],
                 'attenuator': self.attenuators[i + 1],
                 'ether_int': self.packet_senders[i],
                 'iperf_client':
                 ipc.IPerfClientOverAdb(self.android_devices[i + 1])
             }
             tmp_obj = ObjNew(**tmp_dict)
             self.wifi_int_pairs.append(tmp_obj)
         ##Setup connection between WiFi APs and Phones and get DHCP address
         # for the interface
         for obj in self.wifi_int_pairs:
             brconfigs = setup_ap_connection(obj.dut, obj.network, obj.ap)
             iperf_server_address = wputils.wait_for_dhcp(
                 obj.ether_int.interface)
             setattr(obj, 'server_address', iperf_server_address)
             setattr(obj, 'brconfigs', brconfigs)
             obj.attenuator.set_atten(MAX_ATTENUATION)
         # Enable BQR on master and slave Android device
         btutils.enable_bqr(self.dut)
         btutils.enable_bqr(self.bt_device_controller)

     def teardown_class(self):
         super().teardown_class()
         for obj in self.wifi_int_pairs:
             obj.ap.bridge.teardown(obj.brconfigs)
             self.log.info('Stop IPERF server at port {}'.format(
                 obj.iperf_server.port))
             obj.iperf_server.stop()
             self.log.info('Stop IPERF process on {}'.format(obj.dut.serial))
             obj.dut.adb.shell('pkill -9 iperf3')
             #only for glinux machine
             #            wputils.bring_down_interface(obj.ether_int.interface)
             obj.attenuator.set_atten(MAX_ATTENUATION)
             obj.ap.close()

     def teardown_test(self):

         super().teardown_test()

         for obj in self.wifi_int_pairs:
             obj.attenuator.set_atten(MAX_ATTENUATION)

     def play_and_record_audio(self, duration, queue):
         """Play and record audio for a set duration.

         Args:
             duration: duration in seconds for music playing
             que: multiprocess que to store the return value of this function
         Returns:
             audio_captured: captured audio file path
         """

         self.log.info('Play and record audio for {} second'.format(duration))
         self.media.play()
         self.audio_device.start()
         time.sleep(duration)
         audio_captured = self.audio_device.stop()
         self.media.stop()
         self.log.info('Audio play and record stopped')
         asserts.assert_true(audio_captured, 'Audio not recorded')
         queue.put(audio_captured)

     def locate_interference_pair_by_channel(self, interference_channels):
         """Function to find which attenautor to set based on channel info
         Args:
             interference_channels: list of interference channels
         Return:
             interference_pair_indices: list of indices for interference pair
                 in self.wifi_int_pairs
         """
         interference_pair_indices = []
         for chan in interference_channels:
             for i in range(len(self.wifi_int_pairs)):
                 if self.wifi_int_pairs[i].channel == chan:
                     interference_pair_indices.append(i)
         return interference_pair_indices

     def get_interference_rssi(self):
         """Function to read wifi interference RSSI level."""

         bssids = []
         self.interference_rssi = []
         wutils.wifi_toggle_state(self.android_devices[0], True)
         for item in self.wifi_int_pairs:
             ssid = item.network['SSID']
             bssid = item.ap.get_bssid_from_ssid(ssid, '2g')
             bssids.append(bssid)
             interference_rssi_dict = {
                 "ssid": ssid,
                 "bssid": bssid,
                 "chan": item.channel,
                 "rssi": 0
             }
             self.interference_rssi.append(interference_rssi_dict)
         scaned_rssi = wpeutils.get_scan_rssi(self.android_devices[0],
                                              bssids,
                                              num_measurements=2)
         for item in self.interference_rssi:
             item['rssi'] = scaned_rssi[item['bssid']]['mean']
             self.log.info('Interference RSSI at channel {} is {} dBm'.format(
                 item['chan'], item['rssi']))
         wutils.wifi_toggle_state(self.android_devices[0], False)
	#!/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.
	"""Stream music through connected device from phone across different
	attenuations."""

	import json
	import math
	import time
	import acts.controllers.iperf_client as ipc
	import acts.controllers.iperf_server as ipf
	import acts_contrib.test_utils.bt.bt_test_utils as btutils
	from acts import asserts
	from acts_contrib.test_utils.bt.A2dpBaseTest import A2dpBaseTest
	from acts_contrib.test_utils.bt.loggers import bluetooth_metric_logger as log
	from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wpeutils
	from acts_contrib.test_utils.wifi import wifi_power_test_utils as wputils
	from acts_contrib.test_utils.wifi import wifi_test_utils as wutils
	from acts_contrib.test_utils.power.PowerBaseTest import ObjNew

	MAX_ATTENUATION = 95
	TEMP_FILE = '/sdcard/Download/tmp.log'
	IPERF_CLIENT_ERROR = 'the client has unexpectedly closed the connection'


	def setup_ap_connection(dut, network, ap, bandwidth=20):
	"""Setup AP and connect DUT to it.

	Args:
	dut: the android device to connect and run traffic
	network: the network config for the AP to be setup
	ap: access point object
	bandwidth: bandwidth of the WiFi network to be setup
	Returns:
	self.brconfigs: dict for bridge interface configs
	"""
	wutils.wifi_toggle_state(dut, True)
	brconfigs = wputils.ap_setup(ap, network, bandwidth=bandwidth)
	wutils.wifi_connect(dut, network, num_of_tries=3)
	return brconfigs


	def start_iperf_client(traffic_pair_obj, duration):
	"""Setup iperf traffic for TCP downlink.
	Args:
	traffic_pair_obj: obj to contain info on traffic pair
	duration: duration of iperf traffic to run
	"""
	# Construct the iperf command based on the test params
	iperf_cmd = 'iperf3 -c {} -i 1 -t {} -p {} -J -R > {}'.format(
	traffic_pair_obj.server_address, duration,
	traffic_pair_obj.iperf_server.port, TEMP_FILE)
	# Start IPERF client
	traffic_pair_obj.dut.adb.shell_nb(iperf_cmd)


	def unpack_custom_file(file):
	"""Unpack the json file to .

	Args:
	file: custom json file.
	"""
	with open(file, 'r') as f:
	params = json.load(f)
	return params


	def get_iperf_results(iperf_server_obj):
	"""Get the iperf results and process.

	Args:
	iperf_server_obj: the IperfServer object
	Returns:
	throughput: the average throughput during tests.
	"""
	# Get IPERF results and add this to the plot title
	iperf_file = iperf_server_obj.log_files[-1]
	try:
	iperf_result = ipf.IPerfResult(iperf_file)
	# Compute the throughput in Mbit/s
	if iperf_result.error == IPERF_CLIENT_ERROR:
	rates = []
	for item in iperf_result.result['intervals']:
	rates.append(item['sum']['bits_per_second'] / 8 / 1024 / 1024)
	throughput = ((math.fsum(rates) / len(rates))) * 8 * (1.024**2)
	else:
	throughput = (math.fsum(iperf_result.instantaneous_rates) / len(
	iperf_result.instantaneous_rates)) * 8 * (1.024**2)
	except (ValueError, TypeError):
	throughput = 0
	return throughput


	class BtInterferenceBaseTest(A2dpBaseTest):
	def __init__(self, configs):
	super().__init__(configs)
	self.bt_logger = log.BluetoothMetricLogger.for_test_case()
	self.start_time = time.time()
	req_params = [
	'attenuation_vector', 'wifi_networks', 'codecs', 'custom_files',
	'audio_params'
	]
	self.unpack_userparams(req_params)
	for file in self.custom_files:
	if 'static_interference' in file:
	self.static_wifi_interference = unpack_custom_file(file)
	elif 'dynamic_interference' in file:
	self.dynamic_wifi_interference = unpack_custom_file(file)

	def setup_class(self):
	super().setup_class()
	# Build object to store all necessary information for each pair of wifi
	# interference setup: phone, ap, network, channel, iperf server port/ip
	# object and bridge interface configs
	if len(self.android_devices) < 5 or len(self.attenuators) < 4:
	self.log.error('Need a 4 channel attenuator and 5 android phones'
	'please update the config file')
	self.wifi_int_pairs = []
	for i in range(len(self.attenuators) - 1):
	tmp_dict = {
	'dut': self.android_devices[i + 1],
	'ap': self.access_points[i],
	'network': self.wifi_networks[i],
	'channel': self.wifi_networks[i]['channel'],
	'iperf_server': self.iperf_servers[i],
	'attenuator': self.attenuators[i + 1],
	'ether_int': self.packet_senders[i],
	'iperf_client':
	ipc.IPerfClientOverAdb(self.android_devices[i + 1])
	}
	tmp_obj = ObjNew(**tmp_dict)
	self.wifi_int_pairs.append(tmp_obj)
	##Setup connection between WiFi APs and Phones and get DHCP address
	# for the interface
	for obj in self.wifi_int_pairs:
	brconfigs = setup_ap_connection(obj.dut, obj.network, obj.ap)
	iperf_server_address = wputils.wait_for_dhcp(
	obj.ether_int.interface)
	setattr(obj, 'server_address', iperf_server_address)
	setattr(obj, 'brconfigs', brconfigs)
	obj.attenuator.set_atten(MAX_ATTENUATION)
	# Enable BQR on master and slave Android device
	btutils.enable_bqr(self.dut)
	btutils.enable_bqr(self.bt_device_controller)

	def teardown_class(self):
	super().teardown_class()
	for obj in self.wifi_int_pairs:
	obj.ap.bridge.teardown(obj.brconfigs)
	self.log.info('Stop IPERF server at port {}'.format(
	obj.iperf_server.port))
	obj.iperf_server.stop()
	self.log.info('Stop IPERF process on {}'.format(obj.dut.serial))
	obj.dut.adb.shell('pkill -9 iperf3')
	#only for glinux machine
	# wputils.bring_down_interface(obj.ether_int.interface)
	obj.attenuator.set_atten(MAX_ATTENUATION)
	obj.ap.close()

	def teardown_test(self):

	super().teardown_test()

	for obj in self.wifi_int_pairs:
	obj.attenuator.set_atten(MAX_ATTENUATION)

	def play_and_record_audio(self, duration, queue):
	"""Play and record audio for a set duration.

	Args:
	duration: duration in seconds for music playing
	que: multiprocess que to store the return value of this function
	Returns:
	audio_captured: captured audio file path
	"""

	self.log.info('Play and record audio for {} second'.format(duration))
	self.media.play()
	self.audio_device.start()
	time.sleep(duration)
	audio_captured = self.audio_device.stop()
	self.media.stop()
	self.log.info('Audio play and record stopped')
	asserts.assert_true(audio_captured, 'Audio not recorded')
	queue.put(audio_captured)

	def locate_interference_pair_by_channel(self, interference_channels):
	"""Function to find which attenautor to set based on channel info
	Args:
	interference_channels: list of interference channels
	Return:
	interference_pair_indices: list of indices for interference pair
	in self.wifi_int_pairs
	"""
	interference_pair_indices = []
	for chan in interference_channels:
	for i in range(len(self.wifi_int_pairs)):
	if self.wifi_int_pairs[i].channel == chan:
	interference_pair_indices.append(i)
	return interference_pair_indices

	def get_interference_rssi(self):
	"""Function to read wifi interference RSSI level."""

	bssids = []
	self.interference_rssi = []
	wutils.wifi_toggle_state(self.android_devices[0], True)
	for item in self.wifi_int_pairs:
	ssid = item.network['SSID']
	bssid = item.ap.get_bssid_from_ssid(ssid, '2g')
	bssids.append(bssid)
	interference_rssi_dict = {
	"ssid": ssid,
	"bssid": bssid,
	"chan": item.channel,
	"rssi": 0
	}
	self.interference_rssi.append(interference_rssi_dict)
	scaned_rssi = wpeutils.get_scan_rssi(self.android_devices[0],
	bssids,
	num_measurements=2)
	for item in self.interference_rssi:
	item['rssi'] = scaned_rssi[item['bssid']]['mean']
	self.log.info('Interference RSSI at channel {} is {} dBm'.format(
	item['chan'], item['rssi']))
	wutils.wifi_toggle_state(self.android_devices[0], False)