1#!/usr/bin/env python3
2#
3#   Copyright 2019 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the 'License');
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of 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,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import os
18import time
19import numpy as np
20import acts.test_utils.bt.bt_test_utils as bt_utils
21import acts.test_utils.wifi.wifi_performance_test_utils as wifi_utils
22
23from acts import asserts
24from functools import partial
25from acts.test_utils.bt.BtSarBaseTest import BtSarBaseTest
26
27
28class BtSarTpcTest(BtSarBaseTest):
29    """ Class to define BT SAR Transmit power control tests in
30
31    This class defines tests to detect any anomalies in the
32    Transmit power control mechanisms
33    """
34    def __init__(self, controllers):
35        super().__init__(controllers)
36        req_params = ['scenario_count']
37        self.unpack_userparams(req_params)
38        self.tests = self.generate_test_cases()
39
40    def setup_class(self):
41        super().setup_class()
42
43        # Pushing custom table
44        self.backup_sar_path = os.path.join(self.dut.device_log_path,
45                                            self.BACKUP_BT_SAR_TABLE_NAME)
46        self.dut.adb.pull(self.sar_file_path, self.backup_sar_path)
47        self.push_table(self.dut, self.custom_sar_path)
48
49        self.attenuator.set_atten(self.atten_min)
50        self.pathloss = int(self.calibration_params['pathloss'])
51
52        self.sar_df = self.bt_sar_df.copy()
53        self.sar_df['power_cap'] = -128
54        self.sar_df['TPC_result'] = -1
55
56    def setup_test(self):
57        super().setup_test()
58
59        self.tpc_sweep_range = range(self.atten_min, self.pl10_atten)
60        self.log.info(self.current_test_name)
61        self.tpc_plots_figure = wifi_utils.BokehFigure(
62            title='{}_{}'.format(self.current_test_name, 'curve'),
63            x_label='Pathloss(dBm)',
64            primary_y_label='Tx Power(dBm)')
65
66        self.tpc_plots_derivative_figure = wifi_utils.BokehFigure(
67            title='{}_{}'.format(self.current_test_name, 'curve_derivative'),
68            x_label='Pathloss(dBm)',
69            primary_y_label='Tx Power(dB)')
70
71    def teardown_class(self):
72        super().teardown_class()
73        result_file_name = '{}.csv'.format(self.__class__.__name__)
74        result_file_path = os.path.join(self.log_path, result_file_name)
75        self.sar_df.to_csv(result_file_path)
76
77        # Pushing default table back
78        self.push_table(self.dut, self.backup_sar_path)
79
80    def generate_test_cases(self):
81        """Function to generate test cases.
82        Function to generate a test case per BT SAR table row.
83
84        Returns: list of generated test cases.
85        """
86        test_cases = []
87        for scenario in range(int(self.scenario_count)):
88            test_name = 'test_bt_sar_tpc_{}'.format(scenario)
89            setattr(self, test_name, partial(self._test_bt_sar_tpc, scenario))
90            test_cases.append(test_name)
91        return test_cases
92
93    def process_tpc_results(self, tx_power_list, pwlv_list):
94        """Processes the results of tpc sweep.
95
96        Processes tpc sweep to ensure that tx power changes
97        as expected while sweeping.
98
99        Args:
100            tx_power_list: list of tx power measured during the TPC sweep.
101            pwlv_list: list of power levels observed during the TPC sweep.
102
103        Returns:
104            tpc_verdict : result of the tpc sweep; PASS/FAIL.
105            tx_power_derivative : peaks observed during TPC sweep.
106        """
107
108        tpc_verdict = 'FAIL'
109
110        # Locating power level changes in the sweep
111        pwlv_derivative_bool = list(np.diff([pwlv_list[0]] + pwlv_list) == 1)
112
113        # Diff-ing the list to get derivative of the list
114        tx_power_derivative = np.diff([tx_power_list[0]] + tx_power_list)
115
116        # Checking for negative peaks
117        if tx_power_derivative.min() < self.tpc_threshold['negative']:
118            return [tpc_verdict, tx_power_derivative]
119
120        # Locating legitimate tx power changes
121        tx_power_derivative_bool = [
122            self.tpc_threshold['positive'][0] < x <
123            self.tpc_threshold['positive'][1] for x in tx_power_derivative
124        ]
125
126        # Ensuring that changes in power level correspond to tx power changes
127        if pwlv_derivative_bool == tx_power_derivative_bool:
128            tpc_verdict = 'PASS'
129            return [tpc_verdict, tx_power_derivative]
130
131        return [tpc_verdict, tx_power_derivative]
132
133    def _test_bt_sar_tpc(self, scenario):
134        """Performs TCP sweep for the given scenario.
135
136        Function performs and evaluates TPC sweep for a given scenario.
137
138        Args:
139            scenario: row of the BT SAR table.
140        """
141
142        master_tx_power_list = []
143        pwlv_list = []
144
145        # Reading BT SAR Scenario from the table
146        start_time = self.dut.adb.shell('date +%s.%m')
147        time.sleep(1)
148
149        # Forcing the SAR state
150        read_scenario = self.sar_df.loc[scenario].to_dict()
151        self.set_sar_state(self.dut, read_scenario, self.country_code)
152
153        # Reading power cap
154        self.sar_df.loc[scenario, 'power_cap'] = self.get_current_power_cap(
155            self.dut, start_time)
156
157        # TPC sweep
158        for atten in self.tpc_sweep_range:
159
160            self.attenuator.set_atten(atten)
161            self.log.info('Current TPC attenuation {} dB; scenario {}'.format(
162                atten, scenario))
163
164            processed_bqr = bt_utils.get_bt_metric(self.android_devices,
165                                                   self.duration)
166            # Recording master rssi and pwlv
167            master_tx_power_list.append(
168                processed_bqr['rssi'][self.bt_device_controller.serial] +
169                self.pathloss + atten)
170            pwlv_list.append(processed_bqr['pwlv'][self.dut.serial])
171
172        # Processing tpc sweep results
173        [self.sar_df.loc[scenario, 'TPC_result'], tx_power_derivative
174         ] = self.process_tpc_results(master_tx_power_list, pwlv_list)
175
176        # Plot TPC curves
177        self.tpc_plots_figure.add_line(self.tpc_sweep_range,
178                                       master_tx_power_list,
179                                       str(scenario),
180                                       marker='circle')
181
182        self.tpc_plots_derivative_figure.add_line(self.tpc_sweep_range,
183                                                  tx_power_derivative,
184                                                  str(scenario),
185                                                  marker='circle')
186
187        self.tpc_plots_figure.generate_figure()
188        self.tpc_plots_derivative_figure.generate_figure()
189
190        # Saving the TPC curves
191        plot_file_path = os.path.join(self.log_path,
192                                      '{}.html'.format(self.current_test_name))
193        wifi_utils.BokehFigure.save_figures(
194            [self.tpc_plots_figure, self.tpc_plots_derivative_figure],
195            plot_file_path)
196
197        # Asserting pass and fail
198        if self.sar_df.loc[scenario, 'TPC_result'] == 'FAIL':
199            asserts.fail('TPC sweep failed for scenario: {}'.format(scenario))
200
201        else:
202            asserts.explicit_pass(
203                'TPC sweep passed for scenario: {}'.format(scenario))
204