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 csv 19import numpy 20import json 21import re 22import os 23import time 24from acts import context 25from acts import base_test 26from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 27from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils 28from acts_contrib.test_utils.cellular.performance.CellularThroughputBaseTest import CellularThroughputBaseTest 29from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 30from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure 31from functools import partial 32 33VERY_SHORT_SLEEP = 0.1 34SHORT_SLEEP = 1 35TWO_SECOND_SLEEP = 2 36MEDIUM_SLEEP = 3 37LONG_SLEEP = 10 38STOP_COUNTER_LIMIT = 3 39 40 41class CellularPageDecodeTest(CellularThroughputBaseTest): 42 """Class to test ENDC sensitivity""" 43 44 def __init__(self, controllers): 45 base_test.BaseTestClass.__init__(self, controllers) 46 self.testcase_metric_logger = ( 47 BlackboxMappedMetricLogger.for_test_case()) 48 self.testclass_metric_logger = ( 49 BlackboxMappedMetricLogger.for_test_class()) 50 self.publish_testcase_metrics = True 51 self.testclass_params = self.user_params['page_decode_test_params'] 52 self.tests = self.generate_test_cases() 53 54 def process_testcase_results(self): 55 if self.current_test_name not in self.testclass_results: 56 return 57 testcase_data = self.testclass_results[self.current_test_name] 58 results_file_path = os.path.join( 59 context.get_current_context().get_full_output_path(), 60 '{}.json'.format(self.current_test_name)) 61 with open(results_file_path, 'w') as results_file: 62 json.dump(wputils.serialize_dict(testcase_data), 63 results_file, 64 indent=4) 65 66 decode_probability_list = [] 67 average_power_list = [] 68 cell_power_list = testcase_data['testcase_params']['cell_power_sweep'][0] 69 for result in testcase_data['results']: 70 decode_probability_list.append(result['decode_probability']) 71 if self.power_monitor: 72 average_power_list.append(result['average_power']) 73 padding_len = len(cell_power_list) - len(decode_probability_list) 74 decode_probability_list.extend([0] * padding_len) 75 76 testcase_data['decode_probability_list'] = decode_probability_list 77 testcase_data['cell_power_list'] = cell_power_list 78 79 plot = BokehFigure( 80 title='Band {} - Page Decode Probability'.format(testcase_data['testcase_params']['endc_combo_config']['cell_list'][0]['band']), 81 x_label='Cell Power (dBm)', 82 primary_y_label='Decode Probability', 83 secondary_y_label='Power Consumption (mW)' 84 ) 85 86 plot.add_line( 87 testcase_data['cell_power_list'], 88 testcase_data['decode_probability_list'], 89 'Decode Probability', 90 width=1) 91 if self.power_monitor: 92 plot.add_line( 93 testcase_data['testcase_params']['cell_power_sweep'][0], 94 average_power_list, 95 'Power Consumption (mW)', 96 width=1, 97 style='dashdot', 98 y_axis='secondary') 99 plot.generate_figure() 100 output_file_path = os.path.join( 101 context.get_current_context().get_full_output_path(), 102 '{}.html'.format(self.current_test_name)) 103 BokehFigure.save_figure(plot, output_file_path) 104 105 def _test_page_decode(self, testcase_params): 106 """Test function to run cellular throughput and BLER measurements. 107 108 The function runs BLER/throughput measurement after configuring the 109 callbox and DUT. The test supports running PHY or TCP/UDP layer traffic 110 in a variety of band/carrier/mcs/etc configurations. 111 112 Args: 113 testcase_params: dict containing test-specific parameters 114 Returns: 115 result: dict containing throughput results and meta data 116 """ 117 # Prepare results dicts 118 testcase_params = self.compile_test_params(testcase_params) 119 testcase_results = collections.OrderedDict() 120 testcase_results['testcase_params'] = testcase_params 121 testcase_results['results'] = [] 122 123 # Setup ota chamber if needed 124 if hasattr(self, 125 'keysight_chamber') and 'orientation' in testcase_params: 126 self.keysight_chamber.move_theta_phi_abs( 127 self.keysight_chamber.preset_orientations[ 128 testcase_params['orientation']]['theta'], 129 self.keysight_chamber.preset_orientations[ 130 testcase_params['orientation']]['phi']) 131 132 # Setup tester and wait for DUT to connect 133 self.setup_tester(testcase_params) 134 # Put DUT to sleep for power measurements 135 self.dut_utils.go_to_sleep() 136 137 test_cell = testcase_params['endc_combo_config']['cell_list'][0] 138 139 # Release RRC connection 140 self.keysight_test_app.release_rrc_connection(test_cell['cell_type'], 141 test_cell['cell_number']) 142 # Set tester to ignore RACH 143 self.keysight_test_app.enable_rach(test_cell['cell_type'], 144 test_cell['cell_number'], 145 enabled=0) 146 self.keysight_test_app.enable_preamble_report(test_cell['cell_type'], 147 1) 148 stop_counter = 0 149 for power_idx in range(len(testcase_params['cell_power_sweep'][0])): 150 result = collections.OrderedDict() 151 # Set DL cell power 152 for cell_idx, cell in enumerate( 153 testcase_params['endc_combo_config']['cell_list']): 154 cell_power_array = [] 155 current_cell_power = testcase_params['cell_power_sweep'][ 156 cell_idx][power_idx] 157 cell_power_array.append(current_cell_power) 158 self.keysight_test_app.set_cell_dl_power( 159 cell['cell_type'], cell['cell_number'], current_cell_power, 160 0) 161 self.log.info('Cell Power: {}'.format(cell_power_array)) 162 result['cell_power'] = cell_power_array 163 # Start BLER and throughput measurements 164 if self.power_monitor: 165 measurement_wait = LONG_SLEEP if (power_idx == 0) else 0 166 average_power_future = self.collect_power_data_nonblocking( 167 min(10, self.testclass_params['num_measurements'])*MEDIUM_SLEEP, 168 measurement_wait, 169 reconnect_usb=0, 170 measurement_tag=power_idx) 171 decode_counter = 0 172 for idx in range(self.testclass_params['num_measurements']): 173 # Page device 174 self.keysight_test_app.send_rrc_paging( 175 test_cell['cell_type'], test_cell['cell_number']) 176 time.sleep(MEDIUM_SLEEP) 177 # Fetch page result 178 preamble_report = self.keysight_test_app.fetch_preamble_report( 179 test_cell['cell_type'], test_cell['cell_number']) 180 # If rach attempted, increment decode counter. 181 if preamble_report: 182 decode_counter = decode_counter + 1 183 self.log.info('Decode probability: {}/{}'.format(decode_counter, idx+1)) 184 result[ 185 'decode_probability'] = decode_counter / self.testclass_params[ 186 'num_measurements'] 187 if self.power_monitor: 188 average_power = average_power_future.result() 189 result['average_power'] = average_power 190 191 if self.testclass_params.get('log_rsrp_metrics', 1) and self.dut.is_connected(): 192 lte_rx_meas = self.dut_utils.get_rx_measurements('LTE') 193 nr_rx_meas = self.dut_utils.get_rx_measurements('NR5G') 194 result['lte_rx_measurements'] = lte_rx_meas 195 result['nr_rx_measurements'] = nr_rx_meas 196 self.log.info('LTE Rx Measurements: {}'.format(lte_rx_meas)) 197 self.log.info('NR Rx Measurements: {}'.format(nr_rx_meas)) 198 199 testcase_results['results'].append(result) 200 if result['decode_probability'] == 0: 201 stop_counter = stop_counter + 1 202 else: 203 stop_counter = 0 204 if stop_counter == STOP_COUNTER_LIMIT: 205 break 206 self.keysight_test_app.enable_rach(test_cell['cell_type'], 207 test_cell['cell_number'], 208 enabled=1) 209 210 # Save results 211 self.testclass_results[self.current_test_name] = testcase_results 212 213 def compile_test_params(self, testcase_params): 214 """Function that completes all test params based on the test name. 215 216 Args: 217 testcase_params: dict containing test-specific parameters 218 """ 219 # Cell power sweep 220 # TODO: Make this a function to support single power and sweep modes for each cell 221 testcase_params['cell_power_sweep'] = self.get_per_cell_power_sweeps( 222 testcase_params) 223 return testcase_params 224 225class CellularFr1PageDecodeTest(CellularPageDecodeTest): 226 227 def __init__(self, controllers): 228 base_test.BaseTestClass.__init__(self, controllers) 229 self.testcase_metric_logger = ( 230 BlackboxMappedMetricLogger.for_test_case()) 231 self.testclass_metric_logger = ( 232 BlackboxMappedMetricLogger.for_test_class()) 233 self.publish_testcase_metrics = True 234 self.testclass_params = self.user_params['page_decode_test_params'] 235 self.tests = self.generate_test_cases() 236 237 def get_per_cell_power_sweeps(self, testcase_params): 238 nr_cell_sweep = list( 239 numpy.arange(self.testclass_params['nr_cell_power_start'], 240 self.testclass_params['nr_cell_power_stop'], 241 self.testclass_params['nr_cell_power_step'])) 242 return [nr_cell_sweep] 243 244 def generate_test_cases(self, **kwargs): 245 test_cases = [] 246 with open(self.testclass_params['nr_single_cell_configs'], 247 'r') as csvfile: 248 test_configs = csv.DictReader(csvfile) 249 for test_config in test_configs: 250 if int(test_config['skip_test']): 251 continue 252 endc_combo_config = cputils.generate_endc_combo_config_from_csv_row( 253 test_config) 254 test_name = 'test_fr1_{}'.format(test_config['nr_band']) 255 test_params = collections.OrderedDict( 256 endc_combo_config=endc_combo_config, 257 lte_dl_mcs_table='QAM256', 258 lte_dl_mcs=4, 259 lte_ul_mcs_table='QAM256', 260 lte_ul_mcs=4, 261 nr_dl_mcs=4, 262 nr_ul_mcs=4, 263 transform_precoding=0, 264 nr_dl_mcs_table='Q256', 265 nr_ul_mcs_table='Q64', 266 **kwargs) 267 setattr(self, test_name, 268 partial(self._test_page_decode, test_params)) 269 test_cases.append(test_name) 270 return test_cases 271 272 273class CellularLtePageDecodeTest(CellularPageDecodeTest): 274 275 def __init__(self, controllers): 276 base_test.BaseTestClass.__init__(self, controllers) 277 self.testcase_metric_logger = ( 278 BlackboxMappedMetricLogger.for_test_case()) 279 self.testclass_metric_logger = ( 280 BlackboxMappedMetricLogger.for_test_class()) 281 self.publish_testcase_metrics = True 282 self.testclass_params = self.user_params['page_decode_test_params'] 283 self.tests = self.generate_test_cases() 284 285 def get_per_cell_power_sweeps(self, testcase_params): 286 lte_cell_sweep = list( 287 numpy.arange(self.testclass_params['lte_cell_power_start'], 288 self.testclass_params['lte_cell_power_stop'], 289 self.testclass_params['lte_cell_power_step'])) 290 cell_power_sweeps = [lte_cell_sweep] 291 return cell_power_sweeps 292 293 def generate_test_cases(self, **kwargs): 294 test_cases = [] 295 with open(self.testclass_params['lte_single_cell_configs'], 296 'r') as csvfile: 297 test_configs = csv.DictReader(csvfile) 298 for test_config in test_configs: 299 if int(test_config['skip_test']): 300 continue 301 endc_combo_config = cputils.generate_endc_combo_config_from_csv_row( 302 test_config) 303 test_name = 'test_lte_B{}'.format(test_config['lte_band']) 304 test_params = collections.OrderedDict( 305 endc_combo_config=endc_combo_config, 306 lte_dl_mcs_table='QAM256', 307 lte_dl_mcs=4, 308 lte_ul_mcs_table='QAM256', 309 lte_ul_mcs=4, 310 nr_dl_mcs=4, 311 nr_ul_mcs=4, 312 transform_precoding=0, 313 **kwargs) 314 setattr(self, test_name, 315 partial(self._test_page_decode, test_params)) 316 test_cases.append(test_name) 317 return test_cases 318