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.cellular.performance import cellular_performance_test_utils as cputils 28from acts_contrib.test_utils.cellular.performance.CellularThroughputBaseTest import CellularThroughputBaseTest 29from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 30from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure 31from functools import partial 32 33 34class CellularLteSensitivityTest(CellularThroughputBaseTest): 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/SCS)', 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/SCS)', 80 primary_y_label='PHY Rate (Mbps)', 81 secondary_y_label='Power Consumption (mW)') 82 # Compile test id data and metrics 83 compiled_data[test_id]['average_throughput'].append( 84 testcase_data['average_throughput_list']) 85 compiled_data[test_id]['cell_power'].append( 86 testcase_data['cell_power_list']) 87 compiled_data[test_id]['mcs'].append( 88 testcase_data['testcase_params']['lte_dl_mcs']) 89 # Add test id to plots 90 plots[test_id].add_line( 91 testcase_data['cell_power_list'], 92 testcase_data['bler_list'], 93 'MCS {}'.format( 94 testcase_data['testcase_params']['lte_dl_mcs']), 95 width=1) 96 plots[test_id_rvr].add_line( 97 testcase_data['cell_power_list'], 98 testcase_data['average_throughput_list'], 99 'MCS {}'.format( 100 testcase_data['testcase_params']['lte_dl_mcs']), 101 width=1, 102 style='dashed') 103 if self.power_monitor: 104 plots[test_id_rvr].add_line( 105 testcase_data['cell_power_list'], 106 testcase_data['average_power_list'], 107 'MCS {} - Power'.format( 108 testcase_data['testcase_params']['lte_dl_mcs']), 109 width=1, 110 style='dashdot', 111 y_axis='secondary') 112 113 for test_id, test_data in compiled_data.items(): 114 test_id_rvr = test_id + tuple('RvR') 115 cell_power_interp = sorted(set(sum(test_data['cell_power'], []))) 116 average_throughput_interp = [] 117 for mcs, cell_power, throughput in zip( 118 test_data['mcs'], test_data['cell_power'], 119 test_data['average_throughput']): 120 throughput_interp = numpy.interp(cell_power_interp, 121 cell_power[::-1], 122 throughput[::-1]) 123 average_throughput_interp.append(throughput_interp) 124 rvr = numpy.max(average_throughput_interp, 0) 125 plots[test_id_rvr].add_line(cell_power_interp, rvr, 126 'Rate vs. Range') 127 128 figure_list = [] 129 for plot_id, plot in plots.items(): 130 plot.generate_figure() 131 figure_list.append(plot) 132 output_file_path = os.path.join(self.log_path, 'results.html') 133 BokehFigure.save_figures(figure_list, output_file_path) 134 135 """Saves CSV with all test results to enable comparison.""" 136 results_file_path = os.path.join( 137 context.get_current_context().get_full_output_path(), 138 'results.csv') 139 with open(results_file_path, 'w', newline='') as csvfile: 140 field_names = [ 141 'Test Name', 'Sensitivity' 142 ] 143 writer = csv.DictWriter(csvfile, fieldnames=field_names) 144 writer.writeheader() 145 146 for testcase_name, testcase_results in self.testclass_results.items( 147 ): 148 row_dict = { 149 'Test Name': testcase_name, 150 'Sensitivity': testcase_results['sensitivity'] 151 } 152 writer.writerow(row_dict) 153 154 def process_testcase_results(self): 155 if self.current_test_name not in self.testclass_results: 156 return 157 testcase_data = self.testclass_results[self.current_test_name] 158 159 bler_list = [] 160 average_throughput_list = [] 161 theoretical_throughput_list = [] 162 average_power_list = [] 163 cell_power_list = testcase_data['testcase_params']['cell_power_sweep'][ 164 0] 165 for result in testcase_data['results']: 166 bler_list.append(result['throughput_measurements'] 167 ['lte_bler_result']['total']['DL']['nack_ratio']) 168 average_throughput_list.append( 169 result['throughput_measurements']['lte_tput_result']['total'] 170 ['DL']['average_tput']) 171 theoretical_throughput_list.append( 172 result['throughput_measurements']['lte_tput_result']['total'] 173 ['DL']['theoretical_tput']) 174 if self.power_monitor: 175 average_power_list.append(result['average_power']) 176 padding_len = len(cell_power_list) - len(average_throughput_list) 177 average_throughput_list.extend([0] * padding_len) 178 theoretical_throughput_list.extend([0] * padding_len) 179 if self.power_monitor: 180 average_power_list.extend([0] * padding_len) 181 182 bler_above_threshold = [ 183 bler > self.testclass_params['bler_threshold'] 184 for bler in bler_list 185 ] 186 for idx in range(len(bler_above_threshold)): 187 if all(bler_above_threshold[idx:]): 188 sensitivity_idx = max(idx, 1) - 1 189 break 190 else: 191 sensitivity_idx = -1 192 sensitivity = cell_power_list[sensitivity_idx] 193 self.log.info('LTE Band {} Table {} MCS {} Sensitivity = {}dBm'.format( 194 testcase_data['testcase_params']['endc_combo_config']['cell_list'] 195 [0]['band'], testcase_data['testcase_params']['lte_dl_mcs_table'], 196 testcase_data['testcase_params']['lte_dl_mcs'], sensitivity)) 197 198 testcase_data['bler_list'] = bler_list 199 testcase_data['average_throughput_list'] = average_throughput_list 200 testcase_data[ 201 'theoretical_throughput_list'] = theoretical_throughput_list 202 testcase_data['cell_power_list'] = cell_power_list 203 testcase_data['average_power_list'] = average_power_list 204 testcase_data['sensitivity'] = sensitivity 205 206 207 results_file_path = os.path.join( 208 context.get_current_context().get_full_output_path(), 209 '{}.json'.format(self.current_test_name)) 210 with open(results_file_path, 'w') as results_file: 211 json.dump(wputils.serialize_dict(testcase_data), 212 results_file, 213 indent=4) 214 215 def get_per_cell_power_sweeps(self, testcase_params): 216 # get reference test 217 current_band = testcase_params['endc_combo_config']['cell_list'][0][ 218 'band'] 219 reference_test = None 220 reference_sensitivity = None 221 for testcase_name, testcase_data in self.testclass_results.items(): 222 if testcase_data['testcase_params']['endc_combo_config'][ 223 'cell_list'][0]['band'] == current_band: 224 reference_test = testcase_name 225 reference_sensitivity = testcase_data['sensitivity'] 226 if reference_test and reference_sensitivity and not self.retry_flag: 227 start_atten = reference_sensitivity + self.testclass_params[ 228 'adjacent_mcs_gap'] 229 self.log.info( 230 "Reference test {} found. Sensitivity {} dBm. Starting at {} dBm" 231 .format(reference_test, reference_sensitivity, start_atten)) 232 else: 233 start_atten = self.testclass_params['lte_cell_power_start'] 234 self.log.info( 235 "Reference test not found. Starting at {} dBm".format( 236 start_atten)) 237 # get current cell power start 238 cell_power_sweeps = [ 239 list( 240 numpy.arange(start_atten, 241 self.testclass_params['lte_cell_power_stop'], 242 self.testclass_params['lte_cell_power_step'])) 243 ] 244 return cell_power_sweeps 245 246 def generate_test_cases(self, dl_mcs_list, lte_dl_mcs_table, 247 lte_ul_mcs_table, lte_ul_mcs, **kwargs): 248 test_cases = [] 249 with open(self.testclass_params['lte_single_cell_configs'], 250 'r') as csvfile: 251 test_configs = csv.DictReader(csvfile) 252 for test_config, lte_dl_mcs in itertools.product( 253 test_configs, dl_mcs_list): 254 if int(test_config['skip_test']): 255 continue 256 endc_combo_config = cputils.generate_endc_combo_config_from_csv_row( 257 test_config) 258 test_name = 'test_lte_B{}_dl_{}_mcs{}'.format( 259 test_config['lte_band'], lte_dl_mcs_table, lte_dl_mcs) 260 test_params = collections.OrderedDict( 261 endc_combo_config=endc_combo_config, 262 lte_dl_mcs_table=lte_dl_mcs_table, 263 lte_dl_mcs=lte_dl_mcs, 264 lte_ul_mcs_table=lte_ul_mcs_table, 265 lte_ul_mcs=lte_ul_mcs, 266 **kwargs) 267 setattr(self, test_name, 268 partial(self._test_throughput_bler, test_params)) 269 test_cases.append(test_name) 270 return test_cases 271 272 273class CellularLteSensitivity_QAM256_Test(CellularLteSensitivityTest): 274 """Class to test single cell LTE sensitivity""" 275 276 def __init__(self, controllers): 277 base_test.BaseTestClass.__init__(self, controllers) 278 self.testcase_metric_logger = ( 279 BlackboxMappedMetricLogger.for_test_case()) 280 self.testclass_metric_logger = ( 281 BlackboxMappedMetricLogger.for_test_class()) 282 self.publish_testcase_metrics = True 283 self.testclass_params = self.user_params['lte_sensitivity_test_params'] 284 self.tests = self.generate_test_cases(list( 285 numpy.arange(27, -1, -1)), 286 lte_dl_mcs_table='QAM256', 287 lte_ul_mcs_table='QAM256', 288 lte_ul_mcs=4, 289 transform_precoding=0) 290 291class CellularLteSensitivity_QAM64_Test(CellularLteSensitivityTest): 292 """Class to test single cell LTE sensitivity""" 293 294 def __init__(self, controllers): 295 base_test.BaseTestClass.__init__(self, controllers) 296 self.testcase_metric_logger = ( 297 BlackboxMappedMetricLogger.for_test_case()) 298 self.testclass_metric_logger = ( 299 BlackboxMappedMetricLogger.for_test_class()) 300 self.publish_testcase_metrics = True 301 self.testclass_params = self.user_params['lte_sensitivity_test_params'] 302 self.tests = self.generate_test_cases(list( 303 numpy.arange(27, -1, -1)), 304 lte_dl_mcs_table='QAM64', 305 lte_ul_mcs_table='QAM64', 306 lte_ul_mcs=4, 307 transform_precoding=0) 308 309class CellularLteSensitivity_SampleMCS_Test(CellularLteSensitivityTest): 310 """Class to test single cell LTE sensitivity""" 311 312 def __init__(self, controllers): 313 base_test.BaseTestClass.__init__(self, controllers) 314 self.testcase_metric_logger = ( 315 BlackboxMappedMetricLogger.for_test_case()) 316 self.testclass_metric_logger = ( 317 BlackboxMappedMetricLogger.for_test_class()) 318 self.publish_testcase_metrics = True 319 self.testclass_params = self.user_params['lte_sensitivity_test_params'] 320 self.tests = self.generate_test_cases(dl_mcs_list=[27,25,16,9], 321 lte_dl_mcs_table='QAM256', 322 lte_ul_mcs_table='QAM256', 323 lte_ul_mcs=4, 324 transform_precoding=0)