| #!/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.""" |
| |
| from acts_contrib.test_utils.bt.BtInterferenceBaseTest import BtInterferenceBaseTest |
| from acts.metrics.loggers.blackbox import BlackboxMetricLogger |
| from acts_contrib.test_utils.bt.BtInterferenceBaseTest import get_iperf_results |
| from acts_contrib.test_utils.bt.BtInterferenceBaseTest import inject_static_wifi_interference |
| from multiprocessing import Process, Queue |
| from acts.signals import TestPass |
| |
| DEFAULT_THDN_THRESHOLD = 0.9 |
| MAX_ATTENUATION = 95 |
| TIME_OVERHEAD = 2 |
| |
| |
| class BtInterferenceStaticTest(BtInterferenceBaseTest): |
| |
| def __init__(self, configs): |
| super().__init__(configs) |
| self.bt_attenuation_range = range(self.attenuation_vector['start'], |
| self.attenuation_vector['stop'] + 1, |
| self.attenuation_vector['step']) |
| self.iperf_duration = self.audio_params['duration'] + TIME_OVERHEAD |
| test_metrics = [ |
| 'wifi_chan1_rssi', 'wifi_chan6_rssi', 'wifi_chan11_rssi', 'bt_range' |
| ] |
| for metric in test_metrics: |
| setattr(self, '{}_metric'.format(metric), |
| BlackboxMetricLogger.for_test_case(metric_name=metric)) |
| |
| def setup_generated_tests(self): |
| for level in list( |
| self.static_wifi_interference['interference_level'].values()): |
| for channels in self.static_wifi_interference['channels']: |
| arg_set = [(level, channels)] |
| self.generate_tests( |
| test_logic=self.bt_range_with_static_wifi_interference, |
| name_func=self.create_test_name, |
| arg_sets=arg_set) |
| |
| def create_test_name(self, level, channels): |
| str_channel_test = '' |
| for i in channels: |
| str_channel_test = str_channel_test + str(i) + "_" |
| test_case_name = ('test_bt_range_with_static_interference_level_{}_' |
| 'channel_{}'.format(level, str_channel_test)) |
| return test_case_name |
| |
| def bt_range_with_static_wifi_interference(self, level, channels): |
| """Test function to measure bt range under interference. |
| |
| Args: |
| interference_level: wifi interference level |
| channels: wifi interference channels |
| """ |
| #setup wifi interference by setting the correct attenuator |
| inject_static_wifi_interference(self.wifi_int_pairs, level, channels) |
| # Read interference RSSI |
| self.get_interference_rssi() |
| self.wifi_chan1_rssi_metric.metric_value = self.interference_rssi[0][ |
| 'rssi'] |
| self.wifi_chan6_rssi_metric.metric_value = self.interference_rssi[1][ |
| 'rssi'] |
| self.wifi_chan11_rssi_metric.metric_value = self.interference_rssi[2][ |
| 'rssi'] |
| for atten in self.bt_attenuation_range: |
| # Set attenuation for BT link |
| self.attenuator.set_atten(atten) |
| [ |
| rssi_master, pwl_master, rssi_c0_master, rssi_c1_master, |
| txpw_c0_master, txpw_c1_master, bftx_master, divtx_master |
| ], [rssi_slave] = self._get_bt_link_metrics() |
| rssi_primary = rssi_master.get(self.dut.serial, -127) |
| pwl_primary = pwl_master.get(self.dut.serial, -127) |
| rssi_secondary = rssi_slave.get(self.bt_device_controller.serial, |
| -127) |
| tag = 'attenuation_{}dB'.format(atten) |
| self.log.info( |
| 'BT attenuation set to {} dB and start A2DP streaming'.format( |
| atten)) |
| procs_iperf = [] |
| for obj in self.wifi_int_pairs: |
| obj.iperf_server.start() |
| self.log.info('Started IPERF server at port {}'.format( |
| obj.iperf_server.port)) |
| iperf_args = '-i 1 -t {} -p {} -J -R'.format( |
| self.iperf_duration, obj.iperf_server.port) |
| tag = 'chan_{}'.format(obj.channel) |
| proc_iperf = Process(target=obj.iperf_client.start, |
| args=(obj.server_address, iperf_args, tag)) |
| procs_iperf.append(proc_iperf) |
| |
| #play a2dp streaming and run thdn analysis |
| queue = Queue() |
| proc_bt = Process(target=self.play_and_record_audio, |
| args=(self.audio_params['duration'], queue)) |
| for proc in procs_iperf: |
| proc.start() |
| proc_bt.start() |
| proc_bt.join() |
| for proc in procs_iperf: |
| proc.join() |
| for obj in self.wifi_int_pairs: |
| iperf_throughput = get_iperf_results(obj.iperf_server) |
| self.log.info( |
| 'Throughput for channel {} interference is {} Mbps'.format( |
| obj.channel, iperf_throughput)) |
| obj.iperf_server.stop() |
| self.log.info('Stopped IPERF server at port {}'.format( |
| obj.iperf_server.port)) |
| audio_captured = queue.get() |
| thdns = self.run_thdn_analysis(audio_captured, tag) |
| self.log.info('THDN results are {} at {} dB attenuation'.format( |
| thdns, atten)) |
| self.log.info('DUT rssi {} dBm, master tx power level {}, ' |
| 'RemoteDevice rssi {} dBm'.format( |
| rssi_primary, pwl_primary, rssi_secondary)) |
| for thdn in thdns: |
| if thdn >= self.audio_params['thdn_threshold']: |
| self.log.info('Under the WiFi interference condition: ' |
| 'channel 1 RSSI: {} dBm, ' |
| 'channel 6 RSSI: {} dBm' |
| 'channel 11 RSSI: {} dBm'.format( |
| self.interference_rssi[0]['rssi'], |
| self.interference_rssi[1]['rssi'], |
| self.interference_rssi[2]['rssi'])) |
| raise TestPass( |
| 'Max range for this test is {}, with BT master RSSI at' |
| ' {} dBm'.format(atten, rssi_primary)) |
| |
