1#!/usr/bin/env python3.4 2# 3# Copyright 2018 - 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 math 22import numpy 23import os 24import statistics 25from acts import asserts 26from acts import base_test 27from acts import context 28from acts import utils 29from acts.controllers.utils_lib import ssh 30from acts.controllers import iperf_server as ipf 31from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 32from acts_contrib.test_utils.wifi import ota_chamber 33from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 34from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure 35from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap 36from acts_contrib.test_utils.wifi import wifi_test_utils as wutils 37from concurrent.futures import ThreadPoolExecutor 38from functools import partial 39 40SHORT_SLEEP = 1 41MED_SLEEP = 6 42CONST_3dB = 3.01029995664 43RSSI_ERROR_VAL = float('nan') 44 45 46class WifiRssiTest(base_test.BaseTestClass): 47 """Class to test WiFi RSSI reporting. 48 49 This class tests RSSI reporting on android devices. The class tests RSSI 50 accuracy by checking RSSI over a large attenuation range, checks for RSSI 51 stability over time when attenuation is fixed, and checks that RSSI quickly 52 and reacts to changes attenuation by checking RSSI trajectories over 53 configurable attenuation waveforms.For an example config file to run this 54 test class see example_connectivity_performance_ap_sta.json. 55 """ 56 57 def __init__(self, controllers): 58 base_test.BaseTestClass.__init__(self, controllers) 59 self.testcase_metric_logger = ( 60 BlackboxMappedMetricLogger.for_test_case()) 61 self.testclass_metric_logger = ( 62 BlackboxMappedMetricLogger.for_test_class()) 63 self.publish_test_metrics = True 64 65 def setup_class(self): 66 self.dut = self.android_devices[0] 67 req_params = [ 68 'RemoteServer', 'RetailAccessPoints', 'rssi_test_params', 69 'main_network', 'testbed_params' 70 ] 71 self.unpack_userparams(req_params) 72 self.testclass_params = self.rssi_test_params 73 self.num_atten = self.attenuators[0].instrument.num_atten 74 self.iperf_server = self.iperf_servers[0] 75 self.iperf_client = self.iperf_clients[0] 76 self.remote_server = ssh.connection.SshConnection( 77 ssh.settings.from_config(self.RemoteServer[0]['ssh_config'])) 78 self.access_point = retail_ap.create(self.RetailAccessPoints)[0] 79 self.log_path = os.path.join(logging.log_path, 'results') 80 os.makedirs(self.log_path, exist_ok=True) 81 self.log.info('Access Point Configuration: {}'.format( 82 self.access_point.ap_settings)) 83 self.testclass_results = [] 84 85 # Turn WiFi ON 86 if self.testclass_params.get('airplane_mode', 1): 87 self.log.info('Turning on airplane mode.') 88 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 89 'Can not turn on airplane mode.') 90 wutils.wifi_toggle_state(self.dut, True) 91 92 def teardown_test(self): 93 self.iperf_server.stop() 94 95 def teardown_class(self): 96 # Turn WiFi OFF and reset AP 97 self.access_point.teardown() 98 for dev in self.android_devices: 99 wutils.wifi_toggle_state(dev, False) 100 dev.go_to_sleep() 101 102 def pass_fail_check_rssi_stability(self, testcase_params, 103 postprocessed_results): 104 """Check the test result and decide if it passed or failed. 105 106 Checks the RSSI test result and fails the test if the standard 107 deviation of signal_poll_rssi is beyond the threshold defined in the 108 config file. 109 110 Args: 111 testcase_params: dict containing test-specific parameters 112 postprocessed_results: compiled arrays of RSSI measurements 113 """ 114 # Set Blackbox metric values 115 if self.publish_test_metrics: 116 self.testcase_metric_logger.add_metric( 117 'signal_poll_rssi_stdev', 118 max(postprocessed_results['signal_poll_rssi']['stdev'])) 119 self.testcase_metric_logger.add_metric( 120 'chain_0_rssi_stdev', 121 max(postprocessed_results['chain_0_rssi']['stdev'])) 122 self.testcase_metric_logger.add_metric( 123 'chain_1_rssi_stdev', 124 max(postprocessed_results['chain_1_rssi']['stdev'])) 125 126 # Evaluate test pass/fail 127 test_failed = any([ 128 stdev > self.testclass_params['stdev_tolerance'] 129 for stdev in postprocessed_results['signal_poll_rssi']['stdev'] 130 ]) 131 test_message = ( 132 'RSSI stability {0}. Standard deviation was {1} dB ' 133 '(limit {2}), per chain standard deviation [{3}, {4}] dB'.format( 134 'failed' * test_failed + 'passed' * (not test_failed), [ 135 float('{:.2f}'.format(x)) 136 for x in postprocessed_results['signal_poll_rssi']['stdev'] 137 ], self.testclass_params['stdev_tolerance'], [ 138 float('{:.2f}'.format(x)) 139 for x in postprocessed_results['chain_0_rssi']['stdev'] 140 ], [ 141 float('{:.2f}'.format(x)) 142 for x in postprocessed_results['chain_1_rssi']['stdev'] 143 ])) 144 if test_failed: 145 asserts.fail(test_message) 146 asserts.explicit_pass(test_message) 147 148 def pass_fail_check_rssi_accuracy(self, testcase_params, 149 postprocessed_results): 150 """Check the test result and decide if it passed or failed. 151 152 Checks the RSSI test result and compares and compute its deviation from 153 the predicted RSSI. This computation is done for all reported RSSI 154 values. The test fails if any of the RSSI values specified in 155 rssi_under_test have an average error beyond what is specified in the 156 configuration file. 157 158 Args: 159 postprocessed_results: compiled arrays of RSSI measurements 160 testcase_params: dict containing params such as list of RSSIs under 161 test, i.e., can cause test to fail and boolean indicating whether 162 to look at absolute RSSI accuracy, or centered RSSI accuracy. 163 Centered accuracy is computed after systematic RSSI shifts are 164 removed. 165 """ 166 test_failed = False 167 test_message = '' 168 if testcase_params['absolute_accuracy']: 169 error_type = 'absolute' 170 else: 171 error_type = 'centered' 172 173 for key, val in postprocessed_results.items(): 174 # Compute the error metrics ignoring invalid RSSI readings 175 # If all readings invalid, set error to RSSI_ERROR_VAL 176 if 'rssi' in key and 'predicted' not in key: 177 filtered_error = [x for x in val['error'] if not math.isnan(x)] 178 if filtered_error: 179 avg_shift = statistics.mean(filtered_error) 180 if testcase_params['absolute_accuracy']: 181 avg_error = statistics.mean( 182 [abs(x) for x in filtered_error]) 183 else: 184 avg_error = statistics.mean( 185 [abs(x - avg_shift) for x in filtered_error]) 186 else: 187 avg_error = RSSI_ERROR_VAL 188 avg_shift = RSSI_ERROR_VAL 189 # Set Blackbox metric values 190 if self.publish_test_metrics: 191 self.testcase_metric_logger.add_metric( 192 '{}_error'.format(key), avg_error) 193 self.testcase_metric_logger.add_metric( 194 '{}_shift'.format(key), avg_shift) 195 # Evaluate test pass/fail 196 rssi_failure = (avg_error > 197 self.testclass_params['abs_tolerance'] 198 ) or math.isnan(avg_error) 199 if rssi_failure and key in testcase_params['rssi_under_test']: 200 test_message = test_message + ( 201 '{} failed ({} error = {:.2f} dB, ' 202 'shift = {:.2f} dB)\n').format(key, error_type, 203 avg_error, avg_shift) 204 test_failed = True 205 elif rssi_failure: 206 test_message = test_message + ( 207 '{} failed (ignored) ({} error = {:.2f} dB, ' 208 'shift = {:.2f} dB)\n').format(key, error_type, 209 avg_error, avg_shift) 210 else: 211 test_message = test_message + ( 212 '{} passed ({} error = {:.2f} dB, ' 213 'shift = {:.2f} dB)\n').format(key, error_type, 214 avg_error, avg_shift) 215 if test_failed: 216 asserts.fail(test_message) 217 asserts.explicit_pass(test_message) 218 219 def post_process_rssi_sweep(self, rssi_result): 220 """Postprocesses and saves JSON formatted results. 221 222 Args: 223 rssi_result: dict containing attenuation, rssi and other meta 224 data 225 Returns: 226 postprocessed_results: compiled arrays of RSSI data used in 227 pass/fail check 228 """ 229 # Save output as text file 230 results_file_path = os.path.join(self.log_path, self.current_test_name) 231 with open(results_file_path, 'w') as results_file: 232 json.dump(wputils.serialize_dict(rssi_result), 233 results_file, 234 indent=4) 235 # Compile results into arrays of RSSIs suitable for plotting 236 # yapf: disable 237 postprocessed_results = collections.OrderedDict( 238 [('signal_poll_rssi', {}), 239 ('signal_poll_avg_rssi', {}), 240 ('scan_rssi', {}), 241 ('chain_0_rssi', {}), 242 ('chain_1_rssi', {}), 243 ('total_attenuation', []), 244 ('predicted_rssi', [])]) 245 # yapf: enable 246 for key, val in postprocessed_results.items(): 247 if 'scan_rssi' in key: 248 postprocessed_results[key]['data'] = [ 249 x for data_point in rssi_result['rssi_result'] for x in 250 data_point[key][rssi_result['connected_bssid']]['data'] 251 ] 252 postprocessed_results[key]['mean'] = [ 253 x[key][rssi_result['connected_bssid']]['mean'] 254 for x in rssi_result['rssi_result'] 255 ] 256 postprocessed_results[key]['stdev'] = [ 257 x[key][rssi_result['connected_bssid']]['stdev'] 258 for x in rssi_result['rssi_result'] 259 ] 260 elif 'predicted_rssi' in key: 261 postprocessed_results['total_attenuation'] = [ 262 att + rssi_result['fixed_attenuation'] + 263 rssi_result['dut_front_end_loss'] 264 for att in rssi_result['attenuation'] 265 ] 266 postprocessed_results['predicted_rssi'] = [ 267 rssi_result['ap_tx_power'] - att 268 for att in postprocessed_results['total_attenuation'] 269 ] 270 elif 'rssi' in key: 271 postprocessed_results[key]['data'] = [ 272 x for data_point in rssi_result['rssi_result'] 273 for x in data_point[key]['data'] 274 ] 275 postprocessed_results[key]['mean'] = [ 276 x[key]['mean'] for x in rssi_result['rssi_result'] 277 ] 278 postprocessed_results[key]['stdev'] = [ 279 x[key]['stdev'] for x in rssi_result['rssi_result'] 280 ] 281 # Compute RSSI errors 282 for key, val in postprocessed_results.items(): 283 if 'chain' in key: 284 postprocessed_results[key]['error'] = [ 285 postprocessed_results[key]['mean'][idx] + CONST_3dB - 286 postprocessed_results['predicted_rssi'][idx] 287 for idx in range( 288 len(postprocessed_results['predicted_rssi'])) 289 ] 290 elif 'rssi' in key and 'predicted' not in key: 291 postprocessed_results[key]['error'] = [ 292 postprocessed_results[key]['mean'][idx] - 293 postprocessed_results['predicted_rssi'][idx] 294 for idx in range( 295 len(postprocessed_results['predicted_rssi'])) 296 ] 297 return postprocessed_results 298 299 def plot_rssi_vs_attenuation(self, postprocessed_results): 300 """Function to plot RSSI vs attenuation sweeps 301 302 Args: 303 postprocessed_results: compiled arrays of RSSI data. 304 """ 305 figure = BokehFigure(self.current_test_name, 306 x_label='Attenuation (dB)', 307 primary_y_label='RSSI (dBm)') 308 figure.add_line(postprocessed_results['total_attenuation'], 309 postprocessed_results['signal_poll_rssi']['mean'], 310 'Signal Poll RSSI', 311 marker='circle') 312 figure.add_line(postprocessed_results['total_attenuation'], 313 postprocessed_results['scan_rssi']['mean'], 314 'Scan RSSI', 315 marker='circle') 316 figure.add_line(postprocessed_results['total_attenuation'], 317 postprocessed_results['chain_0_rssi']['mean'], 318 'Chain 0 RSSI', 319 marker='circle') 320 figure.add_line(postprocessed_results['total_attenuation'], 321 postprocessed_results['chain_1_rssi']['mean'], 322 'Chain 1 RSSI', 323 marker='circle') 324 figure.add_line(postprocessed_results['total_attenuation'], 325 postprocessed_results['predicted_rssi'], 326 'Predicted RSSI', 327 marker='circle') 328 329 output_file_path = os.path.join(self.log_path, 330 self.current_test_name + '.html') 331 figure.generate_figure(output_file_path) 332 333 def plot_rssi_vs_time(self, rssi_result, postprocessed_results, 334 center_curves): 335 """Function to plot RSSI vs time. 336 337 Args: 338 rssi_result: dict containing raw RSSI data 339 postprocessed_results: compiled arrays of RSSI data 340 center_curvers: boolean indicating whether to shift curves to align 341 them with predicted RSSIs 342 """ 343 figure = BokehFigure( 344 self.current_test_name, 345 x_label='Time (s)', 346 primary_y_label=center_curves * 'Centered' + 'RSSI (dBm)', 347 ) 348 349 # yapf: disable 350 rssi_time_series = collections.OrderedDict( 351 [('signal_poll_rssi', []), 352 ('signal_poll_avg_rssi', []), 353 ('scan_rssi', []), 354 ('chain_0_rssi', []), 355 ('chain_1_rssi', []), 356 ('predicted_rssi', [])]) 357 # yapf: enable 358 for key, val in rssi_time_series.items(): 359 if 'predicted_rssi' in key: 360 rssi_time_series[key] = [ 361 x for x in postprocessed_results[key] for copies in range( 362 len(rssi_result['rssi_result'][0]['signal_poll_rssi'] 363 ['data'])) 364 ] 365 elif 'rssi' in key: 366 if center_curves: 367 filtered_error = [ 368 x for x in postprocessed_results[key]['error'] 369 if not math.isnan(x) 370 ] 371 if filtered_error: 372 avg_shift = statistics.mean(filtered_error) 373 else: 374 avg_shift = 0 375 rssi_time_series[key] = [ 376 x - avg_shift 377 for x in postprocessed_results[key]['data'] 378 ] 379 else: 380 rssi_time_series[key] = postprocessed_results[key]['data'] 381 time_vec = [ 382 self.testclass_params['polling_frequency'] * x 383 for x in range(len(rssi_time_series[key])) 384 ] 385 if len(rssi_time_series[key]) > 0: 386 figure.add_line(time_vec, rssi_time_series[key], key) 387 388 output_file_path = os.path.join(self.log_path, 389 self.current_test_name + '.html') 390 figure.generate_figure(output_file_path) 391 392 def plot_rssi_distribution(self, postprocessed_results): 393 """Function to plot RSSI distributions. 394 395 Args: 396 postprocessed_results: compiled arrays of RSSI data 397 """ 398 monitored_rssis = ['signal_poll_rssi', 'chain_0_rssi', 'chain_1_rssi'] 399 400 rssi_dist = collections.OrderedDict() 401 for rssi_key in monitored_rssis: 402 rssi_data = postprocessed_results[rssi_key] 403 rssi_dist[rssi_key] = collections.OrderedDict() 404 unique_rssi = sorted(set(rssi_data['data'])) 405 rssi_counts = [] 406 for value in unique_rssi: 407 rssi_counts.append(rssi_data['data'].count(value)) 408 total_count = sum(rssi_counts) 409 rssi_dist[rssi_key]['rssi_values'] = unique_rssi 410 rssi_dist[rssi_key]['rssi_pdf'] = [ 411 x / total_count for x in rssi_counts 412 ] 413 rssi_dist[rssi_key]['rssi_cdf'] = [] 414 cum_prob = 0 415 for prob in rssi_dist[rssi_key]['rssi_pdf']: 416 cum_prob += prob 417 rssi_dist[rssi_key]['rssi_cdf'].append(cum_prob) 418 419 figure = BokehFigure(self.current_test_name, 420 x_label='RSSI (dBm)', 421 primary_y_label='p(RSSI = x)', 422 secondary_y_label='p(RSSI <= x)') 423 for rssi_key, rssi_data in rssi_dist.items(): 424 figure.add_line(x_data=rssi_data['rssi_values'], 425 y_data=rssi_data['rssi_pdf'], 426 legend='{} PDF'.format(rssi_key), 427 y_axis='default') 428 figure.add_line(x_data=rssi_data['rssi_values'], 429 y_data=rssi_data['rssi_cdf'], 430 legend='{} CDF'.format(rssi_key), 431 y_axis='secondary') 432 output_file_path = os.path.join(self.log_path, 433 self.current_test_name + '_dist.html') 434 figure.generate_figure(output_file_path) 435 436 def run_rssi_test(self, testcase_params): 437 """Test function to run RSSI tests. 438 439 The function runs an RSSI test in the current device/AP configuration. 440 Function is called from another wrapper function that sets up the 441 testbed for the RvR test 442 443 Args: 444 testcase_params: dict containing test-specific parameters 445 Returns: 446 rssi_result: dict containing rssi_result and meta data 447 """ 448 # Run test and log result 449 rssi_result = collections.OrderedDict() 450 rssi_result['test_name'] = self.current_test_name 451 rssi_result['testcase_params'] = testcase_params 452 rssi_result['ap_settings'] = self.access_point.ap_settings.copy() 453 rssi_result['attenuation'] = list(testcase_params['rssi_atten_range']) 454 rssi_result['connected_bssid'] = self.main_network[ 455 testcase_params['band']].get('BSSID', '00:00:00:00') 456 channel_mode_combo = '{}_{}'.format(str(testcase_params['channel']), 457 testcase_params['mode']) 458 channel_str = str(testcase_params['channel']) 459 if channel_mode_combo in self.testbed_params['ap_tx_power']: 460 rssi_result['ap_tx_power'] = self.testbed_params['ap_tx_power'][ 461 channel_mode_combo] 462 else: 463 rssi_result['ap_tx_power'] = self.testbed_params['ap_tx_power'][ 464 str(testcase_params['channel'])] 465 rssi_result['fixed_attenuation'] = self.testbed_params[ 466 'fixed_attenuation'][channel_str] 467 rssi_result['dut_front_end_loss'] = self.testbed_params[ 468 'dut_front_end_loss'][channel_str] 469 470 self.log.info('Start running RSSI test.') 471 rssi_result['rssi_result'] = [] 472 rssi_result['llstats'] = [] 473 llstats_obj = wputils.LinkLayerStats(self.dut) 474 # Start iperf traffic if required by test 475 if testcase_params['active_traffic'] and testcase_params[ 476 'traffic_type'] == 'iperf': 477 self.iperf_server.start(tag=0) 478 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 479 iperf_server_address = self.dut_ip 480 else: 481 iperf_server_address = wputils.get_server_address( 482 self.remote_server, self.dut_ip, '255.255.255.0') 483 executor = ThreadPoolExecutor(max_workers=1) 484 thread_future = executor.submit( 485 self.iperf_client.start, iperf_server_address, 486 testcase_params['iperf_args'], 0, 487 testcase_params['traffic_timeout'] + SHORT_SLEEP) 488 executor.shutdown(wait=False) 489 elif testcase_params['active_traffic'] and testcase_params[ 490 'traffic_type'] == 'ping': 491 thread_future = wputils.get_ping_stats_nb( 492 self.remote_server, self.dut_ip, 493 testcase_params['traffic_timeout'], 0.02, 64) 494 else: 495 thread_future = wputils.get_ping_stats_nb( 496 self.remote_server, self.dut_ip, 497 testcase_params['traffic_timeout'], 0.5, 64) 498 llstats_obj.update_stats() 499 for atten in testcase_params['rssi_atten_range']: 500 # Set Attenuation 501 self.log.info('Setting attenuation to {} dB'.format(atten)) 502 for attenuator in self.attenuators: 503 attenuator.set_atten(atten) 504 current_rssi = collections.OrderedDict() 505 current_rssi = wputils.get_connected_rssi( 506 self.dut, testcase_params['connected_measurements'], 507 self.testclass_params['polling_frequency'], 508 testcase_params['first_measurement_delay']) 509 current_rssi['scan_rssi'] = wputils.get_scan_rssi( 510 self.dut, testcase_params['tracked_bssid'], 511 testcase_params['scan_measurements']) 512 rssi_result['rssi_result'].append(current_rssi) 513 llstats_obj.update_stats() 514 curr_llstats = llstats_obj.llstats_incremental.copy() 515 rssi_result['llstats'].append(curr_llstats) 516 self.log.info( 517 'Connected RSSI at {0:.2f} dB is {1:.2f} [{2:.2f}, {3:.2f}] dB' 518 .format(atten, current_rssi['signal_poll_rssi']['mean'], 519 current_rssi['chain_0_rssi']['mean'], 520 current_rssi['chain_1_rssi']['mean'])) 521 # Stop iperf traffic if needed 522 for attenuator in self.attenuators: 523 attenuator.set_atten(0) 524 thread_future.result() 525 if testcase_params['active_traffic'] and testcase_params[ 526 'traffic_type'] == 'iperf': 527 self.iperf_server.stop() 528 return rssi_result 529 530 def setup_ap(self, testcase_params): 531 """Function that gets devices ready for the test. 532 533 Args: 534 testcase_params: dict containing test-specific parameters 535 """ 536 band = self.access_point.band_lookup_by_channel( 537 testcase_params['channel']) 538 if '6G' in band: 539 frequency = wutils.WifiEnums.channel_6G_to_freq[int( 540 testcase_params['channel'].strip('6g'))] 541 else: 542 if testcase_params['channel'] < 13: 543 frequency = wutils.WifiEnums.channel_2G_to_freq[ 544 testcase_params['channel']] 545 else: 546 frequency = wutils.WifiEnums.channel_5G_to_freq[ 547 testcase_params['channel']] 548 if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES: 549 self.access_point.set_region(self.testbed_params['DFS_region']) 550 else: 551 self.access_point.set_region(self.testbed_params['default_region']) 552 self.access_point.set_channel(testcase_params['band'], 553 testcase_params['channel']) 554 self.access_point.set_bandwidth(testcase_params['band'], 555 testcase_params['mode']) 556 self.log.info('Access Point Configuration: {}'.format( 557 self.access_point.ap_settings)) 558 559 def setup_dut(self, testcase_params): 560 """Sets up the DUT in the configuration required by the test.""" 561 # Turn screen off to preserve battery 562 if self.testbed_params.get('screen_on', 563 False) or self.testclass_params.get( 564 'screen_on', False): 565 self.dut.droid.wakeLockAcquireDim() 566 else: 567 self.dut.go_to_sleep() 568 if wputils.validate_network(self.dut, 569 testcase_params['test_network']['SSID']): 570 self.log.info('Already connected to desired network') 571 else: 572 wutils.wifi_toggle_state(self.dut, True) 573 wutils.reset_wifi(self.dut) 574 self.main_network[testcase_params['band']][ 575 'channel'] = testcase_params['channel'] 576 wutils.set_wifi_country_code(self.dut, 577 self.testclass_params['country_code']) 578 if self.testbed_params.get('txbf_off', False): 579 wputils.disable_beamforming(self.dut) 580 wutils.wifi_connect(self.dut, 581 self.main_network[testcase_params['band']], 582 num_of_tries=5) 583 self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] 584 585 def setup_rssi_test(self, testcase_params): 586 """Main function to test RSSI. 587 588 The function sets up the AP in the correct channel and mode 589 configuration and called rssi_test to sweep attenuation and measure 590 RSSI 591 592 Args: 593 testcase_params: dict containing test-specific parameters 594 Returns: 595 rssi_result: dict containing rssi_results and meta data 596 """ 597 # Configure AP 598 self.setup_ap(testcase_params) 599 # Initialize attenuators 600 for attenuator in self.attenuators: 601 attenuator.set_atten(testcase_params['rssi_atten_range'][0]) 602 # Connect DUT to Network 603 self.setup_dut(testcase_params) 604 605 def get_traffic_timeout(self, testcase_params): 606 """Function to comput iperf session length required in RSSI test. 607 608 Args: 609 testcase_params: dict containing test-specific parameters 610 Returns: 611 traffic_timeout: length of iperf session required in rssi test 612 """ 613 atten_step_duration = testcase_params['first_measurement_delay'] + ( 614 testcase_params['connected_measurements'] * 615 self.testclass_params['polling_frequency'] 616 ) + testcase_params['scan_measurements'] * MED_SLEEP 617 timeout = len(testcase_params['rssi_atten_range'] 618 ) * atten_step_duration + MED_SLEEP 619 return timeout 620 621 def compile_rssi_vs_atten_test_params(self, testcase_params): 622 """Function to complete compiling test-specific parameters 623 624 Args: 625 testcase_params: dict containing test-specific parameters 626 """ 627 # Check if test should be skipped. 628 wputils.check_skip_conditions(testcase_params, self.dut, 629 self.access_point, 630 getattr(self, 'ota_chamber', None)) 631 632 testcase_params.update( 633 connected_measurements=self. 634 testclass_params['rssi_vs_atten_connected_measurements'], 635 scan_measurements=self. 636 testclass_params['rssi_vs_atten_scan_measurements'], 637 first_measurement_delay=SHORT_SLEEP, 638 absolute_accuracy=1) 639 rssi_under_test = self.testclass_params['rssi_vs_atten_metrics'] 640 if self.testclass_params[ 641 'rssi_vs_atten_scan_measurements'] == 0 and 'scan_rssi' in rssi_under_test: 642 rssi_under_test.remove('scan_rssi') 643 testcase_params['rssi_under_test'] = rssi_under_test 644 645 testcase_params['band'] = self.access_point.band_lookup_by_channel( 646 testcase_params['channel']) 647 testcase_params['test_network'] = self.main_network[ 648 testcase_params['band']] 649 testcase_params['tracked_bssid'] = [ 650 self.main_network[testcase_params['band']].get( 651 'BSSID', '00:00:00:00') 652 ] 653 654 testcase_params['rssi_atten_range'] = numpy.arange( 655 self.testclass_params['rssi_vs_atten_start'], 656 self.testclass_params['rssi_vs_atten_stop'], 657 self.testclass_params['rssi_vs_atten_step']).tolist() 658 659 testcase_params['traffic_timeout'] = self.get_traffic_timeout( 660 testcase_params) 661 662 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 663 testcase_params['iperf_args'] = '-i 1 -t {} -J'.format( 664 testcase_params['traffic_timeout']) 665 else: 666 testcase_params['iperf_args'] = '-i 1 -t {} -J -R'.format( 667 testcase_params['traffic_timeout']) 668 return testcase_params 669 670 def compile_rssi_stability_test_params(self, testcase_params): 671 """Function to complete compiling test-specific parameters 672 673 Args: 674 testcase_params: dict containing test-specific parameters 675 """ 676 # Check if test should be skipped. 677 wputils.check_skip_conditions(testcase_params, self.dut, 678 self.access_point, 679 getattr(self, 'ota_chamber', None)) 680 testcase_params.update( 681 connected_measurements=int( 682 self.testclass_params['rssi_stability_duration'] / 683 self.testclass_params['polling_frequency']), 684 scan_measurements=0, 685 first_measurement_delay=SHORT_SLEEP, 686 rssi_atten_range=self.testclass_params['rssi_stability_atten']) 687 testcase_params['band'] = self.access_point.band_lookup_by_channel( 688 testcase_params['channel']) 689 testcase_params['test_network'] = self.main_network[ 690 testcase_params['band']] 691 testcase_params['tracked_bssid'] = [ 692 self.main_network[testcase_params['band']].get( 693 'BSSID', '00:00:00:00') 694 ] 695 696 testcase_params['traffic_timeout'] = self.get_traffic_timeout( 697 testcase_params) 698 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 699 testcase_params['iperf_args'] = '-i 1 -t {} -J'.format( 700 testcase_params['traffic_timeout']) 701 else: 702 testcase_params['iperf_args'] = '-i 1 -t {} -J -R'.format( 703 testcase_params['traffic_timeout']) 704 return testcase_params 705 706 def compile_rssi_tracking_test_params(self, testcase_params): 707 """Function to complete compiling test-specific parameters 708 709 Args: 710 testcase_params: dict containing test-specific parameters 711 """ 712 # Check if test should be skipped. 713 wputils.check_skip_conditions(testcase_params, self.dut, 714 self.access_point, 715 getattr(self, 'ota_chamber', None)) 716 717 testcase_params.update(connected_measurements=int( 718 1 / self.testclass_params['polling_frequency']), 719 scan_measurements=0, 720 first_measurement_delay=0, 721 rssi_under_test=['signal_poll_rssi'], 722 absolute_accuracy=0) 723 testcase_params['band'] = self.access_point.band_lookup_by_channel( 724 testcase_params['channel']) 725 testcase_params['test_network'] = self.main_network[ 726 testcase_params['band']] 727 testcase_params['tracked_bssid'] = [ 728 self.main_network[testcase_params['band']].get( 729 'BSSID', '00:00:00:00') 730 ] 731 732 rssi_atten_range = [] 733 for waveform in self.testclass_params['rssi_tracking_waveforms']: 734 waveform_vector = [] 735 for section in range(len(waveform['atten_levels']) - 1): 736 section_limits = waveform['atten_levels'][section:section + 2] 737 up_down = (1 - 2 * (section_limits[1] < section_limits[0])) 738 temp_section = list( 739 range(section_limits[0], section_limits[1] + up_down, 740 up_down * waveform['step_size'])) 741 temp_section = [ 742 temp_section[idx] for idx in range(len(temp_section)) 743 for n in range(waveform['step_duration']) 744 ] 745 waveform_vector += temp_section 746 waveform_vector = waveform_vector * waveform['repetitions'] 747 rssi_atten_range = rssi_atten_range + waveform_vector 748 testcase_params['rssi_atten_range'] = rssi_atten_range 749 testcase_params['traffic_timeout'] = self.get_traffic_timeout( 750 testcase_params) 751 752 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 753 testcase_params['iperf_args'] = '-i 1 -t {} -J'.format( 754 testcase_params['traffic_timeout']) 755 else: 756 testcase_params['iperf_args'] = '-i 1 -t {} -J -R'.format( 757 testcase_params['traffic_timeout']) 758 return testcase_params 759 760 def _test_rssi_vs_atten(self, testcase_params): 761 """Function that gets called for each test case of rssi_vs_atten 762 763 The function gets called in each rssi test case. The function 764 customizes the test based on the test name of the test that called it 765 766 Args: 767 testcase_params: dict containing test-specific parameters 768 """ 769 testcase_params = self.compile_rssi_vs_atten_test_params( 770 testcase_params) 771 772 self.setup_rssi_test(testcase_params) 773 rssi_result = self.run_rssi_test(testcase_params) 774 rssi_result['postprocessed_results'] = self.post_process_rssi_sweep( 775 rssi_result) 776 self.testclass_results.append(rssi_result) 777 self.plot_rssi_vs_attenuation(rssi_result['postprocessed_results']) 778 self.pass_fail_check_rssi_accuracy( 779 testcase_params, rssi_result['postprocessed_results']) 780 781 def _test_rssi_stability(self, testcase_params): 782 """ Function that gets called for each test case of rssi_stability 783 784 The function gets called in each stability test case. The function 785 customizes test based on the test name of the test that called it 786 """ 787 testcase_params = self.compile_rssi_stability_test_params( 788 testcase_params) 789 790 self.setup_rssi_test(testcase_params) 791 rssi_result = self.run_rssi_test(testcase_params) 792 rssi_result['postprocessed_results'] = self.post_process_rssi_sweep( 793 rssi_result) 794 self.testclass_results.append(rssi_result) 795 self.plot_rssi_vs_time(rssi_result, 796 rssi_result['postprocessed_results'], 1) 797 self.plot_rssi_distribution(rssi_result['postprocessed_results']) 798 self.pass_fail_check_rssi_stability( 799 testcase_params, rssi_result['postprocessed_results']) 800 801 def _test_rssi_tracking(self, testcase_params): 802 """ Function that gets called for each test case of rssi_tracking 803 804 The function gets called in each rssi test case. The function 805 customizes the test based on the test name of the test that called it 806 """ 807 testcase_params = self.compile_rssi_tracking_test_params( 808 testcase_params) 809 810 self.setup_rssi_test(testcase_params) 811 rssi_result = self.run_rssi_test(testcase_params) 812 rssi_result['postprocessed_results'] = self.post_process_rssi_sweep( 813 rssi_result) 814 self.testclass_results.append(rssi_result) 815 self.plot_rssi_vs_time(rssi_result, 816 rssi_result['postprocessed_results'], 1) 817 self.pass_fail_check_rssi_accuracy( 818 testcase_params, rssi_result['postprocessed_results']) 819 820 def generate_test_cases(self, test_types, channels, modes, traffic_modes): 821 """Function that auto-generates test cases for a test class.""" 822 test_cases = [] 823 allowed_configs = { 824 20: [ 825 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 826 116, 132, 140, 149, 153, 157, 161, '6g37', '6g117', '6g213' 827 ], 828 40: [36, 44, 100, 149, 157, '6g37', '6g117', '6g213'], 829 80: [36, 100, 149, '6g37', '6g117', '6g213'], 830 160: [36, '6g37', '6g117', '6g213'] 831 } 832 833 for channel, mode, traffic_mode, test_type in itertools.product( 834 channels, modes, traffic_modes, test_types): 835 bandwidth = int(''.join([x for x in mode if x.isdigit()])) 836 if channel not in allowed_configs[bandwidth]: 837 continue 838 test_name = test_type + '_ch{}_{}_{}'.format( 839 channel, mode, traffic_mode) 840 testcase_params = collections.OrderedDict( 841 channel=channel, 842 mode=mode, 843 active_traffic=(traffic_mode == 'ActiveTraffic'), 844 traffic_type=self.user_params['rssi_test_params'] 845 ['traffic_type'], 846 ) 847 test_function = getattr(self, '_{}'.format(test_type)) 848 setattr(self, test_name, partial(test_function, testcase_params)) 849 test_cases.append(test_name) 850 return test_cases 851 852 853class WifiRssi_AllChannels_ActiveTraffic_Test(WifiRssiTest): 854 855 def __init__(self, controllers): 856 super().__init__(controllers) 857 self.tests = self.generate_test_cases( 858 ['test_rssi_stability', 'test_rssi_vs_atten'], [ 859 1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161, '6g37', '6g117', 860 '6g213' 861 ], ['bw20', 'bw40', 'bw80', 'bw160'], ['ActiveTraffic']) 862 863 864class WifiRssi_SampleChannels_NoTraffic_Test(WifiRssiTest): 865 866 def __init__(self, controllers): 867 super().__init__(controllers) 868 self.tests = self.generate_test_cases( 869 ['test_rssi_stability', 'test_rssi_vs_atten'], [6, 36, 149, '6g37'], 870 ['bw20', 'bw40', 'bw80', 'bw160'], ['NoTraffic']) 871 872 873class WifiRssiTrackingTest(WifiRssiTest): 874 875 def __init__(self, controllers): 876 super().__init__(controllers) 877 self.tests = self.generate_test_cases(['test_rssi_tracking'], 878 [6, 36, 149, '6g37'], 879 ['bw20', 'bw40', 'bw80', 'bw160'], 880 ['ActiveTraffic', 'NoTraffic']) 881 882 883# Over-the air version of RSSI tests 884class WifiOtaRssiTest(WifiRssiTest): 885 """Class to test over-the-air rssi tests. 886 887 This class implements measures WiFi RSSI tests in an OTA chamber. 888 It allows setting orientation and other chamber parameters to study 889 performance in varying channel conditions 890 """ 891 892 def __init__(self, controllers): 893 base_test.BaseTestClass.__init__(self, controllers) 894 self.testcase_metric_logger = ( 895 BlackboxMappedMetricLogger.for_test_case()) 896 self.testclass_metric_logger = ( 897 BlackboxMappedMetricLogger.for_test_class()) 898 self.publish_test_metrics = False 899 900 def setup_class(self): 901 WifiRssiTest.setup_class(self) 902 self.ota_chamber = ota_chamber.create( 903 self.user_params['OTAChamber'])[0] 904 905 def teardown_class(self): 906 WifiRssiTest.teardown_class(self) 907 self.ota_chamber.reset_chamber() 908 self.process_testclass_results() 909 910 def teardown_test(self): 911 if self.ota_chamber.current_mode == 'continuous': 912 self.ota_chamber.reset_chamber() 913 914 def extract_test_id(self, testcase_params, id_fields): 915 test_id = collections.OrderedDict( 916 (param, testcase_params[param]) for param in id_fields) 917 return test_id 918 919 def process_testclass_results(self): 920 """Saves all test results to enable comparison.""" 921 testclass_data = collections.OrderedDict() 922 for test_result in self.testclass_results: 923 current_params = test_result['testcase_params'] 924 925 channel = current_params['channel'] 926 channel_data = testclass_data.setdefault( 927 channel, 928 collections.OrderedDict(orientation=[], 929 rssi=collections.OrderedDict( 930 signal_poll_rssi=[], 931 chain_0_rssi=[], 932 chain_1_rssi=[]))) 933 934 channel_data['orientation'].append(current_params['orientation']) 935 channel_data['rssi']['signal_poll_rssi'].append( 936 test_result['postprocessed_results']['signal_poll_rssi'] 937 ['mean'][0]) 938 channel_data['rssi']['chain_0_rssi'].append( 939 test_result['postprocessed_results']['chain_0_rssi']['mean'] 940 [0]) 941 channel_data['rssi']['chain_1_rssi'].append( 942 test_result['postprocessed_results']['chain_1_rssi']['mean'] 943 [0]) 944 945 # Publish test class metrics 946 for channel, channel_data in testclass_data.items(): 947 for rssi_metric, rssi_metric_value in channel_data['rssi'].items(): 948 metric_name = 'ota_summary_ch{}.avg_{}'.format( 949 channel, rssi_metric) 950 metric_value = numpy.mean(rssi_metric_value) 951 self.testclass_metric_logger.add_metric( 952 metric_name, metric_value) 953 954 # Plot test class results 955 chamber_mode = self.testclass_results[0]['testcase_params'][ 956 'chamber_mode'] 957 if chamber_mode == 'orientation': 958 x_label = 'Angle (deg)' 959 elif chamber_mode == 'stepped stirrers': 960 x_label = 'Position Index' 961 elif chamber_mode == 'StirrersOn': 962 return 963 plots = [] 964 for channel, channel_data in testclass_data.items(): 965 current_plot = BokehFigure( 966 title='Channel {} - Rssi vs. Position'.format(channel), 967 x_label=x_label, 968 primary_y_label='RSSI (dBm)', 969 ) 970 for rssi_metric, rssi_metric_value in channel_data['rssi'].items(): 971 legend = rssi_metric 972 current_plot.add_line(channel_data['orientation'], 973 rssi_metric_value, legend) 974 current_plot.generate_figure() 975 plots.append(current_plot) 976 current_context = context.get_current_context().get_full_output_path() 977 plot_file_path = os.path.join(current_context, 'results.html') 978 BokehFigure.save_figures(plots, plot_file_path) 979 980 def setup_rssi_test(self, testcase_params): 981 # Test setup 982 WifiRssiTest.setup_rssi_test(self, testcase_params) 983 if testcase_params['chamber_mode'] == 'StirrersOn': 984 self.ota_chamber.start_continuous_stirrers() 985 else: 986 self.ota_chamber.set_orientation(testcase_params['orientation']) 987 988 def compile_ota_rssi_test_params(self, testcase_params): 989 """Function to complete compiling test-specific parameters 990 991 Args: 992 testcase_params: dict containing test-specific parameters 993 """ 994 # Check if test should be skipped. 995 wputils.check_skip_conditions(testcase_params, self.dut, 996 self.access_point, 997 getattr(self, 'ota_chamber', None)) 998 999 if 'rssi_over_orientation' in self.test_name: 1000 rssi_test_duration = self.testclass_params[ 1001 'rssi_over_orientation_duration'] 1002 rssi_ota_test_attenuation = [ 1003 self.testclass_params['rssi_ota_test_attenuation'] 1004 ] 1005 elif 'rssi_variation' in self.test_name: 1006 rssi_test_duration = self.testclass_params[ 1007 'rssi_variation_duration'] 1008 rssi_ota_test_attenuation = [ 1009 self.testclass_params['rssi_ota_test_attenuation'] 1010 ] 1011 elif 'rssi_vs_atten' in self.test_name: 1012 rssi_test_duration = self.testclass_params[ 1013 'rssi_over_orientation_duration'] 1014 rssi_ota_test_attenuation = numpy.arange( 1015 self.testclass_params['rssi_vs_atten_start'], 1016 self.testclass_params['rssi_vs_atten_stop'], 1017 self.testclass_params['rssi_vs_atten_step']).tolist() 1018 1019 testcase_params.update(connected_measurements=int( 1020 rssi_test_duration / self.testclass_params['polling_frequency']), 1021 scan_measurements=0, 1022 first_measurement_delay=SHORT_SLEEP, 1023 rssi_atten_range=rssi_ota_test_attenuation) 1024 testcase_params['band'] = self.access_point.band_lookup_by_channel( 1025 testcase_params['channel']) 1026 testcase_params['test_network'] = self.main_network[ 1027 testcase_params['band']] 1028 testcase_params['tracked_bssid'] = [ 1029 self.main_network[testcase_params['band']].get( 1030 'BSSID', '00:00:00:00') 1031 ] 1032 1033 testcase_params['traffic_timeout'] = self.get_traffic_timeout( 1034 testcase_params) 1035 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 1036 testcase_params['iperf_args'] = '-i 1 -t {} -J'.format( 1037 testcase_params['traffic_timeout']) 1038 else: 1039 testcase_params['iperf_args'] = '-i 1 -t {} -J -R'.format( 1040 testcase_params['traffic_timeout']) 1041 return testcase_params 1042 1043 def _test_ota_rssi(self, testcase_params): 1044 testcase_params = self.compile_ota_rssi_test_params(testcase_params) 1045 1046 self.setup_rssi_test(testcase_params) 1047 rssi_result = self.run_rssi_test(testcase_params) 1048 rssi_result['postprocessed_results'] = self.post_process_rssi_sweep( 1049 rssi_result) 1050 self.testclass_results.append(rssi_result) 1051 self.plot_rssi_vs_time(rssi_result, 1052 rssi_result['postprocessed_results'], 1) 1053 if 'rssi_vs_atten' in self.test_name: 1054 self.plot_rssi_vs_attenuation(rssi_result['postprocessed_results']) 1055 elif 'rssi_variation' in self.test_name: 1056 self.plot_rssi_distribution(rssi_result['postprocessed_results']) 1057 1058 def generate_test_cases(self, test_types, channels, modes, traffic_modes, 1059 chamber_modes, orientations): 1060 test_cases = [] 1061 allowed_configs = { 1062 20: [ 1063 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 1064 116, 132, 140, 149, 153, 157, 161, '6g37', '6g117', '6g213' 1065 ], 1066 40: [36, 44, 100, 149, 157, '6g37', '6g117', '6g213'], 1067 80: [36, 100, 149, '6g37', '6g117', '6g213'], 1068 160: [36, '6g37', '6g117', '6g213'] 1069 } 1070 1071 for (channel, mode, traffic, chamber_mode, orientation, 1072 test_type) in itertools.product(channels, modes, traffic_modes, 1073 chamber_modes, orientations, 1074 test_types): 1075 bandwidth = int(''.join([x for x in mode if x.isdigit()])) 1076 if channel not in allowed_configs[bandwidth]: 1077 continue 1078 test_name = test_type + '_ch{}_{}_{}_{}deg'.format( 1079 channel, mode, traffic, orientation) 1080 testcase_params = collections.OrderedDict( 1081 channel=channel, 1082 mode=mode, 1083 active_traffic=(traffic == 'ActiveTraffic'), 1084 traffic_type=self.user_params['rssi_test_params'] 1085 ['traffic_type'], 1086 chamber_mode=chamber_mode, 1087 orientation=orientation) 1088 test_function = self._test_ota_rssi 1089 setattr(self, test_name, partial(test_function, testcase_params)) 1090 test_cases.append(test_name) 1091 return test_cases 1092 1093class WifiOtaRssi_StirrerVariation_Test(WifiOtaRssiTest): 1094 1095 def __init__(self, controllers): 1096 WifiRssiTest.__init__(self, controllers) 1097 self.tests = self.generate_test_cases(['test_rssi_variation'], 1098 [6, 36, 149, '6g37'], 1099 ['bw20', 'bw80', 'bw160'], 1100 ['ActiveTraffic'], 1101 ['StirrersOn'], [0]) 1102 1103 1104class WifiOtaRssi_TenDegree_Test(WifiOtaRssiTest): 1105 1106 def __init__(self, controllers): 1107 WifiRssiTest.__init__(self, controllers) 1108 self.tests = self.generate_test_cases(['test_rssi_over_orientation'], 1109 [6, 36, 149, '6g37'], 1110 ['bw20', 'bw80', 'bw160'], 1111 ['ActiveTraffic'], 1112 ['orientation'], 1113 list(range(0, 360, 10))) 1114