1#!/usr/bin/env python3.4 2# 3# Copyright 2017 - 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 itertools 19import json 20import logging 21import numpy 22import os 23import time 24from acts import asserts 25from acts import base_test 26from acts import utils 27from acts.controllers import iperf_server as ipf 28from acts.controllers.utils_lib import ssh 29from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 30from acts.test_utils.wifi import ota_chamber 31from acts.test_utils.wifi import ota_sniffer 32from acts.test_utils.wifi import wifi_performance_test_utils as wputils 33from acts.test_utils.wifi import wifi_retail_ap as retail_ap 34from acts.test_utils.wifi import wifi_test_utils as wutils 35from functools import partial 36 37 38class WifiRvrTest(base_test.BaseTestClass): 39 """Class to test WiFi rate versus range. 40 41 This class implements WiFi rate versus range tests on single AP single STA 42 links. The class setups up the AP in the desired configurations, configures 43 and connects the phone to the AP, and runs iperf throughput test while 44 sweeping attenuation. For an example config file to run this test class see 45 example_connectivity_performance_ap_sta.json. 46 """ 47 48 TEST_TIMEOUT = 6 49 MAX_CONSECUTIVE_ZEROS = 3 50 51 def __init__(self, controllers): 52 base_test.BaseTestClass.__init__(self, controllers) 53 self.testcase_metric_logger = ( 54 BlackboxMappedMetricLogger.for_test_case()) 55 self.testclass_metric_logger = ( 56 BlackboxMappedMetricLogger.for_test_class()) 57 self.publish_testcase_metrics = True 58 59 def setup_class(self): 60 """Initializes common test hardware and parameters. 61 62 This function initializes hardwares and compiles parameters that are 63 common to all tests in this class. 64 """ 65 self.dut = self.android_devices[-1] 66 req_params = [ 67 'RetailAccessPoints', 'rvr_test_params', 'testbed_params', 68 'RemoteServer', 'main_network' 69 ] 70 opt_params = ['golden_files_list', 'OTASniffer'] 71 self.unpack_userparams(req_params, opt_params) 72 self.testclass_params = self.rvr_test_params 73 self.num_atten = self.attenuators[0].instrument.num_atten 74 self.iperf_server = self.iperf_servers[0] 75 self.remote_server = ssh.connection.SshConnection( 76 ssh.settings.from_config(self.RemoteServer[0]['ssh_config'])) 77 self.iperf_client = self.iperf_clients[0] 78 self.access_point = retail_ap.create(self.RetailAccessPoints)[0] 79 if hasattr(self, 80 'OTASniffer') and self.testbed_params['sniffer_enable']: 81 self.sniffer = ota_sniffer.create(self.OTASniffer)[0] 82 self.log.info('Access Point Configuration: {}'.format( 83 self.access_point.ap_settings)) 84 self.log_path = os.path.join(logging.log_path, 'results') 85 os.makedirs(self.log_path, exist_ok=True) 86 if not hasattr(self, 'golden_files_list'): 87 if 'golden_results_path' in self.testbed_params: 88 self.golden_files_list = [ 89 os.path.join(self.testbed_params['golden_results_path'], 90 file) for file in 91 os.listdir(self.testbed_params['golden_results_path']) 92 ] 93 else: 94 self.log.warning('No golden files found.') 95 self.golden_files_list = [] 96 self.testclass_results = [] 97 98 # Turn WiFi ON 99 if self.testclass_params.get('airplane_mode', 1): 100 self.log.info('Turning on airplane mode.') 101 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 102 "Can not turn on airplane mode.") 103 wutils.wifi_toggle_state(self.dut, True) 104 105 def teardown_test(self): 106 self.iperf_server.stop() 107 108 def teardown_class(self): 109 # Turn WiFi OFF 110 for dev in self.android_devices: 111 wutils.wifi_toggle_state(dev, False) 112 self.process_testclass_results() 113 114 def process_testclass_results(self): 115 """Saves plot with all test results to enable comparison.""" 116 # Plot and save all results 117 plots = collections.OrderedDict() 118 for result in self.testclass_results: 119 plot_id = (result['testcase_params']['channel'], 120 result['testcase_params']['mode']) 121 if plot_id not in plots: 122 plots[plot_id] = wputils.BokehFigure( 123 title='Channel {} {} ({})'.format( 124 result['testcase_params']['channel'], 125 result['testcase_params']['mode'], 126 result['testcase_params']['traffic_type']), 127 x_label='Attenuation (dB)', 128 primary_y_label='Throughput (Mbps)') 129 plots[plot_id].add_line(result['total_attenuation'], 130 result['throughput_receive'], 131 result['test_name'], 132 marker='circle') 133 figure_list = [] 134 for plot_id, plot in plots.items(): 135 plot.generate_figure() 136 figure_list.append(plot) 137 output_file_path = os.path.join(self.log_path, 'results.html') 138 wputils.BokehFigure.save_figures(figure_list, output_file_path) 139 140 def pass_fail_check(self, rvr_result): 141 """Check the test result and decide if it passed or failed. 142 143 Checks the RvR test result and compares to a throughput limites for 144 the same configuration. The pass/fail tolerances are provided in the 145 config file. 146 147 Args: 148 rvr_result: dict containing attenuation, throughput and other data 149 """ 150 try: 151 throughput_limits = self.compute_throughput_limits(rvr_result) 152 except: 153 asserts.fail('Test failed: Golden file not found') 154 155 failure_count = 0 156 for idx, current_throughput in enumerate( 157 rvr_result['throughput_receive']): 158 if (current_throughput < throughput_limits['lower_limit'][idx] 159 or current_throughput > 160 throughput_limits['upper_limit'][idx]): 161 failure_count = failure_count + 1 162 163 # Set test metrics 164 rvr_result['metrics']['failure_count'] = failure_count 165 if self.publish_testcase_metrics: 166 self.testcase_metric_logger.add_metric('failure_count', 167 failure_count) 168 169 # Assert pass or fail 170 if failure_count >= self.testclass_params['failure_count_tolerance']: 171 asserts.fail('Test failed. Found {} points outside limits.'.format( 172 failure_count)) 173 asserts.explicit_pass( 174 'Test passed. Found {} points outside throughput limits.'.format( 175 failure_count)) 176 177 def compute_throughput_limits(self, rvr_result): 178 """Compute throughput limits for current test. 179 180 Checks the RvR test result and compares to a throughput limites for 181 the same configuration. The pass/fail tolerances are provided in the 182 config file. 183 184 Args: 185 rvr_result: dict containing attenuation, throughput and other meta 186 data 187 Returns: 188 throughput_limits: dict containing attenuation and throughput limit data 189 """ 190 test_name = self.current_test_name 191 golden_path = next(file_name for file_name in self.golden_files_list 192 if test_name in file_name) 193 with open(golden_path, 'r') as golden_file: 194 golden_results = json.load(golden_file) 195 golden_attenuation = [ 196 att + golden_results['fixed_attenuation'] 197 for att in golden_results['attenuation'] 198 ] 199 attenuation = [] 200 lower_limit = [] 201 upper_limit = [] 202 for idx, current_throughput in enumerate( 203 rvr_result['throughput_receive']): 204 current_att = rvr_result['attenuation'][idx] + rvr_result[ 205 'fixed_attenuation'] 206 att_distances = [ 207 abs(current_att - golden_att) 208 for golden_att in golden_attenuation 209 ] 210 sorted_distances = sorted(enumerate(att_distances), 211 key=lambda x: x[1]) 212 closest_indeces = [dist[0] for dist in sorted_distances[0:3]] 213 closest_throughputs = [ 214 golden_results['throughput_receive'][index] 215 for index in closest_indeces 216 ] 217 closest_throughputs.sort() 218 219 attenuation.append(current_att) 220 lower_limit.append( 221 max( 222 closest_throughputs[0] - max( 223 self.testclass_params['abs_tolerance'], 224 closest_throughputs[0] * 225 self.testclass_params['pct_tolerance'] / 100), 0)) 226 upper_limit.append(closest_throughputs[-1] + max( 227 self.testclass_params['abs_tolerance'], closest_throughputs[-1] 228 * self.testclass_params['pct_tolerance'] / 100)) 229 throughput_limits = { 230 'attenuation': attenuation, 231 'lower_limit': lower_limit, 232 'upper_limit': upper_limit 233 } 234 return throughput_limits 235 236 def process_test_results(self, rvr_result): 237 """Saves plots and JSON formatted results. 238 239 Args: 240 rvr_result: dict containing attenuation, throughput and other meta 241 data 242 """ 243 # Save output as text file 244 test_name = self.current_test_name 245 results_file_path = os.path.join( 246 self.log_path, '{}.json'.format(self.current_test_name)) 247 with open(results_file_path, 'w') as results_file: 248 json.dump(rvr_result, results_file, indent=4) 249 # Plot and save 250 figure = wputils.BokehFigure(title=test_name, 251 x_label='Attenuation (dB)', 252 primary_y_label='Throughput (Mbps)') 253 try: 254 golden_path = next(file_name 255 for file_name in self.golden_files_list 256 if test_name in file_name) 257 with open(golden_path, 'r') as golden_file: 258 golden_results = json.load(golden_file) 259 golden_attenuation = [ 260 att + golden_results['fixed_attenuation'] 261 for att in golden_results['attenuation'] 262 ] 263 throughput_limits = self.compute_throughput_limits(rvr_result) 264 shaded_region = { 265 'x_vector': throughput_limits['attenuation'], 266 'lower_limit': throughput_limits['lower_limit'], 267 'upper_limit': throughput_limits['upper_limit'] 268 } 269 figure.add_line(golden_attenuation, 270 golden_results['throughput_receive'], 271 'Golden Results', 272 color='green', 273 marker='circle', 274 shaded_region=shaded_region) 275 except: 276 self.log.warning('ValueError: Golden file not found') 277 278 # Generate graph annotatios 279 hover_text = [ 280 'TX MCS = {0} ({1:.1f}%). RX MCS = {2} ({3:.1f}%)'.format( 281 curr_llstats['summary']['common_tx_mcs'], 282 curr_llstats['summary']['common_tx_mcs_freq'] * 100, 283 curr_llstats['summary']['common_rx_mcs'], 284 curr_llstats['summary']['common_rx_mcs_freq'] * 100) 285 for curr_llstats in rvr_result['llstats'] 286 ] 287 figure.add_line(rvr_result['total_attenuation'], 288 rvr_result['throughput_receive'], 289 'Test Results', 290 hover_text=hover_text, 291 color='red', 292 marker='circle') 293 294 output_file_path = os.path.join(self.log_path, 295 '{}.html'.format(test_name)) 296 figure.generate_figure(output_file_path) 297 298 #Set test metrics 299 rvr_result['metrics'] = {} 300 rvr_result['metrics']['peak_tput'] = max( 301 rvr_result['throughput_receive']) 302 if self.publish_testcase_metrics: 303 self.testcase_metric_logger.add_metric( 304 'peak_tput', rvr_result['metrics']['peak_tput']) 305 306 tput_below_limit = [ 307 tput < self.testclass_params['tput_metric_targets'][ 308 rvr_result['testcase_params']['mode']]['high'] 309 for tput in rvr_result['throughput_receive'] 310 ] 311 rvr_result['metrics']['high_tput_range'] = -1 312 for idx in range(len(tput_below_limit)): 313 if all(tput_below_limit[idx:]): 314 if idx == 0: 315 #Throughput was never above limit 316 rvr_result['metrics']['high_tput_range'] = -1 317 else: 318 rvr_result['metrics']['high_tput_range'] = rvr_result[ 319 'total_attenuation'][max(idx, 1) - 1] 320 break 321 if self.publish_testcase_metrics: 322 self.testcase_metric_logger.add_metric( 323 'high_tput_range', rvr_result['metrics']['high_tput_range']) 324 325 tput_below_limit = [ 326 tput < self.testclass_params['tput_metric_targets'][ 327 rvr_result['testcase_params']['mode']]['low'] 328 for tput in rvr_result['throughput_receive'] 329 ] 330 for idx in range(len(tput_below_limit)): 331 if all(tput_below_limit[idx:]): 332 rvr_result['metrics']['low_tput_range'] = rvr_result[ 333 'total_attenuation'][max(idx, 1) - 1] 334 break 335 else: 336 rvr_result['metrics']['low_tput_range'] = -1 337 if self.publish_testcase_metrics: 338 self.testcase_metric_logger.add_metric( 339 'low_tput_range', rvr_result['metrics']['low_tput_range']) 340 341 def run_rvr_test(self, testcase_params): 342 """Test function to run RvR. 343 344 The function runs an RvR test in the current device/AP configuration. 345 Function is called from another wrapper function that sets up the 346 testbed for the RvR test 347 348 Args: 349 testcase_params: dict containing test-specific parameters 350 Returns: 351 rvr_result: dict containing rvr_results and meta data 352 """ 353 self.log.info('Start running RvR') 354 # Refresh link layer stats before test 355 llstats_obj = wputils.LinkLayerStats(self.dut) 356 zero_counter = 0 357 throughput = [] 358 llstats = [] 359 rssi = [] 360 for atten in testcase_params['atten_range']: 361 for dev in self.android_devices: 362 if not wputils.health_check(dev, 5, 50): 363 asserts.skip('DUT health check failed. Skipping test.') 364 # Set Attenuation 365 for attenuator in self.attenuators: 366 attenuator.set_atten(atten, strict=False) 367 # Refresh link layer stats 368 llstats_obj.update_stats() 369 # Setup sniffer 370 if self.testbed_params['sniffer_enable']: 371 self.sniffer.start_capture( 372 network=testcase_params['test_network'], 373 chan=int(testcase_params['channel']), 374 bw=int(testcase_params['mode'][3:]), 375 duration=self.testclass_params['iperf_duration'] / 5) 376 # Start iperf session 377 self.iperf_server.start(tag=str(atten)) 378 rssi_future = wputils.get_connected_rssi_nb( 379 self.dut, self.testclass_params['iperf_duration'] - 1, 1, 1) 380 client_output_path = self.iperf_client.start( 381 testcase_params['iperf_server_address'], 382 testcase_params['iperf_args'], str(atten), 383 self.testclass_params['iperf_duration'] + self.TEST_TIMEOUT) 384 server_output_path = self.iperf_server.stop() 385 rssi_result = rssi_future.result() 386 current_rssi = { 387 'signal_poll_rssi': rssi_result['signal_poll_rssi']['mean'], 388 'chain_0_rssi': rssi_result['chain_0_rssi']['mean'], 389 'chain_1_rssi': rssi_result['chain_1_rssi']['mean'] 390 } 391 rssi.append(current_rssi) 392 # Stop sniffer 393 if self.testbed_params['sniffer_enable']: 394 self.sniffer.stop_capture(tag=str(atten)) 395 # Parse and log result 396 if testcase_params['use_client_output']: 397 iperf_file = client_output_path 398 else: 399 iperf_file = server_output_path 400 try: 401 iperf_result = ipf.IPerfResult(iperf_file) 402 curr_throughput = numpy.mean(iperf_result.instantaneous_rates[ 403 self.testclass_params['iperf_ignored_interval']:-1] 404 ) * 8 * (1.024**2) 405 except: 406 self.log.warning( 407 'ValueError: Cannot get iperf result. Setting to 0') 408 curr_throughput = 0 409 throughput.append(curr_throughput) 410 llstats_obj.update_stats() 411 curr_llstats = llstats_obj.llstats_incremental.copy() 412 llstats.append(curr_llstats) 413 self.log.info( 414 ('Throughput at {0:.2f} dB is {1:.2f} Mbps. ' 415 'RSSI = {2:.2f} [{3:.2f}, {4:.2f}].').format( 416 atten, curr_throughput, current_rssi['signal_poll_rssi'], 417 current_rssi['chain_0_rssi'], 418 current_rssi['chain_1_rssi'])) 419 if curr_throughput == 0 and ( 420 current_rssi['signal_poll_rssi'] < -80 421 or numpy.isnan(current_rssi['signal_poll_rssi'])): 422 zero_counter = zero_counter + 1 423 else: 424 zero_counter = 0 425 if zero_counter == self.MAX_CONSECUTIVE_ZEROS: 426 self.log.info( 427 'Throughput stable at 0 Mbps. Stopping test now.') 428 throughput.extend( 429 [0] * 430 (len(testcase_params['atten_range']) - len(throughput))) 431 break 432 for attenuator in self.attenuators: 433 attenuator.set_atten(0, strict=False) 434 # Compile test result and meta data 435 rvr_result = collections.OrderedDict() 436 rvr_result['test_name'] = self.current_test_name 437 rvr_result['testcase_params'] = testcase_params.copy() 438 rvr_result['ap_settings'] = self.access_point.ap_settings.copy() 439 rvr_result['fixed_attenuation'] = self.testbed_params[ 440 'fixed_attenuation'][str(testcase_params['channel'])] 441 rvr_result['attenuation'] = list(testcase_params['atten_range']) 442 rvr_result['total_attenuation'] = [ 443 att + rvr_result['fixed_attenuation'] 444 for att in rvr_result['attenuation'] 445 ] 446 rvr_result['rssi'] = rssi 447 rvr_result['throughput_receive'] = throughput 448 rvr_result['llstats'] = llstats 449 return rvr_result 450 451 def setup_ap(self, testcase_params): 452 """Sets up the access point in the configuration required by the test. 453 454 Args: 455 testcase_params: dict containing AP and other test params 456 """ 457 band = self.access_point.band_lookup_by_channel( 458 testcase_params['channel']) 459 if '2G' in band: 460 frequency = wutils.WifiEnums.channel_2G_to_freq[ 461 testcase_params['channel']] 462 else: 463 frequency = wutils.WifiEnums.channel_5G_to_freq[ 464 testcase_params['channel']] 465 if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES: 466 self.access_point.set_region(self.testbed_params['DFS_region']) 467 else: 468 self.access_point.set_region(self.testbed_params['default_region']) 469 self.access_point.set_channel(band, testcase_params['channel']) 470 self.access_point.set_bandwidth(band, testcase_params['mode']) 471 self.log.info('Access Point Configuration: {}'.format( 472 self.access_point.ap_settings)) 473 474 def setup_dut(self, testcase_params): 475 """Sets up the DUT in the configuration required by the test. 476 477 Args: 478 testcase_params: dict containing AP and other test params 479 """ 480 # Check battery level before test 481 if not wputils.health_check( 482 self.dut, 20) and testcase_params['traffic_direction'] == 'UL': 483 asserts.skip('Overheating or Battery level low. Skipping test.') 484 # Turn screen off to preserve battery 485 self.dut.go_to_sleep() 486 if wputils.validate_network(self.dut, 487 testcase_params['test_network']['SSID']): 488 self.log.info('Already connected to desired network') 489 else: 490 wutils.reset_wifi(self.dut) 491 wutils.set_wifi_country_code(self.dut, 492 self.testclass_params['country_code']) 493 testcase_params['test_network']['channel'] = testcase_params[ 494 'channel'] 495 wutils.wifi_connect(self.dut, 496 testcase_params['test_network'], 497 num_of_tries=5, 498 check_connectivity=True) 499 self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] 500 501 def setup_rvr_test(self, testcase_params): 502 """Function that gets devices ready for the test. 503 504 Args: 505 testcase_params: dict containing test-specific parameters 506 """ 507 # Configure AP 508 self.setup_ap(testcase_params) 509 # Set attenuator to 0 dB 510 for attenuator in self.attenuators: 511 attenuator.set_atten(0, strict=False) 512 # Reset, configure, and connect DUT 513 self.setup_dut(testcase_params) 514 # Wait before running the first wifi test 515 first_test_delay = self.testclass_params.get('first_test_delay', 600) 516 if first_test_delay > 0 and len(self.testclass_results) == 0: 517 self.log.info('Waiting before the first RvR test.') 518 time.sleep(first_test_delay) 519 self.setup_dut(testcase_params) 520 # Get iperf_server address 521 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 522 testcase_params['iperf_server_address'] = self.dut_ip 523 else: 524 testcase_params[ 525 'iperf_server_address'] = wputils.get_server_address( 526 self.remote_server, self.dut_ip, '255.255.255.0') 527 528 def compile_test_params(self, testcase_params): 529 """Function that completes all test params based on the test name. 530 531 Args: 532 testcase_params: dict containing test-specific parameters 533 """ 534 num_atten_steps = int((self.testclass_params['atten_stop'] - 535 self.testclass_params['atten_start']) / 536 self.testclass_params['atten_step']) 537 testcase_params['atten_range'] = [ 538 self.testclass_params['atten_start'] + 539 x * self.testclass_params['atten_step'] 540 for x in range(0, num_atten_steps) 541 ] 542 band = self.access_point.band_lookup_by_channel( 543 testcase_params['channel']) 544 testcase_params['test_network'] = self.main_network[band] 545 if (testcase_params['traffic_direction'] == 'DL' 546 and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb) 547 ) or (testcase_params['traffic_direction'] == 'UL' 548 and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)): 549 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 550 duration=self.testclass_params['iperf_duration'], 551 reverse_direction=1, 552 traffic_type=testcase_params['traffic_type']) 553 testcase_params['use_client_output'] = True 554 else: 555 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 556 duration=self.testclass_params['iperf_duration'], 557 reverse_direction=0, 558 traffic_type=testcase_params['traffic_type']) 559 testcase_params['use_client_output'] = False 560 return testcase_params 561 562 def _test_rvr(self, testcase_params): 563 """ Function that gets called for each test case 564 565 Args: 566 testcase_params: dict containing test-specific parameters 567 """ 568 # Compile test parameters from config and test name 569 testcase_params = self.compile_test_params(testcase_params) 570 571 # Prepare devices and run test 572 self.setup_rvr_test(testcase_params) 573 rvr_result = self.run_rvr_test(testcase_params) 574 575 # Post-process results 576 self.testclass_results.append(rvr_result) 577 self.process_test_results(rvr_result) 578 self.pass_fail_check(rvr_result) 579 580 def generate_test_cases(self, channels, modes, traffic_types, 581 traffic_directions): 582 """Function that auto-generates test cases for a test class.""" 583 test_cases = [] 584 allowed_configs = { 585 'VHT20': [ 586 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 587 116, 132, 140, 149, 153, 157, 161 588 ], 589 'VHT40': [36, 44, 100, 149, 157], 590 'VHT80': [36, 100, 149] 591 } 592 593 for channel, mode, traffic_type, traffic_direction in itertools.product( 594 channels, modes, traffic_types, traffic_directions): 595 if channel not in allowed_configs[mode]: 596 continue 597 test_name = 'test_rvr_{}_{}_ch{}_{}'.format( 598 traffic_type, traffic_direction, channel, mode) 599 test_params = collections.OrderedDict( 600 channel=channel, 601 mode=mode, 602 traffic_type=traffic_type, 603 traffic_direction=traffic_direction) 604 setattr(self, test_name, partial(self._test_rvr, test_params)) 605 test_cases.append(test_name) 606 return test_cases 607 608 609# Classes defining test suites 610class WifiRvr_2GHz_Test(WifiRvrTest): 611 def __init__(self, controllers): 612 super().__init__(controllers) 613 self.tests = self.generate_test_cases(channels=[1, 6, 11], 614 modes=['VHT20'], 615 traffic_types=['TCP'], 616 traffic_directions=['DL', 'UL']) 617 618 619class WifiRvr_UNII1_Test(WifiRvrTest): 620 def __init__(self, controllers): 621 super().__init__(controllers) 622 self.tests = self.generate_test_cases( 623 channels=[36, 40, 44, 48], 624 modes=['VHT20', 'VHT40', 'VHT80'], 625 traffic_types=['TCP'], 626 traffic_directions=['DL', 'UL']) 627 628 629class WifiRvr_UNII3_Test(WifiRvrTest): 630 def __init__(self, controllers): 631 super().__init__(controllers) 632 self.tests = self.generate_test_cases( 633 channels=[149, 153, 157, 161], 634 modes=['VHT20', 'VHT40', 'VHT80'], 635 traffic_types=['TCP'], 636 traffic_directions=['DL', 'UL']) 637 638 639class WifiRvr_SampleDFS_Test(WifiRvrTest): 640 def __init__(self, controllers): 641 super().__init__(controllers) 642 self.tests = self.generate_test_cases( 643 channels=[64, 100, 116, 132, 140], 644 modes=['VHT20', 'VHT40', 'VHT80'], 645 traffic_types=['TCP'], 646 traffic_directions=['DL', 'UL']) 647 648 649class WifiRvr_SampleUDP_Test(WifiRvrTest): 650 def __init__(self, controllers): 651 super().__init__(controllers) 652 self.tests = self.generate_test_cases( 653 channels=[6, 36, 149], 654 modes=['VHT20', 'VHT40', 'VHT80'], 655 traffic_types=['UDP'], 656 traffic_directions=['DL', 'UL']) 657 658 659class WifiRvr_TCP_All_Test(WifiRvrTest): 660 def __init__(self, controllers): 661 super().__init__(controllers) 662 self.tests = self.generate_test_cases( 663 channels=[1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], 664 modes=['VHT20', 'VHT40', 'VHT80'], 665 traffic_types=['TCP'], 666 traffic_directions=['DL', 'UL']) 667 668 669class WifiRvr_TCP_Downlink_Test(WifiRvrTest): 670 def __init__(self, controllers): 671 super().__init__(controllers) 672 self.tests = self.generate_test_cases( 673 channels=[1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], 674 modes=['VHT20', 'VHT40', 'VHT80'], 675 traffic_types=['TCP'], 676 traffic_directions=['DL']) 677 678 679class WifiRvr_TCP_Uplink_Test(WifiRvrTest): 680 def __init__(self, controllers): 681 super().__init__(controllers) 682 self.tests = self.generate_test_cases( 683 channels=[1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], 684 modes=['VHT20', 'VHT40', 'VHT80'], 685 traffic_types=['TCP'], 686 traffic_directions=['UL']) 687 688 689# Over-the air version of RVR tests 690class WifiOtaRvrTest(WifiRvrTest): 691 """Class to test over-the-air RvR 692 693 This class implements measures WiFi RvR tests in an OTA chamber. It enables 694 setting turntable orientation and other chamber parameters to study 695 performance in varying channel conditions 696 """ 697 def __init__(self, controllers): 698 base_test.BaseTestClass.__init__(self, controllers) 699 self.testcase_metric_logger = ( 700 BlackboxMappedMetricLogger.for_test_case()) 701 self.testclass_metric_logger = ( 702 BlackboxMappedMetricLogger.for_test_class()) 703 self.publish_testcase_metrics = False 704 705 def setup_class(self): 706 WifiRvrTest.setup_class(self) 707 self.ota_chamber = ota_chamber.create( 708 self.user_params['OTAChamber'])[0] 709 710 def teardown_class(self): 711 WifiRvrTest.teardown_class(self) 712 self.ota_chamber.reset_chamber() 713 714 def extract_test_id(self, testcase_params, id_fields): 715 test_id = collections.OrderedDict( 716 (param, testcase_params[param]) for param in id_fields) 717 return test_id 718 719 def process_testclass_results(self): 720 """Saves plot with all test results to enable comparison.""" 721 # Plot individual test id results raw data and compile metrics 722 plots = collections.OrderedDict() 723 compiled_data = collections.OrderedDict() 724 for result in self.testclass_results: 725 test_id = tuple( 726 self.extract_test_id( 727 result['testcase_params'], 728 ['channel', 'mode', 'traffic_type', 'traffic_direction' 729 ]).items()) 730 if test_id not in plots: 731 # Initialize test id data when not present 732 compiled_data[test_id] = {'throughput': [], 'metrics': {}} 733 compiled_data[test_id]['metrics'] = { 734 key: [] 735 for key in result['metrics'].keys() 736 } 737 plots[test_id] = wputils.BokehFigure( 738 title='Channel {} {} ({} {})'.format( 739 result['testcase_params']['channel'], 740 result['testcase_params']['mode'], 741 result['testcase_params']['traffic_type'], 742 result['testcase_params']['traffic_direction']), 743 x_label='Attenuation (dB)', 744 primary_y_label='Throughput (Mbps)') 745 # Compile test id data and metrics 746 compiled_data[test_id]['throughput'].append( 747 result['throughput_receive']) 748 compiled_data[test_id]['total_attenuation'] = result[ 749 'total_attenuation'] 750 for metric_key, metric_value in result['metrics'].items(): 751 compiled_data[test_id]['metrics'][metric_key].append( 752 metric_value) 753 # Add test id to plots 754 plots[test_id].add_line(result['total_attenuation'], 755 result['throughput_receive'], 756 result['test_name'], 757 width=1, 758 style='dashed', 759 marker='circle') 760 761 # Compute average RvRs and compount metrics over orientations 762 for test_id, test_data in compiled_data.items(): 763 test_id_dict = dict(test_id) 764 metric_tag = '{}_{}_ch{}_{}'.format( 765 test_id_dict['traffic_type'], 766 test_id_dict['traffic_direction'], test_id_dict['channel'], 767 test_id_dict['mode']) 768 high_tput_hit_freq = numpy.mean( 769 numpy.not_equal(test_data['metrics']['high_tput_range'], -1)) 770 self.testclass_metric_logger.add_metric( 771 '{}.high_tput_hit_freq'.format(metric_tag), high_tput_hit_freq) 772 for metric_key, metric_value in test_data['metrics'].items(): 773 metric_key = "{}.avg_{}".format(metric_tag, metric_key) 774 metric_value = numpy.mean(metric_value) 775 self.testclass_metric_logger.add_metric( 776 metric_key, metric_value) 777 test_data['avg_rvr'] = numpy.mean(test_data['throughput'], 0) 778 test_data['median_rvr'] = numpy.median(test_data['throughput'], 0) 779 plots[test_id].add_line(test_data['total_attenuation'], 780 test_data['avg_rvr'], 781 legend='Average Throughput', 782 marker='circle') 783 plots[test_id].add_line(test_data['total_attenuation'], 784 test_data['median_rvr'], 785 legend='Median Throughput', 786 marker='square') 787 788 figure_list = [] 789 for test_id, plot in plots.items(): 790 plot.generate_figure() 791 figure_list.append(plot) 792 output_file_path = os.path.join(self.log_path, 'results.html') 793 wputils.BokehFigure.save_figures(figure_list, output_file_path) 794 795 def setup_rvr_test(self, testcase_params): 796 # Set turntable orientation 797 self.ota_chamber.set_orientation(testcase_params['orientation']) 798 # Continue test setup 799 WifiRvrTest.setup_rvr_test(self, testcase_params) 800 801 def generate_test_cases(self, channels, modes, angles, traffic_types, 802 directions): 803 test_cases = [] 804 allowed_configs = { 805 'VHT20': [ 806 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 149, 153, 807 157, 161 808 ], 809 'VHT40': [36, 44, 149, 157], 810 'VHT80': [36, 149] 811 } 812 for channel, mode, angle, traffic_type, direction in itertools.product( 813 channels, modes, angles, traffic_types, directions): 814 if channel not in allowed_configs[mode]: 815 continue 816 testcase_name = 'test_rvr_{}_{}_ch{}_{}_{}deg'.format( 817 traffic_type, direction, channel, mode, angle) 818 test_params = collections.OrderedDict(channel=channel, 819 mode=mode, 820 traffic_type=traffic_type, 821 traffic_direction=direction, 822 orientation=angle) 823 setattr(self, testcase_name, partial(self._test_rvr, test_params)) 824 test_cases.append(testcase_name) 825 return test_cases 826 827 828class WifiOtaRvr_StandardOrientation_Test(WifiOtaRvrTest): 829 def __init__(self, controllers): 830 WifiOtaRvrTest.__init__(self, controllers) 831 self.tests = self.generate_test_cases( 832 [1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], 833 ['VHT20', 'VHT40', 'VHT80'], list(range(0, 360, 834 45)), ['TCP'], ['DL']) 835 836 837class WifiOtaRvr_SampleChannel_Test(WifiOtaRvrTest): 838 def __init__(self, controllers): 839 WifiOtaRvrTest.__init__(self, controllers) 840 self.tests = self.generate_test_cases([6], ['VHT20'], 841 list(range(0, 360, 45)), ['TCP'], 842 ['DL']) 843 self.tests.extend( 844 self.generate_test_cases([36, 149], ['VHT80'], 845 list(range(0, 360, 45)), ['TCP'], ['DL'])) 846 847 848class WifiOtaRvr_SingleOrientation_Test(WifiOtaRvrTest): 849 def __init__(self, controllers): 850 WifiOtaRvrTest.__init__(self, controllers) 851 self.tests = self.generate_test_cases( 852 [6, 36, 40, 44, 48, 149, 153, 157, 161], 853 ['VHT20', 'VHT40', 'VHT80'], [0], ['TCP'], ['DL', 'UL']) 854