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"""Stream music through connected device from phone test implementation.""" 17import logging 18import os 19 20from acts import asserts 21from acts.test_utils.abstract_devices.bluetooth_handsfree_abstract_device import BluetoothHandsfreeAbstractDeviceFactory as Factory 22from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest 23from acts.test_utils.bt.bt_test_utils import connect_phone_to_headset 24from acts.test_utils.bt.bt_test_utils import set_bluetooth_codec 25from acts.test_utils.coex.audio_test_utils import SshAudioCapture 26 27ADB_FILE_EXISTS = 'test -e %s && echo True' 28ADB_VOL_UP = 'input keyevent 24' 29 30 31class A2dpCodecBaseTest(BluetoothBaseTest): 32 """Stream audio file over desired Bluetooth codec configurations. 33 34 Audio file should be a sine wave. Other audio files will not work for the 35 test analysis metrics. 36 37 Device under test is Android phone, connected to headset with a controller 38 that can generate a BluetoothHandsfreeAbstractDevice from test_utils. 39 abstract_devices.bluetooth_handsfree_abstract_device. 40 BuetoothHandsfreeAbstractDeviceFactory. 41 """ 42 43 def __init__(self, configs): 44 super(A2dpCodecBaseTest, self).__init__(configs) 45 required_params = ['audio_params', 'dut'] 46 self.unpack_userparams(required_params) 47 48 # Instantiate test devices 49 self.android = self.android_devices[0] 50 attr, idx = self.dut.split(':') 51 self.dut_controller = getattr(self, attr)[int(idx)] 52 self.bt_device = Factory().generate(self.dut_controller) 53 if 'input_device' in self.audio_params: 54 self.audio_output_path = '' 55 self.mic = SshAudioCapture(self.audio_params, 56 logging.log_path) 57 self.log.info('Recording device %s initialized.' % 58 self.mic.name) 59 else: 60 raise KeyError('Config audio_params specify input_device.') 61 self.phone_music_file = os.path.join( 62 self.user_params['phone_music_file_dir'], 63 self.user_params['music_file_name']) 64 self.host_music_file = os.path.join( 65 self.user_params['host_music_file_dir'], 66 self.user_params['music_file_name']) 67 68 def setup_class(self): 69 super().setup_class() 70 self.bt_device.power_on() 71 72 def teardown_class(self): 73 super().teardown_class() 74 self.android.droid.mediaPlayStop() 75 76 def setup_test(self): 77 """Pair and connect headset before test, make sure phone has audio file. 78 """ 79 # Key test metrics to determine quality of recorded audio file. 80 self.metrics = {'anomalies': [], 81 'dut_type': self.dut_controller.__class__.__name__, 82 'thdn': []} 83 84 self.log.info('Pairing and connecting to headset...') 85 asserts.assert_true( 86 connect_phone_to_headset(self.android, self.bt_device, 600), 87 'Could not connect to device at address %s' 88 % self.bt_device.mac_address, 89 extras=self.metrics) 90 91 # Ensure audio file exists on phone. 92 self.ensure_phone_has_music_file() 93 94 if 'volume' in self.user_params: 95 pct = self.user_params['volume'] 96 vol = self.android.droid.getMaxMediaVolume() * pct 97 self.android.droid.setMediaVolume(int(vol)) 98 # TODO (aidanhb): this is a weird way to work around the fact that the 99 # above SL4A commands don't actually max out the volume. Fix this. 100 if 'volume_up' in self.user_params: 101 for i in range(self.user_params['volume_up']): 102 self.android.adb.shell(ADB_VOL_UP) 103 104 def ensure_phone_has_music_file(self): 105 """Make sure music file (based on config values) is on the phone.""" 106 if not self.android.adb.shell(ADB_FILE_EXISTS % self.phone_music_file): 107 self.android.adb.push(self.host_music_file, self.phone_music_file) 108 has_file = self.android.adb.shell( 109 ADB_FILE_EXISTS % self.phone_music_file) 110 asserts.assert_true(has_file, 'Audio file not pushed to phone.', 111 extras=self.metrics) 112 self.log.info('Music file successfully pushed to phone.') 113 else: 114 self.log.info( 115 'Music file already on phone. Skipping file transfer.') 116 117 def play_and_record_audio(self): 118 """Play audio file on android phone and record through self.mic. 119 """ 120 playing = self.android.droid.mediaPlayOpen( 121 'file://%s' % self.phone_music_file, 122 'default', 123 True) 124 asserts.assert_true(playing, 125 'Failed to play file %s' % self.phone_music_file, 126 extras=self.metrics) 127 128 looping = self.android.droid.mediaPlaySetLooping(True) 129 if not looping: 130 self.log.warning('Could not loop %s' % self.phone_music_file) 131 132 try: 133 self.log.info('Capturing audio through %s' % 134 self.mic.name) 135 except AttributeError as e: 136 self.log.error('Mic not initialized correctly. Check your ' 137 '"input_device" parameter in config.') 138 raise e 139 audio_captured = self.mic.capture_audio(self.audio_params['trim']) 140 self.android.droid.mediaPlayStop() 141 stopped = not self.android.droid.mediaIsPlaying() 142 asserts.assert_true(audio_captured, 'Audio not recorded', 143 extras=self.metrics) 144 145 if stopped: 146 self.log.info('Finished playing audio.') 147 else: 148 self.log.warning('Failed to stop audio.') 149 150 def stream_music_on_codec(self, 151 codec_type, 152 sample_rate, 153 bits_per_sample, 154 channel_mode, 155 codec_specific_1=0): 156 """Pair phone and headset, set codec, and stream music file. 157 Ensure devices are connected and that music actually plays. 158 159 Args: 160 codec_type (str): the desired codec type. For reference, see 161 test_utils.bt.bt_constants.codec_types 162 sample_rate (int|str): the desired sample rate. For reference, see 163 test_utils.bt.bt_constants.sample_rates 164 bits_per_sample (int|str): the desired bits per sample. For 165 reference, see test_utils.bt.bt_constants.bits_per_samples 166 channel_mode (str): the desired channel mode. For reference, see 167 test_utils.bt.bt_constants.channel_modes 168 codec_specific_1: any codec specific value, such as LDAC quality. 169 """ 170 171 self.log.info('Setting Bluetooth codec to %s...' % codec_type) 172 codec_set = set_bluetooth_codec(android_device=self.android, 173 codec_type=codec_type, 174 sample_rate=sample_rate, 175 bits_per_sample=bits_per_sample, 176 channel_mode=channel_mode, 177 codec_specific_1=codec_specific_1) 178 asserts.assert_true(codec_set, 'Codec configuration failed.', 179 extras=self.metrics) 180 181 self.play_and_record_audio() 182 183 def run_thdn_analysis(self): 184 """Calculate Total Harmonic Distortion plus Noise for latest recording. 185 186 Store result in self.metrics. 187 """ 188 # Calculate Total Harmonic Distortion + Noise 189 thdn = self.mic.THDN(**self.audio_params['thdn_params']) 190 for ch_no, t in enumerate(thdn): 191 self.log.info('THD+N percent for channel %s: %.4f%%' % 192 (ch_no, t * 100)) 193 metrics_key = 'channel_%s_thdn' % ch_no 194 self.metrics[metrics_key] = t 195 self.metrics['thdn'] = thdn 196 return thdn 197 198 def run_anomaly_detection(self): 199 """Detect anomalies in latest recording. 200 201 Store result in self.metrics. 202 """ 203 # Detect Anomalies 204 anom = self.mic.detect_anomalies(**self.audio_params['anomaly_params']) 205 num_anom = 0 206 for ch_no, anomalies in enumerate(anom): 207 if anomalies: 208 for anomaly in anomalies: 209 num_anom += 1 210 start, end = anomaly 211 self.log.warning('Anomaly on channel {} at {}:{}. Duration ' 212 '{} sec'.format(ch_no, 213 start // 60, 214 start % 60, 215 end -start)) 216 metrics_key = 'channel_%s_num_anomalies' % ch_no 217 self.metrics[metrics_key] = len(anomalies) 218 else: 219 self.log.info('%i anomalies detected.' % num_anom) 220 self.metrics['anomalies'] = anom 221 return anom 222