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. 16import os 17import pandas as pd 18import acts_contrib.test_utils.bt.bt_test_utils as btutils 19import acts_contrib.test_utils.wifi.wifi_performance_test_utils as wifi_utils 20from acts import asserts 21from acts_contrib.test_utils.bt import bt_constants 22from acts_contrib.test_utils.bt import BtEnum 23from acts_contrib.test_utils.bt.A2dpBaseTest import A2dpBaseTest 24from acts_contrib.test_utils.bt.loggers import bluetooth_metric_logger as log 25from acts.test_utils.power.PowerBTBaseTest import ramp_attenuation 26from acts.signals import TestPass 27 28 29class BtA2dpRangeTest(A2dpBaseTest): 30 def __init__(self, configs): 31 super().__init__(configs) 32 self.bt_logger = log.BluetoothMetricLogger.for_test_case() 33 req_params = ['attenuation_vector', 'codecs'] 34 #'attenuation_vector' is a dict containing: start, stop and step of 35 #attenuation changes 36 #'codecs' is a list containing all codecs required in the tests 37 self.unpack_userparams(req_params) 38 for codec_config in self.codecs: 39 self.generate_test_case(codec_config) 40 41 def setup_class(self): 42 super().setup_class() 43 # Enable BQR on all android devices 44 btutils.enable_bqr(self.android_devices) 45 46 def generate_test_case(self, codec_config): 47 def test_case_fn(): 48 self.run_a2dp_to_max_range(codec_config) 49 50 test_case_name = 'test_bt_a2dp_range_codec_{}'.format( 51 codec_config['codec_type']) 52 setattr(self, test_case_name, test_case_fn) 53 54 def generate_proto(self, data_points, codec_type, sample_rate, 55 bits_per_sample, channel_mode): 56 """Generate a results protobuf. 57 58 Args: 59 data_points: list of dicts representing info to go into 60 AudioTestDataPoint protobuffer message. 61 codec_type: The codec type config to store in the proto. 62 sample_rate: The sample rate config to store in the proto. 63 bits_per_sample: The bits per sample config to store in the proto. 64 channel_mode: The channel mode config to store in the proto. 65 Returns: 66 dict: Dictionary with key 'proto' mapping to serialized protobuf, 67 'proto_ascii' mapping to human readable protobuf info, and 'test' 68 mapping to the test class name that generated the results. 69 """ 70 71 # Populate protobuf 72 test_case_proto = self.bt_logger.proto_module.BluetoothAudioTestResult( 73 ) 74 75 for data_point in data_points: 76 audio_data_proto = test_case_proto.data_points.add() 77 log.recursive_assign(audio_data_proto, data_point) 78 79 codec_proto = test_case_proto.a2dp_codec_config 80 codec_proto.codec_type = bt_constants.codec_types[codec_type] 81 codec_proto.sample_rate = int(sample_rate) 82 codec_proto.bits_per_sample = int(bits_per_sample) 83 codec_proto.channel_mode = bt_constants.channel_modes[channel_mode] 84 85 self.bt_logger.add_config_data_to_proto(test_case_proto, self.dut, 86 self.bt_device) 87 88 self.bt_logger.add_proto_to_results(test_case_proto, 89 self.__class__.__name__) 90 91 proto_dict = self.bt_logger.get_proto_dict(self.__class__.__name__, 92 test_case_proto) 93 del proto_dict["proto_ascii"] 94 return proto_dict 95 96 def plot_graph(self, df): 97 """ Plotting A2DP DUT RSSI, remote RSSI and TX Power with Attenuation. 98 99 Args: 100 df: Summary of results contains attenuation, DUT RSSI, remote RSSI and Tx Power 101 """ 102 self.plot = wifi_utils.BokehFigure(title='{}'.format( 103 self.current_test_name), 104 x_label='Pathloss (dBm)', 105 primary_y_label='RSSI (dBm)', 106 secondary_y_label='TX Power (dBm)', 107 axis_label_size='16pt') 108 self.plot.add_line(df.index, 109 df['rssi_primary'], 110 legend='DUT RSSI (dBm)', 111 marker='circle_x') 112 self.plot.add_line(df.index, 113 df['rssi_secondary'], 114 legend='Remote RSSI (dBm)', 115 marker='square_x') 116 self.plot.add_line(df.index, 117 df['tx_power_level_master'], 118 legend='DUT TX Power (dBm)', 119 marker='hex', 120 y_axis='secondary') 121 122 results_file_path = os.path.join( 123 self.log_path, '{}.html'.format(self.current_test_name)) 124 self.plot.generate_figure() 125 wifi_utils.BokehFigure.save_figures([self.plot], results_file_path) 126 127 def run_a2dp_to_max_range(self, codec_config): 128 attenuation_range = range(self.attenuation_vector['start'], 129 self.attenuation_vector['stop'] + 1, 130 self.attenuation_vector['step']) 131 132 data_points = [] 133 self.file_output = os.path.join( 134 self.log_path, '{}.csv'.format(self.current_test_name)) 135 136 # Set Codec if needed 137 current_codec = self.dut.droid.bluetoothA2dpGetCurrentCodecConfig() 138 current_codec_type = BtEnum.BluetoothA2dpCodecType( 139 current_codec['codecType']).name 140 if current_codec_type != codec_config['codec_type']: 141 codec_set = btutils.set_bluetooth_codec(self.dut, **codec_config) 142 asserts.assert_true(codec_set, 'Codec configuration failed.') 143 else: 144 self.log.info('Current codec is {}, no need to change'.format( 145 current_codec_type)) 146 147 #loop RSSI with the same codec setting 148 for atten in attenuation_range: 149 ramp_attenuation(self.attenuator, atten) 150 self.log.info('Set attenuation to %d dB', atten) 151 152 tag = 'codec_{}_attenuation_{}dB_'.format( 153 codec_config['codec_type'], atten) 154 recorded_file = self.play_and_record_audio( 155 self.audio_params['duration']) 156 [rssi_master, pwl_master, rssi_slave] = self._get_bt_link_metrics() 157 thdns = self.run_thdn_analysis(recorded_file, tag) 158 # Collect Metrics for dashboard 159 data_point = { 160 'attenuation_db': int(self.attenuator.get_atten()), 161 'rssi_primary': rssi_master[self.dut.serial], 162 'tx_power_level_master': pwl_master[self.dut.serial], 163 'rssi_secondary': rssi_slave[self.bt_device_controller.serial], 164 'total_harmonic_distortion_plus_noise_percent': thdns[0] * 100 165 } 166 data_points.append(data_point) 167 self.log.info(data_point) 168 A2dpRange_df = pd.DataFrame(data_points) 169 170 # Check thdn for glitches, stop if max range reached 171 for thdn in thdns: 172 if thdn >= self.audio_params['thdn_threshold']: 173 self.log.info( 174 'Max range at attenuation {} dB'.format(atten)) 175 self.log.info('DUT rssi {} dBm, DUT tx power level {}, ' 176 'Remote rssi {} dBm'.format( 177 rssi_master, pwl_master, rssi_slave)) 178 proto_dict = self.generate_proto(data_points, 179 **codec_config) 180 A2dpRange_df.to_csv(self.file_output, index=False) 181 self.plot_graph(A2dpRange_df) 182 raise TestPass('Max range reached and move to next codec', 183 extras=proto_dict) 184 # Save Data points to csv 185 A2dpRange_df.to_csv(self.file_output, index=False) 186 # Plot graph 187 self.plot_graph(A2dpRange_df) 188 proto_dict = self.generate_proto(data_points, **codec_config) 189 raise TestPass('Could not reach max range, need extra attenuation.', 190 extras=proto_dict)