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 itertools 20import json 21import re 22import numpy 23import os 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 30 31from functools import partial 32 33LONG_SLEEP = 10 34MEDIUM_SLEEP = 2 35IPERF_TIMEOUT = 10 36SHORT_SLEEP = 1 37SUBFRAME_LENGTH = 0.001 38STOP_COUNTER_LIMIT = 3 39 40 41class CellularFr2UplinkPowerSweepTest(CellularThroughputBaseTest): 42 """Base class to test cellular FR2 throughput 43 44 This class implements cellular FR2 throughput tests on a callbox setup. 45 The class setups up the callbox in the desired configurations, configures 46 and connects the phone, and runs traffic/iperf throughput. 47 """ 48 49 def __init__(self, controllers): 50 super().__init__(controllers) 51 base_test.BaseTestClass.__init__(self, controllers) 52 self.testcase_metric_logger = ( 53 BlackboxMappedMetricLogger.for_test_case()) 54 self.testclass_metric_logger = ( 55 BlackboxMappedMetricLogger.for_test_class()) 56 self.publish_testcase_metrics = True 57 58 def process_testcase_results(self): 59 """Publish test case metrics and save results""" 60 if self.current_test_name not in self.testclass_results: 61 return 62 testcase_data = self.testclass_results[self.current_test_name] 63 results_file_path = os.path.join( 64 context.get_current_context().get_full_output_path(), 65 '{}.json'.format(self.current_test_name)) 66 with open(results_file_path, 'w') as results_file: 67 json.dump(wputils.serialize_dict(testcase_data), 68 results_file, 69 indent=4) 70 testcase_result = testcase_data['results'][0] 71 72 def process_testclass_results(self): 73 pass 74 75 def get_per_cell_power_sweeps(self, testcase_params): 76 cell_power_sweeps = [] 77 for cell in testcase_params['endc_combo_config']['cell_list']: 78 if cell['cell_type'] == 'LTE': 79 sweep = [self.testclass_params['lte_cell_power']] 80 else: 81 sweep = [self.testclass_params['nr_cell_power']] 82 cell_power_sweeps.append(sweep) 83 return cell_power_sweeps 84 85 def generate_endc_combo_config(self, test_config): 86 """Function to generate ENDC combo config from CSV test config 87 88 Args: 89 test_config: dict containing ENDC combo config from CSV 90 Returns: 91 endc_combo_config: dictionary with all ENDC combo settings 92 """ 93 endc_combo_config = collections.OrderedDict() 94 cell_config_list = [] 95 96 lte_cell_count = 1 97 lte_carriers = [1] 98 lte_scc_list = [] 99 endc_combo_config['lte_pcc'] = 1 100 lte_cell = { 101 'cell_type': 'LTE', 102 'cell_number': 1, 103 'pcc': 1, 104 'band': self.testclass_params['lte_anchor_band'], 105 'dl_bandwidth': self.testclass_params['lte_anchor_bandwidth'], 106 'ul_enabled': 1, 107 'duplex_mode': self.testclass_params['lte_anchor_duplex_mode'], 108 'dl_mimo_config': 'D{nss}U{nss}'.format(nss=1), 109 'ul_mimo_config': 'D{nss}U{nss}'.format(nss=1), 110 'transmission_mode': 'TM1', 111 'num_codewords': 1, 112 'num_layers': 1, 113 'dl_subframe_allocation': [1] * 10, 114 } 115 cell_config_list.append(lte_cell) 116 117 nr_cell_count = 0 118 nr_dl_carriers = [] 119 nr_ul_carriers = [] 120 for nr_cell_idx in range(1, test_config['num_dl_cells'] + 1): 121 nr_cell = { 122 'cell_type': 123 'NR5G', 124 'cell_number': 125 nr_cell_idx, 126 'nr_cell_type': 127 'NSA', 128 'band': 129 test_config['nr_band'], 130 'duplex_mode': 131 test_config['nr_duplex_mode'], 132 'channel': 133 test_config['nr_channel'], 134 'dl_mimo_config': 135 'N{nss}X{nss}'.format(nss=test_config['nr_dl_mimo_config']), 136 'dl_bandwidth_class': 137 'A', 138 'dl_bandwidth': 139 test_config['nr_bandwidth'], 140 'ul_enabled': 141 1 if nr_cell_idx <= test_config['num_ul_cells'] else 0, 142 'ul_bandwidth_class': 143 'A', 144 'ul_mimo_config': 145 'N{nss}X{nss}'.format(nss=test_config['nr_ul_mimo_config']), 146 'subcarrier_spacing': 147 'MU3' 148 } 149 cell_config_list.append(nr_cell) 150 nr_cell_count = nr_cell_count + 1 151 nr_dl_carriers.append(nr_cell_idx) 152 if nr_cell_idx <= test_config['num_ul_cells']: 153 nr_ul_carriers.append(nr_cell_idx) 154 155 endc_combo_config['lte_cell_count'] = lte_cell_count 156 endc_combo_config['nr_cell_count'] = nr_cell_count 157 endc_combo_config['nr_dl_carriers'] = nr_dl_carriers 158 endc_combo_config['nr_ul_carriers'] = nr_ul_carriers 159 endc_combo_config['cell_list'] = cell_config_list 160 endc_combo_config['lte_scc_list'] = lte_scc_list 161 endc_combo_config['lte_dl_carriers'] = lte_carriers 162 endc_combo_config['lte_ul_carriers'] = lte_carriers 163 return endc_combo_config 164 165 def _test_throughput_bler_sweep_ul_power(self, testcase_params): 166 """Test function to run cellular throughput and BLER measurements. 167 168 The function runs BLER/throughput measurement after configuring the 169 callbox and DUT. The test supports running PHY or TCP/UDP layer traffic 170 in a variety of band/carrier/mcs/etc configurations. 171 172 Args: 173 testcase_params: dict containing test-specific parameters 174 Returns: 175 result: dict containing throughput results and metadata 176 """ 177 # Prepare results dicts 178 testcase_params = self.compile_test_params(testcase_params) 179 testcase_params['nr_target_power_sweep'] = list( 180 numpy.arange(self.testclass_params['nr_target_power_start'], 181 self.testclass_params['nr_target_power_stop'], 182 self.testclass_params['nr_target_power_step'])) 183 184 testcase_results = collections.OrderedDict() 185 testcase_results['testcase_params'] = testcase_params 186 testcase_results['results'] = [] 187 188 # Setup ota chamber if needed 189 if hasattr(self, 190 'keysight_chamber') and 'orientation' in testcase_params: 191 self.keysight_chamber.move_theta_phi_abs( 192 self.keysight_chamber.preset_orientations[ 193 testcase_params['orientation']]['theta'], 194 self.keysight_chamber.preset_orientations[ 195 testcase_params['orientation']]['phi']) 196 197 # Setup tester and wait for DUT to connect 198 self.setup_tester(testcase_params) 199 200 # Run throughput test loop 201 stop_counter = 0 202 if testcase_params['endc_combo_config']['nr_cell_count']: 203 self.keysight_test_app.select_display_tab('NR5G', 1, 'BTHR', 204 'OTAGRAPH') 205 else: 206 self.keysight_test_app.select_display_tab('LTE', 1, 'BTHR', 207 'OTAGRAPH') 208 for power_idx in range(len(testcase_params['nr_target_power_sweep'])): 209 result = collections.OrderedDict() 210 # Check that cells are still connected 211 connected = 1 212 for cell in testcase_params['endc_combo_config']['cell_list']: 213 if not self.keysight_test_app.wait_for_cell_status( 214 cell['cell_type'], cell['cell_number'], 215 ['ACT', 'CONN'], SHORT_SLEEP, SHORT_SLEEP): 216 connected = 0 217 if not connected: 218 self.log.info('DUT lost connection to cells. Ending test.') 219 break 220 # Set DL cell power 221 current_target_power = testcase_params['nr_target_power_sweep'][ 222 power_idx] 223 self.log.info( 224 'Setting target power to {}dBm'.format(current_target_power)) 225 for cell_idx, cell in enumerate( 226 testcase_params['endc_combo_config']['cell_list']): 227 self.keysight_test_app.set_cell_ul_power_control( 228 cell['cell_type'], cell['cell_number'], 'TARget', 229 current_target_power) 230 # Start BLER and throughput measurements 231 current_throughput = self.run_single_throughput_measurement( 232 testcase_params) 233 result['throughput_measurements'] = current_throughput 234 result['nr_target_power'] = current_target_power 235 self.print_throughput_result(current_throughput) 236 237 tx_power = self.dut_utils.get_fr2_tx_power() 238 239 testcase_results['results'].append(result) 240 if (('lte_bler_result' in result['throughput_measurements'] 241 and result['throughput_measurements']['lte_bler_result'] 242 ['total']['DL']['nack_ratio'] * 100 > 99) 243 or ('nr_bler_result' in result['throughput_measurements'] 244 and result['throughput_measurements']['nr_bler_result'] 245 ['total']['DL']['nack_ratio'] * 100 > 99)): 246 stop_counter = stop_counter + 1 247 else: 248 stop_counter = 0 249 if stop_counter == STOP_COUNTER_LIMIT: 250 break 251 252 # Save results 253 self.testclass_results[self.current_test_name] = testcase_results 254 255 def generate_test_cases(self, bands, channels, nr_mcs_pair_list, 256 num_dl_cells_list, num_ul_cells_list, 257 orientation_list, dl_mimo_config, ul_mimo_config, 258 **kwargs): 259 """Function that auto-generates test cases for a test class.""" 260 test_cases = [] 261 for orientation, band, channel, num_ul_cells, num_dl_cells, nr_mcs_pair in itertools.product( 262 orientation_list, bands, channels, num_ul_cells_list, 263 num_dl_cells_list, nr_mcs_pair_list): 264 if num_ul_cells > num_dl_cells: 265 continue 266 if channel not in cputils.PCC_PRESET_MAPPING[band]: 267 continue 268 test_config = { 269 'nr_band': band, 270 'nr_bandwidth': 'BW100', 271 'nr_duplex_mode': 'TDD', 272 'nr_cell_type': 'NSA', 273 'nr_channel': cputils.PCC_PRESET_MAPPING[band][channel], 274 'num_dl_cells': num_dl_cells, 275 'num_ul_cells': num_ul_cells, 276 'nr_dl_mimo_config': dl_mimo_config, 277 'nr_ul_mimo_config': ul_mimo_config 278 } 279 endc_combo_config = self.generate_endc_combo_config(test_config) 280 test_name = 'test_fr2_ul_power_sweep_{}_{}_{}_DL_{}CC_mcs{}_{}x{}_UL_{}CC_mcs{}_{}x{}'.format( 281 orientation, band, channel, num_dl_cells, nr_mcs_pair[0], 282 dl_mimo_config, dl_mimo_config, num_ul_cells, nr_mcs_pair[1], 283 ul_mimo_config, ul_mimo_config) 284 test_params = collections.OrderedDict( 285 endc_combo_config=endc_combo_config, 286 nr_dl_mcs=nr_mcs_pair[0], 287 nr_ul_mcs=nr_mcs_pair[1], 288 orientation=orientation, 289 **kwargs) 290 setattr( 291 self, test_name, 292 partial(self._test_throughput_bler_sweep_ul_power, 293 test_params)) 294 test_cases.append(test_name) 295 return test_cases 296 297 298class CellularFr2CpOfdmUplinkPowerSweepTest(CellularFr2UplinkPowerSweepTest): 299 300 def __init__(self, controllers): 301 super().__init__(controllers) 302 self.testclass_params = self.user_params[ 303 'fr2_uplink_power_sweep_test_params'] 304 self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 305 ['low', 'mid', 'high'], 306 [(4, 16), (4, 25), (4, 27), 307 (4, 28)], [1], [1], 308 ['A_Plane', 'B_Plane'], 309 force_contiguous_nr_channel=True, 310 dl_mimo_config=2, 311 ul_mimo_config=2, 312 schedule_scenario="FULL_TPUT", 313 schedule_slot_ratio=80, 314 traffic_direction='UL', 315 transform_precoding=0, 316 lte_dl_mcs=4, 317 lte_dl_mcs_table='QAM64', 318 lte_ul_mcs=4, 319 lte_ul_mcs_table='QAM64', 320 nr_dl_mcs_table='Q256', 321 nr_ul_mcs_table='Q64') 322 323 self.tests.extend( 324 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 325 ['low', 'mid', 'high'], [(4, 16), (4, 25), 326 (4, 27), 327 (4, 28)], [2], 328 [2], ['A_Plane', 'B_Plane'], 329 force_contiguous_nr_channel=True, 330 dl_mimo_config=2, 331 ul_mimo_config=2, 332 schedule_scenario="FULL_TPUT", 333 schedule_slot_ratio=80, 334 traffic_direction='UL', 335 transform_precoding=0, 336 lte_dl_mcs=4, 337 lte_dl_mcs_table='QAM64', 338 lte_ul_mcs=4, 339 lte_ul_mcs_table='QAM64', 340 nr_dl_mcs_table='Q256', 341 nr_ul_mcs_table='Q64')) 342 self.tests.extend( 343 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 344 ['low', 'mid', 'high'], [(4, 16), (4, 25), 345 (4, 27), 346 (4, 28)], [4], 347 [4], ['A_Plane', 'B_Plane'], 348 force_contiguous_nr_channel=True, 349 dl_mimo_config=2, 350 ul_mimo_config=2, 351 schedule_scenario="FULL_TPUT", 352 schedule_slot_ratio=80, 353 traffic_direction='UL', 354 transform_precoding=0, 355 lte_dl_mcs=4, 356 lte_dl_mcs_table='QAM64', 357 lte_ul_mcs=4, 358 lte_ul_mcs_table='QAM64', 359 nr_dl_mcs_table='Q256', 360 nr_ul_mcs_table='Q64')) 361 362 363class CellularFr2DftsOfdmUplinkPowerSweepTest(CellularFr2UplinkPowerSweepTest): 364 365 def __init__(self, controllers): 366 super().__init__(controllers) 367 self.testclass_params = self.user_params[ 368 'fr2_uplink_power_sweep_test_params'] 369 self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 370 ['low', 'mid', 'high'], 371 [(4, 16), (4, 25), (4, 27), 372 (4, 28)], [1], [1], 373 ['A_Plane', 'B_Plane'], 374 force_contiguous_nr_channel=True, 375 dl_mimo_config=2, 376 ul_mimo_config=1, 377 schedule_scenario="FULL_TPUT", 378 schedule_slot_ratio=80, 379 traffic_direction='UL', 380 transform_precoding=1, 381 lte_dl_mcs=4, 382 lte_dl_mcs_table='QAM64', 383 lte_ul_mcs=4, 384 lte_ul_mcs_table='QAM64', 385 nr_dl_mcs_table='Q256', 386 nr_ul_mcs_table='Q64') 387 388 self.tests.extend( 389 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 390 ['low', 'mid', 'high'], [(4, 16), (4, 25), 391 (4, 27), 392 (4, 28)], [2], 393 [2], ['A_Plane', 'B_Plane'], 394 force_contiguous_nr_channel=True, 395 dl_mimo_config=2, 396 ul_mimo_config=2, 397 schedule_scenario="FULL_TPUT", 398 schedule_slot_ratio=80, 399 traffic_direction='UL', 400 transform_precoding=1, 401 lte_dl_mcs=4, 402 lte_dl_mcs_table='QAM64', 403 lte_ul_mcs=4, 404 lte_ul_mcs_table='QAM64', 405 nr_dl_mcs_table='Q256', 406 nr_ul_mcs_table='Q64')) 407 self.tests.extend( 408 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 409 ['low', 'mid', 'high'], [(4, 16), (4, 25), 410 (4, 27), 411 (4, 28)], [4], 412 [4], ['A_Plane', 'B_Plane'], 413 force_contiguous_nr_channel=True, 414 dl_mimo_config=2, 415 ul_mimo_config=2, 416 schedule_scenario="FULL_TPUT", 417 schedule_slot_ratio=80, 418 traffic_direction='UL', 419 transform_precoding=1, 420 lte_dl_mcs=4, 421 lte_dl_mcs_table='QAM64', 422 lte_ul_mcs=4, 423 lte_ul_mcs_table='QAM64', 424 nr_dl_mcs_table='Q256', 425 nr_ul_mcs_table='Q64')) 426