#!/usr/bin/env python3.4 # # Copyright 2018 - The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the 'License'); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an 'AS IS' BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import collections import itertools import json import logging import math import numpy import os import statistics from acts import asserts from acts import base_test from acts import context from acts import utils from acts.controllers.utils_lib import ssh from acts.controllers import iperf_server as ipf from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger from acts_contrib.test_utils.wifi import ota_chamber from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap from acts_contrib.test_utils.wifi import wifi_test_utils as wutils from concurrent.futures import ThreadPoolExecutor from functools import partial SHORT_SLEEP = 1 MED_SLEEP = 6 CONST_3dB = 3.01029995664 RSSI_ERROR_VAL = float('nan') class WifiRssiTest(base_test.BaseTestClass): """Class to test WiFi RSSI reporting. This class tests RSSI reporting on android devices. The class tests RSSI accuracy by checking RSSI over a large attenuation range, checks for RSSI stability over time when attenuation is fixed, and checks that RSSI quickly and reacts to changes attenuation by checking RSSI trajectories over configurable attenuation waveforms.For an example config file to run this test class see example_connectivity_performance_ap_sta.json. """ def __init__(self, controllers): base_test.BaseTestClass.__init__(self, controllers) self.testcase_metric_logger = ( BlackboxMappedMetricLogger.for_test_case()) self.testclass_metric_logger = ( BlackboxMappedMetricLogger.for_test_class()) self.publish_test_metrics = True def setup_class(self): self.dut = self.android_devices[0] req_params = [ 'RemoteServer', 'RetailAccessPoints', 'rssi_test_params', 'main_network', 'testbed_params' ] self.unpack_userparams(req_params) self.testclass_params = self.rssi_test_params self.num_atten = self.attenuators[0].instrument.num_atten self.iperf_server = self.iperf_servers[0] self.iperf_client = self.iperf_clients[0] self.remote_server = ssh.connection.SshConnection( ssh.settings.from_config(self.RemoteServer[0]['ssh_config'])) self.access_point = retail_ap.create(self.RetailAccessPoints)[0] self.log_path = os.path.join(logging.log_path, 'results') os.makedirs(self.log_path, exist_ok=True) self.log.info('Access Point Configuration: {}'.format( self.access_point.ap_settings)) self.testclass_results = [] # Turn WiFi ON if self.testclass_params.get('airplane_mode', 1): self.log.info('Turning on airplane mode.') asserts.assert_true(utils.force_airplane_mode(self.dut, True), 'Can not turn on airplane mode.') wutils.wifi_toggle_state(self.dut, True) def teardown_test(self): self.iperf_server.stop() def teardown_class(self): # Turn WiFi OFF and reset AP self.access_point.teardown() for dev in self.android_devices: wutils.wifi_toggle_state(dev, False) dev.go_to_sleep() def pass_fail_check_rssi_stability(self, testcase_params, postprocessed_results): """Check the test result and decide if it passed or failed. Checks the RSSI test result and fails the test if the standard deviation of signal_poll_rssi is beyond the threshold defined in the config file. Args: testcase_params: dict containing test-specific parameters postprocessed_results: compiled arrays of RSSI measurements """ # Set Blackbox metric values if self.publish_test_metrics: self.testcase_metric_logger.add_metric( 'signal_poll_rssi_stdev', max(postprocessed_results['signal_poll_rssi']['stdev'])) self.testcase_metric_logger.add_metric( 'chain_0_rssi_stdev', max(postprocessed_results['chain_0_rssi']['stdev'])) self.testcase_metric_logger.add_metric( 'chain_1_rssi_stdev', max(postprocessed_results['chain_1_rssi']['stdev'])) # Evaluate test pass/fail test_failed = any([ stdev > self.testclass_params['stdev_tolerance'] for stdev in postprocessed_results['signal_poll_rssi']['stdev'] ]) test_message = ( 'RSSI stability {0}. Standard deviation was {1} dB ' '(limit {2}), per chain standard deviation [{3}, {4}] dB'.format( 'failed' * test_failed + 'passed' * (not test_failed), [ float('{:.2f}'.format(x)) for x in postprocessed_results['signal_poll_rssi']['stdev'] ], self.testclass_params['stdev_tolerance'], [ float('{:.2f}'.format(x)) for x in postprocessed_results['chain_0_rssi']['stdev'] ], [ float('{:.2f}'.format(x)) for x in postprocessed_results['chain_1_rssi']['stdev'] ])) if test_failed: asserts.fail(test_message) asserts.explicit_pass(test_message) def pass_fail_check_rssi_accuracy(self, testcase_params, postprocessed_results): """Check the test result and decide if it passed or failed. Checks the RSSI test result and compares and compute its deviation from the predicted RSSI. This computation is done for all reported RSSI values. The test fails if any of the RSSI values specified in rssi_under_test have an average error beyond what is specified in the configuration file. Args: postprocessed_results: compiled arrays of RSSI measurements testcase_params: dict containing params such as list of RSSIs under test, i.e., can cause test to fail and boolean indicating whether to look at absolute RSSI accuracy, or centered RSSI accuracy. Centered accuracy is computed after systematic RSSI shifts are removed. """ test_failed = False test_message = '' if testcase_params['absolute_accuracy']: error_type = 'absolute' else: error_type = 'centered' for key, val in postprocessed_results.items(): # Compute the error metrics ignoring invalid RSSI readings # If all readings invalid, set error to RSSI_ERROR_VAL if 'rssi' in key and 'predicted' not in key: filtered_error = [x for x in val['error'] if not math.isnan(x)] if filtered_error: avg_shift = statistics.mean(filtered_error) if testcase_params['absolute_accuracy']: avg_error = statistics.mean( [abs(x) for x in filtered_error]) else: avg_error = statistics.mean( [abs(x - avg_shift) for x in filtered_error]) else: avg_error = RSSI_ERROR_VAL avg_shift = RSSI_ERROR_VAL # Set Blackbox metric values if self.publish_test_metrics: self.testcase_metric_logger.add_metric( '{}_error'.format(key), avg_error) self.testcase_metric_logger.add_metric( '{}_shift'.format(key), avg_shift) # Evaluate test pass/fail rssi_failure = (avg_error > self.testclass_params['abs_tolerance'] ) or math.isnan(avg_error) if rssi_failure and key in testcase_params['rssi_under_test']: test_message = test_message + ( '{} failed ({} error = {:.2f} dB, ' 'shift = {:.2f} dB)\n').format(key, error_type, avg_error, avg_shift) test_failed = True elif rssi_failure: test_message = test_message + ( '{} failed (ignored) ({} error = {:.2f} dB, ' 'shift = {:.2f} dB)\n').format(key, error_type, avg_error, avg_shift) else: test_message = test_message + ( '{} passed ({} error = {:.2f} dB, ' 'shift = {:.2f} dB)\n').format(key, error_type, avg_error, avg_shift) if test_failed: asserts.fail(test_message) asserts.explicit_pass(test_message) def post_process_rssi_sweep(self, rssi_result): """Postprocesses and saves JSON formatted results. Args: rssi_result: dict containing attenuation, rssi and other meta data Returns: postprocessed_results: compiled arrays of RSSI data used in pass/fail check """ # Save output as text file results_file_path = os.path.join(self.log_path, self.current_test_name) with open(results_file_path, 'w') as results_file: json.dump(wputils.serialize_dict(rssi_result), results_file, indent=4) # Compile results into arrays of RSSIs suitable for plotting # yapf: disable postprocessed_results = collections.OrderedDict( [('signal_poll_rssi', {}), ('signal_poll_avg_rssi', {}), ('scan_rssi', {}), ('chain_0_rssi', {}), ('chain_1_rssi', {}), ('total_attenuation', []), ('predicted_rssi', [])]) # yapf: enable for key, val in postprocessed_results.items(): if 'scan_rssi' in key: postprocessed_results[key]['data'] = [ x for data_point in rssi_result['rssi_result'] for x in data_point[key][rssi_result['connected_bssid']]['data'] ] postprocessed_results[key]['mean'] = [ x[key][rssi_result['connected_bssid']]['mean'] for x in rssi_result['rssi_result'] ] postprocessed_results[key]['stdev'] = [ x[key][rssi_result['connected_bssid']]['stdev'] for x in rssi_result['rssi_result'] ] elif 'predicted_rssi' in key: postprocessed_results['total_attenuation'] = [ att + rssi_result['fixed_attenuation'] + rssi_result['dut_front_end_loss'] for att in rssi_result['attenuation'] ] postprocessed_results['predicted_rssi'] = [ rssi_result['ap_tx_power'] - att for att in postprocessed_results['total_attenuation'] ] elif 'rssi' in key: postprocessed_results[key]['data'] = [ x for data_point in rssi_result['rssi_result'] for x in data_point[key]['data'] ] postprocessed_results[key]['mean'] = [ x[key]['mean'] for x in rssi_result['rssi_result'] ] postprocessed_results[key]['stdev'] = [ x[key]['stdev'] for x in rssi_result['rssi_result'] ] # Compute RSSI errors for key, val in postprocessed_results.items(): if 'chain' in key: postprocessed_results[key]['error'] = [ postprocessed_results[key]['mean'][idx] + CONST_3dB - postprocessed_results['predicted_rssi'][idx] for idx in range( len(postprocessed_results['predicted_rssi'])) ] elif 'rssi' in key and 'predicted' not in key: postprocessed_results[key]['error'] = [ postprocessed_results[key]['mean'][idx] - postprocessed_results['predicted_rssi'][idx] for idx in range( len(postprocessed_results['predicted_rssi'])) ] return postprocessed_results def plot_rssi_vs_attenuation(self, postprocessed_results): """Function to plot RSSI vs attenuation sweeps Args: postprocessed_results: compiled arrays of RSSI data. """ figure = BokehFigure(self.current_test_name, x_label='Attenuation (dB)', primary_y_label='RSSI (dBm)') figure.add_line(postprocessed_results['total_attenuation'], postprocessed_results['signal_poll_rssi']['mean'], 'Signal Poll RSSI', marker='circle') figure.add_line(postprocessed_results['total_attenuation'], postprocessed_results['scan_rssi']['mean'], 'Scan RSSI', marker='circle') figure.add_line(postprocessed_results['total_attenuation'], postprocessed_results['chain_0_rssi']['mean'], 'Chain 0 RSSI', marker='circle') figure.add_line(postprocessed_results['total_attenuation'], postprocessed_results['chain_1_rssi']['mean'], 'Chain 1 RSSI', marker='circle') figure.add_line(postprocessed_results['total_attenuation'], postprocessed_results['predicted_rssi'], 'Predicted RSSI', marker='circle') output_file_path = os.path.join(self.log_path, self.current_test_name + '.html') figure.generate_figure(output_file_path) def plot_rssi_vs_time(self, rssi_result, postprocessed_results, center_curves): """Function to plot RSSI vs time. Args: rssi_result: dict containing raw RSSI data postprocessed_results: compiled arrays of RSSI data center_curvers: boolean indicating whether to shift curves to align them with predicted RSSIs """ figure = BokehFigure( self.current_test_name, x_label='Time (s)', primary_y_label=center_curves * 'Centered' + 'RSSI (dBm)', ) # yapf: disable rssi_time_series = collections.OrderedDict( [('signal_poll_rssi', []), ('signal_poll_avg_rssi', []), ('scan_rssi', []), ('chain_0_rssi', []), ('chain_1_rssi', []), ('predicted_rssi', [])]) # yapf: enable for key, val in rssi_time_series.items(): if 'predicted_rssi' in key: rssi_time_series[key] = [ x for x in postprocessed_results[key] for copies in range( len(rssi_result['rssi_result'][0]['signal_poll_rssi'] ['data'])) ] elif 'rssi' in key: if center_curves: filtered_error = [ x for x in postprocessed_results[key]['error'] if not math.isnan(x) ] if filtered_error: avg_shift = statistics.mean(filtered_error) else: avg_shift = 0 rssi_time_series[key] = [ x - avg_shift for x in postprocessed_results[key]['data'] ] else: rssi_time_series[key] = postprocessed_results[key]['data'] time_vec = [ self.testclass_params['polling_frequency'] * x for x in range(len(rssi_time_series[key])) ] if len(rssi_time_series[key]) > 0: figure.add_line(time_vec, rssi_time_series[key], key) output_file_path = os.path.join(self.log_path, self.current_test_name + '.html') figure.generate_figure(output_file_path) def plot_rssi_distribution(self, postprocessed_results): """Function to plot RSSI distributions. Args: postprocessed_results: compiled arrays of RSSI data """ monitored_rssis = ['signal_poll_rssi', 'chain_0_rssi', 'chain_1_rssi'] rssi_dist = collections.OrderedDict() for rssi_key in monitored_rssis: rssi_data = postprocessed_results[rssi_key] rssi_dist[rssi_key] = collections.OrderedDict() unique_rssi = sorted(set(rssi_data['data'])) rssi_counts = [] for value in unique_rssi: rssi_counts.append(rssi_data['data'].count(value)) total_count = sum(rssi_counts) rssi_dist[rssi_key]['rssi_values'] = unique_rssi rssi_dist[rssi_key]['rssi_pdf'] = [ x / total_count for x in rssi_counts ] rssi_dist[rssi_key]['rssi_cdf'] = [] cum_prob = 0 for prob in rssi_dist[rssi_key]['rssi_pdf']: cum_prob += prob rssi_dist[rssi_key]['rssi_cdf'].append(cum_prob) figure = BokehFigure(self.current_test_name, x_label='RSSI (dBm)', primary_y_label='p(RSSI = x)', secondary_y_label='p(RSSI <= x)') for rssi_key, rssi_data in rssi_dist.items(): figure.add_line(x_data=rssi_data['rssi_values'], y_data=rssi_data['rssi_pdf'], legend='{} PDF'.format(rssi_key), y_axis='default') figure.add_line(x_data=rssi_data['rssi_values'], y_data=rssi_data['rssi_cdf'], legend='{} CDF'.format(rssi_key), y_axis='secondary') output_file_path = os.path.join(self.log_path, self.current_test_name + '_dist.html') figure.generate_figure(output_file_path) def run_rssi_test(self, testcase_params): """Test function to run RSSI tests. The function runs an RSSI test in the current device/AP configuration. Function is called from another wrapper function that sets up the testbed for the RvR test Args: testcase_params: dict containing test-specific parameters Returns: rssi_result: dict containing rssi_result and meta data """ # Run test and log result rssi_result = collections.OrderedDict() rssi_result['test_name'] = self.current_test_name rssi_result['testcase_params'] = testcase_params rssi_result['ap_settings'] = self.access_point.ap_settings.copy() rssi_result['attenuation'] = list(testcase_params['rssi_atten_range']) rssi_result['connected_bssid'] = self.main_network[ testcase_params['band']].get('BSSID', '00:00:00:00') channel_mode_combo = '{}_{}'.format(str(testcase_params['channel']), testcase_params['mode']) channel_str = str(testcase_params['channel']) if channel_mode_combo in self.testbed_params['ap_tx_power']: rssi_result['ap_tx_power'] = self.testbed_params['ap_tx_power'][ channel_mode_combo] else: rssi_result['ap_tx_power'] = self.testbed_params['ap_tx_power'][ str(testcase_params['channel'])] rssi_result['fixed_attenuation'] = self.testbed_params[ 'fixed_attenuation'][channel_str] rssi_result['dut_front_end_loss'] = self.testbed_params[ 'dut_front_end_loss'][channel_str] self.log.info('Start running RSSI test.') rssi_result['rssi_result'] = [] rssi_result['llstats'] = [] llstats_obj = wputils.LinkLayerStats(self.dut) # Start iperf traffic if required by test if testcase_params['active_traffic'] and testcase_params[ 'traffic_type'] == 'iperf': self.iperf_server.start(tag=0) if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): iperf_server_address = self.dut_ip else: iperf_server_address = wputils.get_server_address( self.remote_server, self.dut_ip, '255.255.255.0') executor = ThreadPoolExecutor(max_workers=1) thread_future = executor.submit( self.iperf_client.start, iperf_server_address, testcase_params['iperf_args'], 0, testcase_params['traffic_timeout'] + SHORT_SLEEP) executor.shutdown(wait=False) elif testcase_params['active_traffic'] and testcase_params[ 'traffic_type'] == 'ping': thread_future = wputils.get_ping_stats_nb( self.remote_server, self.dut_ip, testcase_params['traffic_timeout'], 0.02, 64) else: thread_future = wputils.get_ping_stats_nb( self.remote_server, self.dut_ip, testcase_params['traffic_timeout'], 0.5, 64) llstats_obj.update_stats() for atten in testcase_params['rssi_atten_range']: # Set Attenuation self.log.info('Setting attenuation to {} dB'.format(atten)) for attenuator in self.attenuators: attenuator.set_atten(atten) current_rssi = collections.OrderedDict() current_rssi = wputils.get_connected_rssi( self.dut, testcase_params['connected_measurements'], self.testclass_params['polling_frequency'], testcase_params['first_measurement_delay']) current_rssi['scan_rssi'] = wputils.get_scan_rssi( self.dut, testcase_params['tracked_bssid'], testcase_params['scan_measurements']) rssi_result['rssi_result'].append(current_rssi) llstats_obj.update_stats() curr_llstats = llstats_obj.llstats_incremental.copy() rssi_result['llstats'].append(curr_llstats) self.log.info( 'Connected RSSI at {0:.2f} dB is {1:.2f} [{2:.2f}, {3:.2f}] dB' .format(atten, current_rssi['signal_poll_rssi']['mean'], current_rssi['chain_0_rssi']['mean'], current_rssi['chain_1_rssi']['mean'])) # Stop iperf traffic if needed for attenuator in self.attenuators: attenuator.set_atten(0) thread_future.result() if testcase_params['active_traffic'] and testcase_params[ 'traffic_type'] == 'iperf': self.iperf_server.stop() return rssi_result def setup_ap(self, testcase_params): """Function that gets devices ready for the test. Args: testcase_params: dict containing test-specific parameters """ band = self.access_point.band_lookup_by_channel( testcase_params['channel']) if '6G' in band: frequency = wutils.WifiEnums.channel_6G_to_freq[int( testcase_params['channel'].strip('6g'))] else: if testcase_params['channel'] < 13: frequency = wutils.WifiEnums.channel_2G_to_freq[ testcase_params['channel']] else: frequency = wutils.WifiEnums.channel_5G_to_freq[ testcase_params['channel']] if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES: self.access_point.set_region(self.testbed_params['DFS_region']) else: self.access_point.set_region(self.testbed_params['default_region']) self.access_point.set_channel(testcase_params['band'], testcase_params['channel']) self.access_point.set_bandwidth(testcase_params['band'], testcase_params['mode']) self.log.info('Access Point Configuration: {}'.format( self.access_point.ap_settings)) def setup_dut(self, testcase_params): """Sets up the DUT in the configuration required by the test.""" # Turn screen off to preserve battery if self.testbed_params.get('screen_on', False) or self.testclass_params.get( 'screen_on', False): self.dut.droid.wakeLockAcquireDim() else: self.dut.go_to_sleep() if wputils.validate_network(self.dut, testcase_params['test_network']['SSID']): self.log.info('Already connected to desired network') else: wutils.wifi_toggle_state(self.dut, True) wutils.reset_wifi(self.dut) self.main_network[testcase_params['band']][ 'channel'] = testcase_params['channel'] wutils.set_wifi_country_code(self.dut, self.testclass_params['country_code']) if self.testbed_params.get('txbf_off', False): wputils.disable_beamforming(self.dut) wutils.wifi_connect(self.dut, self.main_network[testcase_params['band']], num_of_tries=5) self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] def setup_rssi_test(self, testcase_params): """Main function to test RSSI. The function sets up the AP in the correct channel and mode configuration and called rssi_test to sweep attenuation and measure RSSI Args: testcase_params: dict containing test-specific parameters Returns: rssi_result: dict containing rssi_results and meta data """ # Configure AP self.setup_ap(testcase_params) # Initialize attenuators for attenuator in self.attenuators: attenuator.set_atten(testcase_params['rssi_atten_range'][0]) # Connect DUT to Network self.setup_dut(testcase_params) def get_traffic_timeout(self, testcase_params): """Function to comput iperf session length required in RSSI test. Args: testcase_params: dict containing test-specific parameters Returns: traffic_timeout: length of iperf session required in rssi test """ atten_step_duration = testcase_params['first_measurement_delay'] + ( testcase_params['connected_measurements'] * self.testclass_params['polling_frequency'] ) + testcase_params['scan_measurements'] * MED_SLEEP timeout = len(testcase_params['rssi_atten_range'] ) * atten_step_duration + MED_SLEEP return timeout def compile_rssi_vs_atten_test_params(self, testcase_params): """Function to complete compiling test-specific parameters Args: testcase_params: dict containing test-specific parameters """ # Check if test should be skipped. wputils.check_skip_conditions(testcase_params, self.dut, self.access_point, getattr(self, 'ota_chamber', None)) testcase_params.update( connected_measurements=self. testclass_params['rssi_vs_atten_connected_measurements'], scan_measurements=self. testclass_params['rssi_vs_atten_scan_measurements'], first_measurement_delay=SHORT_SLEEP, absolute_accuracy=1) rssi_under_test = self.testclass_params['rssi_vs_atten_metrics'] if self.testclass_params[ 'rssi_vs_atten_scan_measurements'] == 0 and 'scan_rssi' in rssi_under_test: rssi_under_test.remove('scan_rssi') testcase_params['rssi_under_test'] = rssi_under_test testcase_params['band'] = self.access_point.band_lookup_by_channel( testcase_params['channel']) testcase_params['test_network'] = self.main_network[ testcase_params['band']] testcase_params['tracked_bssid'] = [ self.main_network[testcase_params['band']].get( 'BSSID', '00:00:00:00') ] testcase_params['rssi_atten_range'] = numpy.arange( self.testclass_params['rssi_vs_atten_start'], self.testclass_params['rssi_vs_atten_stop'], self.testclass_params['rssi_vs_atten_step']).tolist() testcase_params['traffic_timeout'] = self.get_traffic_timeout( testcase_params) if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): testcase_params['iperf_args'] = '-i 1 -t {} -J'.format( testcase_params['traffic_timeout']) else: testcase_params['iperf_args'] = '-i 1 -t {} -J -R'.format( testcase_params['traffic_timeout']) return testcase_params def compile_rssi_stability_test_params(self, testcase_params): """Function to complete compiling test-specific parameters Args: testcase_params: dict containing test-specific parameters """ # Check if test should be skipped. wputils.check_skip_conditions(testcase_params, self.dut, self.access_point, getattr(self, 'ota_chamber', None)) testcase_params.update( connected_measurements=int( self.testclass_params['rssi_stability_duration'] / self.testclass_params['polling_frequency']), scan_measurements=0, first_measurement_delay=SHORT_SLEEP, rssi_atten_range=self.testclass_params['rssi_stability_atten']) testcase_params['band'] = self.access_point.band_lookup_by_channel( testcase_params['channel']) testcase_params['test_network'] = self.main_network[ testcase_params['band']] testcase_params['tracked_bssid'] = [ self.main_network[testcase_params['band']].get( 'BSSID', '00:00:00:00') ] testcase_params['traffic_timeout'] = self.get_traffic_timeout( testcase_params) if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): testcase_params['iperf_args'] = '-i 1 -t {} -J'.format( testcase_params['traffic_timeout']) else: testcase_params['iperf_args'] = '-i 1 -t {} -J -R'.format( testcase_params['traffic_timeout']) return testcase_params def compile_rssi_tracking_test_params(self, testcase_params): """Function to complete compiling test-specific parameters Args: testcase_params: dict containing test-specific parameters """ # Check if test should be skipped. wputils.check_skip_conditions(testcase_params, self.dut, self.access_point, getattr(self, 'ota_chamber', None)) testcase_params.update(connected_measurements=int( 1 / self.testclass_params['polling_frequency']), scan_measurements=0, first_measurement_delay=0, rssi_under_test=['signal_poll_rssi'], absolute_accuracy=0) testcase_params['band'] = self.access_point.band_lookup_by_channel( testcase_params['channel']) testcase_params['test_network'] = self.main_network[ testcase_params['band']] testcase_params['tracked_bssid'] = [ self.main_network[testcase_params['band']].get( 'BSSID', '00:00:00:00') ] rssi_atten_range = [] for waveform in self.testclass_params['rssi_tracking_waveforms']: waveform_vector = [] for section in range(len(waveform['atten_levels']) - 1): section_limits = waveform['atten_levels'][section:section + 2] up_down = (1 - 2 * (section_limits[1] < section_limits[0])) temp_section = list( range(section_limits[0], section_limits[1] + up_down, up_down * waveform['step_size'])) temp_section = [ temp_section[idx] for idx in range(len(temp_section)) for n in range(waveform['step_duration']) ] waveform_vector += temp_section waveform_vector = waveform_vector * waveform['repetitions'] rssi_atten_range = rssi_atten_range + waveform_vector testcase_params['rssi_atten_range'] = rssi_atten_range testcase_params['traffic_timeout'] = self.get_traffic_timeout( testcase_params) if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): testcase_params['iperf_args'] = '-i 1 -t {} -J'.format( testcase_params['traffic_timeout']) else: testcase_params['iperf_args'] = '-i 1 -t {} -J -R'.format( testcase_params['traffic_timeout']) return testcase_params def _test_rssi_vs_atten(self, testcase_params): """Function that gets called for each test case of rssi_vs_atten The function gets called in each rssi test case. The function customizes the test based on the test name of the test that called it Args: testcase_params: dict containing test-specific parameters """ testcase_params = self.compile_rssi_vs_atten_test_params( testcase_params) self.setup_rssi_test(testcase_params) rssi_result = self.run_rssi_test(testcase_params) rssi_result['postprocessed_results'] = self.post_process_rssi_sweep( rssi_result) self.testclass_results.append(rssi_result) self.plot_rssi_vs_attenuation(rssi_result['postprocessed_results']) self.pass_fail_check_rssi_accuracy( testcase_params, rssi_result['postprocessed_results']) def _test_rssi_stability(self, testcase_params): """ Function that gets called for each test case of rssi_stability The function gets called in each stability test case. The function customizes test based on the test name of the test that called it """ testcase_params = self.compile_rssi_stability_test_params( testcase_params) self.setup_rssi_test(testcase_params) rssi_result = self.run_rssi_test(testcase_params) rssi_result['postprocessed_results'] = self.post_process_rssi_sweep( rssi_result) self.testclass_results.append(rssi_result) self.plot_rssi_vs_time(rssi_result, rssi_result['postprocessed_results'], 1) self.plot_rssi_distribution(rssi_result['postprocessed_results']) self.pass_fail_check_rssi_stability( testcase_params, rssi_result['postprocessed_results']) def _test_rssi_tracking(self, testcase_params): """ Function that gets called for each test case of rssi_tracking The function gets called in each rssi test case. The function customizes the test based on the test name of the test that called it """ testcase_params = self.compile_rssi_tracking_test_params( testcase_params) self.setup_rssi_test(testcase_params) rssi_result = self.run_rssi_test(testcase_params) rssi_result['postprocessed_results'] = self.post_process_rssi_sweep( rssi_result) self.testclass_results.append(rssi_result) self.plot_rssi_vs_time(rssi_result, rssi_result['postprocessed_results'], 1) self.pass_fail_check_rssi_accuracy( testcase_params, rssi_result['postprocessed_results']) def generate_test_cases(self, test_types, channels, modes, traffic_modes): """Function that auto-generates test cases for a test class.""" test_cases = [] allowed_configs = { 20: [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 116, 132, 140, 149, 153, 157, 161, '6g37', '6g117', '6g213' ], 40: [36, 44, 100, 149, 157, '6g37', '6g117', '6g213'], 80: [36, 100, 149, '6g37', '6g117', '6g213'], 160: [36, '6g37', '6g117', '6g213'] } for channel, mode, traffic_mode, test_type in itertools.product( channels, modes, traffic_modes, test_types): bandwidth = int(''.join([x for x in mode if x.isdigit()])) if channel not in allowed_configs[bandwidth]: continue test_name = test_type + '_ch{}_{}_{}'.format( channel, mode, traffic_mode) testcase_params = collections.OrderedDict( channel=channel, mode=mode, active_traffic=(traffic_mode == 'ActiveTraffic'), traffic_type=self.user_params['rssi_test_params'] ['traffic_type'], ) test_function = getattr(self, '_{}'.format(test_type)) setattr(self, test_name, partial(test_function, testcase_params)) test_cases.append(test_name) return test_cases class WifiRssi_AllChannels_ActiveTraffic_Test(WifiRssiTest): def __init__(self, controllers): super().__init__(controllers) self.tests = self.generate_test_cases( ['test_rssi_stability', 'test_rssi_vs_atten'], [ 1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161, '6g37', '6g117', '6g213' ], ['bw20', 'bw40', 'bw80', 'bw160'], ['ActiveTraffic']) class WifiRssi_SampleChannels_NoTraffic_Test(WifiRssiTest): def __init__(self, controllers): super().__init__(controllers) self.tests = self.generate_test_cases( ['test_rssi_stability', 'test_rssi_vs_atten'], [6, 36, 149, '6g37'], ['bw20', 'bw40', 'bw80', 'bw160'], ['NoTraffic']) class WifiRssiTrackingTest(WifiRssiTest): def __init__(self, controllers): super().__init__(controllers) self.tests = self.generate_test_cases(['test_rssi_tracking'], [6, 36, 149, '6g37'], ['bw20', 'bw40', 'bw80', 'bw160'], ['ActiveTraffic', 'NoTraffic']) # Over-the air version of RSSI tests class WifiOtaRssiTest(WifiRssiTest): """Class to test over-the-air rssi tests. This class implements measures WiFi RSSI tests in an OTA chamber. It allows setting orientation and other chamber parameters to study performance in varying channel conditions """ def __init__(self, controllers): base_test.BaseTestClass.__init__(self, controllers) self.testcase_metric_logger = ( BlackboxMappedMetricLogger.for_test_case()) self.testclass_metric_logger = ( BlackboxMappedMetricLogger.for_test_class()) self.publish_test_metrics = False def setup_class(self): WifiRssiTest.setup_class(self) self.ota_chamber = ota_chamber.create( self.user_params['OTAChamber'])[0] def teardown_class(self): WifiRssiTest.teardown_class(self) self.ota_chamber.reset_chamber() self.process_testclass_results() def teardown_test(self): if self.ota_chamber.current_mode == 'continuous': self.ota_chamber.reset_chamber() def extract_test_id(self, testcase_params, id_fields): test_id = collections.OrderedDict( (param, testcase_params[param]) for param in id_fields) return test_id def process_testclass_results(self): """Saves all test results to enable comparison.""" testclass_data = collections.OrderedDict() for test_result in self.testclass_results: current_params = test_result['testcase_params'] channel = current_params['channel'] channel_data = testclass_data.setdefault( channel, collections.OrderedDict(orientation=[], rssi=collections.OrderedDict( signal_poll_rssi=[], chain_0_rssi=[], chain_1_rssi=[]))) channel_data['orientation'].append(current_params['orientation']) channel_data['rssi']['signal_poll_rssi'].append( test_result['postprocessed_results']['signal_poll_rssi'] ['mean'][0]) channel_data['rssi']['chain_0_rssi'].append( test_result['postprocessed_results']['chain_0_rssi']['mean'] [0]) channel_data['rssi']['chain_1_rssi'].append( test_result['postprocessed_results']['chain_1_rssi']['mean'] [0]) # Publish test class metrics for channel, channel_data in testclass_data.items(): for rssi_metric, rssi_metric_value in channel_data['rssi'].items(): metric_name = 'ota_summary_ch{}.avg_{}'.format( channel, rssi_metric) metric_value = numpy.mean(rssi_metric_value) self.testclass_metric_logger.add_metric( metric_name, metric_value) # Plot test class results chamber_mode = self.testclass_results[0]['testcase_params'][ 'chamber_mode'] if chamber_mode == 'orientation': x_label = 'Angle (deg)' elif chamber_mode == 'stepped stirrers': x_label = 'Position Index' elif chamber_mode == 'StirrersOn': return plots = [] for channel, channel_data in testclass_data.items(): current_plot = BokehFigure( title='Channel {} - Rssi vs. Position'.format(channel), x_label=x_label, primary_y_label='RSSI (dBm)', ) for rssi_metric, rssi_metric_value in channel_data['rssi'].items(): legend = rssi_metric current_plot.add_line(channel_data['orientation'], rssi_metric_value, legend) current_plot.generate_figure() plots.append(current_plot) current_context = context.get_current_context().get_full_output_path() plot_file_path = os.path.join(current_context, 'results.html') BokehFigure.save_figures(plots, plot_file_path) def setup_rssi_test(self, testcase_params): # Test setup WifiRssiTest.setup_rssi_test(self, testcase_params) if testcase_params['chamber_mode'] == 'StirrersOn': self.ota_chamber.start_continuous_stirrers() else: self.ota_chamber.set_orientation(testcase_params['orientation']) def compile_ota_rssi_test_params(self, testcase_params): """Function to complete compiling test-specific parameters Args: testcase_params: dict containing test-specific parameters """ # Check if test should be skipped. wputils.check_skip_conditions(testcase_params, self.dut, self.access_point, getattr(self, 'ota_chamber', None)) if 'rssi_over_orientation' in self.test_name: rssi_test_duration = self.testclass_params[ 'rssi_over_orientation_duration'] rssi_ota_test_attenuation = [ self.testclass_params['rssi_ota_test_attenuation'] ] elif 'rssi_variation' in self.test_name: rssi_test_duration = self.testclass_params[ 'rssi_variation_duration'] rssi_ota_test_attenuation = [ self.testclass_params['rssi_ota_test_attenuation'] ] elif 'rssi_vs_atten' in self.test_name: rssi_test_duration = self.testclass_params[ 'rssi_over_orientation_duration'] rssi_ota_test_attenuation = numpy.arange( self.testclass_params['rssi_vs_atten_start'], self.testclass_params['rssi_vs_atten_stop'], self.testclass_params['rssi_vs_atten_step']).tolist() testcase_params.update(connected_measurements=int( rssi_test_duration / self.testclass_params['polling_frequency']), scan_measurements=0, first_measurement_delay=SHORT_SLEEP, rssi_atten_range=rssi_ota_test_attenuation) testcase_params['band'] = self.access_point.band_lookup_by_channel( testcase_params['channel']) testcase_params['test_network'] = self.main_network[ testcase_params['band']] testcase_params['tracked_bssid'] = [ self.main_network[testcase_params['band']].get( 'BSSID', '00:00:00:00') ] testcase_params['traffic_timeout'] = self.get_traffic_timeout( testcase_params) if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): testcase_params['iperf_args'] = '-i 1 -t {} -J'.format( testcase_params['traffic_timeout']) else: testcase_params['iperf_args'] = '-i 1 -t {} -J -R'.format( testcase_params['traffic_timeout']) return testcase_params def _test_ota_rssi(self, testcase_params): testcase_params = self.compile_ota_rssi_test_params(testcase_params) self.setup_rssi_test(testcase_params) rssi_result = self.run_rssi_test(testcase_params) rssi_result['postprocessed_results'] = self.post_process_rssi_sweep( rssi_result) self.testclass_results.append(rssi_result) self.plot_rssi_vs_time(rssi_result, rssi_result['postprocessed_results'], 1) if 'rssi_vs_atten' in self.test_name: self.plot_rssi_vs_attenuation(rssi_result['postprocessed_results']) elif 'rssi_variation' in self.test_name: self.plot_rssi_distribution(rssi_result['postprocessed_results']) def generate_test_cases(self, test_types, channels, modes, traffic_modes, chamber_modes, orientations): test_cases = [] allowed_configs = { 20: [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 116, 132, 140, 149, 153, 157, 161, '6g37', '6g117', '6g213' ], 40: [36, 44, 100, 149, 157, '6g37', '6g117', '6g213'], 80: [36, 100, 149, '6g37', '6g117', '6g213'], 160: [36, '6g37', '6g117', '6g213'] } for (channel, mode, traffic, chamber_mode, orientation, test_type) in itertools.product(channels, modes, traffic_modes, chamber_modes, orientations, test_types): bandwidth = int(''.join([x for x in mode if x.isdigit()])) if channel not in allowed_configs[bandwidth]: continue test_name = test_type + '_ch{}_{}_{}_{}deg'.format( channel, mode, traffic, orientation) testcase_params = collections.OrderedDict( channel=channel, mode=mode, active_traffic=(traffic == 'ActiveTraffic'), traffic_type=self.user_params['rssi_test_params'] ['traffic_type'], chamber_mode=chamber_mode, orientation=orientation) test_function = self._test_ota_rssi setattr(self, test_name, partial(test_function, testcase_params)) test_cases.append(test_name) return test_cases class WifiOtaRssi_StirrerVariation_Test(WifiOtaRssiTest): def __init__(self, controllers): WifiRssiTest.__init__(self, controllers) self.tests = self.generate_test_cases(['test_rssi_variation'], [6, 36, 149, '6g37'], ['bw20', 'bw80', 'bw160'], ['ActiveTraffic'], ['StirrersOn'], [0]) class WifiOtaRssi_TenDegree_Test(WifiOtaRssiTest): def __init__(self, controllers): WifiRssiTest.__init__(self, controllers) self.tests = self.generate_test_cases(['test_rssi_over_orientation'], [6, 36, 149, '6g37'], ['bw20', 'bw80', 'bw160'], ['ActiveTraffic'], ['orientation'], list(range(0, 360, 10)))