1#!/usr/bin/env python3.4 2# 3# Copyright 2022 - 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 numpy 21import json 22import re 23import os 24from acts import context 25from acts import base_test 26from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 27from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 28from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure 29from CellularLtePlusFr1PeakThroughputTest import CellularLteSingleCellPeakThroughputTest 30 31from functools import partial 32 33 34class CellularLteSensitivityTest(CellularLteSingleCellPeakThroughputTest): 35 """Class to test single cell LTE sensitivity""" 36 37 def __init__(self, controllers): 38 base_test.BaseTestClass.__init__(self, controllers) 39 self.testcase_metric_logger = ( 40 BlackboxMappedMetricLogger.for_test_case()) 41 self.testclass_metric_logger = ( 42 BlackboxMappedMetricLogger.for_test_class()) 43 self.publish_testcase_metrics = True 44 self.testclass_params = self.user_params['lte_sensitivity_test_params'] 45 self.tests = self.generate_test_cases(dl_mcs_list=list( 46 numpy.arange(27, -1, -1)), 47 lte_dl_mcs_table='QAM256', 48 lte_ul_mcs_table='QAM256', 49 lte_ul_mcs=4, 50 transform_precoding=0) 51 52 def process_testclass_results(self): 53 # Plot individual test id results raw data and compile metrics 54 plots = collections.OrderedDict() 55 compiled_data = collections.OrderedDict() 56 for testcase_name, testcase_data in self.testclass_results.items(): 57 cell_config = testcase_data['testcase_params'][ 58 'endc_combo_config']['cell_list'][0] 59 test_id = tuple(('band', cell_config['band'])) 60 if test_id not in plots: 61 # Initialize test id data when not present 62 compiled_data[test_id] = { 63 'mcs': [], 64 'average_throughput': [], 65 'theoretical_throughput': [], 66 'cell_power': [], 67 } 68 plots[test_id] = BokehFigure( 69 title='Band {} ({}) - BLER Curves'.format( 70 cell_config['band'], 71 testcase_data['testcase_params']['lte_dl_mcs_table']), 72 x_label='Cell Power (dBm)', 73 primary_y_label='BLER (Mbps)') 74 test_id_rvr = test_id + tuple('RvR') 75 plots[test_id_rvr] = BokehFigure( 76 title='Band {} ({}) - RvR'.format( 77 cell_config['band'], 78 testcase_data['testcase_params']['lte_dl_mcs_table']), 79 x_label='Cell Power (dBm)', 80 primary_y_label='PHY Rate (Mbps)') 81 # Compile test id data and metrics 82 compiled_data[test_id]['average_throughput'].append( 83 testcase_data['average_throughput_list']) 84 compiled_data[test_id]['cell_power'].append( 85 testcase_data['cell_power_list']) 86 compiled_data[test_id]['mcs'].append( 87 testcase_data['testcase_params']['lte_dl_mcs']) 88 # Add test id to plots 89 plots[test_id].add_line( 90 testcase_data['cell_power_list'], 91 testcase_data['bler_list'], 92 'MCS {}'.format( 93 testcase_data['testcase_params']['lte_dl_mcs']), 94 width=1) 95 plots[test_id_rvr].add_line( 96 testcase_data['cell_power_list'], 97 testcase_data['average_throughput_list'], 98 'MCS {}'.format( 99 testcase_data['testcase_params']['lte_dl_mcs']), 100 width=1, 101 style='dashed') 102 103 # Compute average RvRs and compute metrics over orientations 104 for test_id, test_data in compiled_data.items(): 105 test_id_rvr = test_id + tuple('RvR') 106 cell_power_interp = sorted(set(sum(test_data['cell_power'], []))) 107 average_throughput_interp = [] 108 for mcs, cell_power, throughput in zip( 109 test_data['mcs'], test_data['cell_power'], 110 test_data['average_throughput']): 111 throughput_interp = numpy.interp(cell_power_interp, 112 cell_power[::-1], 113 throughput[::-1]) 114 average_throughput_interp.append(throughput_interp) 115 rvr = numpy.max(average_throughput_interp, 0) 116 plots[test_id_rvr].add_line(cell_power_interp, rvr, 117 'Rate vs. Range') 118 119 figure_list = [] 120 for plot_id, plot in plots.items(): 121 plot.generate_figure() 122 figure_list.append(plot) 123 output_file_path = os.path.join(self.log_path, 'results.html') 124 BokehFigure.save_figures(figure_list, output_file_path) 125 126 def process_testcase_results(self): 127 if self.current_test_name not in self.testclass_results: 128 return 129 testcase_data = self.testclass_results[self.current_test_name] 130 results_file_path = os.path.join( 131 context.get_current_context().get_full_output_path(), 132 '{}.json'.format(self.current_test_name)) 133 with open(results_file_path, 'w') as results_file: 134 json.dump(wputils.serialize_dict(testcase_data), 135 results_file, 136 indent=4) 137 138 bler_list = [] 139 average_throughput_list = [] 140 theoretical_throughput_list = [] 141 cell_power_list = testcase_data['testcase_params']['cell_power_sweep'][ 142 0] 143 for result in testcase_data['results']: 144 bler_list.append( 145 result['lte_bler_result']['total']['DL']['nack_ratio']) 146 average_throughput_list.append( 147 result['lte_tput_result']['total']['DL']['average_tput']) 148 theoretical_throughput_list.append( 149 result['lte_tput_result']['total']['DL']['theoretical_tput']) 150 padding_len = len(cell_power_list) - len(average_throughput_list) 151 average_throughput_list.extend([0] * padding_len) 152 theoretical_throughput_list.extend([0] * padding_len) 153 154 bler_above_threshold = [ 155 bler > self.testclass_params['bler_threshold'] 156 for bler in bler_list 157 ] 158 for idx in range(len(bler_above_threshold)): 159 if all(bler_above_threshold[idx:]): 160 sensitivity_idx = max(idx, 1) - 1 161 break 162 else: 163 sensitivity_idx = -1 164 sensitivity = cell_power_list[sensitivity_idx] 165 self.log.info('LTE Band {} Table {} MCS {} Sensitivity = {}dBm'.format( 166 testcase_data['testcase_params']['endc_combo_config']['cell_list'] 167 [0]['band'], testcase_data['testcase_params']['lte_dl_mcs_table'], 168 testcase_data['testcase_params']['lte_dl_mcs'], sensitivity)) 169 170 testcase_data['bler_list'] = bler_list 171 testcase_data['average_throughput_list'] = average_throughput_list 172 testcase_data[ 173 'theoretical_throughput_list'] = theoretical_throughput_list 174 testcase_data['cell_power_list'] = cell_power_list 175 testcase_data['sensitivity'] = sensitivity 176 177 def get_per_cell_power_sweeps(self, testcase_params): 178 # get reference test 179 current_band = testcase_params['endc_combo_config']['cell_list'][0][ 180 'band'] 181 reference_test = None 182 reference_sensitivity = None 183 for testcase_name, testcase_data in self.testclass_results.items(): 184 if testcase_data['testcase_params']['endc_combo_config'][ 185 'cell_list'][0]['band'] == current_band: 186 reference_test = testcase_name 187 reference_sensitivity = testcase_data['sensitivity'] 188 if reference_test and reference_sensitivity and not self.retry_flag: 189 start_atten = reference_sensitivity + self.testclass_params[ 190 'adjacent_mcs_gap'] 191 self.log.info( 192 "Reference test {} found. Sensitivity {} dBm. Starting at {} dBm" 193 .format(reference_test, reference_sensitivity, start_atten)) 194 else: 195 start_atten = self.testclass_params['lte_cell_power_start'] 196 self.log.info( 197 "Reference test not found. Starting at {} dBm".format( 198 start_atten)) 199 # get current cell power start 200 cell_power_sweeps = [ 201 list( 202 numpy.arange(start_atten, 203 self.testclass_params['lte_cell_power_stop'], 204 self.testclass_params['lte_cell_power_step'])) 205 ] 206 return cell_power_sweeps 207 208 def generate_test_cases(self, dl_mcs_list, lte_dl_mcs_table, 209 lte_ul_mcs_table, lte_ul_mcs, **kwargs): 210 test_cases = [] 211 with open(self.testclass_params['lte_single_cell_configs'], 212 'r') as csvfile: 213 test_configs = csv.DictReader(csvfile) 214 for test_config, lte_dl_mcs in itertools.product( 215 test_configs, dl_mcs_list): 216 if int(test_config['skip_test']): 217 continue 218 endc_combo_config = self.generate_endc_combo_config( 219 test_config) 220 test_name = 'test_lte_B{}_dl_{}_mcs{}'.format( 221 test_config['lte_band'], lte_dl_mcs_table, lte_dl_mcs) 222 test_params = collections.OrderedDict( 223 endc_combo_config=endc_combo_config, 224 lte_dl_mcs_table=lte_dl_mcs_table, 225 lte_dl_mcs=lte_dl_mcs, 226 lte_ul_mcs_table=lte_ul_mcs_table, 227 lte_ul_mcs=lte_ul_mcs, 228 **kwargs) 229 setattr(self, test_name, 230 partial(self._test_throughput_bler, test_params)) 231 test_cases.append(test_name) 232 return test_cases 233