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 csv 19import itertools 20import json 21import logging 22import os 23from acts import asserts 24from acts import base_test 25from acts import utils 26from acts.controllers import iperf_server as ipf 27from acts.controllers.utils_lib import ssh 28from acts.metrics.loggers.blackbox import BlackboxMetricLogger 29from acts.test_utils.wifi import wifi_test_utils as wutils 30from acts.test_utils.wifi import wifi_retail_ap as retail_ap 31from WifiRvrTest import WifiRvrTest 32from WifiPingTest import WifiPingTest 33 34 35class WifiSensitivityTest(WifiRvrTest, WifiPingTest): 36 """Class to test WiFi sensitivity tests. 37 38 This class implements measures WiFi sensitivity per rate. It heavily 39 leverages the WifiRvrTest class and introduced minor differences to set 40 specific rates and the access point, and implements a different pass/fail 41 check. For an example config file to run this test class see 42 example_connectivity_performance_ap_sta.json. 43 """ 44 45 VALID_TEST_CONFIGS = { 46 1: ["legacy", "VHT20"], 47 2: ["legacy", "VHT20"], 48 6: ["legacy", "VHT20"], 49 10: ["legacy", "VHT20"], 50 11: ["legacy", "VHT20"], 51 36: ["legacy", "VHT20"], 52 40: ["legacy", "VHT20"], 53 44: ["legacy", "VHT20"], 54 48: ["legacy", "VHT20"], 55 149: ["legacy", "VHT20"], 56 153: ["legacy", "VHT20"], 57 157: ["legacy", "VHT20"], 58 161: ["legacy", "VHT20"] 59 } 60 VALID_RATES = { 61 "legacy_2GHz": [[54, 1], [48, 1], [36, 1], [24, 1], [18, 1], [12, 1], 62 [11, 1], [9, 1], [6, 1], [5.5, 1], [2, 1], [1, 1]], 63 "legacy_5GHz": [[54, 1], [48, 1], [36, 1], [24, 1], [18, 1], [12, 1], 64 [9, 1], [6, 1]], 65 "HT": [[8, 1], [7, 1], [6, 1], [5, 1], [4, 1], [3, 1], [2, 1], [1, 1], 66 [0, 1], [15, 2], [14, 2], [13, 2], [12, 2], [11, 2], [10, 2], 67 [9, 2], [8, 2]], 68 "VHT": [[9, 1], [8, 1], [7, 1], [6, 1], [5, 1], [4, 1], [3, 1], [2, 1], 69 [1, 1], [0, 1], [9, 2], [8, 2], [7, 2], [6, 2], [5, 2], [4, 2], 70 [3, 2], [2, 2], [1, 2], [0, 2]] 71 } 72 73 def __init__(self, controllers): 74 base_test.BaseTestClass.__init__(self, controllers) 75 self.failure_count_metric = BlackboxMetricLogger.for_test_case( 76 metric_name='sensitivity') 77 78 def setup_class(self): 79 """Initializes common test hardware and parameters. 80 81 This function initializes hardwares and compiles parameters that are 82 common to all tests in this class. 83 """ 84 self.client_dut = self.android_devices[-1] 85 req_params = [ 86 "RetailAccessPoints", "sensitivity_test_params", "testbed_params", 87 "RemoteServer" 88 ] 89 opt_params = ["main_network", "golden_files_list"] 90 self.unpack_userparams(req_params, opt_params) 91 self.testclass_params = self.sensitivity_test_params 92 self.num_atten = self.attenuators[0].instrument.num_atten 93 self.ping_server = ssh.connection.SshConnection( 94 ssh.settings.from_config(self.RemoteServer[0]["ssh_config"])) 95 self.iperf_server = self.iperf_servers[0] 96 self.iperf_client = self.iperf_clients[0] 97 if isinstance(self.iperf_server, ipf.IPerfServerOverSsh): 98 self.ping_server = self.iperf_server 99 else: 100 self.ping_server = self.iperf_client 101 self.access_points = retail_ap.create(self.RetailAccessPoints) 102 self.access_point = self.access_points[0] 103 self.log.info("Access Point Configuration: {}".format( 104 self.access_point.ap_settings)) 105 self.log_path = os.path.join(logging.log_path, "results") 106 utils.create_dir(self.log_path) 107 if not hasattr(self, "golden_files_list"): 108 self.golden_files_list = [ 109 os.path.join(self.testbed_params["golden_results_path"], 110 file) for file in os.listdir( 111 self.testbed_params["golden_results_path"]) 112 ] 113 self.testclass_results = [] 114 115 # Turn WiFi ON 116 for dev in self.android_devices: 117 wutils.wifi_toggle_state(dev, True) 118 119 def teardown_class(self): 120 # Turn WiFi OFF 121 for dev in self.android_devices: 122 wutils.wifi_toggle_state(dev, False) 123 self.process_testclass_results() 124 125 def pass_fail_check(self, result): 126 """Checks sensitivity against golden results and decides on pass/fail. 127 128 Args: 129 result: dict containing attenuation, throughput and other meta 130 data 131 """ 132 try: 133 golden_path = next(file_name 134 for file_name in self.golden_files_list 135 if "sensitivity_targets" in file_name) 136 with open(golden_path, 'r') as golden_file: 137 golden_results = json.load(golden_file) 138 golden_sensitivity = golden_results[self.current_test_name][ 139 "sensitivity"] 140 except: 141 golden_sensitivity = float("nan") 142 143 result_string = "Througput = {}, Sensitivity = {}. Target Sensitivity = {}".format( 144 result["peak_throughput"], result["sensitivity"], 145 golden_sensitivity) 146 if result["sensitivity"] - golden_sensitivity < self.testclass_params["sensitivity_tolerance"]: 147 asserts.explicit_pass("Test Passed. {}".format(result_string)) 148 else: 149 asserts.fail("Test Failed. {}".format(result_string)) 150 151 def process_testclass_results(self): 152 """Saves and plots test results from all executed test cases.""" 153 # write json output 154 testclass_results_dict = collections.OrderedDict() 155 for result in self.testclass_results: 156 testclass_results_dict[result["test_name"]] = { 157 "peak_throughput": result["peak_throughput"], 158 "range": result["range"], 159 "sensitivity": result["sensitivity"] 160 } 161 results_file_path = os.path.join(self.log_path, 'results.json') 162 with open(results_file_path, 'w') as results_file: 163 json.dump(testclass_results_dict, results_file, indent=4) 164 # write csv 165 results_file_path = os.path.join(self.log_path, 'results.csv') 166 with open(results_file_path, mode='w') as csv_file: 167 csv_header = [ 168 "Channel", "Mode", "MCS", "Streams", "Chain", "Sensitivity", 169 "Range", "Peak Throughput" 170 ] 171 writer = csv.DictWriter(csv_file, fieldnames=csv_header) 172 writer.writeheader() 173 for result in self.testclass_results: 174 testcase_params = self.parse_test_params(result["test_name"]) 175 writer.writerow({ 176 "Channel": testcase_params["channel"], 177 "Mode": testcase_params["mode"], 178 "MCS": testcase_params["rate"], 179 "Streams": testcase_params["num_streams"], 180 "Chain": testcase_params["chain_mask"], 181 "Sensitivity": result["sensitivity"], 182 "Range": result["range"], 183 "Peak Throughput": result["peak_throughput"] 184 }) 185 186 if not self.testclass_params["traffic_type"].lower() == "ping": 187 WifiRvrTest.process_testclass_results(self) 188 189 def process_rvr_test_results(self, testcase_params, rvr_result): 190 """Post processes RvR results to compute sensitivity. 191 192 Takes in the results of the RvR tests and computes the sensitivity of 193 the current rate by looking at the point at which throughput drops 194 below the percentage specified in the config file. The function then 195 calls on its parent class process_test_results to plot the result. 196 197 Args: 198 rvr_result: dict containing attenuation, throughput and other meta 199 data 200 """ 201 rvr_result["peak_throughput"] = max(rvr_result["throughput_receive"]) 202 throughput_check = [ 203 throughput < rvr_result["peak_throughput"] * 204 (self.testclass_params["throughput_pct_at_sensitivity"] / 100) 205 for throughput in rvr_result["throughput_receive"] 206 ] 207 consistency_check = [ 208 idx for idx in range(len(throughput_check)) 209 if all(throughput_check[idx:]) 210 ] 211 rvr_result["atten_at_range"] = rvr_result["attenuation"][ 212 consistency_check[0] - 1] 213 rvr_result["range"] = rvr_result["fixed_attenuation"] + ( 214 rvr_result["atten_at_range"]) 215 rvr_result["sensitivity"] = self.testclass_params["ap_tx_power"] + ( 216 self.testbed_params["ap_tx_power_offset"][str( 217 testcase_params["channel"])] - rvr_result["range"]) 218 WifiRvrTest.process_test_results(self, rvr_result) 219 220 def process_ping_test_results(self, testcase_params, ping_result): 221 """Post processes RvR results to compute sensitivity. 222 223 Takes in the results of the RvR tests and computes the sensitivity of 224 the current rate by looking at the point at which throughput drops 225 below the percentage specified in the config file. The function then 226 calls on its parent class process_test_results to plot the result. 227 228 Args: 229 rvr_result: dict containing attenuation, throughput and other meta 230 data 231 """ 232 testcase_params[ 233 "range_ping_loss_threshold"] = 100 - testcase_params["throughput_pct_at_sensitivity"] 234 WifiPingTest.process_ping_results(self, testcase_params, ping_result) 235 ping_result["sensitivity"] = self.testclass_params["ap_tx_power"] + ( 236 self.testbed_params["ap_tx_power_offset"][str( 237 testcase_params["channel"])] - ping_result["range"]) 238 239 def setup_ap(self, testcase_params): 240 """Sets up the AP and attenuator to compensate for AP chain imbalance. 241 242 Args: 243 testcase_params: dict containing AP and other test params 244 """ 245 band = self.access_point.band_lookup_by_channel( 246 testcase_params["channel"]) 247 if "2G" in band: 248 frequency = wutils.WifiEnums.channel_2G_to_freq[testcase_params[ 249 "channel"]] 250 else: 251 frequency = wutils.WifiEnums.channel_5G_to_freq[testcase_params[ 252 "channel"]] 253 if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES: 254 self.access_point.set_region(self.testbed_params["DFS_region"]) 255 else: 256 self.access_point.set_region(self.testbed_params["default_region"]) 257 self.access_point.set_channel(band, testcase_params["channel"]) 258 self.access_point.set_bandwidth(band, testcase_params["mode"]) 259 self.access_point.set_power(band, testcase_params["ap_tx_power"]) 260 self.access_point.set_rate( 261 band, testcase_params["mode"], testcase_params["num_streams"], 262 testcase_params["rate"], testcase_params["short_gi"]) 263 # Set attenuator offsets and set attenuators to initial condition 264 atten_offsets = self.testbed_params['chain_offset'][str( 265 testcase_params['channel'])] 266 for atten in self.attenuators: 267 if 'AP-Chain-0' in atten.path: 268 atten.offset = atten_offsets[0] 269 elif 'AP-Chain-1' in atten.path: 270 atten.offset = atten_offsets[1] 271 if testcase_params["attenuated_chain"] in atten.path: 272 atten.offset = atten.instrument.max_atten 273 self.log.info("Access Point Configuration: {}".format( 274 self.access_point.ap_settings)) 275 276 def get_start_atten(self): 277 """Gets the starting attenuation for this sensitivity test. 278 279 The function gets the starting attenuation by checking whether a test 280 as the next higher MCS has been executed. If so it sets the starting 281 point a configurable number of dBs below the next MCS's sensitivity. 282 283 Returns: 284 start_atten: starting attenuation for current test 285 """ 286 # Get the current and reference test config. The reference test is the 287 # one performed at the current MCS+1 288 current_test_params = self.parse_test_params(self.current_test_name) 289 ref_test_params = current_test_params.copy() 290 if "legacy" in current_test_params["mode"] and current_test_params["rate"] < 54: 291 if current_test_params["channel"] <= 13: 292 ref_index = self.VALID_RATES["legacy_2GHz"].index( 293 [current_test_params["rate"], 1]) - 1 294 ref_test_params["rate"] = self.VALID_RATES["legacy_2GHz"][ 295 ref_index][0] 296 else: 297 ref_index = self.VALID_RATES["legacy_5GHz"].index( 298 [current_test_params["rate"], 1]) - 1 299 ref_test_params["rate"] = self.VALID_RATES["legacy_5GHz"][ 300 ref_index][0] 301 else: 302 ref_test_params["rate"] = ref_test_params["rate"] + 1 303 304 # Check if reference test has been run and set attenuation accordingly 305 previous_params = [ 306 self.parse_test_params(result["test_name"]) 307 for result in self.testclass_results 308 ] 309 try: 310 ref_index = previous_params.index(ref_test_params) 311 start_atten = self.testclass_results[ref_index]["atten_at_range"] - ( 312 self.testclass_params["adjacent_mcs_range_gap"]) 313 except: 314 print("Reference test not found. Starting from {} dB".format( 315 self.testclass_params["atten_start"])) 316 start_atten = self.testclass_params["atten_start"] 317 return start_atten 318 319 def parse_test_params(self, test_name): 320 """Function that generates test params based on the test name.""" 321 test_name_params = test_name.split("_") 322 testcase_params = collections.OrderedDict() 323 testcase_params["channel"] = int(test_name_params[2][2:]) 324 testcase_params["mode"] = test_name_params[3] 325 326 if "legacy" in testcase_params["mode"].lower(): 327 testcase_params["rate"] = float( 328 str(test_name_params[4]).replace("p", ".")) 329 else: 330 testcase_params["rate"] = int(test_name_params[4][3:]) 331 testcase_params["num_streams"] = int(test_name_params[5][3:]) 332 testcase_params["short_gi"] = 0 333 testcase_params["chain_mask"] = test_name_params[6][2:] 334 if testcase_params["chain_mask"] in ["0", "1"]: 335 testcase_params["attenuated_chain"] = "DUT-Chain-{}".format( 336 1 if testcase_params['chain_mask'] == "0" else 0) 337 else: 338 testcase_params["attenuated_chain"] = None 339 340 if self.testclass_params["traffic_type"] == "UDP": 341 testcase_params["iperf_args"] = '-i 1 -t {} -J -u -b {}'.format( 342 self.testclass_params["iperf_duration"], 343 self.testclass_params["UDP_rates"][testcase_params["mode"]]) 344 elif self.testclass_params["traffic_type"] == "TCP": 345 testcase_params["iperf_args"] = '-i 1 -t {} -J'.format( 346 self.testclass_params["iperf_duration"]) 347 348 if not isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 349 testcase_params["iperf_args"] += ' -R' 350 testcase_params["use_client_output"] = True 351 else: 352 testcase_params["use_client_output"] = False 353 354 return testcase_params 355 356 def _test_sensitivity(self): 357 """ Function that gets called for each test case 358 359 The function gets called in each rvr test case. The function customizes 360 the rvr test based on the test name of the test that called it 361 """ 362 # Compile test parameters from config and test name 363 testcase_params = self.parse_test_params(self.current_test_name) 364 testcase_params.update(self.testclass_params) 365 testcase_params["atten_start"] = self.get_start_atten() 366 num_atten_steps = int( 367 (testcase_params["atten_stop"] - testcase_params["atten_start"]) / 368 testcase_params["atten_step"]) 369 testcase_params["atten_range"] = [ 370 testcase_params["atten_start"] + x * testcase_params["atten_step"] 371 for x in range(0, num_atten_steps) 372 ] 373 374 # Prepare devices and run test 375 if testcase_params["traffic_type"].lower() == "ping": 376 self.setup_ping_test(testcase_params) 377 result = self.run_ping_test(testcase_params) 378 self.process_ping_test_results(testcase_params, result) 379 else: 380 self.setup_rvr_test(testcase_params) 381 result = self.run_rvr_test(testcase_params) 382 self.process_rvr_test_results(testcase_params, result) 383 # Post-process results 384 self.testclass_results.append(result) 385 self.pass_fail_check(result) 386 387 def generate_test_cases(self, channels, chain_mask): 388 """Function that auto-generates test cases for a test class.""" 389 testcase_wrapper = self._test_sensitivity 390 for channel in channels: 391 for mode in self.VALID_TEST_CONFIGS[channel]: 392 if "VHT" in mode: 393 rates = self.VALID_RATES["VHT"] 394 elif "HT" in mode: 395 rates = self.VALID_RATES["HT"] 396 elif "legacy" in mode and channel < 14: 397 rates = self.VALID_RATES["legacy_2GHz"] 398 elif "legacy" in mode and channel > 14: 399 rates = self.VALID_RATES["legacy_5GHz"] 400 else: 401 raise ValueError("Invalid test mode.") 402 for chain, rate in itertools.product(chain_mask, rates): 403 if str(chain) in ["0", "1"] and rate[1] == 2: 404 # Do not test 2-stream rates in single chain mode 405 continue 406 if "legacy" in mode: 407 testcase_name = "test_sensitivity_ch{}_{}_{}_nss{}_ch{}".format( 408 channel, mode, 409 str(rate[0]).replace(".", "p"), rate[1], chain) 410 else: 411 testcase_name = "test_sensitivity_ch{}_{}_mcs{}_nss{}_ch{}".format( 412 channel, mode, rate[0], rate[1], chain) 413 setattr(self, testcase_name, testcase_wrapper) 414 self.tests.append(testcase_name) 415 416 417class WifiSensitivity_AllChannels_Test(WifiSensitivityTest): 418 def __init__(self, controllers): 419 base_test.BaseTestClass.__init__(self, controllers) 420 self.generate_test_cases( 421 [1, 2, 6, 10, 11, 36, 40, 44, 48, 149, 153, 157, 161], 422 ["0", "1", "2x2"]) 423 424 425class WifiSensitivity_2GHz_Test(WifiSensitivityTest): 426 def __init__(self, controllers): 427 base_test.BaseTestClass.__init__(self, controllers) 428 self.generate_test_cases([1, 2, 6, 10, 11], ["0", "1", "2x2"]) 429 430 431class WifiSensitivity_5GHz_Test(WifiSensitivityTest): 432 def __init__(self, controllers): 433 base_test.BaseTestClass.__init__(self, controllers) 434 self.generate_test_cases([36, 40, 44, 48, 149, 153, 157, 161], 435 ["0", "1", "2x2"]) 436 437 438class WifiSensitivity_UNII1_Test(WifiSensitivityTest): 439 def __init__(self, controllers): 440 base_test.BaseTestClass.__init__(self, controllers) 441 self.generate_test_cases([36, 40, 44, 48], ["0", "1", "2x2"]) 442 443 444class WifiSensitivity_UNII3_Test(WifiSensitivityTest): 445 def __init__(self, controllers): 446 base_test.BaseTestClass.__init__(self, controllers) 447 self.generate_test_cases([149, 153, 157, 161], ["0", "1", "2x2"]) 448 449 450class WifiSensitivity_ch1_Test(WifiSensitivityTest): 451 def __init__(self, controllers): 452 base_test.BaseTestClass.__init__(self, controllers) 453 self.generate_test_cases([1], ["0", "1", "2x2"]) 454 455 456class WifiSensitivity_ch2_Test(WifiSensitivityTest): 457 def __init__(self, controllers): 458 base_test.BaseTestClass.__init__(self, controllers) 459 self.generate_test_cases([2], ["0", "1", "2x2"]) 460 461 462class WifiSensitivity_ch6_Test(WifiSensitivityTest): 463 def __init__(self, controllers): 464 base_test.BaseTestClass.__init__(self, controllers) 465 self.generate_test_cases([6], ["0", "1", "2x2"]) 466 467 468class WifiSensitivity_ch10_Test(WifiSensitivityTest): 469 def __init__(self, controllers): 470 base_test.BaseTestClass.__init__(self, controllers) 471 self.generate_test_cases([10], ["0", "1", "2x2"]) 472 473 474class WifiSensitivity_ch11_Test(WifiSensitivityTest): 475 def __init__(self, controllers): 476 base_test.BaseTestClass.__init__(self, controllers) 477 self.generate_test_cases([11], ["0", "1", "2x2"]) 478 479 480class WifiSensitivity_ch36_Test(WifiSensitivityTest): 481 def __init__(self, controllers): 482 base_test.BaseTestClass.__init__(self, controllers) 483 self.generate_test_cases([36], ["0", "1", "2x2"]) 484 485 486class WifiSensitivity_ch40_Test(WifiSensitivityTest): 487 def __init__(self, controllers): 488 base_test.BaseTestClass.__init__(self, controllers) 489 self.generate_test_cases([40], ["0", "1", "2x2"]) 490 491 492class WifiSensitivity_ch44_Test(WifiSensitivityTest): 493 def __init__(self, controllers): 494 base_test.BaseTestClass.__init__(self, controllers) 495 self.generate_test_cases([44], ["0", "1", "2x2"]) 496 497 498class WifiSensitivity_ch48_Test(WifiSensitivityTest): 499 def __init__(self, controllers): 500 base_test.BaseTestClass.__init__(self, controllers) 501 self.generate_test_cases([48], ["0", "1", "2x2"]) 502 503 504class WifiSensitivity_ch149_Test(WifiSensitivityTest): 505 def __init__(self, controllers): 506 base_test.BaseTestClass.__init__(self, controllers) 507 self.generate_test_cases([149], ["0", "1", "2x2"]) 508 509 510class WifiSensitivity_ch153_Test(WifiSensitivityTest): 511 def __init__(self, controllers): 512 base_test.BaseTestClass.__init__(self, controllers) 513 self.generate_test_cases([153], ["0", "1", "2x2"]) 514 515 516class WifiSensitivity_ch157_Test(WifiSensitivityTest): 517 def __init__(self, controllers): 518 base_test.BaseTestClass.__init__(self, controllers) 519 self.generate_test_cases([157], ["0", "1", "2x2"]) 520 521 522class WifiSensitivity_ch161_Test(WifiSensitivityTest): 523 def __init__(self, controllers): 524 base_test.BaseTestClass.__init__(self, controllers) 525 self.generate_test_cases([161], ["0", "1", "2x2"]) 526