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 22 23import numpy 24import os 25import time 26from acts import asserts 27from acts import context 28from acts import base_test 29from acts import utils 30from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 31from acts.controllers.utils_lib import ssh 32from acts.controllers import iperf_server as ipf 33from acts_contrib.test_utils.cellular.keysight_5g_testapp import Keysight5GTestApp 34from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils 35from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 36from functools import partial 37 38LONG_SLEEP = 10 39MEDIUM_SLEEP = 2 40IPERF_TIMEOUT = 10 41SHORT_SLEEP = 1 42SUBFRAME_LENGTH = 0.001 43STOP_COUNTER_LIMIT = 3 44 45 46class CellularThroughputBaseTest(base_test.BaseTestClass): 47 """Base class for Cellular Throughput Testing 48 49 This base class enables cellular throughput tests on a lab/callbox setup 50 with PHY layer or iperf traffic. 51 """ 52 53 def __init__(self, controllers): 54 base_test.BaseTestClass.__init__(self, controllers) 55 self.testcase_metric_logger = ( 56 BlackboxMappedMetricLogger.for_test_case()) 57 self.testclass_metric_logger = ( 58 BlackboxMappedMetricLogger.for_test_class()) 59 self.publish_testcase_metrics = True 60 self.testclass_params = None 61 62 def setup_class(self): 63 """Initializes common test hardware and parameters. 64 65 This function initializes hardwares and compiles parameters that are 66 common to all tests in this class. 67 """ 68 # Setup controllers 69 self.dut = self.android_devices[-1] 70 self.keysight_test_app = Keysight5GTestApp( 71 self.user_params['Keysight5GTestApp']) 72 self.iperf_server = self.iperf_servers[0] 73 self.iperf_client = self.iperf_clients[0] 74 self.remote_server = ssh.connection.SshConnection( 75 ssh.settings.from_config( 76 self.user_params['RemoteServer']['ssh_config'])) 77 78 # Configure Tester 79 if self.testclass_params.get('reload_scpi', 1): 80 self.keysight_test_app.import_scpi_file( 81 self.testclass_params['scpi_file']) 82 83 # Declare testclass variables 84 self.testclass_results = collections.OrderedDict() 85 86 # Configure test retries 87 self.user_params['retry_tests'] = [self.__class__.__name__] 88 89 # Turn Airplane mode on 90 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 91 'Can not turn on airplane mode.') 92 93 def teardown_class(self): 94 self.log.info('Turning airplane mode on') 95 try: 96 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 97 'Can not turn on airplane mode.') 98 except: 99 self.log.warning('Cannot perform teardown operations on DUT.') 100 try: 101 self.keysight_test_app.set_cell_state('LTE', 1, 0) 102 self.keysight_test_app.destroy() 103 except: 104 self.log.warning('Cannot perform teardown operations on tester.') 105 self.process_testclass_results() 106 107 def setup_test(self): 108 self.retry_flag = False 109 if self.testclass_params['enable_pixel_logs']: 110 cputils.start_pixel_logger(self.dut) 111 112 def teardown_test(self): 113 self.retry_flag = False 114 self.log.info('Turing airplane mode on') 115 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 116 'Can not turn on airplane mode.') 117 if self.keysight_test_app.get_cell_state('LTE', 'CELL1'): 118 self.log.info('Turning LTE off.') 119 self.keysight_test_app.set_cell_state('LTE', 'CELL1', 0) 120 log_path = os.path.join( 121 context.get_current_context().get_full_output_path(), 'pixel_logs') 122 os.makedirs(self.log_path, exist_ok=True) 123 if self.testclass_params['enable_pixel_logs']: 124 cputils.stop_pixel_logger(self.dut, log_path) 125 self.process_testcase_results() 126 self.pass_fail_check() 127 128 def on_retry(self): 129 """Function to control test logic on retried tests. 130 131 This function is automatically executed on tests that are being 132 retried. In this case the function resets wifi, toggles it off and on 133 and sets a retry_flag to enable further tweaking the test logic on 134 second attempts. 135 """ 136 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 137 'Can not turn on airplane mode.') 138 if self.keysight_test_app.get_cell_state('LTE', 'CELL1'): 139 self.log.info('Turning LTE off.') 140 self.keysight_test_app.set_cell_state('LTE', 'CELL1', 0) 141 self.retry_flag = True 142 143 def pass_fail_check(self): 144 pass 145 146 def process_testcase_results(self): 147 pass 148 149 def process_testclass_results(self): 150 pass 151 152 def get_per_cell_power_sweeps(self, testcase_params): 153 raise NotImplementedError( 154 'get_per_cell_power_sweeps must be implemented.') 155 156 def compile_test_params(self, testcase_params): 157 """Function that completes all test params based on the test name. 158 159 Args: 160 testcase_params: dict containing test-specific parameters 161 """ 162 # Measurement Duration 163 testcase_params['bler_measurement_length'] = int( 164 self.testclass_params['traffic_duration'] / SUBFRAME_LENGTH) 165 # Cell power sweep 166 # TODO: Make this a function to support single power and sweep modes for each cell 167 testcase_params['cell_power_sweep'] = self.get_per_cell_power_sweeps( 168 testcase_params) 169 # Traffic & iperf params 170 if self.testclass_params['traffic_type'] == 'PHY': 171 return testcase_params 172 if self.testclass_params['traffic_type'] == 'TCP': 173 testcase_params['iperf_socket_size'] = self.testclass_params.get( 174 'tcp_socket_size', None) 175 testcase_params['iperf_processes'] = self.testclass_params.get( 176 'tcp_processes', 1) 177 elif self.testclass_params['traffic_type'] == 'UDP': 178 testcase_params['iperf_socket_size'] = self.testclass_params.get( 179 'udp_socket_size', None) 180 testcase_params['iperf_processes'] = self.testclass_params.get( 181 'udp_processes', 1) 182 adb_iperf_server = isinstance(self.iperf_server, 183 ipf.IPerfServerOverAdb) 184 if testcase_params['traffic_direction'] == 'DL': 185 reverse_direction = 0 if adb_iperf_server else 1 186 testcase_params[ 187 'use_client_output'] = False if adb_iperf_server else True 188 elif testcase_params['traffic_direction'] == 'UL': 189 reverse_direction = 1 if adb_iperf_server else 0 190 testcase_params[ 191 'use_client_output'] = True if adb_iperf_server else False 192 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 193 duration=self.testclass_params['traffic_duration'], 194 reverse_direction=reverse_direction, 195 traffic_type=self.testclass_params['traffic_type'], 196 socket_size=testcase_params['iperf_socket_size'], 197 num_processes=testcase_params['iperf_processes'], 198 udp_throughput=self.testclass_params['UDP_rates'].get( 199 testcase_params['num_dl_cells'], 200 self.testclass_params['UDP_rates']["default"]), 201 udp_length=1440) 202 return testcase_params 203 204 def run_iperf_traffic(self, testcase_params): 205 self.iperf_server.start(tag=0) 206 dut_ip = self.dut.droid.connectivityGetIPv4Addresses('rmnet0')[0] 207 if 'iperf_server_address' in self.testclass_params: 208 iperf_server_address = self.testclass_params[ 209 'iperf_server_address'] 210 elif isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 211 iperf_server_address = dut_ip 212 else: 213 iperf_server_address = wputils.get_server_address( 214 self.remote_server, dut_ip, '255.255.255.0') 215 client_output_path = self.iperf_client.start( 216 iperf_server_address, testcase_params['iperf_args'], 0, 217 self.testclass_params['traffic_duration'] + IPERF_TIMEOUT) 218 server_output_path = self.iperf_server.stop() 219 # Parse and log result 220 if testcase_params['use_client_output']: 221 iperf_file = client_output_path 222 else: 223 iperf_file = server_output_path 224 try: 225 iperf_result = ipf.IPerfResult(iperf_file) 226 current_throughput = numpy.mean(iperf_result.instantaneous_rates[ 227 self.testclass_params['iperf_ignored_interval']:-1]) * 8 * ( 228 1.024**2) 229 except: 230 self.log.warning( 231 'ValueError: Cannot get iperf result. Setting to 0') 232 current_throughput = 0 233 return current_throughput 234 235 def run_single_throughput_measurement(self, testcase_params): 236 result = collections.OrderedDict() 237 self.log.info('Starting BLER & throughput tests.') 238 if testcase_params['endc_combo_config']['nr_cell_count']: 239 self.keysight_test_app.start_bler_measurement( 240 'NR5G', testcase_params['endc_combo_config']['nr_dl_carriers'], 241 testcase_params['bler_measurement_length']) 242 if testcase_params['endc_combo_config']['lte_cell_count']: 243 self.keysight_test_app.start_bler_measurement( 244 'LTE', testcase_params['endc_combo_config']['lte_carriers'][0], 245 testcase_params['bler_measurement_length']) 246 247 if self.testclass_params['traffic_type'] != 'PHY': 248 result['iperf_throughput'] = self.run_iperf_traffic( 249 testcase_params) 250 251 if testcase_params['endc_combo_config']['nr_cell_count']: 252 result['nr_bler_result'] = self.keysight_test_app.get_bler_result( 253 'NR5G', testcase_params['endc_combo_config']['nr_dl_carriers'], 254 testcase_params['bler_measurement_length']) 255 result['nr_tput_result'] = self.keysight_test_app.get_throughput( 256 'NR5G', testcase_params['endc_combo_config']['nr_dl_carriers']) 257 if testcase_params['endc_combo_config']['lte_cell_count']: 258 result['lte_bler_result'] = self.keysight_test_app.get_bler_result( 259 'LTE', testcase_params['endc_combo_config']['lte_carriers'], 260 testcase_params['bler_measurement_length']) 261 result['lte_tput_result'] = self.keysight_test_app.get_throughput( 262 'LTE', testcase_params['endc_combo_config']['lte_carriers']) 263 return result 264 265 def print_throughput_result(self, result): 266 # Print Test Summary 267 if 'nr_tput_result' in result: 268 self.log.info( 269 "----NR5G STATS-------NR5G STATS-------NR5G STATS---") 270 self.log.info( 271 "DL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}" 272 .format( 273 result['nr_tput_result']['total']['DL']['min_tput'], 274 result['nr_tput_result']['total']['DL']['average_tput'], 275 result['nr_tput_result']['total']['DL']['max_tput'], 276 result['nr_tput_result']['total']['DL'] 277 ['theoretical_tput'])) 278 self.log.info( 279 "UL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}" 280 .format( 281 result['nr_tput_result']['total']['UL']['min_tput'], 282 result['nr_tput_result']['total']['UL']['average_tput'], 283 result['nr_tput_result']['total']['UL']['max_tput'], 284 result['nr_tput_result']['total']['UL'] 285 ['theoretical_tput'])) 286 self.log.info("DL BLER: {:.2f}%\tUL BLER: {:.2f}%".format( 287 result['nr_bler_result']['total']['DL']['nack_ratio'] * 100, 288 result['nr_bler_result']['total']['UL']['nack_ratio'] * 100)) 289 if 'lte_tput_result' in result: 290 self.log.info("----LTE STATS-------LTE STATS-------LTE STATS---") 291 self.log.info( 292 "DL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}" 293 .format( 294 result['lte_tput_result']['total']['DL']['min_tput'], 295 result['lte_tput_result']['total']['DL']['average_tput'], 296 result['lte_tput_result']['total']['DL']['max_tput'], 297 result['lte_tput_result']['total']['DL'] 298 ['theoretical_tput'])) 299 if self.testclass_params['lte_ul_mac_padding']: 300 self.log.info( 301 "UL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}" 302 .format( 303 result['lte_tput_result']['total']['UL']['min_tput'], 304 result['lte_tput_result']['total']['UL'] 305 ['average_tput'], 306 result['lte_tput_result']['total']['UL']['max_tput'], 307 result['lte_tput_result']['total']['UL'] 308 ['theoretical_tput'])) 309 self.log.info("DL BLER: {:.2f}%\tUL BLER: {:.2f}%".format( 310 result['lte_bler_result']['total']['DL']['nack_ratio'] * 100, 311 result['lte_bler_result']['total']['UL']['nack_ratio'] * 100)) 312 if self.testclass_params['traffic_type'] != 'PHY': 313 self.log.info("{} Tput: {:.2f} Mbps".format( 314 self.testclass_params['traffic_type'], 315 result['iperf_throughput'])) 316 317 def setup_tester(self, testcase_params): 318 # Configure all cells 319 for cell_idx, cell in enumerate( 320 testcase_params['endc_combo_config']['cell_list']): 321 self.keysight_test_app.set_cell_duplex_mode( 322 cell['cell_type'], cell['cell_number'], cell['duplex_mode']) 323 self.keysight_test_app.set_cell_band(cell['cell_type'], 324 cell['cell_number'], 325 cell['band']) 326 self.keysight_test_app.set_cell_dl_power( 327 cell['cell_type'], cell['cell_number'], 328 testcase_params['cell_power_sweep'][cell_idx][0], 1) 329 if cell['cell_type'] == 'NR5G': 330 self.keysight_test_app.set_nr_subcarrier_spacing( 331 cell['cell_number'], cell['subcarrier_spacing']) 332 if 'channel' in cell: 333 self.keysight_test_app.set_cell_channel( 334 cell['cell_type'], cell['cell_number'], cell['channel']) 335 self.keysight_test_app.set_cell_bandwidth(cell['cell_type'], 336 cell['cell_number'], 337 cell['dl_bandwidth']) 338 self.keysight_test_app.set_cell_mimo_config( 339 cell['cell_type'], cell['cell_number'], 'DL', 340 cell['dl_mimo_config']) 341 if cell['cell_type'] == 'LTE': 342 self.keysight_test_app.set_lte_cell_transmission_mode( 343 cell['cell_number'], cell['transmission_mode']) 344 self.keysight_test_app.set_lte_control_region_size( 345 cell['cell_number'], 1) 346 if cell['ul_enabled'] and cell['cell_type'] == 'NR5G': 347 self.keysight_test_app.set_cell_mimo_config( 348 cell['cell_type'], cell['cell_number'], 'UL', 349 cell['ul_mimo_config']) 350 351 if testcase_params.get('force_contiguous_nr_channel', False): 352 self.keysight_test_app.toggle_contiguous_nr_channels(1) 353 354 if testcase_params['endc_combo_config']['lte_cell_count']: 355 self.keysight_test_app.set_lte_cell_mcs( 356 'CELL1', testcase_params['lte_dl_mcs_table'], 357 testcase_params['lte_dl_mcs'], 358 testcase_params['lte_ul_mcs_table'], 359 testcase_params['lte_ul_mcs']) 360 self.keysight_test_app.set_lte_ul_mac_padding( 361 self.testclass_params['lte_ul_mac_padding']) 362 363 if testcase_params['endc_combo_config']['nr_cell_count']: 364 if 'schedule_scenario' in testcase_params: 365 self.keysight_test_app.set_nr_cell_schedule_scenario( 366 'CELL1', 367 testcase_params['schedule_scenario']) 368 self.keysight_test_app.set_nr_ul_dft_precoding( 369 'CELL1', testcase_params['transform_precoding']) 370 self.keysight_test_app.set_nr_cell_mcs( 371 'CELL1', testcase_params['nr_dl_mcs'], 372 testcase_params['nr_ul_mcs']) 373 self.keysight_test_app.set_dl_carriers( 374 testcase_params['endc_combo_config']['nr_dl_carriers']) 375 self.keysight_test_app.set_ul_carriers( 376 testcase_params['endc_combo_config']['nr_ul_carriers']) 377 378 # Turn on LTE cells 379 for cell in testcase_params['endc_combo_config']['cell_list']: 380 if cell['cell_type'] == 'LTE' and not self.keysight_test_app.get_cell_state( 381 cell['cell_type'], cell['cell_number']): 382 self.log.info('Turning LTE Cell {} on.'.format( 383 cell['cell_number'])) 384 self.keysight_test_app.set_cell_state(cell['cell_type'], 385 cell['cell_number'], 1) 386 387 # Activate LTE aggregation 388 if testcase_params['endc_combo_config']['lte_scc_list']: 389 self.keysight_test_app.apply_lte_carrier_agg( 390 testcase_params['endc_combo_config']['lte_scc_list']) 391 392 self.log.info('Waiting for LTE connections') 393 # Turn airplane mode off 394 num_apm_toggles = 5 395 for idx in range(num_apm_toggles): 396 self.log.info('Turning off airplane mode') 397 asserts.assert_true(utils.force_airplane_mode(self.dut, False), 398 'Can not turn off airplane mode.') 399 if self.keysight_test_app.wait_for_cell_status( 400 'LTE', 'CELL1', 'CONN', 180): 401 break 402 elif idx < num_apm_toggles - 1: 403 self.log.info('Turning on airplane mode') 404 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 405 'Can not turn on airplane mode.') 406 time.sleep(MEDIUM_SLEEP) 407 else: 408 asserts.fail('DUT did not connect to LTE.') 409 410 if testcase_params['endc_combo_config']['nr_cell_count']: 411 self.keysight_test_app.apply_carrier_agg() 412 self.log.info('Waiting for 5G connection') 413 connected = self.keysight_test_app.wait_for_cell_status( 414 'NR5G', testcase_params['endc_combo_config']['nr_cell_count'], 415 ['ACT', 'CONN'], 60) 416 if not connected: 417 asserts.fail('DUT did not connect to NR.') 418 time.sleep(SHORT_SLEEP) 419 420 def _test_throughput_bler(self, testcase_params): 421 """Test function to run cellular throughput and BLER measurements. 422 423 The function runs BLER/throughput measurement after configuring the 424 callbox and DUT. The test supports running PHY or TCP/UDP layer traffic 425 in a variety of band/carrier/mcs/etc configurations. 426 427 Args: 428 testcase_params: dict containing test-specific parameters 429 Returns: 430 result: dict containing throughput results and meta data 431 """ 432 # Prepare results dicts 433 testcase_params = self.compile_test_params(testcase_params) 434 testcase_results = collections.OrderedDict() 435 testcase_results['testcase_params'] = testcase_params 436 testcase_results['results'] = [] 437 438 # Setup tester and wait for DUT to connect 439 self.setup_tester(testcase_params) 440 441 # Run throughput test loop 442 stop_counter = 0 443 if testcase_params['endc_combo_config']['nr_cell_count']: 444 self.keysight_test_app.select_display_tab('NR5G', 1, 'BTHR', 445 'OTAGRAPH') 446 else: 447 self.keysight_test_app.select_display_tab('LTE', 1, 'BTHR', 448 'OTAGRAPH') 449 for power_idx in range(len(testcase_params['cell_power_sweep'][0])): 450 result = collections.OrderedDict() 451 # Set DL cell power 452 result['cell_power'] = [] 453 for cell_idx, cell in enumerate( 454 testcase_params['endc_combo_config']['cell_list']): 455 current_cell_power = testcase_params['cell_power_sweep'][ 456 cell_idx][power_idx] 457 result['cell_power'].append(current_cell_power) 458 self.keysight_test_app.set_cell_dl_power( 459 cell['cell_type'], cell['cell_number'], current_cell_power, 460 1) 461 462 # Start BLER and throughput measurements 463 result = self.run_single_throughput_measurement(testcase_params) 464 testcase_results['results'].append(result) 465 466 self.print_throughput_result(result) 467 468 if (('lte_bler_result' in result 469 and result['lte_bler_result']['total']['DL']['nack_ratio'] * 470 100 > 99) or 471 ('nr_bler_result' in result 472 and result['nr_bler_result']['total']['DL']['nack_ratio'] * 473 100 > 99)): 474 stop_counter = stop_counter + 1 475 else: 476 stop_counter = 0 477 if stop_counter == STOP_COUNTER_LIMIT: 478 break 479 480 # Save results 481 self.testclass_results[self.current_test_name] = testcase_results 482