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 os 23from acts import context 24from acts import base_test 25from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 26from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils 27from acts_contrib.test_utils.cellular.performance.CellularThroughputBaseTest import CellularThroughputBaseTest 28from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 29 30from functools import partial 31 32LONG_SLEEP = 10 33MEDIUM_SLEEP = 2 34IPERF_TIMEOUT = 10 35SHORT_SLEEP = 1 36SUBFRAME_LENGTH = 0.001 37STOP_COUNTER_LIMIT = 3 38 39 40class CellularLtePlusFr1PeakThroughputTest(CellularThroughputBaseTest): 41 """Base class to test cellular LTE and FR1 throughput 42 43 This class implements cellular LTE & FR1 throughput tests on a callbox setup. 44 The class setups up the callbox in the desired configurations, configures 45 and connects the phone, and runs traffic/iperf throughput. 46 """ 47 48 def process_testcase_results(self): 49 """Publish test case metrics and save results""" 50 if self.current_test_name not in self.testclass_results: 51 return 52 testcase_data = self.testclass_results[self.current_test_name] 53 results_file_path = os.path.join( 54 context.get_current_context().get_full_output_path(), 55 '{}.json'.format(self.current_test_name)) 56 with open(results_file_path, 'w') as results_file: 57 json.dump(wputils.serialize_dict(testcase_data), 58 results_file, 59 indent=4) 60 testcase_result = testcase_data['results'][0] 61 metric_map = { 62 'tcp_udp_tput': testcase_result.get('iperf_throughput', 63 float('nan')) 64 } 65 if testcase_data['testcase_params']['endc_combo_config'][ 66 'nr_cell_count']: 67 metric_map.update({ 68 'nr_min_dl_tput': 69 testcase_result['throughput_measurements']['nr_tput_result']['total']['DL']['min_tput'], 70 'nr_max_dl_tput': 71 testcase_result['throughput_measurements']['nr_tput_result']['total']['DL']['max_tput'], 72 'nr_avg_dl_tput': 73 testcase_result['throughput_measurements']['nr_tput_result']['total']['DL'] 74 ['average_tput'], 75 'nr_theoretical_dl_tput': 76 testcase_result['throughput_measurements']['nr_tput_result']['total']['DL'] 77 ['theoretical_tput'], 78 'nr_dl_bler': 79 testcase_result['throughput_measurements']['nr_bler_result']['total']['DL']['nack_ratio'] 80 * 100, 81 'nr_min_dl_tput': 82 testcase_result['throughput_measurements']['nr_tput_result']['total']['UL']['min_tput'], 83 'nr_max_dl_tput': 84 testcase_result['throughput_measurements']['nr_tput_result']['total']['UL']['max_tput'], 85 'nr_avg_dl_tput': 86 testcase_result['throughput_measurements']['nr_tput_result']['total']['UL'] 87 ['average_tput'], 88 'nr_theoretical_dl_tput': 89 testcase_result['throughput_measurements']['nr_tput_result']['total']['UL'] 90 ['theoretical_tput'], 91 'nr_ul_bler': 92 testcase_result['throughput_measurements']['nr_bler_result']['total']['UL']['nack_ratio'] 93 * 100 94 }) 95 if testcase_data['testcase_params']['endc_combo_config'][ 96 'lte_cell_count']: 97 metric_map.update({ 98 'lte_min_dl_tput': 99 testcase_result['throughput_measurements']['lte_tput_result']['total']['DL']['min_tput'], 100 'lte_max_dl_tput': 101 testcase_result['throughput_measurements']['lte_tput_result']['total']['DL']['max_tput'], 102 'lte_avg_dl_tput': 103 testcase_result['throughput_measurements']['lte_tput_result']['total']['DL'] 104 ['average_tput'], 105 'lte_theoretical_dl_tput': 106 testcase_result['throughput_measurements']['lte_tput_result']['total']['DL'] 107 ['theoretical_tput'], 108 'lte_dl_bler': 109 testcase_result['throughput_measurements']['lte_bler_result']['total']['DL']['nack_ratio'] 110 * 100, 111 'lte_min_dl_tput': 112 testcase_result['throughput_measurements']['lte_tput_result']['total']['UL']['min_tput'], 113 'lte_max_dl_tput': 114 testcase_result['throughput_measurements']['lte_tput_result']['total']['UL']['max_tput'], 115 'lte_avg_dl_tput': 116 testcase_result['throughput_measurements']['lte_tput_result']['total']['UL'] 117 ['average_tput'], 118 'lte_theoretical_dl_tput': 119 testcase_result['throughput_measurements']['lte_tput_result']['total']['UL'] 120 ['theoretical_tput'], 121 'lte_ul_bler': 122 testcase_result['throughput_measurements']['lte_bler_result']['total']['UL']['nack_ratio'] 123 * 100 124 }) 125 if self.publish_testcase_metrics: 126 for metric_name, metric_value in metric_map.items(): 127 self.testcase_metric_logger.add_metric(metric_name, 128 metric_value) 129 130 def process_testclass_results(self): 131 """Saves CSV with all test results to enable comparison.""" 132 results_file_path = os.path.join( 133 context.get_current_context().get_full_output_path(), 134 'results.csv') 135 with open(results_file_path, 'w', newline='') as csvfile: 136 field_names = [ 137 'Test Name', 'NR DL Min. Throughput', 'NR DL Max. Throughput', 138 'NR DL Avg. Throughput', 'NR DL Theoretical Throughput', 139 'NR UL Min. Throughput', 'NR UL Max. Throughput', 140 'NR UL Avg. Throughput', 'NR UL Theoretical Throughput', 141 'NR DL BLER (%)', 'NR UL BLER (%)', 'LTE DL Min. Throughput', 142 'LTE DL Max. Throughput', 'LTE DL Avg. Throughput', 143 'LTE DL Theoretical Throughput', 'LTE UL Min. Throughput', 144 'LTE UL Max. Throughput', 'LTE UL Avg. Throughput', 145 'LTE UL Theoretical Throughput', 'LTE DL BLER (%)', 146 'LTE UL BLER (%)', 'TCP/UDP Throughput' 147 ] 148 writer = csv.DictWriter(csvfile, fieldnames=field_names) 149 writer.writeheader() 150 151 for testcase_name, testcase_results in self.testclass_results.items( 152 ): 153 for result in testcase_results['results']: 154 row_dict = { 155 'Test Name': testcase_name, 156 'TCP/UDP Throughput': 157 result.get('iperf_throughput', 0) 158 } 159 if testcase_results['testcase_params'][ 160 'endc_combo_config']['nr_cell_count']: 161 row_dict.update({ 162 'NR DL Min. Throughput': 163 result['throughput_measurements']['nr_tput_result']['total']['DL'] 164 ['min_tput'], 165 'NR DL Max. Throughput': 166 result['throughput_measurements']['nr_tput_result']['total']['DL'] 167 ['max_tput'], 168 'NR DL Avg. Throughput': 169 result['throughput_measurements']['nr_tput_result']['total']['DL'] 170 ['average_tput'], 171 'NR DL Theoretical Throughput': 172 result['throughput_measurements']['nr_tput_result']['total']['DL'] 173 ['theoretical_tput'], 174 'NR UL Min. Throughput': 175 result['throughput_measurements']['nr_tput_result']['total']['UL'] 176 ['min_tput'], 177 'NR UL Max. Throughput': 178 result['throughput_measurements']['nr_tput_result']['total']['UL'] 179 ['max_tput'], 180 'NR UL Avg. Throughput': 181 result['throughput_measurements']['nr_tput_result']['total']['UL'] 182 ['average_tput'], 183 'NR UL Theoretical Throughput': 184 result['throughput_measurements']['nr_tput_result']['total']['UL'] 185 ['theoretical_tput'], 186 'NR DL BLER (%)': 187 result['throughput_measurements']['nr_bler_result']['total']['DL'] 188 ['nack_ratio'] * 100, 189 'NR UL BLER (%)': 190 result['throughput_measurements']['nr_bler_result']['total']['UL'] 191 ['nack_ratio'] * 100 192 }) 193 if testcase_results['testcase_params'][ 194 'endc_combo_config']['lte_cell_count']: 195 row_dict.update({ 196 'LTE DL Min. Throughput': 197 result['throughput_measurements']['lte_tput_result']['total']['DL'] 198 ['min_tput'], 199 'LTE DL Max. Throughput': 200 result['throughput_measurements']['lte_tput_result']['total']['DL'] 201 ['max_tput'], 202 'LTE DL Avg. Throughput': 203 result['throughput_measurements']['lte_tput_result']['total']['DL'] 204 ['average_tput'], 205 'LTE DL Theoretical Throughput': 206 result['throughput_measurements']['lte_tput_result']['total']['DL'] 207 ['theoretical_tput'], 208 'LTE UL Min. Throughput': 209 result['throughput_measurements']['lte_tput_result']['total']['UL'] 210 ['min_tput'], 211 'LTE UL Max. Throughput': 212 result['throughput_measurements']['lte_tput_result']['total']['UL'] 213 ['max_tput'], 214 'LTE UL Avg. Throughput': 215 result['throughput_measurements']['lte_tput_result']['total']['UL'] 216 ['average_tput'], 217 'LTE UL Theoretical Throughput': 218 result['throughput_measurements']['lte_tput_result']['total']['UL'] 219 ['theoretical_tput'], 220 'LTE DL BLER (%)': 221 result['throughput_measurements']['lte_bler_result']['total']['DL'] 222 ['nack_ratio'] * 100, 223 'LTE UL BLER (%)': 224 result['throughput_measurements']['lte_bler_result']['total']['UL'] 225 ['nack_ratio'] * 100 226 }) 227 writer.writerow(row_dict) 228 229 def get_per_cell_power_sweeps(self, testcase_params): 230 """Function to get per cell power sweep lists 231 232 Args: 233 testcase_params: dict containing all test case params 234 Returns: 235 cell_power_sweeps: list of cell power sweeps for each cell under test 236 """ 237 cell_power_sweeps = [] 238 for cell in testcase_params['endc_combo_config']['cell_list']: 239 if cell['cell_type'] == 'LTE': 240 sweep = [self.testclass_params['lte_cell_power']] 241 else: 242 sweep = [self.testclass_params['nr_cell_power']] 243 cell_power_sweeps.append(sweep) 244 return cell_power_sweeps 245 246 247class CellularLteFr1EndcPeakThroughputTest(CellularLtePlusFr1PeakThroughputTest 248 ): 249 """Class to test cellular LTE/FR1 ENDC combo list""" 250 251 def __init__(self, controllers): 252 base_test.BaseTestClass.__init__(self, controllers) 253 self.testcase_metric_logger = ( 254 BlackboxMappedMetricLogger.for_test_case()) 255 self.testclass_metric_logger = ( 256 BlackboxMappedMetricLogger.for_test_class()) 257 self.publish_testcase_metrics = True 258 self.testclass_params = self.user_params['throughput_test_params'] 259 self.tests = self.generate_test_cases([(27, 4), (4, 27)], 260 lte_dl_mcs_table='QAM256', 261 lte_ul_mcs_table='QAM256', 262 transform_precoding=0, 263 schedule_scenario='FULL_TPUT', 264 schedule_slot_ratio=80, 265 nr_dl_mcs_table='Q256', 266 nr_ul_mcs_table='Q64') 267 268 def generate_test_cases(self, mcs_pair_list, **kwargs): 269 test_cases = [] 270 271 with open(self.testclass_params['endc_combo_file'], 272 'r') as endc_combos: 273 for endc_combo_str in endc_combos: 274 if endc_combo_str[0] == '#': 275 continue 276 endc_combo_config = cputils.generate_endc_combo_config_from_string( 277 endc_combo_str) 278 special_chars = '+[]=;,\n' 279 for char in special_chars: 280 endc_combo_str = endc_combo_str.replace(char, '_') 281 endc_combo_str = endc_combo_str.replace('__', '_') 282 endc_combo_str = endc_combo_str.strip('_') 283 for mcs_pair in mcs_pair_list: 284 test_name = 'test_lte_fr1_endc_{}_dl_mcs{}_ul_mcs{}'.format( 285 endc_combo_str, mcs_pair[0], mcs_pair[1]) 286 test_params = collections.OrderedDict( 287 endc_combo_config=endc_combo_config, 288 nr_dl_mcs=mcs_pair[0], 289 nr_ul_mcs=mcs_pair[1], 290 lte_dl_mcs=mcs_pair[0], 291 lte_ul_mcs=mcs_pair[1], 292 **kwargs) 293 setattr(self, test_name, 294 partial(self._test_throughput_bler, test_params)) 295 test_cases.append(test_name) 296 return test_cases 297 298 299class CellularFr1SingleCellPeakThroughputTest(CellularLtePlusFr1PeakThroughputTest 300 ): 301 """Class to test single cell FR1 NSA mode""" 302 303 def __init__(self, controllers): 304 base_test.BaseTestClass.__init__(self, controllers) 305 self.testcase_metric_logger = ( 306 BlackboxMappedMetricLogger.for_test_case()) 307 self.testclass_metric_logger = ( 308 BlackboxMappedMetricLogger.for_test_class()) 309 self.publish_testcase_metrics = True 310 self.testclass_params = self.user_params['throughput_test_params'] 311 self.tests = self.generate_test_cases( 312 nr_mcs_pair_list=[(27, 4), (4, 27)], 313 nr_channel_list=['LOW', 'MID', 'HIGH'], 314 schedule_scenario='FULL_TPUT', 315 schedule_slot_ratio=80, 316 transform_precoding=0, 317 lte_dl_mcs=4, 318 lte_dl_mcs_table='QAM256', 319 lte_ul_mcs=4, 320 lte_ul_mcs_table='QAM64', 321 nr_dl_mcs_table='Q256', 322 nr_ul_mcs_table='Q64') 323 324 def generate_test_cases(self, nr_mcs_pair_list, nr_channel_list, **kwargs): 325 326 test_cases = [] 327 with open(self.testclass_params['nr_single_cell_configs'], 328 'r') as csvfile: 329 test_configs = csv.DictReader(csvfile) 330 for test_config, nr_channel, nr_mcs_pair in itertools.product( 331 test_configs, nr_channel_list, nr_mcs_pair_list): 332 if int(test_config['skip_test']): 333 continue 334 endc_combo_config = cputils.generate_endc_combo_config_from_csv_row( 335 test_config) 336 endc_combo_config['cell_list'][endc_combo_config['lte_cell_count']]['channel'] = nr_channel 337 test_name = 'test_fr1_{}_{}_dl_mcs{}_ul_mcs{}'.format( 338 test_config['nr_band'], nr_channel.lower(), nr_mcs_pair[0], 339 nr_mcs_pair[1]) 340 test_params = collections.OrderedDict( 341 endc_combo_config=endc_combo_config, 342 nr_dl_mcs=nr_mcs_pair[0], 343 nr_ul_mcs=nr_mcs_pair[1], 344 **kwargs) 345 setattr(self, test_name, 346 partial(self._test_throughput_bler, test_params)) 347 test_cases.append(test_name) 348 return test_cases 349 350 351class CellularLteSingleCellPeakThroughputTest(CellularLtePlusFr1PeakThroughputTest 352 ): 353 """Class to test single cell LTE""" 354 355 def __init__(self, controllers): 356 base_test.BaseTestClass.__init__(self, controllers) 357 self.testcase_metric_logger = ( 358 BlackboxMappedMetricLogger.for_test_case()) 359 self.testclass_metric_logger = ( 360 BlackboxMappedMetricLogger.for_test_class()) 361 self.publish_testcase_metrics = True 362 self.testclass_params = self.user_params['throughput_test_params'] 363 self.tests = self.generate_test_cases(lte_mcs_pair_list=[ 364 (('QAM256', 28), ('QAM256', 23)), 365 (('QAM256', 27), ('QAM256', 4)), (('QAM256', 4), ('QAM256', 27)) 366 ], 367 transform_precoding=0) 368 369 def generate_test_cases(self, lte_mcs_pair_list, **kwargs): 370 test_cases = [] 371 with open(self.testclass_params['lte_single_cell_configs'], 372 'r') as csvfile: 373 test_configs = csv.DictReader(csvfile) 374 for test_config, lte_mcs_pair in itertools.product( 375 test_configs, lte_mcs_pair_list): 376 if int(test_config['skip_test']): 377 continue 378 endc_combo_config = cputils.generate_endc_combo_config_from_csv_row( 379 test_config) 380 test_name = 'test_lte_B{}_dl_{}_mcs{}_ul_{}_mcs{}'.format( 381 test_config['lte_band'], lte_mcs_pair[0][0], 382 lte_mcs_pair[0][1], lte_mcs_pair[1][0], lte_mcs_pair[1][1]) 383 test_params = collections.OrderedDict( 384 endc_combo_config=endc_combo_config, 385 lte_dl_mcs_table=lte_mcs_pair[0][0], 386 lte_dl_mcs=lte_mcs_pair[0][1], 387 lte_ul_mcs_table=lte_mcs_pair[1][0], 388 lte_ul_mcs=lte_mcs_pair[1][1], 389 **kwargs) 390 setattr(self, test_name, 391 partial(self._test_throughput_bler, test_params)) 392 test_cases.append(test_name) 393 return test_cases 394