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 CellularLteFr1EndcSensitivityTest(CellularThroughputBaseTest): 35 """Class to test ENDC 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['endc_sensitivity_test_params'] 45 self.tests = self.generate_test_cases(lte_dl_mcs_list=list(numpy.arange(27,0,-1)), 46 lte_dl_mcs_table='QAM256', 47 lte_ul_mcs_table='QAM256', 48 lte_ul_mcs=4, 49 nr_dl_mcs_list=list(numpy.arange(27,0,-1)), 50 nr_ul_mcs=4, 51 transform_precoding=0, 52 schedule_scenario='FULL_TPUT', 53 schedule_slot_ratio=80, 54 nr_dl_mcs_table='Q256', 55 nr_ul_mcs_table='Q64') 56 57 def process_testclass_results(self): 58 """Saves CSV with all test results to enable comparison.""" 59 results_file_path = os.path.join( 60 context.get_current_context().get_full_output_path(), 61 'results.csv') 62 with open(results_file_path, 'w', newline='') as csvfile: 63 field_names = [ 64 'Test Name', 'Sensitivity' 65 ] 66 writer = csv.DictWriter(csvfile, fieldnames=field_names) 67 writer.writeheader() 68 69 for testcase_name, testcase_results in self.testclass_results.items( 70 ): 71 row_dict = { 72 'Test Name': testcase_name, 73 'Sensitivity': testcase_results['sensitivity'] 74 } 75 writer.writerow(row_dict) 76 77 def process_testcase_results(self): 78 if self.current_test_name not in self.testclass_results: 79 return 80 testcase_data = self.testclass_results[self.current_test_name] 81 82 bler_list = [] 83 average_throughput_list = [] 84 theoretical_throughput_list = [] 85 average_power_list = [] 86 test_cell_idx = testcase_data['testcase_params']['test_cell_idx'] 87 test_cell_config = testcase_data['testcase_params']['endc_combo_config']['cell_list'][test_cell_idx] 88 cell_power_list = testcase_data['testcase_params']['cell_power_sweep'][ 89 test_cell_idx] 90 91 for result in testcase_data['results']: 92 if test_cell_config['cell_type'] == 'LTE': 93 bler_list.append(1-result['throughput_measurements'] 94 ['lte_bler_result'][test_cell_config['cell_number']]['DL']['ack_ratio']) 95 average_throughput_list.append( 96 result['throughput_measurements']['lte_tput_result'][test_cell_config['cell_number']] 97 ['DL']['average_tput']) 98 theoretical_throughput_list.append( 99 result['throughput_measurements']['lte_tput_result'][test_cell_config['cell_number']] 100 ['DL']['theoretical_tput']) 101 else: 102 bler_list.append(1-result['throughput_measurements'] 103 ['nr_bler_result'][test_cell_config['cell_number']]['DL']['ack_ratio']) 104 average_throughput_list.append( 105 result['throughput_measurements']['nr_tput_result'][test_cell_config['cell_number']] 106 ['DL']['average_tput']) 107 theoretical_throughput_list.append( 108 result['throughput_measurements']['nr_tput_result'][test_cell_config['cell_number']] 109 ['DL']['theoretical_tput']) 110 if self.power_monitor: 111 average_power_list.append(result['average_power']) 112 padding_len = len(cell_power_list) - len(average_throughput_list) 113 average_throughput_list.extend([0] * padding_len) 114 theoretical_throughput_list.extend([0] * padding_len) 115 average_throughput_list.extend([0] * padding_len) 116 117 118 bler_above_threshold = [ 119 bler > self.testclass_params['bler_threshold'] 120 for bler in bler_list 121 ] 122 123 for idx in range(len(bler_above_threshold)): 124 if all(bler_above_threshold[idx:]): 125 sensitivity_idx = max(idx, 1) - 1 126 sensitivity = cell_power_list[sensitivity_idx] 127 break 128 else: 129 sensitivity = float('nan') 130 131 132 if test_cell_config['cell_type'] == 'LTE': 133 test_mcs = testcase_data['testcase_params']['lte_dl_mcs'] 134 else: 135 test_mcs = testcase_data['testcase_params']['nr_dl_mcs'] 136 self.log.info('{} Band {} MCS {} Sensitivity = {}dBm'.format( 137 test_cell_config['cell_type'], 138 test_cell_config['band'], 139 test_mcs, 140 sensitivity)) 141 142 testcase_data['bler_list'] = bler_list 143 testcase_data['average_throughput_list'] = average_throughput_list 144 testcase_data[ 145 'theoretical_throughput_list'] = theoretical_throughput_list 146 testcase_data['cell_power_list'] = cell_power_list 147 testcase_data['average_power_list'] = average_power_list 148 testcase_data['sensitivity'] = sensitivity 149 150 results_file_path = os.path.join( 151 context.get_current_context().get_full_output_path(), 152 '{}.json'.format(self.current_test_name)) 153 with open(results_file_path, 'w') as results_file: 154 json.dump(wputils.serialize_dict(testcase_data), 155 results_file, 156 indent=4) 157 158 def get_per_cell_power_sweeps(self, testcase_params): 159 cell_power_sweeps = [] 160 # Construct test cell sweep 161 test_cell = testcase_params['endc_combo_config']['cell_list'][testcase_params['test_cell_idx']] 162 if test_cell['cell_type'] == 'LTE': 163 test_cell_sweep = list( 164 numpy.arange(self.testclass_params['lte_cell_power_start'], 165 self.testclass_params['lte_cell_power_stop'], 166 self.testclass_params['lte_cell_power_step'])) 167 else: 168 test_cell_sweep = list( 169 numpy.arange(self.testclass_params['nr_cell_power_start'], 170 self.testclass_params['nr_cell_power_stop'], 171 self.testclass_params['nr_cell_power_step'])) 172 173 for cell_idx, cell_config in enumerate(testcase_params['endc_combo_config']['cell_list']): 174 if cell_idx == testcase_params['test_cell_idx']: 175 cell_power_sweeps.append(test_cell_sweep) 176 elif cell_config['cell_type'] == 'LTE': 177 lte_sweep = [self.testclass_params['lte_cell_power_start'] 178 ] * len(test_cell_sweep) 179 cell_power_sweeps.append(lte_sweep) 180 elif cell_config['cell_type'] == 'NR5G': 181 nr_sweep = [self.testclass_params['nr_cell_power_start'] 182 ] * len(test_cell_sweep) 183 cell_power_sweeps.append(nr_sweep) 184 return cell_power_sweeps 185 186 def generate_test_cases(self, lte_dl_mcs_list, lte_dl_mcs_table, 187 lte_ul_mcs_table, lte_ul_mcs, nr_dl_mcs_list, 188 nr_ul_mcs, **kwargs): 189 test_cases = [] 190 with open(self.testclass_params['endc_combo_file'], 191 'r') as endc_combos: 192 for endc_combo_str in endc_combos: 193 if endc_combo_str[0] == '#': 194 continue 195 endc_combo_config = cputils.generate_endc_combo_config_from_string( 196 endc_combo_str) 197 special_chars = '+[]=;,\n' 198 for char in special_chars: 199 endc_combo_str = endc_combo_str.replace(char, '_') 200 endc_combo_str = endc_combo_str.replace('__', '_') 201 endc_combo_str = endc_combo_str.strip('_') 202 for cell_idx, cell_config in enumerate(endc_combo_config['cell_list']): 203 if cell_config['cell_type'] == 'LTE': 204 dl_mcs_list = lte_dl_mcs_list 205 else: 206 dl_mcs_list = nr_dl_mcs_list 207 for dl_mcs in dl_mcs_list: 208 test_name = 'test_sensitivity_{}_cell_{}_mcs{}'.format( 209 endc_combo_str, cell_idx, dl_mcs) 210 if cell_config['cell_type'] == 'LTE': 211 test_params = collections.OrderedDict( 212 endc_combo_config=endc_combo_config, 213 test_cell_idx=cell_idx, 214 lte_dl_mcs_table=lte_dl_mcs_table, 215 lte_dl_mcs=dl_mcs, 216 lte_ul_mcs_table=lte_ul_mcs_table, 217 lte_ul_mcs=lte_ul_mcs, 218 nr_dl_mcs=4, 219 nr_ul_mcs=nr_ul_mcs, 220 **kwargs) 221 else: 222 test_params = collections.OrderedDict( 223 endc_combo_config=endc_combo_config, 224 test_cell_idx=cell_idx, 225 lte_dl_mcs_table=lte_dl_mcs_table, 226 lte_dl_mcs=4, 227 lte_ul_mcs_table=lte_ul_mcs_table, 228 lte_ul_mcs=lte_ul_mcs, 229 nr_dl_mcs=dl_mcs, 230 nr_ul_mcs=nr_ul_mcs, 231 **kwargs) 232 setattr(self, test_name, 233 partial(self._test_throughput_bler, test_params)) 234 test_cases.append(test_name) 235 return test_cases 236 237 238class CellularLteFr1EndcSensitivity_SampleMCS_Test(CellularLteFr1EndcSensitivityTest): 239 """Class to test single cell LTE sensitivity""" 240 241 def __init__(self, controllers): 242 base_test.BaseTestClass.__init__(self, controllers) 243 self.testcase_metric_logger = ( 244 BlackboxMappedMetricLogger.for_test_case()) 245 self.testclass_metric_logger = ( 246 BlackboxMappedMetricLogger.for_test_class()) 247 self.publish_testcase_metrics = True 248 self.testclass_params = self.user_params['endc_sensitivity_test_params'] 249 self.tests = self.generate_test_cases(lte_dl_mcs_list=[27,25,16,9], 250 lte_dl_mcs_table='QAM256', 251 lte_ul_mcs_table='QAM256', 252 lte_ul_mcs=4, 253 nr_dl_mcs_list=[27,25,16,9], 254 nr_ul_mcs=4, 255 transform_precoding=0, 256 schedule_scenario='FULL_TPUT', 257 schedule_slot_ratio=80, 258 nr_dl_mcs_table='Q256', 259 nr_ul_mcs_table='Q64')