1#!/usr/bin/env python3.4 2# 3# Copyright 2022 - 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 collections 18import itertools 19import json 20import numpy 21import os 22import time 23from acts import asserts 24from acts import context 25from acts import base_test 26from acts import utils 27from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 28from acts_contrib.test_utils.cellular.keysight_5g_testapp import Keysight5GTestApp 29from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils 30from acts_contrib.test_utils.cellular.performance.shannon_log_parser import ShannonLogger 31from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 32from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure 33from functools import partial 34 35 36class CellularRxPowerTest(base_test.BaseTestClass): 37 """Class to test cellular throughput.""" 38 39 def __init__(self, controllers): 40 base_test.BaseTestClass.__init__(self, controllers) 41 self.testcase_metric_logger = ( 42 BlackboxMappedMetricLogger.for_test_case()) 43 self.testclass_metric_logger = ( 44 BlackboxMappedMetricLogger.for_test_class()) 45 self.publish_testcase_metrics = True 46 self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 47 list(range(1, 9))) 48 49 def setup_class(self): 50 """Initializes common test hardware and parameters. 51 52 This function initializes hardwares and compiles parameters that are 53 common to all tests in this class. 54 """ 55 self.dut = self.android_devices[-1] 56 self.testclass_params = self.user_params['rx_power_params'] 57 self.keysight_test_app = Keysight5GTestApp( 58 self.user_params['Keysight5GTestApp']) 59 self.sdm_logger = ShannonLogger(self.dut) 60 self.testclass_results = collections.OrderedDict() 61 # Configure test retries 62 self.user_params['retry_tests'] = [self.__class__.__name__] 63 64 # Turn Airplane mode on 65 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 66 'Can not turn on airplane mode.') 67 68 def teardown_class(self): 69 self.log.info('Turning airplane mode on') 70 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 71 'Can not turn on airplane mode.') 72 self.keysight_test_app.set_cell_state('LTE', 1, 0) 73 self.keysight_test_app.destroy() 74 75 def setup_test(self): 76 self.dut_utils.start_pixel_logger() 77 78 def on_retry(self): 79 """Function to control test logic on retried tests. 80 81 This function is automatically executed on tests that are being 82 retried. In this case the function resets wifi, toggles it off and on 83 and sets a retry_flag to enable further tweaking the test logic on 84 second attempts. 85 """ 86 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 87 'Can not turn on airplane mode.') 88 if self.keysight_test_app.get_cell_state('LTE', 'CELL1'): 89 self.log.info('Turning LTE off.') 90 self.keysight_test_app.set_cell_state('LTE', 'CELL1', 0) 91 92 def teardown_test(self): 93 self.log.info('Turning airplane mode on') 94 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 95 'Can not turn on airplane mode.') 96 log_path = os.path.join( 97 context.get_current_context().get_full_output_path(), 'pixel_logs') 98 os.makedirs(log_path, exist_ok=True) 99 self.log.info(self.current_test_info) 100 self.testclass_results.setdefault(self.current_test_name, 101 collections.OrderedDict()) 102 self.testclass_results[self.current_test_name].setdefault( 103 'log_path', []) 104 self.testclass_results[self.current_test_name]['log_path'].append( 105 self.dut_utils.stop_pixel_logger(log_path)) 106 self.process_test_results() 107 108 def process_test_results(self): 109 test_result = self.testclass_results[self.current_test_name] 110 111 # Save output as text file 112 results_file_path = os.path.join( 113 self.log_path, '{}.json'.format(self.current_test_name)) 114 with open(results_file_path, 'w') as results_file: 115 json.dump(wputils.serialize_dict(test_result), 116 results_file, 117 indent=4) 118 # Plot and save 119 if test_result['log_path']: 120 log_data = self.sdm_logger.process_log(test_result['log_path'][-1]) 121 else: 122 return 123 figure = BokehFigure(title=self.current_test_name, 124 x_label='Cell Power Setting (dBm)', 125 primary_y_label='Time') 126 figure.add_line(log_data.lte.rsrp_time, log_data.lte.rsrp_rx0, 127 'LTE RSRP (Rx0)') 128 figure.add_line(log_data.lte.rsrp_time, log_data.lte.rsrp_rx1, 129 'LTE RSRP (Rx1)') 130 figure.add_line(log_data.lte.rsrp2_time, log_data.lte.rsrp2_rx0, 131 'LTE RSRP2 (Rx0)') 132 figure.add_line(log_data.lte.rsrp2_time, log_data.lte.rsrp2_rx1, 133 'LTE RSRP2 (Rx0)') 134 figure.add_line(log_data.nr.rsrp_time, log_data.nr.rsrp_rx0, 135 'NR RSRP (Rx0)') 136 figure.add_line(log_data.nr.rsrp_time, log_data.nr.rsrp_rx1, 137 'NR RSRP (Rx1)') 138 figure.add_line(log_data.nr.rsrp2_time, log_data.nr.rsrp2_rx0, 139 'NR RSRP2 (Rx0)') 140 figure.add_line(log_data.nr.rsrp2_time, log_data.nr.rsrp2_rx1, 141 'NR RSRP2 (Rx0)') 142 figure.add_line(log_data.fr2.rsrp0_time, log_data.fr2.rsrp0, 143 'NR RSRP (Rx0)') 144 figure.add_line(log_data.fr2.rsrp1_time, log_data.fr2.rsrp1, 145 'NR RSRP2 (Rx1)') 146 output_file_path = os.path.join( 147 self.log_path, '{}.html'.format(self.current_test_name)) 148 figure.generate_figure(output_file_path) 149 150 def _test_nr_rsrp(self, testcase_params): 151 """Test function to run cellular RSRP tests. 152 153 The function runs a sweep of cell powers while collecting pixel logs 154 for later postprocessing and RSRP analysis. 155 156 Args: 157 testcase_params: dict containing test-specific parameters 158 """ 159 160 result = collections.OrderedDict() 161 testcase_params['power_range_vector'] = list( 162 numpy.arange(self.testclass_params['cell_power_start'], 163 self.testclass_params['cell_power_stop'], 164 self.testclass_params['cell_power_step'])) 165 166 if not self.keysight_test_app.get_cell_state('LTE', 'CELL1'): 167 self.log.info('Turning LTE on.') 168 self.keysight_test_app.set_cell_state('LTE', 'CELL1', 1) 169 self.log.info('Turning off airplane mode') 170 asserts.assert_true(utils.force_airplane_mode(self.dut, False), 171 'Can not turn on airplane mode.') 172 173 for cell in testcase_params['dl_cell_list']: 174 self.keysight_test_app.set_cell_band('NR5G', cell, 175 testcase_params['band']) 176 # Consider configuring schedule quick config 177 self.keysight_test_app.set_nr_cell_schedule_scenario( 178 testcase_params['dl_cell_list'][0], 'BASIC') 179 self.keysight_test_app.set_dl_carriers(testcase_params['dl_cell_list']) 180 self.keysight_test_app.set_ul_carriers( 181 testcase_params['dl_cell_list'][0]) 182 self.log.info('Waiting for LTE and applying aggregation') 183 if not self.keysight_test_app.wait_for_cell_status( 184 'LTE', 'CELL1', 'CONN', 60): 185 asserts.fail('DUT did not connect to LTE.') 186 self.keysight_test_app.apply_carrier_agg() 187 self.log.info('Waiting for 5G connection') 188 connected = self.keysight_test_app.wait_for_cell_status( 189 'NR5G', testcase_params['dl_cell_list'][-1], ['ACT', 'CONN'], 60) 190 if not connected: 191 asserts.fail('DUT did not connect to NR.') 192 for cell_power in testcase_params['power_range_vector']: 193 self.log.info('Setting power to {} dBm'.format(cell_power)) 194 for cell in testcase_params['dl_cell_list']: 195 self.keysight_test_app.set_cell_dl_power( 196 'NR5G', cell, cell_power, True) 197 #measure RSRP 198 self.keysight_test_app.start_nr_rsrp_measurement( 199 testcase_params['dl_cell_list'], 200 self.testclass_params['rsrp_measurement_duration']) 201 time.sleep(self.testclass_params['rsrp_measurement_duration'] * 202 1.5 / 1000) 203 self.keysight_test_app.get_nr_rsrp_measurement_state( 204 testcase_params['dl_cell_list']) 205 self.keysight_test_app.get_nr_rsrp_measurement_results( 206 testcase_params['dl_cell_list']) 207 208 for cell in testcase_params['dl_cell_list'][::-1]: 209 self.keysight_test_app.set_cell_state('NR5G', cell, 0) 210 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 211 'Can not turn on airplane mode.') 212 # Save results 213 result['testcase_params'] = testcase_params 214 self.testclass_results[self.current_test_name] = result 215 results_file_path = os.path.join( 216 context.get_current_context().get_full_output_path(), 217 '{}.json'.format(self.current_test_name)) 218 with open(results_file_path, 'w') as results_file: 219 json.dump(wputils.serialize_dict(result), results_file, indent=4) 220 221 def generate_test_cases(self, bands, num_cells_list): 222 """Function that auto-generates test cases for a test class.""" 223 test_cases = [] 224 225 for band, num_cells in itertools.product(bands, num_cells_list): 226 test_name = 'test_nr_rsrp_{}_{}CC'.format(band, num_cells) 227 test_params = collections.OrderedDict(band=band, 228 num_cells=num_cells, 229 dl_cell_list=list( 230 range(1, num_cells + 1))) 231 setattr(self, test_name, partial(self._test_nr_rsrp, test_params)) 232 test_cases.append(test_name) 233 return test_cases 234