• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 time
17import os
18import logging
19import acts_contrib.test_utils.bt.bt_test_utils as btutils
20import acts_contrib.test_utils.coex.audio_test_utils as atu
21from acts import asserts
22from acts_contrib.test_utils.bt.A2dpBaseTest import A2dpBaseTest
23from acts.signals import TestFailure
24
25INIT_ATTEN = 0
26WAIT_TIME = 2
27
28
29class BtA2dpDynamicChannelTest(A2dpBaseTest):
30    def __init__(self, configs):
31        super().__init__(configs)
32        req_params = ['codecs', 'rssi_profile_params']
33        # 'rssi_profile_params' is a dict containing,a list of upper_bound,
34        # lower_bound attenuation values, Dwell time for RSSI and the test duration
35        # ex:- "rssi_profile_params": {
36        #   "upper_bound": [15, 25],
37        #    "RSSI_Dwell_time": [1, 1],
38        #    "lower_bound": [35, 45],
39        #    "test_duration": 30}
40        # 'codecs' is a list containing all codecs required in the tests
41        self.unpack_userparams(req_params)
42        self.upper_bound = self.rssi_profile_params['upper_bound']
43        self.lower_bound = self.rssi_profile_params['lower_bound']
44        self.dwell_time = self.rssi_profile_params['RSSI_Dwell_time']
45        for upper_bound, lower_bound, dwell_time in zip(
46                self.upper_bound, self.lower_bound, self.dwell_time):
47            for codec_config in self.codecs:
48                self.generate_test_case(codec_config, upper_bound, lower_bound,
49                                        dwell_time)
50
51    def setup_class(self):
52        super().setup_class()
53        # Enable BQR on all android devices
54        btutils.enable_bqr(self.android_devices)
55        self.log_path = os.path.join(logging.log_path, 'results')
56
57    def teardown_class(self):
58        super().teardown_class()
59
60    def generate_test_case(self, codec_config, upper_bound, lower_bound,
61                           dwell_time):
62        def test_case_fn():
63            self.check_audio_quality_dynamic_rssi(upper_bound, lower_bound,
64                                                  dwell_time)
65
66        test_case_name = 'test_bt_a2dp_Dynamic_channel_between_attenuation_{}dB_and_{}dB' \
67                         '_codec_{}'.format(upper_bound, lower_bound, codec_config['codec_type'])
68        setattr(self, test_case_name, test_case_fn)
69
70    def check_audio_quality_dynamic_rssi(self, upper_bound, lower_bound,
71                                         dwell_time):
72        tag = 'Dynamic_RSSI'
73        self.media.play()
74        proc = self.audio_device.start()
75        self.inject_rssi_profile(upper_bound, lower_bound, dwell_time)
76        proc.kill()
77        time.sleep(WAIT_TIME)
78        proc.kill()
79        audio_captured = self.audio_device.stop()
80        self.media.stop()
81        self.log.info('Audio play and record stopped')
82        asserts.assert_true(audio_captured, 'Audio not recorded')
83        audio_result = atu.AudioCaptureResult(audio_captured,
84                                              self.audio_params)
85        thdn = audio_result.THDN(**self.audio_params['thdn_params'])
86        self.log.info('THDN is {}'.format(thdn[0]))
87        # Reading DUT RSSI to check the RSSI fluctuation from
88        # upper and lower bound attenuation values
89        self.attenuator.set_atten(upper_bound)
90        [
91            rssi_master, pwl_master, rssi_c0_master, rssi_c1_master,
92            txpw_c0_master, txpw_c1_master, bftx_master, divtx_master
93        ], [rssi_slave] = self._get_bt_link_metrics(tag)
94        rssi_l1 = rssi_master.get(self.dut.serial, -127)
95        pwlv_l1 = pwl_master.get(self.dut.serial, -127)
96        self.attenuator.set_atten(lower_bound)
97        [
98            rssi_master, pwl_master, rssi_c0_master, rssi_c1_master,
99            txpw_c0_master, txpw_c1_master, bftx_master, divtx_master
100        ], [rssi_slave] = self._get_bt_link_metrics(tag)
101        rssi_l2 = rssi_master.get(self.dut.serial, -127)
102        pwlv_l2 = pwl_master.get(self.dut.serial, -127)
103        self.log.info(
104            "DUT RSSI is fluctuating between {} and {} dBm with {}sec interval"
105            .format(rssi_l1, rssi_l2, dwell_time))
106        if thdn[0] > self.audio_params['thdn_threshold'] or thdn[0] == 0:
107            raise TestFailure('Observed audio glitches!')
108
109    def inject_rssi_profile(self, upper_bound, lower_bound, dwell_time):
110        end_time = time.time() + self.rssi_profile_params['test_duration']
111        self.log.info("Testing dynamic channel RSSI")
112        while time.time() < end_time:
113            self.attenuator.set_atten(upper_bound)
114            time.sleep(dwell_time)
115            self.attenuator.set_atten(lower_bound)
116            time.sleep(dwell_time)
117