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.
16"""Stream music through connected device from phone test implementation."""
17import os
18import shutil
19import time
20
21import acts.test_utils.coex.audio_test_utils as atu
22import acts.test_utils.bt.bt_test_utils as btutils
23from acts import asserts
24from acts.test_utils.abstract_devices.bluetooth_handsfree_abstract_device import BluetoothHandsfreeAbstractDeviceFactory as bt_factory
25from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
26
27PHONE_MUSIC_FILE_DIRECTORY = '/sdcard/Music'
28INIT_ATTEN = 0
29
30
31class A2dpBaseTest(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    def setup_class(self):
43
44        super().setup_class()
45        self.dut = self.android_devices[0]
46        req_params = ['audio_params', 'music_files']
47        #'audio_params' is a dict, contains the audio device type, audio streaming
48        #settings such as volumn, duration, audio recording parameters such as
49        #channel, sampling rate/width, and thdn parameters for audio processing
50        self.unpack_userparams(req_params)
51        # Find music file and push it to the dut
52        music_src = self.music_files[0]
53        music_dest = PHONE_MUSIC_FILE_DIRECTORY
54        success = self.dut.push_system_file(music_src, music_dest)
55        if success:
56            self.music_file = os.path.join(PHONE_MUSIC_FILE_DIRECTORY,
57                                           os.path.basename(music_src))
58        # Initialize media_control class
59        self.media = btutils.MediaControlOverSl4a(self.dut, self.music_file)
60        # Set attenuator to minimum attenuation
61        if hasattr(self, 'attenuators'):
62            self.attenuator = self.attenuators[0]
63            self.attenuator.set_atten(INIT_ATTEN)
64        # Create the BTOE(Bluetooth-Other-End) device object
65        bt_devices = self.user_params.get('bt_devices', [])
66        if bt_devices:
67            attr, idx = bt_devices.split(':')
68            self.bt_device_controller = getattr(self, attr)[int(idx)]
69            self.bt_device = bt_factory().generate(self.bt_device_controller)
70        else:
71            self.log.error('No BT devices config is provided!')
72
73    def teardown_class(self):
74
75        super().teardown_class()
76        if hasattr(self, 'media'):
77            self.media.stop()
78        if hasattr(self, 'attenuator'):
79            self.attenuator.set_atten(INIT_ATTEN)
80        self.dut.droid.bluetoothFactoryReset()
81        self.bt_device.reset()
82        self.bt_device.power_off()
83        btutils.disable_bluetooth(self.dut.droid)
84
85    def setup_test(self):
86
87        super().setup_test()
88        # Initialize audio capture devices
89        self.audio_device = atu.get_audio_capture_device(
90            self.bt_device_controller, self.audio_params)
91        # Reset BT to factory defaults
92        self.dut.droid.bluetoothFactoryReset()
93        self.bt_device.reset()
94        self.bt_device.power_on()
95        btutils.enable_bluetooth(self.dut.droid, self.dut.ed)
96        btutils.connect_phone_to_headset(self.dut, self.bt_device, 60)
97        vol = self.dut.droid.getMaxMediaVolume() * self.audio_params['volume']
98        self.dut.droid.setMediaVolume(0)
99        time.sleep(1)
100        self.dut.droid.setMediaVolume(int(vol))
101
102    def teardown_test(self):
103
104        super().teardown_test()
105        self.dut.droid.bluetoothFactoryReset()
106        self.media.stop()
107        # Set Attenuator to the initial attenuation
108        if hasattr(self, 'attenuator'):
109            self.attenuator.set_atten(INIT_ATTEN)
110        self.bt_device.reset()
111        self.bt_device.power_off()
112        btutils.disable_bluetooth(self.dut.droid)
113
114    def play_and_record_audio(self, duration):
115        """Play and record audio for a set duration.
116
117        Args:
118            duration: duration in seconds for music playing
119        Returns:
120            audio_captured: captured audio file path
121        """
122
123        self.log.info('Play and record audio for {} second'.format(duration))
124        self.media.play()
125        self.audio_device.start()
126        time.sleep(duration)
127        audio_captured = self.audio_device.stop()
128        self.media.stop()
129        self.log.info('Audio play and record stopped')
130        asserts.assert_true(audio_captured, 'Audio not recorded')
131        return audio_captured
132
133    def run_thdn_analysis(self, audio_captured, tag):
134        """Calculate Total Harmonic Distortion plus Noise for latest recording.
135
136        Store result in self.metrics.
137
138        Args:
139            audio_captured: the captured audio file
140        Returns:
141            thdn: thdn value in a list
142        """
143        # Calculate Total Harmonic Distortion + Noise
144        audio_result = atu.AudioCaptureResult(audio_captured,
145                                              self.audio_params)
146        thdn = audio_result.THDN(**self.audio_params['thdn_params'])
147        file_name = tag + os.path.basename(audio_result.path)
148        file_new = os.path.join(os.path.dirname(audio_result.path), file_name)
149        shutil.copyfile(audio_result.path, file_new)
150        for ch_no, t in enumerate(thdn):
151            self.log.info('THD+N for channel %s: %.4f%%' % (ch_no, t * 100))
152        return thdn
153
154    def run_anomaly_detection(self, audio_captured):
155        """Detect anomalies in latest recording.
156
157        Store result in self.metrics.
158
159        Args:
160            audio_captured: the captured audio file
161        Returns:
162            anom: anom detected in the captured file
163        """
164        # Detect Anomalies
165        audio_result = atu.AudioCaptureResult(audio_captured)
166        anom = audio_result.detect_anomalies(
167            **self.audio_params['anomaly_params'])
168        num_anom = 0
169        for ch_no, anomalies in enumerate(anom):
170            if anomalies:
171                for anomaly in anomalies:
172                    num_anom += 1
173                    start, end = anomaly
174                    self.log.warning(
175                        'Anomaly on channel {} at {}:{}. Duration '
176                        '{} sec'.format(ch_no, start // 60, start % 60,
177                                        end - start))
178        else:
179            self.log.info('%i anomalies detected.' % num_anom)
180        return anom
181