1#!/usr/bin/env python3 2# 3# Copyright (C) 2018 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 json 18import os 19import time 20 21from acts.base_test import BaseTestClass 22from acts.test_utils.bt import BtEnum 23from acts.test_utils.bt.bt_test_utils import bluetooth_enabled_check 24from acts.test_utils.bt.bt_test_utils import disable_bluetooth 25from acts.test_utils.bt.bt_test_utils import enable_bluetooth 26from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test 27from acts.test_utils.coex.coex_test_utils import A2dpDumpsysParser 28from acts.test_utils.coex.audio_test_utils import SshAudioCapture 29from acts.test_utils.coex.coex_test_utils import ( 30 collect_bluetooth_manager_dumpsys_logs) 31from acts.test_utils.coex.coex_test_utils import is_a2dp_connected 32from acts.test_utils.coex.coex_test_utils import multithread_func 33from acts.test_utils.coex.coex_test_utils import pair_and_connect_headset 34from acts.test_utils.coex.coex_test_utils import push_music_to_android_device 35from acts.utils import create_dir 36 37AVRCP_WAIT_TIME = 5 38 39 40class BtRangeBaseTest(BaseTestClass): 41 42 def __init__(self, controllers): 43 super().__init__(controllers) 44 self.pri_ad = self.android_devices[0] 45 46 def setup_class(self): 47 if not setup_multiple_devices_for_bt_test(self.android_devices): 48 self.log.error("Failed to setup devices for bluetooth test") 49 return False 50 req_params = ["test_params", "Attenuator"] 51 opt_params = ["RelayDevice", "required_devices", "audio_params"] 52 self.unpack_userparams(req_params, opt_params) 53 if hasattr(self, "Attenuator"): 54 self.num_atten = self.attenuators[0].instrument.num_atten 55 else: 56 self.log.error("Attenuator should be connected to run tests.") 57 return False 58 if hasattr(self, "RelayDevice"): 59 self.audio_receiver = self.relay_devices[0] 60 self.headset_mac_address = self.audio_receiver.mac_address 61 else: 62 self.log.warning("Missing Relay config file.") 63 if self.audio_params["music_file"]: 64 self.music_file_to_play = push_music_to_android_device( 65 self.pri_ad, self.audio_params) 66 if not self.music_file_to_play: 67 return False 68 self.bt_attenuation_range = range(self.test_params["bt_atten_start"], 69 self.test_params["bt_atten_stop"], 70 self.test_params["bt_atten_step"]) 71 72 def setup_test(self): 73 self.result = {} 74 self.a2dp_dumpsys = A2dpDumpsysParser() 75 self.a2dp_dropped_list = [] 76 self.log_path = os.path.join(self.pri_ad.log_path, 77 self.current_test_name) 78 create_dir(self.log_path) 79 self.json_file = os.path.join(self.log_path, "test_results.json") 80 self.attenuators[self.num_atten - 1].set_atten(0) 81 if not enable_bluetooth(self.pri_ad.droid, self.pri_ad.ed): 82 self.log.error("Failed to enable bluetooth") 83 return False 84 if hasattr(self, "RelayDevice"): 85 self.audio_receiver.power_on() 86 self.audio_receiver.enter_pairing_mode() 87 time.sleep(5) 88 if "a2dp_streaming" in self.current_test_name: 89 self.audio = SshAudioCapture(self.audio_params, self.log_path) 90 if not pair_and_connect_headset( 91 self.pri_ad, self.headset_mac_address, 92 set([BtEnum.BluetoothProfile.A2DP.value])): 93 self.log.error("Failed to pair and connect to headset") 94 return False 95 96 def teardown_test(self): 97 if "a2dp_streaming" in self.current_test_name: 98 self.audio.terminate_and_store_audio_results() 99 with open(self.json_file, 'a') as results_file: 100 json.dump(self.result, results_file, indent=4, sort_keys=True) 101 self.attenuators[self.num_atten - 1].set_atten(0) 102 if not disable_bluetooth(self.pri_ad.droid): 103 self.log.error("Failed to disable bluetooth") 104 return False 105 106 def set_bt_attenuation(self, called_func): 107 """Sets bt attenuation and runs the bluetooth functions.""" 108 for bt_atten in self.bt_attenuation_range: 109 self.attenuators[self.num_atten - 1].set_atten(bt_atten) 110 self.log.info("Setting bt attenuation = {}".format(bt_atten)) 111 self.result[bt_atten] = {} 112 if not bluetooth_enabled_check(self.pri_ad): 113 return False 114 if called_func: 115 if not multithread_func(self.log, called_func): 116 if not is_a2dp_connected(self.pri_ad, 117 self.headset_mac_address): 118 self.log.error( 119 'A2DP Connection dropped at %s attenuation', 120 bt_atten) 121 return False 122 if "a2dp_streaming" in self.current_test_name: 123 file_path = collect_bluetooth_manager_dumpsys_logs( 124 self.pri_ad, self.current_test_name) 125 self.a2dp_dropped_list.append( 126 self.a2dp_dumpsys.parse(file_path)) 127 if self.a2dp_dropped_list[-1] > 0: 128 self.result[bt_atten]["a2dp_packet_drop"] = ( 129 self.a2dp_dropped_list[-1]) 130 self.log.info("A2dp packets dropped = {}".format( 131 self.a2dp_dropped_list)) 132 analysis_path = self.audio.audio_quality_analysis(self.log_path) 133 with open(analysis_path) as f: 134 self.result[bt_atten]["audio_artifacts"] = f.readline() 135 return True 136