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