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 266 def generate_test_cases(self, mcs_pair_list, **kwargs): 267 test_cases = [] 268 269 with open(self.testclass_params['endc_combo_file'], 270 'r') as endc_combos: 271 for endc_combo_str in endc_combos: 272 if endc_combo_str[0] == '#': 273 continue 274 endc_combo_config = cputils.generate_endc_combo_config_from_string( 275 endc_combo_str) 276 special_chars = '+[]=;,\n' 277 for char in special_chars: 278 endc_combo_str = endc_combo_str.replace(char, '_') 279 endc_combo_str = endc_combo_str.replace('__', '_') 280 endc_combo_str = endc_combo_str.strip('_') 281 for mcs_pair in mcs_pair_list: 282 test_name = 'test_lte_fr1_endc_{}_dl_mcs{}_ul_mcs{}'.format( 283 endc_combo_str, mcs_pair[0], mcs_pair[1]) 284 test_params = collections.OrderedDict( 285 endc_combo_config=endc_combo_config, 286 nr_dl_mcs=mcs_pair[0], 287 nr_ul_mcs=mcs_pair[1], 288 lte_dl_mcs=mcs_pair[0], 289 lte_ul_mcs=mcs_pair[1], 290 **kwargs) 291 setattr(self, test_name, 292 partial(self._test_throughput_bler, test_params)) 293 test_cases.append(test_name) 294 return test_cases 295 296 297class CellularFr1SingleCellPeakThroughputTest(CellularLtePlusFr1PeakThroughputTest 298 ): 299 """Class to test single cell FR1 NSA mode""" 300 301 def __init__(self, controllers): 302 base_test.BaseTestClass.__init__(self, controllers) 303 self.testcase_metric_logger = ( 304 BlackboxMappedMetricLogger.for_test_case()) 305 self.testclass_metric_logger = ( 306 BlackboxMappedMetricLogger.for_test_class()) 307 self.publish_testcase_metrics = True 308 self.testclass_params = self.user_params['throughput_test_params'] 309 self.tests = self.generate_test_cases( 310 nr_mcs_pair_list=[(27, 4), (4, 27)], 311 nr_channel_list=['LOW', 'MID', 'HIGH'], 312 schedule_scenario='FULL_TPUT', 313 schedule_slot_ratio=80, 314 transform_precoding=0, 315 lte_dl_mcs=4, 316 lte_dl_mcs_table='QAM256', 317 lte_ul_mcs=4, 318 lte_ul_mcs_table='QAM64') 319 320 def generate_test_cases(self, nr_mcs_pair_list, nr_channel_list, **kwargs): 321 322 test_cases = [] 323 with open(self.testclass_params['nr_single_cell_configs'], 324 'r') as csvfile: 325 test_configs = csv.DictReader(csvfile) 326 for test_config, nr_channel, nr_mcs_pair in itertools.product( 327 test_configs, nr_channel_list, nr_mcs_pair_list): 328 if int(test_config['skip_test']): 329 continue 330 endc_combo_config = cputils.generate_endc_combo_config_from_csv_row( 331 test_config) 332 endc_combo_config['cell_list'][endc_combo_config['lte_cell_count']]['channel'] = nr_channel 333 test_name = 'test_fr1_{}_{}_dl_mcs{}_ul_mcs{}'.format( 334 test_config['nr_band'], nr_channel.lower(), nr_mcs_pair[0], 335 nr_mcs_pair[1]) 336 test_params = collections.OrderedDict( 337 endc_combo_config=endc_combo_config, 338 nr_dl_mcs=nr_mcs_pair[0], 339 nr_ul_mcs=nr_mcs_pair[1], 340 **kwargs) 341 setattr(self, test_name, 342 partial(self._test_throughput_bler, test_params)) 343 test_cases.append(test_name) 344 return test_cases 345 346 347class CellularLteSingleCellPeakThroughputTest(CellularLtePlusFr1PeakThroughputTest 348 ): 349 """Class to test single cell LTE""" 350 351 def __init__(self, controllers): 352 base_test.BaseTestClass.__init__(self, controllers) 353 self.testcase_metric_logger = ( 354 BlackboxMappedMetricLogger.for_test_case()) 355 self.testclass_metric_logger = ( 356 BlackboxMappedMetricLogger.for_test_class()) 357 self.publish_testcase_metrics = True 358 self.testclass_params = self.user_params['throughput_test_params'] 359 self.tests = self.generate_test_cases(lte_mcs_pair_list=[ 360 (('QAM256', 27), ('QAM256', 4)), (('QAM256', 4), ('QAM256', 27)) 361 ], 362 transform_precoding=0) 363 364 def generate_test_cases(self, lte_mcs_pair_list, **kwargs): 365 test_cases = [] 366 with open(self.testclass_params['lte_single_cell_configs'], 367 'r') as csvfile: 368 test_configs = csv.DictReader(csvfile) 369 for test_config, lte_mcs_pair in itertools.product( 370 test_configs, lte_mcs_pair_list): 371 if int(test_config['skip_test']): 372 continue 373 endc_combo_config = cputils.generate_endc_combo_config_from_csv_row( 374 test_config) 375 test_name = 'test_lte_B{}_dl_{}_mcs{}_ul_{}_mcs{}'.format( 376 test_config['lte_band'], lte_mcs_pair[0][0], 377 lte_mcs_pair[0][1], lte_mcs_pair[1][0], lte_mcs_pair[1][1]) 378 test_params = collections.OrderedDict( 379 endc_combo_config=endc_combo_config, 380 lte_dl_mcs_table=lte_mcs_pair[0][0], 381 lte_dl_mcs=lte_mcs_pair[0][1], 382 lte_ul_mcs_table=lte_mcs_pair[1][0], 383 lte_ul_mcs=lte_mcs_pair[1][1], 384 **kwargs) 385 setattr(self, test_name, 386 partial(self._test_throughput_bler, test_params)) 387 test_cases.append(test_name) 388 return test_cases 389